🛠️ CrewAI自定义工具实战：五步封装百度搜索工具，让Agent真正”联网”！

💡 导读：想让AI Agent具备实时搜索能力？本文带你从零封装一个基于百度千帆API的CrewAI工具，掌握自定义工具的标准SOP，附完整代码+避坑指南，建议收藏！

🔍 为什么需要自定义工具？

CrewAI官方提供了基础工具集，但真实业务场景中，我们往往需要：

✅ 对接企业内部API
✅ 集成第三方服务（如搜索、地图、天气）
✅ 定制业务逻辑和输出格式

自定义工具 = Agent的”超能力扩展包” 🚀

今天我们就以百度搜索工具为例，拆解自定义工具的五步标准SOP。

📋 五步标准SOP详解

第1️⃣步：定义输入Schema（Pydantic验证）

from pydantic import BaseModel, Field, field_validator
from typing import Optional, List, Literal

class BaiduSearchInput(BaseModel):
    """百度搜索工具的输入参数模式"""
    query: str = Field(
        ..., 
        description="搜索查询内容，不能为空"
    )
    top_k: Optional[int] = Field(
        20,
        description="返回结果数量，推荐：精确搜索≤5，广泛调研≥10"
    )
    recency_filter: Optional[Literal["week", "month", "semiyear", "year"]] = Field(
        None,
        description="时间筛选：week(7天)/month(30天)/semiyear(180天)/year(365天)"
    )
    sites: Optional[List[str]] = Field(
        None,
        description="指定搜索站点，最多20个，如['github.com', 'zhihu.com']"
    )

🎯 关键技巧：

使用Field(...)标记必填项，Optional标记可选项
description不仅是注释，更是告诉Agent何时使用该工具的关键！
用@field_validator做业务规则校验，提前拦截非法输入

@field_validator('query')
def validate_query(cls, v: str) -> str:
    if not v or not v.strip():
        raise ValueError("查询内容不能为空，请提供有效关键词")
    return v.strip()

第2️⃣步：继承BaseTool，声明工具元信息

from crewai.tools import BaseTool

class BaiduSearchTool(BaseTool):
    name: str = "search_web"  # ⚠️ 必须是英文！CrewAI会过滤中文工具名
    description: str = (
        "使用百度搜索引擎查找相关信息，支持时间/站点筛选。\n"
        "触发时机：查找最新信息、特定网站内容、按时间筛选结果时使用。\n"
        "适用边界：仅搜索公开通用信息，专业领域请用专用工具。"
    )
    args_schema: Type[BaseModel] = BaiduSearchInput  # 关联输入验证器

📝 Description编写心法：

✅ 包含三要素：
1. 功能说明：工具能做什么
2. 触发时机：什么时候该用（帮助Agent决策）
3. 适用边界：什么时候不该用（避免误调用）

❌ 避免：
- 过于技术化的实现细节
- 模糊的"可以搜索信息"描述

第3️⃣步：实现`_run`方法（核心业务逻辑）

