Agent 联网能力：Tavily 和 SerpAPI 搜索集成

一、概念速查

搜索 API 选型对比

特性	Tavily	SerpAPI	Brave Search
定价	每月 1000 次免费，之后按量计费	按次计费，$0.01/次起	每月 2000 次免费（Rate Limited）
语义搜索	✅ 原生支持（NLP 理解查询意图）	❌ 仅关键词	✅ 支持
结构化输出	✅ JSON，含摘要、评分、来源	✅ JSON，结构丰富	✅ JSON
内容提取	✅ TavilyExtract 内置	❌ 需额外解析	❌
AI 代理优化	✅ 专为 Agent 设计	❌ 通用搜索 API	❌ 通用搜索 API
实时性	✅ 实时网络爬取	✅ 实时 Google 索引	✅
响应速度	快（1-3s）	中（2-5s，含反爬）	快（1-2s）
国内替代	无	Baidu/360 类似 SDK	无

选型建议： Agent 场景首选 Tavily——语义理解让 Agent 可以直接用自然语言提问而非拼关键词，内置内容提取省去二次解析步骤，专为 ReAct 模式优化的接口设计减少工具调用失败。需要 Google 排名特定数据（如 SEO 分析、本地商家排名、广告投放监控）选 SerpAPI，它返回的原始 Google 搜索结果结构与用户浏览器所见一致。预算敏感且 JSON 格式够用选 Brave Search，但缺乏内容提取能力，需要配合 trafilatura 或 newspaper3k 等库自行解析。

核心参数速查

TavilySearch：

参数	说明	默认值
max_results	最大返回结果数	5
topic	搜索主题：general / news / finance	general
search_depth	basic 快速 / advanced 深度	basic
include_domains	限定来源域名	无
exclude_domains	排除来源域名	无
include_answer	是否返回 AI 摘要	False
raw_search	True 绕过 AI 语义拦截	False

SerpAPI：

参数	说明	默认值
q	查询关键词	必填
engine	google / bing / baidu 等	google
num	返回结果数	10
hl	搜索结果语言	zh-cn
tbm	搜索类型：nws(新闻) / isch(图片)	无

快速集成示例

Tavily： LangChain 通过 langchain_tavily 包提供一等的工具类支持，Agent 可以直接绑定。

from langchain_tavily import TavilySearch, TavilyExtract

# 单独使用搜索——适合在管道中先检索后生成
search = TavilySearch(
    max_results=3,           # 返回前 3 条结果，减少 token 消耗
    topic="general",         # 可选 general / news / finance
    search_depth="advanced"  # advanced 会爬取目标页摘要，basic 仅返回片段
)
result = search.invoke("2026 年大模型技术趋势")

# 集成到 LangChain Agent——自动触发搜索→思考→再搜索循环
from langchain_openai import ChatOpenAI
from langchain.agents import create_openai_tools_agent, AgentExecutor

llm = ChatOpenAI(model="gpt-4o", temperature=0)
tools = [
    TavilySearch(max_results=5, topic="general"),  # 搜索工具
    TavilyExtract()                                  # 提取工具——获取整页内容而非摘要
]

