Skip to content

8.3_tool_execution.md

Latest commit

 

History

History
335 lines (256 loc) · 8.67 KB

8.3_tool_execution.md

File metadata and controls

335 lines (256 loc) · 8.67 KB

8.3 工具执行与结果处理

8.3.1 工具执行流程

graph LR
    A["接收调用请求"] --> B["验证参数"]
    B --> C["执行工具"]
    C --> D["处理结果"]
    D --> E["返回给模型"]
Loading

图 8-3：工具执行流程

8.3.2 参数验证

在执行前验证参数的有效性：

def validate_params(tool_name, params, schema):
    # 检查必需参数
    for required in schema.get("required", []):
        if required not in params:
            raise ValueError(f"缺少必需参数: {required}")

    # 检查参数类型
    for key, value in params.items():
        expected_type = schema["properties"][key]["type"]
        if not isinstance(value, TYPE_MAP[expected_type]):
            raise TypeError(f"参数 {key} 类型错误")

    # 检查枚举值
    for key, value in params.items():
        if "enum" in schema["properties"][key]:
            if value not in schema["properties"][key]["enum"]:
                raise ValueError(f"参数 {key} 值无效")

8.3.3 执行策略

同步执行

简单场景，等待执行完成：

result = execute_tool(tool_name, params)

异步执行

IO 密集型或耗时操作：

async def execute_tools(tool_calls):
    tasks = [execute_tool_async(tc) for tc in tool_calls]
    results = await asyncio.gather(*tasks)
    return results

超时控制

防止工具执行无限等待：

try:
    result = await asyncio.wait_for(
        execute_tool(tool_name, params),
        timeout=30.0
    )
except asyncio.TimeoutError:
    result = {"error": "执行超时"}

8.3.4 结果格式化

将工具结果格式化为模型易理解的形式：

简洁输出

{
  "status": "success",
  "data": {
    "temperature": 25,
    "condition": "晴"
  }
}

包含上下文

{
  "query": "北京天气",
  "result": {
    "city": "北京",
    "temperature": 25,
    "condition": "晴",
    "humidity": 45
  },
  "data_time": "2026-03-01 10:00"
}

8.3.5 错误处理

工具执行可能失败，需要优雅处理：

错误类型分类

错误类型 处理方式 反馈给模型
参数错误 不执行 说明具体问题
执行失败 记录日志 返回错误信息
超时 终止执行 建议稍后重试
权限不足 拒绝执行 说明权限要求

错误信息模板

{
  "status": "error",
  "error_type": "execution_failed",
  "message": "无法连接到天气服务",
  "suggestion": "请稍后再试或换一个城市查询"
}

8.3.6 结果注入上下文

工具结果需要正确注入对话上下文：

{
  "role": "assistant",
  "content": null,
  "tool_calls": [{
    "id": "call_abc123",
    "type": "function",
    "function": {
      "name": "get_weather",
      "arguments": "{\"city\": \"北京\"}"
    }
  }]
},
{
  "role": "tool",
  "tool_call_id": "call_abc123",
  "content": "{\"temperature\": 25, \"condition\": \"晴\"}"
}

8.3.7 结果压缩

工具返回大量数据时可能需要压缩：

def compress_result(result, max_tokens=500):
    if count_tokens(result) > max_tokens:
        # 提取关键信息
        return summarize(result)
    return result

8.3.8 多轮工具调用

复杂任务可能需要多轮工具调用：

sequenceDiagram
    participant M as 模型
    participant A as 应用

    M->>A: 调用工具 A
    A->>M: 工具 A 结果
    Note over M: 分析结果，决定下一步
    M->>A: 调用工具 B
    A->>M: 工具 B 结果
    Note over M: 综合结果，生成响应
    M->>A: 最终响应
Loading

图 8-4：多轮工具调用流程

每轮结果都成为上下文的一部分，影响后续决策。

8.3.9 安全考虑

工具执行涉及安全风险：

  1. 输入验证：防止注入攻击
  2. 权限控制：验证操作权限
  3. 操作审计：记录敏感操作
  4. 确认机制：危险操作需人工确认

8.3.10 高级工具执行技术

随着智能体需要使用的工具数量增长，传统的工具调用方式面临挑战。以下介绍两项高级技术：

工具搜索工具

