【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）标准，定义统一接口：

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(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
            with open(cache_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        return None

    def _write_cache(self, key: str, data: Dict):
        cache_file = os.path.join(self.cache_dir, f"{key}.json")
        import json
        with open(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)
        if len(tokens) > self.max_tokens:
            combined = self.encoder.decode(tokens[:self.max_tokens])
        return combined

    def search(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 in enumerate(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 Tool
def create_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")
    if not 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)

环境配置：

export SERPER_API_KEY=your_serper_api_key
pip install langchain-community serpapi tiktoken

---

实战案例

案例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()

---

性能优化

1. 缓存策略：

   • 使用LRU缓存（内存） + 文件缓存（持久化）
   • TTL设置：新闻类1小时，静态事实24小时

2. 并发处理：

   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))
   

3. 结果截断：

   • 限制返回结果数（默认5条）
   • 摘要Token数控制（避免LLM上下文溢出）

---

安全考量

1. 输入校验：

   • 过滤SQL注入字符（如;, --）
   • 限制查询长度（≤200字符）

2. 权限控制：

   • 敏感查询拦截（如“how to hack”）
   • 企业环境中按角色控制搜索范围

3. 沙箱隔离：

   • 搜索结果不直接执行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%

---

最佳实践

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可突破大模型知识时效限制，实时获取互联网权威信息，显著提升在动态场景下的实用性与可靠性，为构建企业级智能体奠定基础。