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

技能概述

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 字符）
权限控制：
- 敏感查询拦截（如'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 协议设计接口，确保技能可移植性

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

技能概述

架构设计

接口设计

代码实现（Python + LangChain）

实战案例

案例 1：实时财经问答 Agent

案例 2：多语言新闻聚合器

错误处理

性能优化

安全考量

测试方案

最佳实践

扩展方向

总结

更多推荐文章

相关免费在线工具

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

技能概述

架构设计

接口设计

代码实现（Python + LangChain）

实战案例

案例 1：实时财经问答 Agent

案例 2：多语言新闻聚合器

错误处理

性能优化

安全考量

测试方案

最佳实践

扩展方向

总结

微信扫一扫，关注极客日志

更多推荐文章

相关免费在线工具