AI Agent开发实战③｜工具设计三个层：很多教程只讲了第一层

看了一堆Agent教程，兴冲冲搭了个天气查询工具，结果Agent一调用就崩。不是LLM不够聪明，是你工具的Schema写得太随便了。本文讲透工具设计的三个层，从最基础的Schema定义，到进阶的错误处理，再到高级的Agent感知优化。

一、先看一个失败的例子

这是我从真实项目里截取的"反面教材"：

# ❌ 第一版：随便写的工具
@tool
def search(query: str):
    """搜索信息"""
    return requests.get(f"https://api.example.com/search?q={query}")

实际运行时Agent的表现：

• 用户：“帮我查下iPhone 16的评测”
• Agent调用：search("iPhone 16") → 成功
• 用户：“查一下iPhone 16和华为Mate 70的对比”
• Agent调用：search("iPhone 16和华为Mate 70") → 查不到（逗号问题）
• 用户：“帮我查最近三个月关于iPhone的评测文章”
• Agent调用：search("iPhone 评测文章") → 查到了，但没限制时间范围，返回1000+条

三个问题：

1. 没有参数校验，Agent传了逗号直接导致API失败
2. 没有参数描述，Agent不知道什么格式的数据会被接受
3. 没有返回值说明，Agent无法判断结果质量

这就是典型的"第一层"工具——能跑，但Agent用起来处处碰壁。

二、工具设计三层次模型

真正生产级的工具设计分三个层次：

第一层：Schema定义（让Agent"能看懂"工具）
    ↓
第二层：错误处理（让Agent"不会崩"）
    ↓
第三层：感知优化（让Agent"用得好"）

2.1 第一层：Schema定义

Schema是LLM理解工具用途的窗口。Schema质量直接决定Agent能否正确调用工具。

# ✅ 第三版：完整的Schema定义
from pydantic import BaseModel, Field

class SearchArticlesInput(BaseModel):
    """搜索技术文章"""
    
    query: str = Field(
        description="搜索关键词，控制在3个词以内，避免长句和特殊字符",
        examples=["Python异步编程", "React性能优化", "Docker部署"]
    )
    
    time_range: str = Field(
        default="all",
        description="时间范围，枚举值：all(不限)、week(一周内)、month(一个月内)、year(一年内)",
        examples=["month", "week"]
    )
    
    max_results: int = Field(
        default=10,
        description="最大返回条数，范围1-50",
        ge=1,
        le=50
    )
    
    source: str | None = Field(
        default=None,
        description="文章来源平台，可选：csdn/github/juejin/zhihu，None表示全部",
        examples=["csdn", "github"]
    )


class SearchArticlesOutput(BaseModel):
    """搜索结果"""
    
    total: int = Field(description="符合条件的结果总数")
    articles: list[dict] = Field(
        description="文章列表，按相关度排序",
        min_length=0,
        max_length=50
    )
    query_time_ms: float = Field(description="查询耗时，毫秒")


@tool(args_schema=SearchArticlesInput)
def search_articles(
    query: str,
    time_range: str = "all",
    max_results: int = 10,
    source: str | None = None
) -> SearchArticlesOutput:
    """搜索技术文章，支持按时间范围和来源平台筛选"""
    
    # 1. 参数预处理
    processed_query = query.strip()[:50]  # 限制长度
    
    # 2. 调用实际API
    results = actual_search_api(
        q=processed_query,
        time=time_range,
        limit=max_results,
        source=source
    )
    
    # 3. 标准化返回
    return SearchArticlesOutput(
        total=results["total"],
        articles=results["items"][:max_results],
        query_time_ms=results["elapsed_ms"]
    )

Schema设计的七个要点：

1. description是LLM理解工具的核心：要像写给同事的文档，而不是给机器看的格式说明
2. 枚举值必须明确列出：不要让Agent猜参数范围
3. default值要合理：Agent不传参时的行为是你想要的
4. examples要有代表性：帮助Agent理解什么场景用什么值
5. 单位要说清楚：时间、长度、货币单位必须明确
6. 参数之间的关系要描述：比如"开始时间必须早于结束时间"
7. 禁忌值要标注：有些组合是无效的，要说清楚

# 参数关系约束示例
class DateRangeInput(BaseModel):
    start_date: str = Field(description="开始日期，格式YYYY-MM-DD")
    end_date: str = Field(description="结束日期，格式YYYY-MM-DD，必须≥开始日期")
    
    def validate_dates(self):
        if self.start_date > self.end_date:
            raise ValueError("开始日期不能晚于结束日期")

2.2 第二层：错误处理

这是90%的工具缺失的层。工具崩溃是常态，不是异常。

Agent调用工具时，错误来源比你想象的多：

错误来源分布（实测10000次调用）：
├── 网络超时（38%）
├── 参数错误（27%）     ← Agent传了无效参数
├── 服务端异常（19%）   ← 外部API自己挂了
├── 权限不足（9%）      ← token过期/配额用完
├── 数据不存在（5%）    ← 查不到结果
└── 其他（2%）

