PydanticAI 的流式传输实践

PydanticAI 的流式传输实践

学习目标

补充理解：

• PartStart/Delta/End 事件模型的设计动机
• AgentStream 的防抖机制与渐进验证
• cooperative hand-off 协议的实现

前置知识

• 流式传输架构 — 通用流式传输概念

---

1. Part 事件模型

PydanticAI 使用三种事件类型表示流式更新：

PartStartEvent(index, part)    # 新 part 开始（TextPart、ToolCallPart、ThinkingPart）
PartDeltaEvent(index, delta)   # 增量更新（TextPartDelta、ToolCallPartDelta）
PartEndEvent(index, part)      # part 完成

与通用流式传输的差异：

• 通用流式传输通常基于纯 token 流——每个 token 是一个事件
• PydanticAI 的 Part 事件模型是结构化的——支持 thinking、tool call、text 等不同类型的增量更新
• 每个 part 有 index 标识符——支持并行多个 part 的增量更新（如同时流式生成多个工具调用）

2. AgentStream 的四层 API

AgentStream 包裹 models.StreamedResponse，提供四种消费方式：

stream_response(debounce_by=0.1)  # 返回 ModelResponse 快照（state='incomplete'）
stream_output(debounce_by=0.1)    # 渐进验证的输出（partial → full validation）
stream_text(delta=False)          # 文本增量/累积（可选验证）
get_output()                      # 排空流，返回最终验证输出

防抖机制：group_by_temporal 时间分组，默认 debounce_by=0.1 秒——平衡了实时更新和过度分片。

3. Cooperative Hand-off 协议

流式路径的核心协议：

1. ModelRequestNode.stream() 打开模型请求流
2. 设置 stream_ready 事件，通知消费者可以开始消费
3. 消费者（用户代码）获取 AgentStream 并开始消费
4. 当检测到 FinalResultEvent 时，停止流式
5. 调用 process_tool_calls() 处理剩余工具调用
6. 进入 SetFinalResult 节点结束图遍历

为什么需要 cooperative hand-off？

• 流式消费和工具处理是异步的——消费者需要在流式输出的同时决定是否提前结束
• 协议确保了流的生命周期管理——打开、消费、关闭由两个独立方（节点和消费者）协调

4. 流式与并行

AgentStream 使用 anyio.Lock 进行序列化访问——防止并发消费者竞争。流的状态（is_complete）通过锁保护。

注意：流被取消时，消息历史可能不完整——cancel() 记录 interrupted 状态。