技能概述
Web Search 技能是指 AI Agent 在接收到用户查询后,自动调用外部搜索引擎接口,获取相关网页摘要、链接及结构化信息,并将结果进行清洗、去重、排序和语义压缩后返回给大模型,以支持后续推理或直接回答的能力。
功能边界:
- 输入:自然语言查询(如'2024 年苹果 WWDC 发布了哪些新功能?')
- 输出:结构化搜索结果(标题、URL、摘要) + 聚合后的文本摘要
- 不包含:网页全文抓取(需配合 Document Parser 技能)、登录态访问、付费内容获取
核心能力:
- 关键词提取与查询优化:从用户问题中识别实体与意图,生成高效搜索 Query
- 多源结果聚合:支持多个搜索引擎结果融合(可选)
- 结果过滤与可信度评估:排除低质量、广告或虚假信息
- 上下文感知排序:结合对话历史调整结果相关性
架构设计
Web Search 技能采用分层架构,确保高内聚低耦合:
+---------------------+
| User Query |
+----------+----------+
v
+---------------------+
| Query Preprocessor |
| - 实体识别 |
| - 意图分类 |
| - 查询重写 |
+----------+----------+
v
+---------------------+
| Search API Gateway |
| - 多引擎适配器 |
| - 请求限流/重试 |
| - 缓存代理 |
+----------+----------+
v
+---------------------+
| Result Postprocessor|
| - 去重 |
| - 摘要提取 |
| - 可信度打分 |
| - 结果截断(TopK)
v
Output Formatter
JSON Markdown
引用标注
该架构支持插件式扩展,例如替换 Google 为 DuckDuckGo 只需实现新的 SearchEngineAdapter。
接口设计
遵循 MCP(Model Context Protocol)标准,定义统一接口:
class WebSearchSkill:
def __init__(self, api_key: str, engine: str = "google", max_results: int = 5):
self.api_key = api_key
self.engine = engine
self.max_results = max_results
def search(self, query: str, region: str = "us", lang: str = "en") -> dict:
"""
执行网络搜索并返回结构化结果
Args:
query (str): 搜索关键词
region (str): 搜索区域(如"cn", "us")
lang (str): 返回语言
Returns:
dict: {
"query": str,
"results": [
{ "title": str, "url": str, "snippet": str, "position": int },
...
],
"summary": str # 聚合摘要
}
"""
pass
返回格式示例:
{
"query": "2024 年苹果 WWDC 新功能",
"results": [
{
"title": "Apple WWDC 2024: iOS 18, macOS Sequoia...",
"url": "https://www.apple.com/newsroom/2024/wwdc/",
"snippet": "Apple announced iOS 18 with Apple Intelligence...",
"position": 1
}
],
"summary": "在 2024 年 WWDC 上,苹果发布了 iOS 18、macOS Sequoia,并首次推出 Apple Intelligence..."
}
代码实现(Python + LangChain)
以下基于 LangChain 实现完整 Web Search 技能,集成 SerpAPI 并支持缓存:
import os
import hashlib
from typing import List, Dict, Optional
from langchain_community.utilities import GoogleSerperAPIWrapper
from langchain_core.tools import Tool
from langchain_core.callbacks import CallbackManagerForToolRun
from langchain_core.pydantic_v1 import BaseModel, Field
import tiktoken
# 安装依赖:pip install langchain-community serpapi tiktoken
class WebSearchInput(BaseModel):
query: str = Field(description="The search query in natural language")
max_results: int = Field(default=5, ge=1, le=10)
class WebSearchSkill:
def __init__(self, serper_api_key: str, cache_dir: str = "./search_cache", max_tokens: int = 2000):
self.search = GoogleSerperAPIWrapper(
serper_api_key=serper_api_key,
gl="us",
hl="en"
)
self.cache_dir = cache_dir
self.max_tokens = max_tokens
os.makedirs(cache_dir, exist_ok=True)
self.encoder = tiktoken.encoding_for_model("gpt-3.5-turbo")
def _get_cache_key() -> :
hashlib.md5(query.encode()).hexdigest()
() -> []:
cache_file = os.path.join(.cache_dir, )
os.path.exists(cache_file):
json
(cache_file, , encoding=) f:
json.load(f)
():
cache_file = os.path.join(.cache_dir, )
json
(cache_file, , encoding=) f:
json.dump(data, f, ensure_ascii=, indent=)
() -> :
snippets = [r.get(, ) r results r.get()]
combined = .join(snippets[:])
tokens = .encoder.encode(combined)
(tokens) > .max_tokens:
combined = .encoder.decode(tokens[:.max_tokens])
combined
() -> :
cache_key = ._get_cache_key(query)
cached = ._read_cache(cache_key)
cached:
()
cached
:
raw_results = .search.results(query, num_results=max_results)
organic = raw_results.get(, [])
formatted_results = []
i, res (organic):
formatted_results.append({
: res.get(, ),
: res.get(, ),
: res.get(, ),
: i +
})
summary = ._summarize_results(formatted_results)
output = {
: query,
: formatted_results,
: summary
}
._write_cache(cache_key, output)
output
Exception e:
()
{: query, : [], : }
() -> Tool:
skill = WebSearchSkill(serper_api_key=serper_api_key)
() -> :
result = skill.search(query, max_results=)
result[]
Tool(
name=,
description=,
func=_run,
args_schema=WebSearchInput
)
__name__ == :
SERPER_API_KEY = os.getenv()
SERPER_API_KEY:
ValueError()
tool = create_web_search_tool(SERPER_API_KEY)
result = tool.run()
(result)
实战案例
案例 1:实时财经问答 Agent
业务背景:构建一个能回答股票价格、财报日期、行业新闻的金融助手。
技术选型:
- 搜索引擎:SerpAPI(支持 Google Finance)
- LLM:OpenAI GPT-4
- 框架:LangChain AgentExecutor
完整实现:
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# 初始化 LLM 和工具
llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
search_tool = create_web_search_tool(os.getenv("SERPER_API_KEY"))
# 定义 Agent 提示词
prompt = ChatPromptTemplate.from_messages([
("system", "You are a financial assistant. Use web_search to get real-time data."),
MessagesPlaceholder("chat_history"),
("human", "{input}"),
MessagesPlaceholder("agent_scratchpad")
])
# 创建 Agent
agent = create_openai_tools_agent(llm, [search_tool], prompt)
agent_executor = AgentExecutor(agent=agent, tools=[search_tool], verbose=True)
# 测试
response = agent_executor.invoke({"input": "What is Tesla's current stock price and market cap?"})
print(response["output"])
运行结果:
Tesla Inc. (TSLA) is currently trading at $178.50 per share with a market capitalization of approximately $568 billion as of June 2024.
问题与解决:
- 问题:搜索结果包含广告链接。
- 解决:在
_summarize_results中过滤url包含/ads/或googleads的结果。
案例 2:多语言新闻聚合器
业务背景:为跨国企业提供当日全球新闻摘要,支持中英双语。
关键技术点:
- 动态设置
hl(host language)和gl(geolocation) - 结果翻译(调用翻译 API 或 LLM)
代码片段:
def multilingual_news_summary(topic: str, languages: List[str] = ["en", "zh"]):
all_snippets = []
skill = WebSearchSkill(serper_api_key=os.getenv("SERPER_API_KEY"))
for lang in languages:
region = "cn" if lang == "zh" else "us"
results = skill.search(f"{topic} site:news.google.com", max_results=3)
for r in results["results"]:
all_snippets.append(r["snippet"])
# 调用 LLM 生成多语言摘要
llm = ChatOpenAI(model="gpt-4")
summary_prompt = f"Summarize the following news snippets about '{topic}' in both English and Chinese:\n\n" + "\n".join(all_snippets)
return llm.invoke(summary_prompt).content
错误处理
常见异常及处理策略:
| 异常类型 | 原因 | 处理机制 |
|---|---|---|
| API Rate Limit | 超出 QPS 限制 | 指数退避重试(最多 3 次) |
| Invalid Query | 查询含非法字符 | 自动清理(移除特殊符号) |
| Empty Results | 无相关结果 | 返回友好提示 + 建议改写查询 |
| Network Timeout | 网络波动 | 设置 10 秒超时,失败后降级缓存 |
在代码中通过 try-except 捕获异常,并记录结构化日志:
import logging
logger = logging.getLogger(__name__)
try:
# 搜索逻辑
except requests.exceptions.Timeout:
logger.warning("Search timeout", extra={"query": query})
return fallback_response()
性能优化
- 缓存策略:
- 使用 LRU 缓存(内存) + 文件缓存(持久化)
- TTL 设置:新闻类 1 小时,静态事实 24 小时
- 结果截断:
- 限制返回结果数(默认 5 条)
- 摘要 Token 数控制(避免 LLM 上下文溢出)
并发处理:
from concurrent.futures import ThreadPoolExecutor
def batch_search(queries: List[str]) -> List[Dict]:
with ThreadPoolExecutor(max_workers=5) as executor:
return list(executor.map(self.search, queries))
安全考量
- 输入校验:
- 过滤 SQL 注入字符(如
;,--) - 限制查询长度(≤200 字符)
- 过滤 SQL 注入字符(如
- 权限控制:
- 敏感查询拦截(如'how to hack')
- 企业环境中按角色控制搜索范围
- 沙箱隔离:
- 搜索结果不直接执行 JavaScript
- URL 白名单机制(仅允许可信域名)
测试方案
单元测试示例(pytest):
def test_web_search_skill():
skill = WebSearchSkill(serper_api_key="fake_key")
# Mock Serper API
with patch.object(skill.search, 'results') as mock_search:
mock_search.return_value = {
"organic": [{"title": "Test", "link": "http://test.com", "snippet": "Test snippet"}]
}
result = skill.search("test query")
assert len(result["results"]) == 1
assert "Test snippet" in result["summary"]
集成测试:
- 验证端到端流程:用户提问 → 搜索 → LLM 生成答案
- 使用真实 API Key 在 CI 中运行(每日限额内)
端到端测试指标:
- 成功率 ≥ 95%
- P95 延迟 ≤ 2s
- 缓存命中率 ≥ 30%
最佳实践
- 查询重写优于原始输入:使用 LLM 先将模糊问题转为精准关键词
- 结果必须标注来源:避免幻觉,增强可信度
- 避免过度搜索:非时效性问题优先查本地知识库
- 成本意识:监控 API 调用量,设置预算告警
- 用户隐私:不在查询中泄露 PII(个人身份信息)
扩展方向
- 多引擎融合:同时调用 Google、Bing、DuckDuckGo,加权融合结果
- 垂直搜索:集成 arXiv(学术)、GitHub(代码)、Patent(专利)等专业引擎
- 语义搜索增强:将搜索结果向量化,与用户问题做相似度匹配
- MCP 标准化:实现符合 Model Context Protocol 的通用接口,便于跨平台复用
总结
Web Search 技能是 AI Agent 连接动态世界的桥梁。通过合理设计查询预处理、结果后处理和缓存机制,可显著提升信息获取的准确性与时效性。本文提供了基于 LangChain 的完整实现,涵盖错误处理、安全控制和性能优化,适用于生产环境。
技能开发实践要点:
- 搜索查询需经过意图识别与关键词提取,避免直接透传用户原始输入
- 必须实现缓存机制以降低 API 成本并提升响应速度
- 结果摘要应保留关键事实并标注信息来源,防止幻觉
- 对搜索结果进行可信度过滤,排除低质量或广告内容
- 严格控制输入长度与敏感词,保障系统安全
- 在非必要场景下优先使用本地知识库,减少对外部依赖
- 监控 API 调用量与延迟,设置熔断与降级策略
- 遵循 MCP 协议设计接口,确保技能可移植性