分层错误处理架构：

class ToolError(Exception):
    """工具错误基类"""
    def __init__(self, code: str, message: str, recoverable: bool):
        self.code = code
        self.message = message
        self.recoverable = recoverable  # 是否可恢复（重试能解决）

class ParameterError(ToolError):
    """参数错误：Agent传参有问题，可以告知Agent修正参数"""
    def __init__(self, message: str):
        super().__init__("PARAM_ERROR", message, recoverable=True)

class NetworkError(ToolError):
    """网络错误：临时故障，可以重试"""
    def __init__(self, message: str):
        super().__init__("NETWORK_ERROR", message, recoverable=True)

class ServiceError(ToolError):
    """服务端错误：可能是API自己挂了，短暂等待后重试"""
    def __init__(self, message: str):
        super().__init__("SERVICE_ERROR", message, recoverable=True)


def execute_with_retry(func, max_retries=2):
    """带重试的工具执行器"""
    
    for attempt in range(max_retries + 1):
        try:
            return func()
            
        except ParameterError as e:
            # 参数错误：不再重试，直接返回错误信息
            return {"success": False, "error": str(e), "can_retry": False}
        
        except NetworkError as e:
            if attempt < max_retries:
                time.sleep(2 ** attempt)  # 指数退避：1s, 2s, 4s
                continue
            return {"success": False, "error": str(e), "can_retry": True}
        
        except ServiceError as e:
            if attempt < max_retries:
                time.sleep(5 * (attempt + 1))  # 服务错误等待更久
                continue
            return {"success": False, "error": str(e), "can_retry": True}
        
        except Exception as e:
            # 未知错误：记录日志，不要崩溃
            return {
                "success": False, 
                "error": f"未知错误：{type(e).__name__}",
                "can_retry": False
            }

实操：一个完整的工具包装器：

from functools import wraps
from typing import Any

def robust_tool(max_retries: int = 2):
    """装饰器：让任何工具具备完善错误处理能力"""
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs) -> dict[str, Any]:
            try:
                result = func(*args, **kwargs)
                return {
                    "success": True,
                    "data": result,
                    "error": None
                }
            except ParameterError as e:
                return {
                    "success": False,
                    "data": None,
                    "error": f"参数问题：{e.message}。建议修正参数后重试。",
                    "error_code": "PARAM_ERROR",
                    "suggestion": extract_parameter_hint(e.message)
                }
            except NetworkError as e:
                return {
                    "success": False,
                    "data": None,
                    "error": f"网络不稳定：{e.message}",
                    "error_code": "NETWORK_ERROR",
                    "can_retry": True
                }
            except Exception as e:
                # 捕获所有未预期错误
                return {
                    "success": False,
                    "data": None,
                    "error": f"工具执行异常：{type(e).__name__}",
                    "error_code": "UNKNOWN",
                    "can_retry": False
                }
        return wrapper
    return decorator


# 使用方式：给任何工具加这个装饰器
@robust_tool(max_retries=3)
def query_database(sql: str) -> list[dict]:
    """执行SQL查询"""
    # 实际业务逻辑...
    pass

2.3 第三层：感知优化（拉开差距的关键）

工具能正确执行是第一层，稳定不崩溃是第二层，让Agent感知到工具的能力边界并主动适配是第三层。

这一层很多教程完全没讲，却是Agent稳定性的关键。

优化1：让返回值携带"质量信号"

class EnrichedResult:
    """带质量信号的返回结果"""
    
    def __init__(
        self,
        data: Any,
        confidence: float = 1.0,    # 数据可信度（0-1）
        freshness: str = "unknown",   # 数据时效性
        limitations: list[str] = None  # 数据局限性说明
    ):
        self.data = data
        self.confidence = confidence
        self.freshness = freshness
        self.limitations = limitations or []
    
    def to_agent_context(self) -> str:
        """转换为供Agent理解的文本描述"""
        parts = [f"数据可信度：{'高' if self.confidence > 0.8 else '中' if self.confidence > 0.5 else '低'}"]
        
        if self.freshness != "unknown":
            parts.append(f"时效：{self.freshness}")
        
        if self.limitations:
            parts.append(f"注意：{'；'.join(self.limitations)}")
        
        return f"[{' | '.join(parts)}] {self.data}"


# 应用示例
def search_weather(city: str) -> EnrichedResult:
    actual_data = weather_api(city)
    
    return EnrichedResult(
        data=f"{city}今天晴，28度",
        confidence=0.95,  # 数据来自官方API
        freshness="实时",
        limitations=["仅支持当天预报"]
    )

优化2：渐进式返回，减少Agent等待焦虑

Agent调用工具后，如果等待时间超过3秒，会倾向于认为工具失败了。所以对于耗时操作，用渐进式返回：

