Deer-Flow 的嵌入式 Python 客户端与流式传输

Deer-Flow 的嵌入式 Python 客户端与流式传输

学习目标

• 理解 DeerFlowClient 的架构设计和 Gateway 等价方法
• 掌握 LangGraph stream_mode 的语义和 per-id 去重机制
• 学会分析 Gateway Conformance 测试的 schema 漂移检测

---

项目实践

DeerFlowClient 架构

DeerFlowClient 绕过 HTTP 直接调用 Harness 内部模块，与 Gateway API 保持响应 schema 一致：

from deerflow.client import DeerFlowClient

client = DeerFlowClient()

# 同步聊天
response = client.chat("分析这篇论文", thread_id="my-thread")

# 流式
for event in client.stream("hello"):
    if event.type == "messages-tuple" and event.data.get("type") == "ai":
        print(event.data["content"])

# 管理操作
models = client.list_models()
skills = client.list_skills()
client.update_skill("web-search", enabled=True)

关键设计：

• 导入与 Gateway API 相同的 deerflow 模块
• 共享相同的配置文件和数据目录
• 无 FastAPI 依赖

LangGraph stream_mode 语义

client.stream() 订阅 LangGraph 的 stream_mode=["values", "messages-tuple", "custom"]：

模式	语义
values	完整状态快照（标题、消息、产出物）；已交付的 AI 文本不重新合成
messages-tuple	逐块更新：AI 文本为 delta（按 id 拼接重建完整消息）；工具调用和结果各发射一次
custom	转发自 StreamWriter
end	流结束（携带每个消息 id 的累计 usage，仅计数一次）

per-id 去重

在流式传输中，同一消息的多个 delta 片段需要按 id 去重拼接：

• messages-tuple 模式为 AI 文本输出 delta 而非完整消息
• 调用者按消息 id 累积 delta 重建完整消息
• values 模式不重新合成已投递的 AI 文本，避免重复交付
• end 事件的 usage 按消息 id 累计，每个 id 仅计数一次

Gateway Conformance 测试

TestGatewayConformance 类（77 个单元测试）确保 DeerFlowClient 的每个 dict-returning 方法与对应的 Gateway Pydantic 响应模型一致：

• 每个测试通过 Gateway 模型解析客户端输出
• 如果 Gateway 新增必填字段而客户端不提供 → Pydantic 报 ValidationError → CI 拦截
• 覆盖：ModelsListResponse、ModelResponse、SkillsListResponse、SkillResponse 等

Gateway 与 DeerFlowClient 的平行路径

Gateway 和 DeerFlowClient 作为平行路径，共享相同的事件源和 Lead Agent 实例。

---

问题与规避

schema 漂移

Gateway 和 DeerFlowClient 可能在不同步的情况下独立演进。

Deer-Flow 的规避：TestGatewayConformance 在 CI 中自动检测 schema 漂移。Gateway 新增必填字段时客户端测试自动失败。

---

参考来源

• Deer-Flow CLAUDE.md — Embedded Client 章节
• docs/STREAMING.md — 流式传输设计文档
• packages/harness/deerflow/client.py — 客户端源码