agent = create_openai_tools_agent(llm=llm, tools=tools, prompt=prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

response = executor.invoke({
    "messages": [{"role": "user", "content": "研究 2026 年大模型落地趋势"}]
})

SerpAPI： 需要自行包装成 LangChain 工具，适合需要 Google 原生排名结构的场景。

from langchain.tools import tool
from serpapi import GoogleSearch
import os

@tool
def serpapi_search(query: str, num_results: int = 5) -> list[dict]:
    """返回 Google 搜索结果（标题/链接/摘要/排名位置）"""
    params = {
        "q": query,
        "api_key": os.environ["SERPAPI_API_KEY"],
        "num": num_results,
        "hl": "zh-cn",
        "engine": "google"
    }
    search = GoogleSearch(params)
    results = search.get_dict()
    return [
        {
            "title": r.get("title"),
            "link": r.get("link"),
            "snippet": r.get("snippet"),
            "position": r.get("position")   # Google 原始排名
        }
        for r in results.get("organic_results", [])
    ]

# 绑定到 Agent
tools = [serpapi_search, TavilySearch(max_results=3)]

二、底层原理

搜索 Agent 的构建流程

查询重写： Agent 自动检测查询是否需要联网。像"今天天气"、"最新新闻"这类时效性问题路由到搜索工具；纯内部知识类问题走本地 RAG。重写策略包括去除修饰词、补充搜索关键词、中英文同义扩展。

多源融合： 先用 Tavily 做主搜索，若结果不足 3 条或置信度低于阈值，自动 fallback 到 SerpAPI。融合阶段做 URL 去重，保留最高分的重复结果。

搜索结果解析与质量排序

def rank_results(raw_results: list[dict]) -> list[dict]:
    """对搜索结果做质量排序"""
    scored = []
    for r in raw_results:
        score = 0
        # 1. 域名权威性
        if any(domain in r["url"] for domain in DOMAIN_WHITELIST):
            score += 30
        # 2. 内容完整度：有摘要 + 有正文片段
        if r.get("content") and len(r["content"]) > 100:
            score += 25
        # 3. 时效性：近 7 天加分
        if is_recent(r.get("published_date"), days=7):
            score += 20
        # 4. 与查询的语义相关性（如果 API 返回了评分）
        score += r.get("score", 0) * 25
        scored.append((score, r))

    scored.sort(key=lambda x: x[0], reverse=True)
    return [r for _, r in scored[:5]]

排序时，白名单域名（官方站点、权威媒体如 arxiv.org、github.com、docs.python.org）得到 30 分权重加成。低质量网站（采集站、SEO 垃圾内容、AI 生成农场）预置排除域名列表——这些站通常标题与内容不匹配、缺乏作者署名、发布日期模糊。实际部署时应维护一个动态黑名单，每轮检索后自动标记被 Agent 最终回答"忽略"的结果来源，逐步积累排除库。

多引擎结果融合时的 URL 去重逻辑：Tavily 和 SerpAPI 可能返回同一篇博客的不同版本（HTTP/HTTPS、带/不带 www、带 UTM 参数）。统一规范化 URL——移除 www. 前缀、移除 utm_* 参数、统一协议为 HTTPS——再以规范化后的 URL 做哈希去重，保留评分最高的那条。

搜索幻觉的成因与缓解手段

搜索 Agent 特有的幻觉有三种：

幻觉类型	成因	缓解手段
截断幻觉	Agent 只读摘要不读全文，断章取义	TavilyExtract 拉取完整内容后再推理
过期幻觉	搜到时事问题但返回了过时结果	检查 published_date，过滤超过阈值的结果
编造来源	LLM 自行生成不存在的 URL	Agent 输出时绑定实际检索到的 URL 列表，强制引用校验

核心防御手段：temperature=0 + stop 序列。 设置 stop=["\nObservation:", "\nThought:", "\nFinal Answer:"] 防止 Agent 自说自话编造 Observation 字段。当 LLM 在生成过程中遇到这些序列时立即停止，迫使它等待工具真实的返回内容。temperature=0 消除随机性，确保同一查询每次走相同的推理路径——这在工具调用场景中至关重要，因为即便是微小的随机波动也可能让 Agent 跳过搜索直接凭记忆回答。

来源校验层： 在 Agent 输出前追加一道拦截——检查最终回答中的所有 URL 是否确实出现在搜索结果列表中。LangChain 的 AgentExecutor 可以通过自定义输出解析器挂载校验逻辑：

from langchain.agents import AgentExecutor
from langchain_core.agents import AgentFinish

def validate_citations(output: AgentFinish) -> AgentFinish:
    """校验最终回答的 URL 是否来自本次搜索"""
    import re
    urls_in_response = re.findall(r'https?://[^\s)]+', output.return_values["output"])
    if not urls_in_response:
        return output  # 无引用则跳过校验
    # 从 Agent 的中间步骤中提取实际检索到的 URL
    searched_urls = set()
    for step in output.messages:
        if hasattr(step, "artifact") and isinstance(step.artifact, list):
            for r in step.artifact:
                searched_urls.add(r.get("url", ""))
    fake_urls = [u for u in urls_in_response if u not in searched_urls]
    if fake_urls:
        output.return_values["output"] += (
            f"\n\n⚠️ 以下引用来源未在搜索结果中确认：{', '.join(fake_urls)}"
        )
    return output

executor = AgentExecutor(
    agent=agent, tools=tools,
    custom_output_parser=validate_citations
)

# 防御性 LLM 配置
llm = ChatOpenAI(
    temperature=0,       # 消除随机性，确保工具调用路径稳定
    max_tokens=2048,
    model_kwargs={
        "stop": ["\nObservation:", "\nThought:", "\nFinal Answer:"]
    }
)

三、架构设计原则

多源备份策略

实线为主链路，虚线为降级链路。 Tavily 免费额度 1000 次/月，建议作为主链路；SerpAPI 按量付费做备份（无此场景无需开通）；Brave Search 完全免费但结果质量略低，作为最末兜底。如果需要国内可访问的方案，可在 SerpAPI 备用链路中配置 engine=baidu，用百度搜索结果填补国内网络场景下的盲区。整体判断逻辑应打包为一个 SearchRouter 类，对外提供统一的 search(query) 接口，内部管理重试、降级和熔断状态——详见下文的搜索幻觉缓解中的校验层。

缓存策略

缓存是控制搜索 API 成本的核心手段。Tavily 每月仅 1000 次免费额度，不加缓存的 Agent 可能在一次深度对话中就消耗 10% 的月度配额。

• 同查询去重： 5 分钟内完全相同的查询直接返回缓存结果，以 {query}:{max_results}:{topic} 为缓存键。最简单的实现是用 functools.lru_cache 加时间戳检查。
• 语义缓存： 相似查询（使用 sentence-transformers 计算的余弦相似度 > 0.95）复用缓存，避免重复扣费。例如"2026 AI 趋势"和"2026 年人工智能发展方向"命中同一缓存。需要引入向量存储（如 FAISS 或 Chroma）存储嵌入，每次搜索前先做最近邻检索。
• 缓存有效期： 针对查询类型区分 TTL——新闻类 5 分钟（结果变化极快，过期后应强制刷新），通用类 1 小时（大部分技术类内容在这个窗口内稳定），知识百科类 24 小时（事实性内容几乎不变，如"Python 列表推导式语法"）。
• 冷热分离： 热点查询（同一查询 > 3 次/小时）迁移到长期缓存区，用 LRU 保留最近 100 个热点条目。冷查询则按标准 TTL 过期，释放内存。

缓存命中时应标记来源为 cache，日志中记录节省的 API 调用次数，便于评估缓存 ROI。

频率控制

• QPS 限制： Tavily 建议 < 10 QPS（否则返回 429），SerpAPI 建议 < 1 QPS（超限直接封 IP 几分钟）。两者的免费套餐限制更严格——Tavily 免费账号约 1 QPS，SerpAPI 免费约 0.5 QPS。生产环境务必通过中间代理做请求排队。
• 分批检索： 批量查询时使用 asyncio.gather 并发但控制最大并发数。Semaphore 设置为对应 API 的 QPS 上限；单 Agent 会话中应限制工具调用次数（max_iterations=6 即最多搜索 6 次），防止 Agent 失控循环扣费。
• 退避策略： 遇到 429 错误，指数退避重试——初始 1s，每次翻倍到最大 30s，加入随机抖动（±500ms）避免惊群效应。重试 3 次仍失败则跳过该查询，返回兜底提示而非阻塞整个 Agent。
• 日额度预警： 用量达 80% 时自动降级到备用 API，避免主 API 在关键任务中被耗尽。通过 requests 在前一天结束时查询 API 剩余额度，写入本地计数器。当日额度触及阈值，Agent 的路由器自动切换搜索后端。

import asyncio
from langchain_tavily import TavilySearch

tool = TavilySearch(max_results=3, topic="general")

async def controlled_search(queries: list[str], max_concurrency=5):
    sem = asyncio.Semaphore(max_concurrency)

    async def search_one(q):
        async with sem:
            return await tool.ainvoke({"query": q})

    return await asyncio.gather(*[search_one(q) for q in queries])

如果 API 全部不可用，Agent 应优雅提示"当前搜索服务暂不可用，请稍后重试"而非卡死或编造答案。