给Agent增加联网搜索并不难。定义一个搜索工具,把用户问题传进去,再把结果交给大模型,十几行代码就能跑起来。
真正麻烦的是上线以后。
同一个问题有时能找到准确答案,有时全是重复转载;搜到了正确页面,返回的摘要却不包含关键内容;模型明明带了引用,链接里的原文却无法支持它的结论。请求量上来以后,还会遇到超时、限流和网页读取失败。
这些问题通常不是换一个更大的模型就能解决,而是搜索链路本身没有设计完整。
一条完整的联网检索链路
比较实用的处理流程是:
用户问题
↓
判断是否需要联网
↓
查询改写与拆分
↓
并发搜索
↓
结果标准化、去重
↓
相关性与来源排序
↓
读取少量候选网页
↓
正文切分与压缩
↓
生成带引用答案
↓
引用一致性检查
搜索API只是其中一环。最终效果取决于整条链路,而不是接口一次返回多少条结果。
先判断这个问题是否需要搜索
并非所有问题都需要联网。
代码解释、固定知识和已经存在于内部知识库中的内容,可以直接回答。涉及新闻、政策、价格、版本、公司信息和实时状态的问题,则应该强制搜索。
可以先做一个简单分类:
def search_policy(question: str) -> str:
realtime_words = [
"今天", "目前", "最新", "现在",
"价格", "版本", "政策", "公告",
"是否上线", "什么时候发布"
]
if any(word in question for word in realtime_words):
return "required"
return "model_decide"
生产环境可以用一个小模型判断,但最好保留强制规则。否则模型可能认为自己知道答案,直接使用已经过期的内部知识。
不要直接拿用户原话搜索
用户问题和搜索引擎需要的查询往往不是一回事。
例如:
为什么我的流式接口经常过一会儿自己断掉?
这个问题至少包含几个可能方向:
SSE连接超时原因
反向代理SSE超时配置
流式API idle timeout
Nginx proxy_read_timeout SSE
如果只搜索用户原话,结果可能非常分散。比较稳妥的做法是让模型生成两到四个查询,分别覆盖现象、技术术语和可能原因。
async def rewrite_queries(question: str) -> list[str]:
prompt = f"""
将下面的问题改写为最多4个适合搜索引擎的查询。
查询之间要有不同侧重点,不要生成同义句。
用户问题:
{question}
"""
return await small_model.generate_json(prompt)
查询数量也不是越多越好。简单事实问题通常一个查询就够;只有调研、比较和故障排查类问题,才有必要并行搜索多个方向。
搜索结果必须先标准化和去重
不同搜索源返回的字段不完全一样。有的提供标题和摘要,有的返回正文片段,还有的额外提供发布时间、站点名称或相关性得分。
进入后续流程前,最好统一成内部结构:
from dataclasses import dataclass
@dataclass
class SearchResult:
title: str
url: str
snippet: str
source: str
published_at: str | None = None
score: float | None = None
去重时不能只比较完整URL。同一篇文章可能带有不同的跟踪参数,也可能被多个页面转载。
至少需要处理:
- 删除
utm_*等跟踪参数;
-统一HTTP与HTTPS形式; - 识别相同域名下的重复路径;
- 比较标题相似度;
- 比较摘要或正文指纹;
- 合并同一事件的重复新闻。
如果不做去重,模型可能看到五条结果,以为得到了多个来源的交叉验证,实际上五条都来自同一篇稿件。
不要读取所有网页
搜索结果里的摘要通常不足以回答复杂问题,但也没必要读取每一个页面。
更合理的流程是:
- 搜索阶段先获取10条左右结果;
- 根据标题、摘要、时间和来源完成第一次排序;
- 只读取排名靠前的3—5个页面;
- 对正文再次排序;
- 最终向模型提供2—4个真正有用的来源。
简化后的代码大致如下:
async def retrieve(question: str):
queries = await rewrite_queries(question)
result_groups = await parallel_search(
queries=queries,
top_k=10
)
merged = normalize_and_merge(result_groups)
unique = deduplicate(merged)
candidates = rank_search_results(question, unique)
pages = await read_pages(
candidates[:5],
timeout=12
)
passages = split_and_rerank(
question=question,
pages=pages,
top_k=8
)
return passages
读取网页时要设置单独超时。一两个页面解析失败,不应该拖死整个Agent任务。
时间和来源需要显式控制
新闻、版本、价格、政策等问题应设置时间范围,不能完全依赖搜索引擎自行判断。
高风险问题还需要限制来源。例如:
- 法规优先政府和监管机构网站;
- 医疗优先卫生机构、医院及专业组织;
- 软件问题优先官方文档和代码仓库;
- 公司公告优先交易所及企业官网;
- 学术问题优先论文原文和权威数据库。
可以给不同来源配置基础权重:
SOURCE_WEIGHT = {
"official": 1.0,
"academic": 0.9,
"professional_media": 0.75,
"community": 0.55,
"unknown": 0.3,
}
权重不能代替相关性。一个权威但答非所问的页面,不应该排在真正解决问题的技术文章前面。
引用链接不等于答案可靠
很多系统做到“答案后面有链接”就结束了,但真正需要检查的是:
引用页面里的内容,是否真的支持答案中的说法?
一种可操作的方式,是把生成答案拆成若干事实陈述,再逐条寻找支持证据:
陈述:某功能从3.2版本开始支持。
证据:官方更新日志明确写明3.2版本新增该功能。
结论:支持。
陈述:该接口平均延迟低于500ms。
证据:引用页面只写了“低延迟”,没有提供测试数据。
结论:不支持,应删除具体数值。
对于找不到直接证据的内容,可以删除、降低语气强度,或者明确标注为推断。
这一步会增加一次模型调用,但对调研、报告和高风险问答很有价值。
必须处理超时、限流和空结果
联网服务不可避免会失败,Agent不能把搜索失败理解为“网上没有信息”。
建议区分以下情况:
class SearchError(Exception):
pass
class RateLimited(SearchError):
pass
class SearchTimeout(SearchError):
pass
class EmptyResult(SearchError):
pass
对应策略也应不同:
- 限流:指数退避后重试;
- 超时:缩短查询或切换备用搜索源;
- 空结果:改写查询后再搜索;
- 网页读取失败:保留搜索摘要,继续处理其他页面;
- 全部失败:明确告诉用户当前无法完成联网验证。
如果业务对可用性要求较高,最好准备两个搜索源,但不建议每次同时调用。正常情况下使用主搜索源,失败或结果质量不足时再触发备用源。
怎么判断搜索链路有没有变好
只统计请求成功率不够。搜索接口正常返回,不代表Agent找到了正确答案。
可以重点记录下面这些指标:
| 指标 | 说明 |
|---|---|
| Top 5相关率 | 前5条结果中真正相关的比例 |
| 新鲜信息命中率 | 时间敏感问题是否找到最新资料 |
| 权威来源占比 | 候选结果中可信来源的比例 |
| 重复结果率 | 多条结果是否实际来自同一内容 |
| 正文读取成功率 | 候选页面能否提取有效正文 |
| 引用支持率 | 引用内容是否能支持对应结论 |
| P50/P95耗时 | 常规请求与慢请求的实际表现 |
| 单任务成本 | 搜索、读取、重排和模型调用总成本 |
| 任务完成率 | 最终是否正确回答了用户问题 |
测试问题最好来自真实业务,而不是只用“今天天气怎么样”之类容易搜索的问题。
一套基础测试集可以包含新闻、技术文档、政策、产品信息、故障排查、学术资料和长尾问题。每个问题重复运行几次,才能看出结果波动。
结语
联网搜索不是给模型加一个工具按钮,而是一条完整的信息处理链路。
查询改写决定搜什么,去重和排序决定看什么,网页读取决定能拿到多少证据,引用校验决定最后的答案能不能相信。
接口接通只是开始。真正影响Agent体验的,通常是这些看起来不起眼的工程细节。