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

该架构支持插件式扩展，例如替换 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
        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)

实战案例

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

错误处理

常见异常及处理策略：

在代码中通过 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

def batch_search(queries: List[str]) -> List[Dict]:
    with ThreadPoolExecutor(max_workers=5) as executor:
        return list(executor.map(self.search, queries))

安全考量

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 的完整实现，涵盖错误处理、安全控制和性能优化，适用于生产环境。

技能开发实践要点：

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