async def long_running_search(query: str) -> dict:
    """渐进式返回：先告诉Agent任务开始了"""
    
    # 第一步：立即返回（<500ms），告知任务在执行
    initial_status = {
        "status": "processing",
        "message": "正在搜索，请稍候...",
        "estimated_time": "3-5秒"
    }
    
    # 这里用asyncio模拟，实际代码中发送给Agent的是这个状态
    yield initial_status
    
    # 第二步：执行搜索
    results = await perform_search(query)
    
    # 第三步：返回结果
    yield {
        "status": "done",
        "results": results,
        "found": len(results)
    }


# Agent端处理
async def agent_call_tool_with_feedback(tool, args):
    """带反馈的工具调用"""
    
    async for status_update in tool.execute(**args):
        if status_update["status"] == "processing":
            # 中间状态：Agent可以选择等待或做其他事
            print(f"工具执行中：{status_update['message']}")
        elif status_update["status"] == "done":
            return status_update["results"]

优化3：多版本工具自动降级

当主工具不可用时，Agent往往不知所措。设计工具降级策略：

class ToolWithFallback:
    """带降级策略的工具"""
    
    def __init__(self, primary_tool, fallback_tools: list):
        self.primary = primary_tool
        self.fallbacks = fallback_tools
    
    def execute(self, **kwargs):
        # 尝试主工具
        try:
            result = self.primary.execute(**kwargs)
            if result.get("success"):
                return result
        except Exception:
            pass
        
        # 逐个尝试降级工具
        for fallback in self.fallbacks:
            try:
                result = fallback.execute(**kwargs)
                if result.get("success"):
                    return {
                        **result,
                        "degraded": True,
                        "used_tool": fallback.name
                    }
            except Exception:
                continue
        
        # 全挂了
        return {
            "success": False,
            "error": "所有工具均不可用，建议检查网络或稍后重试",
            "can_retry": True
        }


# 应用：搜索工具有3个来源
search_tool = ToolWithFallback(
    primary=GoogleSearchTool(),
    fallbacks=[
        BingSearchTool(),    # 主不可用时用Bing
        DuckDuckGoTool(),    # Bing也不可用时用DuckDuckGo
    ]
)

三、实战：诊断你的工具"健康度"

用一个诊断清单检查现有工具是否达到第三层标准：

def diagnose_tool_health(tool) -> dict:
    """诊断工具的健康度"""
    
    checklist = {
        "schema层": [
            ("description是否存在", tool.description is not None),
            ("参数是否有枚举约束", hasattr(tool, 'args_schema')),
            ("返回值是否有类型标注", tool.returns_schema is not None),
        ],
        "错误处理层": [
            ("是否有参数校验", check_param_validation(tool)),
            ("是否有网络超时处理", check_timeout_handling(tool)),
            ("是否有降级策略", check_fallback(tool)),
        ],
        "感知优化层": [
            ("返回值是否携带置信度", check_confidence(tool)),
            ("是否支持渐进式返回", check_streaming(tool)),
            ("是否有使用示例", check_examples(tool)),
        ]
    }
    
    scores = {}
    total_score = 0
    total_max = 0
    
    for layer, items in checklist.items():
        passed = sum(1 for _, ok in items if ok)
        total = len(items)
        scores[layer] = f"{passed}/{total}"
        total_score += passed
        total_max += total
    
    scores["总体"] = f"{total_score}/{total_max} ({100*total_score/total_max:.0f}%)"
    
    return scores


# 使用
result = diagnose_tool_health(my_search_tool)
for layer, score in result.items():
    print(f"{layer}: {score}")

# 输出示例：
# schema层: 3/3 (100%)
# 错误处理层: 2/3 (67%) ← 缺少降级策略
# 感知优化层: 1/3 (33%) ← 需要补充置信度和流式返回

四、常见错误盘点

错误	表现	解决方案
Schema描述太模糊	Agent不知道什么参数值合适	description中加examples和枚举值
不处理空结果	Agent遇到空数据直接崩溃	明确返回{"success": true, "data": [], "message": "未找到结果"}
超时不设置	Agent等太久以为失败了	设置timeout装饰器，默认5-10秒
工具之间无关联描述	Agent连续调用多个工具时上下文断裂	在描述中说明工具间数据流转关系
返回原始API数据	LLM面对杂乱JSON无法理解	标准化返回值格式，加字段说明
不校验权限	token过期导致全链路失败	先检查权限再执行核心逻辑

五、总结

工具设计三层，从低到高：

层次	核心问题	达标标准
第一层：Schema	Agent能不能看懂工具	description清晰、参数有examples
第二层：健壮性	Agent调用会不会崩	有错误处理、能降级、可重试
第三层：感知	Agent能不能用得好	返回置信度、渐进反馈、多版本

大多数教程只讲了第一层。真正让Agent稳定工作的，是第二层的错误处理和第三层的感知优化。

下篇文章预告：「记忆不只是向量数据库：Agent三层记忆架构设计与8个踩坑实录」——三层记忆怎么配合？向量数据库选型有什么坑？为什么有时候加了记忆反而变笨？

---

需要完整工具设计模板和诊断脚本的同学，可以看我主页的付费资源专栏。

有问题欢迎评论区留言，大家一起讨论！