Pydantic AI 中 Anthropic 模型流式调用工具时的响应处理问题解析

问题背景

在使用 Pydantic AI 框架与 Anthropic 的 Claude 3 模型进行交互时，开发者发现了一个有趣的现象：当使用流式调用（streaming）方式执行包含工具调用的请求时，模型的响应行为与非流式调用存在显著差异。

现象描述

具体表现为：

• 非流式调用时，模型会完整执行工具调用并返回最终结果（如"用户是John，当前时间是..."）
• 流式调用时，模型仅返回初始响应（如"我将帮你查询用户信息..."），而不会继续返回工具调用后的最终结果
• 相同代码在 OpenAI 和 Gemini 模型上表现正常

技术分析

经过深入调试，发现问题根源在于 Anthropic 模型特有的响应模式与 Pydantic AI 流式处理逻辑的交互方式：

1. 模型响应模式：Anthropic 模型（特别是 Claude 系列）通常采用三步走策略：

   • 首先说明将使用哪些工具
   • 然后实际调用这些工具
   • 最后汇总工具结果并生成最终响应

2. 流式处理机制：Pydantic AI 的 run_stream 方法在收到第一个有效文本响应后就会结束，因为：

   • 未指定 output_type 时，系统会接受第一个文本输出作为最终结果
   • 这与 Anthropic 的多阶段响应模式产生了冲突

解决方案

针对这一问题，开发者提供了几种可行的解决方案：

1. 显式提示模型

尝试通过提示工程让 Claude 模型不要解释其工具调用过程，直接返回结果。不过这种方法可靠性有限，可能无法完全解决问题。

2. 使用结构化输出

定义明确的输出类型，强制模型返回结构化数据而非自然语言：

class NameAndTime(BaseModel):
    name: str
    time: str

# 创建 Agent 时指定 output_type
agent = Agent(model="anthropic:claude-3-7-sonnet-latest", output_type=NameAndTime)

这种方法将获得更可预测的输出格式，但需要调整代码以适应结构化数据。

3. 手动迭代处理

最灵活的方法是直接操作执行图（execution graph），手动控制流式处理过程：

async for node in agent.iter("查询用户和时间"):
    if isinstance(node, ModelResponse):
        async for text in node.stream_text():
            print(text)
    
    # 可以在此添加自定义逻辑判断是否继续

这种方法提供了最大的控制权，允许开发者：

• 决定何时结束流式处理
• 检查中间结果
• 实现自定义的终止条件

最佳实践建议

1. 明确需求：如果只需要最终结果，非流式调用可能是更简单的选择
2. 结构化优先：尽可能使用结构化输出类型，提高响应一致性
3. 复杂场景用手动控制：对于需要精细控制或复杂交互的场景，推荐使用迭代方法
4. 跨模型兼容性：如果项目需要支持多种模型，应测试不同提供商的响应模式差异

总结

Pydantic AI 框架与 Anthropic 模型的交互展示了大型语言模型在实际应用中的复杂性。理解模型特定的响应模式和处理机制对于构建稳定的AI应用至关重要。通过本文介绍的方法，开发者可以更有效地处理流式工具调用场景，确保获得预期的完整响应。

这一案例也提醒我们，在构建基于LLM的应用时，考虑不同模型提供商的行为差异是确保系统鲁棒性的关键因素。