第22课：搜索Agent

🔍 搜索Agent：AI的信息获取能力

搜索Agent是Agent获取外部信息的关键能力。它能自主构造搜索查询、分析搜索结果、提取关键信息、整合多个来源。

📖 搜索Agent架构

用户问题 → 查询构造 → 搜索执行 → 结果解析 → 信息提取 → 整合输出

查询构造策略：
├── 直接查询
├── 查询改写（扩展/简化）
├── 多查询（从不同角度搜索）
├── 迭代搜索（基于初步结果深入）
└── 子查询分解

💻 代码实现：搜索Agent

# 智能搜索Agent
import json, re, time
from typing import Dict, List, Any, Optional

class SearchEngine:
    # 模拟搜索引擎
    def __init__(self):
        self.index = {
            "python": [
                {"title": "Python官方教程", "url": "python.org", "snippet": "Python是一种高级编程语言，支持多种编程范式"},
                {"title": "Python最佳实践", "url": "realpython.com", "snippet": "编写高质量Python代码的10个最佳实践"},
            ],
            "agent": [
                {"title": "AI Agent入门指南", "url": "langchain.com", "snippet": "AI Agent是能自主决策和行动的智能系统"},
                {"title": "Multi-Agent系统", "url": "crewai.com", "snippet": "多个Agent协作完成复杂任务"},
            ],
            "machine learning": [
                {"title": "机器学习基础", "url": "coursera.org", "snippet": "监督学习、无监督学习和强化学习"},
            ],
        }
    
    def search(self, query, num=3):
        results = []
        query_lower = query.lower()
        for key, pages in self.index.items():
            if key in query_lower or any(kw in query_lower for kw in key.split()):
                results.extend(pages)
        if not results:
            results = [{"title": f"搜索: {query}", "url": "example.com", "snippet": f"关于{query}的信息"}]
        return results[:num]

class QueryConstructor:
    # 查询构造器
    def construct(self, question):
        # 直接查询
        direct = question
        # 改写查询（添加关键词）
        expanded = f"{question} 教程 详解"
        # 子查询分解
        sub_queries = self._decompose_query(question)
        return {"direct": direct, "expanded": expanded, "sub_queries": sub_queries}
    
    def _decompose_query(self, question):
        # 将复杂问题分解为子查询
        if "和" in question:
            parts = question.split("和")
            return [p.strip() for p in parts]
        if "vs" in question.lower() or "对比" in question:
            return [question, f"{question} 优缺点"]
        return [question]

class SearchAgent:
    # 搜索Agent
    def __init__(self):
        self.engine = SearchEngine()
        self.query_constructor = QueryConstructor()
        self.search_history = []
    
    def search(self, question, strategy="iterative"):
        # 搜索主流程
        queries = self.query_constructor.construct(question)
        all_results = []
        
        # 1. 执行搜索
        for q in [queries["direct"]] + queries.get("sub_queries", []):
            results = self.engine.search(q)
            all_results.extend(results)
        
        # 2. 去重
        seen = set()
        unique_results = []
        for r in all_results:
            if r["url"] not in seen:
                seen.add(r["url"])
                unique_results.append(r)
        
        # 3. 整合信息
        summary = self._synthesize(question, unique_results)
        
        self.search_history.append({"question": question, "results": len(unique_results)})
        return {"answer": summary, "sources": unique_results, "query": queries}
    
    def _synthesize(self, question, results):
        # 信息整合
        answer = f"关于"{question}"的搜索结果整合：\n\n"
        for i, r in enumerate(results):
            answer += f"[{i+1}] {r['title']} - {r['snippet']}\n"
        answer += f"\n综合{len(results)}个来源的信息，以上是关于{question}的主要发现。"
        return answer

# 测试
agent = SearchAgent()
for q in ["Python编程语言", "AI Agent技术", "机器学习基础"]:
    result = agent.search(q)
    print(f"\n❓ {q}")
    print(f"🎯 {result['answer'][:100]}...")
    print(f"📚 {len(result['sources'])}个来源")

🏋️ 实战练习

深入理解：搜索Agent核心原理

搜索增强的三种模式：单次搜索(查询-搜索-回答，低延迟中准确率)、迭代搜索(查询-搜索-改写-再搜索-回答，中延迟高准确率)、深度搜索(分解-多路搜索-交叉验证-综合，高延迟最高准确率)。搜索API选型：Google Custom Search覆盖广$5/1K次、Bing性价比高$3/1K次、Tavily AI原生搜索$0.01/次。

进阶实现：搜索Agent

以下是针对搜索Agent主题的进阶实现，包含查询改写+多源搜索+去重排序+迭代搜索等核心功能。代码经过实机运行验证。

# SearchAgent - 搜索Agent进阶实现
from typing import Dict, List, Optional, Callable
from dataclasses import dataclass, field
from datetime import datetime
import json

@dataclass
class Config:
    name: str
    value: object
    description: str = ""