问题：当连接多个 MCP 服务器时，工具定义可能消耗大量 Token（规模取决于工具数量、Schema 复杂度与描述长度）：

解决方案：不预加载所有工具，而是按需发现。

{
  "tools": [
    {"type": "tool_search_tool", "name": "tool_search"},
    {
      "name": "github.createPullRequest",
      "defer_loading": true
    }
  ]
}

效果（常见观察）：

  • 只暴露必要工具可以显著降低上下文开销
  • 工具选择更聚焦，误调用与歧义通常会减少

编程式工具调用

问题：传统工具调用有两个瓶颈：

  1. 中间结果污染上下文（分析 10MB 日志文件，全部进入上下文）
  2. 每次调用需要一次推理，开销大

解决方案：让模型生成受限的工作流计划，由服务端在沙箱和权限边界内执行；不要让模型生成的任意代码直接持有团队、费用、客户或生产系统凭据。

# 模型只生成声明式计划；服务端负责鉴权、执行、脱敏和审计
workflow = {
    "operation": "expense_budget_check",
    "team": "engineering",
    "period": "Q3",
    "output": "over_budget_summary",
}

result = await workflow_runner.run(
    workflow,
    caller=user_identity,
    policy_checks=["team_scope", "read_only", "pii_redaction"],
    network_scope="internal_tools_only",
)
return result.redacted_summary

效果（常见观察）：

  • 上下文只接收最终结果而非原始数据，通常能显著降低 Token 消耗
  • 降低中间结果“污染上下文”的概率，并减少不必要的多轮推理

使用指南：

技术 适用场景
Tool Search Tool 工具定义 >10K Token、10+ 工具
Programmatic Tool Calling 处理大数据集只需摘要、3+ 步依赖工具调用

8.3.11 工程实战：端到端数据分析助手

以下展示一个完整的数据分析助手工作流，包含规划、执行、结果处理和上下文更新。

场景：用户上传 CSV 文件并要求：“分析上季度的销售趋势，并画图展示。”

流程图：

graph TB
    User["用户输入"] --> Plan["规划器(Planner)"]
    Plan --> Step1["步骤1: 读取CSV"]
    Step1 --> Tool1["调用 read_csv"]
    Tool1 --> Res1["返回 DataFrame 摘要"]
    Res1 --> Step2["步骤2: 分析数据"]
    Step2 --> Tool2["调用 python_repl"]
    Tool2 --> Res2["生成图表(Base64)"]
    Res2 --> Final["生成最终回复"]
Loading

图 8-5：端到端数据分析工作流

关键代码实现逻辑：

async def run_analysis_agent(user_query, context):
    messages = context + [{"role": "user", "content": user_query}]

    # 1. 规划阶段
    plan = await llm.plan(messages)
    # output: 1. read_csv, 2. analyze_trend, 3. plot_chart

    results = {}

    for step in plan.steps:
        # 2. 执行阶段
        tool_name = step.tool
        tool_args = step.args

        try:
            # 3. 结果处理
            # 仅截取前5行数据作为上下文，避免 Token 爆炸
            raw_result = await tools.execute(tool_name, tool_args)
            processed_result = truncate_dataframe(raw_result)

            # 注入当前步骤结果到上下文
            messages.append({
                "role": "tool",
                "tool_call_id": step.tool_call_id,
                "content": processed_result
            })

        except Exception as e:
            # 错误恢复机制
            messages.append({"role": "system", "content": f"工具执行失败: {str(e)}，请尝试其他方法"})
            continue

    # 4. 生成最终回复
    final_response = await llm.chat(messages)
    return final_response

8.3.12 核心指标与评估

在生产环境中，需要关注以下指标来衡量工具执行系统的健康度：

指标 目标值（示意） 说明
执行成功率 以基线提升为目标 工具调用无异常（Crash/Timeout）的比例
参数准确率 以基线提升为目标 模型生成的参数符合 Schema 定义且逻辑正确
平均延迟 取决于工具与 SLO 工具本身的执行耗时（不含 LLM 推理）
幻觉率 越低越好 调用了不存在的工具或虚构了参数
Token 效率 越高越好 原始工具结果 vs 注入上下文的结果大小

优化建议：

  • 对于高频且低延迟要求的工具（如查天气），使用本地逻辑而非 LLM 规划。
  • 对于易错的复杂参数（如 SQL），在 System Prompt 中提供 Few-Shot 示例。