Hotdry.
归档
ai-systems

Claude API 工具调用流式 JSON 解析工程实践：部分验证与实时错误恢复

针对 Claude API 工具调用的流式结构化输出，提供 JSON 解析、增量验证和错误恢复的工程参数与监控要点。

在低延迟代理编排工作流中，Claude API 的流式结构化输出机制为工具调用提供了高效支持。通过工程化实现流式 JSON 解析，可以实现部分验证和实时错误恢复，从而提升动态多轮交互的可靠性和响应速度。这种方法不同于静态 JSON schema 的单次响应场景，专注于增量处理和容错设计，确保代理在复杂环境中稳定运行。

Claude API 的工具调用（Tool Use）允许模型根据用户查询生成结构化 JSON 输入，用于调用外部函数，如天气查询或数据检索。在流式模式下，响应通过 Server-Sent Events (SSE) 传输，工具输入以 partial JSON delta 形式逐步输出。例如，当模型决定调用 “get_weather” 工具时，API 会先发送文本 delta（如 “我来检查旧金山天气”），然后是 tool_use 块的 input_json_delta，如 “{"location": "San Fra”” 和后续片段。这种设计支持细粒度流式，但也引入解析挑战：客户端必须累积不完整 JSON 片段，并在块结束时解析完整对象。

证据显示，这种流式机制在 Anthropic 官方文档中得到明确定义。工具输入 delta 是 partial JSON 字符串，模型通常一次输出一个完整键值对（如 “location”: “San Francisco, CA”），以确保兼容性。相比非流式请求，streaming 减少了整体延迟，但要求客户端实现 robust 的解析逻辑。例如，使用 Pydantic 等库进行部分 JSON 解析，可以在接收到 content_block_stop 事件时验证 schema 符合性。实际测试中，对于多工具并行调用，客户端需按 index 累积多个 tool_use 块，并返回对应的 tool_result，确保顺序一致。

部分验证是工程核心。通过增量校验，客户端可在接收 delta 时初步验证类型和范围，避免完整解析失败。例如，对于 get_weather 工具的 schema（location: string, unit: enum ["celsius", "fahrenheit"]），在累积 “{"location": "San Francisco, CA"}” 时，即可校验 location 为有效字符串，并在 unit delta 到达时验证枚举值。这种方法支持实时错误检测，如类型不匹配（e.g., location 传入数字），可立即触发回滚。

实时错误恢复机制进一步增强鲁棒性。常见错误包括网络中断（overloaded_error, HTTP 529）或 schema 违规。恢复策略包括：捕获 partial 响应，重构 continuation 请求，将已接收的 assistant 消息作为新消息开头续传。使用 SDK 的内置累积功能，可自动处理 text 和 tool_use 块。对于 thinking 块（extended thinking 模式），需验证 signature_delta 以确保完整性。监控要点：缓冲区大小不超过 10KB 以防内存溢出，超时阈值设为 30s（模型思考时间），并记录 delta 事件延迟（>500ms 触发警报）。

可落地参数与清单如下：

解析参数：

  • 缓冲大小：最大 8192 字节，超过时强制解析并恢复。
  • Delta 合并阈值：累积至少 3 个 delta 后尝试 partial 校验。
  • Schema 验证规则：使用 JSON Schema Draft 2020-12，支持 required 字段检查和 enum 验证。

恢复清单：

  1. 捕获中断：监听 SSE error 事件，保存已解析 content 块。
  2. 重构请求：新消息中包含 partial assistant content，设置 stream: true 续传。
  3. 回滚策略：若 schema 失败 3 次，切换到非工具响应模式，并日志记录。
  4. 监控指标：解析成功率 >95%，平均延迟 <200ms，错误恢复时间 <5s。

实现伪代码：

import json
from typing import Dict, Any

class StreamingParser:
    def __init__(self):
        self.buffer: Dict[int, str] = {}  # index -> partial json
        self.schema = {...}  # tool schema

    def on_delta(self, index: int, partial_json: str):
        if index not in self.buffer:
            self.buffer[index] = ""
        self.buffer[index] += partial_json
        # Partial validation
        try:
            partial_obj = json.loads(self.buffer[index])
            # Validate known fields
            if 'location' in partial_obj and not isinstance(partial_obj['location'], str):
                raise ValueError("Location must be string")
        except json.JSONDecodeError:
            pass  # Continue accumulating

    def on_block_stop(self, index: int):
        try:
            full_input = json.loads(self.buffer[index])
            # Full schema validation
            validate_instance(full_input, self.schema)
            # Execute tool
            result = execute_tool(full_input)
            # Return tool_result
        except Exception as e:
            # Error recovery: fallback to text response
            result = f"Error: {e}. Please clarify."
        del self.buffer[index]

在生产环境中，集成 Prometheus 监控 delta 事件速率和解析错误率，确保系统在高并发下稳定。测试场景包括网络抖动模拟和 schema 边界 case，如缺失 required 字段。

资料来源：

查看归档