AI Agent Skill Day 12：Web Search技能：互联网搜索与信息聚合

【AI Agent Skill Day 12】Web Search技能：互联网搜索与信息聚合

在“AI Agent Skill技能开发实战”系列的第12天，我们聚焦于Web Search技能——这一使Agent具备实时获取互联网公开信息能力的核心模块。随着大模型知识存在时效性限制（如训练数据截止至2023年或2024年），仅依赖内部知识库难以应对动态世界中的最新事件、股价、新闻、产品发布等需求。Web Search技能通过集成搜索引擎API（如SerpAPI、Google Programmable Search Engine、DuckDuckGo等），实现对网络信息的精准检索、结果聚合与语义提炼，是构建“活Agent”的关键一环。本技能广泛应用于智能客服、市场情报分析、科研辅助、金融舆情监控等场景，显著提升Agent的信息鲜度与决策质量。

技能概述

Web Search技能是指AI Agent在接收到用户查询后，自动调用外部搜索引擎接口，获取相关网页摘要、链接及结构化信息，并将结果进行清洗、去重、排序和语义压缩后返回给大模型，以支持后续推理或直接回答的能力。

功能边界：

• 输入：自然语言查询（如“2024年苹果WWDC发布了哪些新功能？”）
• 输出：结构化搜索结果（标题、URL、摘要） + 聚合后的文本摘要
• 不包含：网页全文抓取（需配合Document Parser技能）、登录态访问、付费内容获取

核心能力：

1. 关键词提取与查询优化：从用户问题中识别实体与意图，生成高效搜索Query
2. 多源结果聚合：支持多个搜索引擎结果融合（可选）
3. 结果过滤与可信度评估：排除低质量、广告或虚假信息
4. 上下文感知排序：结合对话历史调整结果相关性

架构设计

Web Search技能采用分层架构，确保高内聚低耦合：

+---------------------+ | User Query | +----------+----------+ | v +---------------------+ | Query Preprocessor| | - 实体识别 | | - 意图分类 | | - 查询重写 | +----------+----------+ | v +---------------------+ | Search API Gateway | | - 多引擎适配器 | | - 请求限流/重试 | | - 缓存代理 | +----------+----------+ | v +---------------------+ | Result Postprocessor| | - 去重 | | - 摘要提取 | | - 可信度打分 | | - 结果截断（Top-K） | +----------+----------+ | v +---------------------+ | Output Formatter | | - JSON / Markdown | | - 引用标注 | +---------------------+ 

该架构支持插件式扩展，例如替换Google为DuckDuckGo只需实现新的SearchEngineAdapter。

接口设计

遵循MCP（Model Context Protocol）标准，定义统一接口：

