Skip to content

Latest commit

 

History

History
211 lines (167 loc) · 7.58 KB

File metadata and controls

211 lines (167 loc) · 7.58 KB

3.3 处理工具调用与结果反馈

3.3.1 识别工具调用请求

当 Claude 决定使用工具时,它的响应行为与普通对话有所不同。应用程序的核心逻辑需要能够准确识别这种状态切换。

停止原因

最直接的判断依据是 API 响应中的 stop_reason 字段:

  • tool_use: 表示模型生成了一个或多个确定的工具调用意图,并暂停了生成,等待外部输入。
  • end_turn: 表示模型完成了回答,没有通过工具进行后续操作。
  • max_tokens: 警告!表示被长度截断。如果截断发生在 JSON 生成中间,需要特殊处理。
  • stop_sequence: 命中自定义停止序列,应按业务协议判断是否完整。
  • pause_turn: 服务端工具或长流程需要继续轮询;应把本次 Assistant 响应保留进历史,再继续调用 Messages API。
  • refusal: 模型以 HTTP 200 返回安全拒答,不应按普通最终答案盲取 response.content[0].text
  • model_context_window_exceeded: 生成触达上下文窗口边界,应触发压缩、截断或降级策略。

内容块结构

stop_reason="tool_use" 时,content 列表通常包含两种类型的块:

  1. Text Block(说明性文本): {"type": "text", "text": "..."}。这是模型在调用工具前的说明性文本(朴素的思维链),解释它为什么要调用工具——注意它与 Extended/Adaptive Thinking 产生的专用 thinking 块是两种不同的块类型。
  2. Tool Use Block: 核心指令。
{
  "stop_reason": "tool_use",
  "content": [
    {
      "type": "text",
      "text": "好的,我来查询北京和上海的天气。"
    },
    {
      "type": "tool_use",
      "id": "toolu_01...",
      "name": "get_weather",
      "input": {"city": "Beijing"}
    },
    {
      "type": "tool_use",
      "id": "toolu_02...",
      "name": "get_weather",
      "input": {"city": "Shanghai"}
    }
  ]
}

3.3.2 并行工具执行

如上例所示,Claude Sonnet 5 等先进模型具有强大的 并行调用能力。它可以在一次回复中请求执行多个动作。

关键规则:如果模型一次性发出了 3 个 tool_use 请求,必须在下一次交互中,一次性返回这 3 个请求对应的 tool_result。如果遗漏任何一个,会引发 API 错误。

执行策略

对于并行请求,推荐使用异步编程(如 Python 的 asyncio)并发执行,以减少总等待时间。

3.3.3 构造工具结果

执行完工具逻辑后,需要构造一个特定格式的 tool_result 消息块。

基础文本结果

大多数工具返回 JSON 字符串或纯文本。

{
  "role": "user",
  "content": [
    {
      "type": "tool_result",
      "tool_use_id": "toolu_01...",  // 必须与请求中的 ID 完全匹配
      "content": "25°C, Sunny"
    }
  ]
}

错误处理:is_error

如果工具执行失败(例如数据库连接超时、API 报错),不要抛出异常中断程序。相反,应该捕获异常并告诉 Claude 发生了错误。

{
  "type": "tool_result",
  "tool_use_id": "toolu_01...",
  "content": "Error: Connection timed out after 5000ms",
  "is_error": true
}

当 Claude 收到 is_error: true 时,它会分析错误原因,并尝试自我修正(例如修改参数并重试)。

多模态结果:返回图片或文档

工具不仅能返回文字,还能返回图片或文档块。这对于截图工具、绘图工具和文档检索工具至关重要。

{
  "type": "tool_result",
  "tool_use_id": "toolu_01...",
  "content": [
    {"type": "text", "text": "这是当前的屏幕截图:"},
    {
      "type": "image",
      "source": {
        "type": "base64",
        "media_type": "image/png",
        "data": "iVBORw0KGgoAAAANSUhEUg..."
      }
    }
  ]
}

3.3.4 多轮对话的历史维护

这是开发者最容易出错的地方。 为了让 Claude 理解上下文,必须维护完整的对话历史链条。

正确的历史堆栈顺序

  1. User: “查询北京天气”
  2. Assistant:(包含 tool_use 请求的完整响应)
  3. User:(包含 tool_result 的结果响应)
  4. Assistant:(Claude 根据结果生成的最终回复)

错误示范: 如果在第 3 步发送结果时,忘记在 messages 列表中包含第 2 步 Assistant 的原始请求消息,API 会报错,因为它找不到 tool_use_id 的来源。

3.3.5 完整代码范例

下面是一个健壮的处理循环,涵盖了并行执行和历史维护:

import anthropic

client = anthropic.Anthropic()
messages = []

def process_query(user_input):
    messages.append({"role": "user", "content": user_input})

    while True:
        # 发送请求给 Claude
        response = client.messages.create(
            model="claude-sonnet-5",
            max_tokens=1024,
            tools=tools, # 假设 tools 已定义
            messages=messages
        )

        # 将 Claude 的回复加入历史
        messages.append({"role": "assistant", "content": response.content})

        # 检查是否停止原因为 tool_use
        if response.stop_reason == "tool_use":
            tool_results = []

            # 遍历所有工具调用请求
            for block in response.content:
                if block.type == "tool_use":
                    # 执行具体逻辑 (此处简化为通过名称分发)
                    result_content = execute_function(block.name, block.input)

                    tool_results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(result_content)
                    })

            # 将所有结果作为一条 User 消息发回
            messages.append({
                "role": "user",
                "content": tool_results
            })
            # 循环继续,Claude 将看到结果并决定是继续调用工具还是给出最终回复

        elif response.stop_reason == "pause_turn":
            # 服务端工具或长任务要求继续,同一历史直接重试
            continue
        elif response.stop_reason == "end_turn":
            # 任务完成,只拼接文本块,避免假设 content[0] 一定是 text
            final_response = "".join(
                block.text for block in response.content
                if getattr(block, "type", None) == "text"
            )
            print(f"Claude: {final_response}")
            break
        elif response.stop_reason in {"max_tokens", "model_context_window_exceeded"}:
            raise RuntimeError("Claude response was truncated; compact context or raise max_tokens")
        elif response.stop_reason == "refusal":
            raise RuntimeError("Claude refused the request; return a controlled fallback")
        else:
            # stop_sequence 或未来新增 stop_reason:交给业务层显式处理
            final_response = response.content[0].text
            print(f"Claude: {final_response}")
            break

def execute_function(name, args):
    # 模拟工具执行
    if name == "get_weather":
        return f"Weather in {args.get('city')} is Sunny."
    return "Unknown tool"

掌握了请求与反馈的机制后,下面看看如何将多个工具组合起来,实现复杂的任务编排。

可运行参考:examples/tool_use_loop.py 完整回放 assistant 的 tool_use 块,并用同一个调用 ID 生成 user tool_result;其 fake transport 测试无需 API key 或网络。

➡️ Agent 编排模式