BiGuan/Agent

1"""2tools/web_search.py —— 工具①：联网搜索3 4给 agent 一个"上网搜东西"的能力。5"""6 7import time8 9from langchain_core.tools import tool10 11from config import TAVILY_API_KEY12 13 14def _tavily(query: str):15    """用 Tavily 搜索。函数名以下划线开头，表示这是"内部辅助函数"，只给本文件自己调用。"""16    # 没有配密钥就直接返回 None（表示"用不了 Tavily"），让外层去用备胎 DuckDuckGo。17    if not TAVILY_API_KEY:18        return None19    import requests20 21    # 向 Tavily 的接口发一个请求，带上：密钥、搜索词、最多要 8 条结果、并让它顺便给个总结性答案。22    r = requests.post(23        "https://api.tavily.com/search",24        json={"api_key": TAVILY_API_KEY, "query": query, "max_results": 8, "include_answer": True},25        timeout=30,   # 最多等 30 秒，避免卡死26    )27    r.raise_for_status()   # 如果服务器返回错误状态码，这里会主动报错28    data = r.json()        # 把返回的 JSON 文本解析成 Python 能用的数据29    # 下面把搜索结果拼成一段整齐的文字返回（标题、网址、摘要），方便大模型阅读。30    parts = []31    if data.get("answer"):32        parts.append("Answer: " + data["answer"])33    for it in data.get("results", []):34        parts.append(f"[{it.get('title')}]({it.get('url')})\n{it.get('content', '')}")35    return "## Search Results\n\n" + "\n\n".join(parts) if parts else None36 37 38def _duckduckgo(query: str):39    """备用搜索引擎：DuckDuckGo。同样是内部辅助函数。"""40    # 不同版本的库名字不一样，try/except 是为了"哪个能导入就用哪个"，增强兼容性。41    try:42        from ddgs import DDGS43    except ImportError:44        from duckduckgo_search import DDGS45 46    results = DDGS().text(query, max_results=10)   # 搜索，最多取 10 条47    if not results:48        return None49    # 同样把结果拼成整齐文字返回。50    return "## Search Results\n\n" + "\n\n".join(51        f"[{r.get('title')}]({r.get('href') or r.get('url')})\n{r.get('body') or r.get('content', '')}"52        for r in results53    )54 55 56@tool57def web_search(query: str) -> str:58    """Search the web and return the top results (title, url, snippet). Use for general,59    up-to-date research; follow up with `visit_webpage` to read a result in full. For60    encyclopedic facts prefer `wikipedia_search`, which is more reliable."""61    # 第一步：先试 Tavily。成功拿到结果就直接返回。62    try:63        out = _tavily(query)64        if out:65            return out66    except Exception:67        pass   # Tavily 出错也不报错中断，继续往下用68 69    # 第二步：用 DuckDuckGo，并且最多重试 4 次（免费引擎容易临时失败，多试几次更稳）。70    last_err = None71    for attempt in range(4):72        try:73            out = _duckduckgo(query)74            if out:75                return out76        except Exception as e:77            last_err = e78        # 每次失败后等一会儿再试，且等待时间越来越长（1.5秒、3秒、4.5秒...），避免被对方当成攻击而封锁。79        time.sleep(1.5 * (attempt + 1))80    # 两套引擎都没搜到，就返回一句提示，告诉大模型"换个工具或换个说法再试"。81    return (82        f"Web search returned nothing (last error: {last_err}). "83        "Try `wikipedia_search`, rephrase the query, or `visit_webpage` on a known url."84    )85