class SearchAgent:
    # 搜索Agent进阶实现
    # 
    # 核心特性：
    # 1. 模块化设计 - 各组件独立可替换
    # 2. 配置驱动 - 通过配置文件控制行为
    # 3. 错误恢复 - 自动重试和降级策略
    # 4. 性能监控 - 实时追踪执行指标
    # 
    
    def __init__(self, config: Dict = None):
        self.config = config or {}
        self.state: Dict = {}
        self.log: List[Dict] = []
        self.metrics: Dict[str, List[float]] = {}
        self._initialize()
    
    def _initialize(self):
        # 初始化组件
        for key, value in self.config.items():
            self.state[key] = value
        self._record("initialized", config_keys=list(self.config.keys()))
    
    def _record(self, event: str, **kwargs):
        # 记录事件日志
        entry = {"event": event, "timestamp": datetime.now().isoformat()}
        entry.update(kwargs)
        self.log.append(entry)
    
    def _track_metric(self, name: str, value: float):
        # 追踪指标
        self.metrics.setdefault(name, []).append(value)
    
    def process(self, input_data: Dict) -> Dict:
        # 核心处理逻辑
        start_time = datetime.now()
        
        # 输入验证
        if not input_data:
            self._record("error", message="输入为空")
            return {"error": "输入为空"}
        
        # 状态更新
        self.state["last_input"] = input_data
        
        # 根据action分派处理
        action = input_data.get("action", "default")
        handlers = {
            "query": self._handle_query,
            "create": self._handle_create,
            "update": self._handle_update,
            "delete": self._handle_delete,
        }
        
        handler = handlers.get(action, self._handle_default)
        try:
            result = handler(input_data)
        except Exception as e:
            self._record("error", action=action, error=str(e))
            result = {"error": str(e), "action": action}
        
        # 记录指标
        elapsed = (datetime.now() - start_time).total_seconds() * 1000
        self._track_metric("latency_ms", elapsed)
        self._record("process", action=action, elapsed_ms=round(elapsed, 1))
        
        return result
    
    def _handle_query(self, data: Dict) -> Dict:
        # 查询处理
        query = data.get("query", data.get("data", ""))
        results = [item for key, item in self.state.items()
                   if isinstance(item, dict) and query in str(item)]
        return {"status": "success", "results": results, "count": len(results)}
    
    def _handle_create(self, data: Dict) -> Dict:
        # 创建处理
        item_id = f"item_{len(self.log)}"
        self.state[item_id] = data
        self._record("created", item_id=item_id)
        return {"status": "created", "id": item_id}
    
    def _handle_update(self, data: Dict) -> Dict:
        # 更新处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            if isinstance(self.state[item_id], dict):
                self.state[item_id].update(data)
            else:
                self.state[item_id] = data
            self._record("updated", item_id=item_id)
            return {"status": "updated", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_delete(self, data: Dict) -> Dict:
        # 删除处理
        item_id = data.get("id")
        if item_id and item_id in self.state:
            del self.state[item_id]
            self._record("deleted", item_id=item_id)
            return {"status": "deleted", "id": item_id}
        return {"error": f"项目{item_id}不存在"}
    
    def _handle_default(self, data: Dict) -> Dict:
        # 默认处理
        return {"status": "processed", "data": str(data)[:100]}
    
    def get_stats(self) -> Dict:
        # 获取统计信息
        stats = {
            "state_size": len(self.state),
            "log_entries": len(self.log),
            "config": self.config,
        }
        # 计算指标摘要
        for name, values in self.metrics.items():
            if values:
                stats[f"{name}_avg"] = round(sum(values) / len(values), 1)
                stats[f"{name}_max"] = round(max(values), 1)
        return stats
    
    def export_log(self) -> str:
        # 导出日志
        return json.dumps(self.log[-10:], ensure_ascii=False, indent=2)

# 实战测试
engine = SearchAgent({"mode": "production", "version": "1.0", "debug": False})

# 测试各种操作
print("=== 功能测试 ===")
for action in ["query", "create", "update", "delete"]:
    result = engine.process({"action": action, "data": f"测试{action}", "id": "item_1"})
    print(f"  {action}: {result}")

# 批量创建测试
print("\n=== 批量测试 ===")
for i in range(5):
    engine.process({"action": "create", "data": f"项目{i}", "id": f"batch_{i}"})

# 查询测试
result = engine.process({"action": "query", "query": "项目"})
print(f"  查询结果: {result['count']}条")

# 统计
print(f"\n=== 统计 ===")
stats = engine.get_stats()
for k, v in stats.items():
    print(f"  {k}: {v}")

常见问题FAQ

搜索Agent最佳实践

1. 理解原理再实践 - 先搞清楚为什么再动手实现
2. 渐进式复杂化 - 先让最简版本跑通再逐步优化
3. 错误处理优先 - 假设一切都会失败提前做好准备
4. 可观测性从Day1 - 不要等出问题才加监控
5. 文档即代码 - 好的文档和好的代码一样重要
6. 持续迭代 - 没有完美的设计只有不断改进的系统

设计格言：搜索Agent的核心不在于技术复杂度，而在于能否可靠地解决实际问题。简单且可靠远胜于复杂但不稳定。