classWebSearchSkill: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 defsearch(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 tiktokenclassWebSearchInput(BaseModel): query:str= Field(description="The search query in natural language") max_results:int= Field(default=5, ge=1, le=10)classWebSearchSkill: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(self, query:str)->str:return hashlib.md5(query.encode()).hexdigest()def_read_cache(self, key:str)-> Optional[Dict]: cache_file = os.path.join(self.cache_dir,f"{key}.json")if os.path.exists(cache_file):import json withopen(cache_file,'r', encoding='utf-8')as f:return json.load(f)returnNonedef_write_cache(self, key:str, data: Dict): cache_file = os.path.join(self.cache_dir,f"{key}.json")import json withopen(cache_file,'w', encoding='utf-8')as f: json.dump(data, f, ensure_ascii=False, indent=2)def_summarize_results(self, results: List[Dict])->str: snippets =[r.get("snippet","")for r in results if r.get("snippet")] combined ="\n\n".join(snippets[:5])# 简单截断防止超长 tokens = self.encoder.encode(combined)iflen(tokens)> self.max_tokens: combined = self.encoder.decode(tokens[:self.max_tokens])return combined defsearch(self, query:str, max_results:int=5)-> Dict: cache_key = self._get_cache_key(query) cached = self._read_cache(cache_key)if cached:print(f"[CACHE HIT] {query}")return cached try: raw_results = self.search.results(query, num_results=max_results) organic = raw_results.get("organic",[]) formatted_results =[]for i, res inenumerate(organic): formatted_results.append({"title": res.get("title",""),"url": res.get("link",""),"snippet": res.get("snippet",""),"position": i +1}) summary = self._summarize_results(formatted_results) output ={"query": query,"results": formatted_results,"summary": summary } self._write_cache(cache_key, output)return output except Exception as e:print(f"[SEARCH ERROR] {e}")return{"query": query,"results":[],"summary":"搜索服务暂时不可用，请稍后再试。"}# 封装为LangChain Tooldefcreate_web_search_tool(serper_api_key:str)-> Tool: skill = WebSearchSkill(serper_api_key=serper_api_key)def_run(query:str, run_manager: Optional[CallbackManagerForToolRun]=None)->str: result = skill.search(query, max_results=5)# 返回摘要供LLM使用return result["summary"]return Tool( name="web_search", description="Useful for when you need to answer questions about current events, latest news, or real-time information. Input should be a search query.", func=_run, args_schema=WebSearchInput )# 使用示例if __name__ =="__main__": SERPER_API_KEY = os.getenv("SERPER_API_KEY")ifnot SERPER_API_KEY:raise ValueError("Please set SERPER_API_KEY environment variable") tool = create_web_search_tool(SERPER_API_KEY) result = tool.run("What is the latest stock price of NVIDIA?")print(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）

代码片段：

defmultilingual_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 

错误处理

常见异常及处理策略：

在代码中通过try-except捕获异常，并记录结构化日志：

import logging logger = logging.getLogger(__name__)try:# 搜索逻辑except requests.exceptions.Timeout: logger.warning("Search timeout", extra={"query": query})return fallback_response()

性能优化

1. 缓存策略：
   • 使用LRU缓存（内存） + 文件缓存（持久化）
   • TTL设置：新闻类1小时，静态事实24小时
2. 结果截断：
   • 限制返回结果数（默认5条）
   • 摘要Token数控制（避免LLM上下文溢出）

并发处理：

from concurrent.futures import ThreadPoolExecutor defbatch_search(queries: List[str])-> List[Dict]:with ThreadPoolExecutor(max_workers=5)as executor:returnlist(executor.map(self.search, queries))

安全考量

1. 输入校验：
   • 过滤SQL注入字符（如;, --）
   • 限制查询长度（≤200字符）
2. 权限控制：
   • 敏感查询拦截（如“how to hack”）
   • 企业环境中按角色控制搜索范围
3. 沙箱隔离：
   • 搜索结果不直接执行JavaScript
   • URL白名单机制（仅允许可信域名）

测试方案

单元测试示例（pytest）：

deftest_web_search_skill(): skill = WebSearchSkill(serper_api_key="fake_key")# Mock Serper APIwith 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")assertlen(result["results"])==1assert"Test snippet"in result["summary"]

集成测试：

• 验证端到端流程：用户提问 → 搜索 → LLM生成答案
• 使用真实API Key在CI中运行（每日限额内）

端到端测试指标：

• 成功率 ≥ 95%
• P95延迟 ≤ 2s
• 缓存命中率 ≥ 30%

最佳实践

1. 查询重写优于原始输入：使用LLM先将模糊问题转为精准关键词
2. 结果必须标注来源：避免幻觉，增强可信度
3. 避免过度搜索：非时效性问题优先查本地知识库
4. 成本意识：监控API调用量，设置预算告警
5. 用户隐私：不在查询中泄露PII（个人身份信息）

扩展方向

1. 多引擎融合：同时调用Google、Bing、DuckDuckGo，加权融合结果
2. 垂直搜索：集成arXiv（学术）、GitHub（代码）、Patent（专利）等专业引擎
3. 语义搜索增强：将搜索结果向量化，与用户问题做相似度匹配
4. MCP标准化：实现符合Model Context Protocol的通用接口，便于跨平台复用

总结

Web Search技能是AI Agent连接动态世界的桥梁。通过合理设计查询预处理、结果后处理和缓存机制，可显著提升信息获取的准确性与时效性。本文提供了基于LangChain的完整实现，涵盖错误处理、安全控制和性能优化，适用于生产环境。在Day 13中，我们将深入Knowledge Graph技能，探讨如何利用结构化知识进行复杂推理。

技能开发实践要点：

1. 搜索查询需经过意图识别与关键词提取，避免直接透传用户原始输入
2. 必须实现缓存机制以降低API成本并提升响应速度
3. 结果摘要应保留关键事实并标注信息来源，防止幻觉
4. 对搜索结果进行可信度过滤，排除低质量或广告内容
5. 严格控制输入长度与敏感词，保障系统安全
6. 在非必要场景下优先使用本地知识库，减少对外部依赖
7. 监控API调用量与延迟，设置熔断与降级策略
8. 遵循MCP协议设计接口，确保技能可移植性

进阶学习资源：

1. LangChain官方文档 - Tools and Agents: https://python.langchain.com/docs/modules/agents/
2. SerpAPI开发者指南: https://serpapi.com/documentation
3. Google Programmable Search Engine: https://developers.google.com/custom-search
4. MCP (Model Context Protocol) 规范草案: https://github.com/modelcontextprotocol/spec
5. LlamaIndex Web Search Integration: https://docs.llamaindex.ai/en/stable/examples/query_engine/web_search_query_engine/
6. “Rethinking Web Search for LLMs” - arXiv:2305.14282
7. DuckDuckGo Search API (开源替代): https://pypi.org/project/duckduckgo-search/
8. Open Source Search Aggregator: https://github.com/emptycrown/llama-hub/tree/main/llama_hub/tools

文章标签：AI Agent, Web Search, LangChain, SerpAPI, MCP, 信息聚合, 技能开发, RAG
文章简述：本文详细解析了AI Agent中Web Search技能的设计与实现，涵盖架构设计、接口规范、LangChain代码示例及两大实战案例（财经问答与多语言新闻聚合）。重点讨论了缓存策略、安全控制、错误处理和性能优化等生产级考量，并提供完整的可执行代码与测试方案。通过本技能，Agent可突破大模型知识时效限制，实时获取互联网权威信息，显著提升在动态场景下的实用性与可靠性，为构建企业级智能体奠定基础。