def _run(
    self,
    query: str,
    top_k: int = 20,
    recency_filter: Optional[str] = None,
    sites: Optional[List[str]] = None,
) -> str:
    # 1️⃣ 获取鉴权信息
    api_key = os.getenv("BAIDU_API_KEY")
    if not api_key:
        return "错误：缺少API认证密钥，请配置环境变量BAIDU_API_KEY"
    
    # 2️⃣ 构建请求参数
    payload = {
        "messages": [{"content": query, "role": "user"}],
        "search_source": "baidu_search_v2",
        "resource_type_filter": [{"type": "web", "top_k": top_k}],
    }
    if recency_filter:
        payload["search_recency_filter"] = recency_filter
    if sites:
        payload["search_filter"] = {"match": {"site": sites}}
    
    # 3️⃣ 发送HTTP请求
    headers = {
        "X-Appbuilder-Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    response = requests.post(API_URL, json=payload, headers=headers, timeout=30)
    response.raise_for_status()
    
    # 4️⃣ 解析并格式化结果（见第5步）
    ...

🔧 工程化细节：

✅ 敏感信息（API Key）通过环境变量注入，永不硬编码
✅ 请求体构建采用”按需添加”策略，避免冗余参数
✅ 设置合理timeout，防止Agent卡死

第4️⃣步：错误处理 + 日志记录（稳定性保障）

try:
    response = requests.post(...)
    response.raise_for_status()
    result = response.json()
    
    # 检查业务错误码
    if result.get("code") not in [0, None, ""]:
        return f"API错误：{result.get('message')}"
        
except requests.exceptions.Timeout:
    logger.error("请求超时")
    return "错误：服务器响应超时，请稍后重试"
    
except json.JSONDecodeError:
    logger.error("JSON解析失败")
    return "错误：响应格式异常，请重试"
    
except Exception as e:
    logger.exception(f"未预期错误: {e}")
    return f"错误：系统异常，请稍后重试"

🛡️ 错误处理三层设计：

层级	处理方式	返回给Agent的内容
🔹 业务错误	解析API错误码	友好提示+解决建议
🔹 网络异常	捕获requests异常	“请稍后重试”类通用提示
🔹 未知异常	全局Exception捕获	兜底提示+日志记录

📊 日志最佳实践：

# ✅ 记录关键节点，但隐藏敏感信息
logger.info(f"搜索关键词: {query}，top_k: {top_k}")  # 可记录
logger.info(f"请求体:\n{json.dumps(safe_payload)}")   # 脱敏后记录

# ❌ 避免记录
logger.info(f"API Key: {api_key}")  # 🔥 绝对禁止！

第5️⃣步：格式化输出（让Agent”看懂”结果）

# 原始API返回
{
  "references": [
    {
      "id": "1",
      "title": "Python 3.12新特性",
      "url": "https://example.com",
      "content": "Python 3.12引入了..."
    }
  ]
}

# ✅ 格式化为Agent友好的纯文本
结果1: [ Python 3.12新特性 ] ( https://example.com ) 
  内容摘要: Python 3.12引入了...

结果2: [ ... ]

🎯 格式化原则：

结构清晰：用固定模板，方便Agent解析关键信息
信息完整：标题+链接+摘要，三位一体
长度可控：避免返回超长文本导致Token浪费
带统计信息：找到{N}条结果，帮助Agent判断信息充分性

💡 进阶技巧：让工具更”智能”

技巧1：动态调整top_k策略

# 在_run方法开头添加
if "最新" in query or "新闻" in query:
    top_k = min(top_k, 10)  # 资讯类结果少而精
elif "教程" in query or "指南" in query:
    top_k = min(top_k, 30)  # 教程类需要更多参考

技巧2：结果去重 + 质量排序

# 简单去重示例
seen_urls = set()
unique_refs = []
for ref in references:
    if ref["url"] not in seen_urls:
        seen_urls.add(ref["url"])
        unique_refs.append(ref)
references = unique_refs[:top_k]  # 截断到期望数量

技巧3：缓存高频查询（生产环境必备）

from functools import lru_cache

@lru_cache(maxsize=100)
def _cached_search(query_hash: str, top_k: int) -> str:
    # 实际搜索逻辑
    pass

# 在_run中调用
import hashlib
query_hash = hashlib.md5(f"{query}|{recency_filter}".encode()).hexdigest()
return _cached_search(query_hash, top_k)

🚨 常见坑点排查清单

问题现象	可能原因	解决方案
Agent不调用该工具	description写得太模糊	重写description，明确”触发时机”
工具名被忽略	使用了中文工具名	改为英文，如`search_web`
参数验证失败	Pydantic字段类型不匹配	检查`Optional`/`List`等类型注解
API调用超时	未设置timeout或网络问题	添加`timeout=30`，增加重试机制
返回结果被截断	输出文本过长	限制返回条数，或分段返回

🎓 学习总结：自定义工具核心心法

🔑 一个中心：以Agent理解为中心设计工具
   ├─ 输入：用Pydantic严格约束+清晰description
   ├─ 输出：结构化纯文本，避免复杂嵌套
   └─ 错误：返回"可操作"的错误提示，而非堆栈

🔑 两个基本点：
   1️⃣ 稳定性：完善的异常处理+日志记录
   2️⃣ 可维护性：配置外置+代码注释+类型提示

🔑 三个不要：
   ❌ 不要硬编码敏感信息
   ❌ 不要返回原始JSON给Agent
   ❌ 不要忽略边界情况（空结果/超时/限流）