[Discussion] Subagent 异步执行与 Tracing 完整性讨论

[heart-scalpel]

概述

在为 DeerFlow 实现 CLI 会话持久化功能时，发现了 Subagent 执行架构中的一些问题，希望与维护者讨论：

1. Subagent 使用异步执行时与同步 checkpointer 的兼容性
2. Subagent 的 Langfuse tracing 是否完整
3. 最佳实践建议

背景：问题描述

错误现象

CLI模式下当启用 subagent 委托功能后，遇到以下错误：

NotImplementedError: The SqliteSaver does not support async methods. Consider using AsyncSqliteSaver instead.

错误堆栈

好的，我现在尝试启动一个 subagent：

[DeerFlowClient] Agent calling tools: ['task']
[DeerFlowClient] Agent calling tools: ['']
[trace=e559546b] Subagent general-purpose async execution failed
Traceback (most recent call last):
  File "/harness/deerflow/subagents/executor.py", line 514, in _aexecute
    async for chunk in agent.astream(state, config=run_config, context=context, stream_mode="values"):  # type: ignore[arg-type]
  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/main.py", line 3324, in astream
    async with AsyncPregelLoop(
  File "/usr/local/lib/python3.12/site-packages/langgraph/pregel/_loop.py", line 1888, in __aenter__
    saved = await self.checkpointer.aget_tuple(self.checkpoint_config)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.12/site-packages/langgraph/checkpoint/sqlite/__init__.py", line 592, in aget_tuple
    raise NotImplementedError(_AIO_ERROR_MSG)
NotImplementedError: The SqliteSaver does not support async methods. Consider using AsyncSqliteSaver instead.
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
Note: AsyncSqliteSaver requires the aiosqlite package to use.
Install with:
`pip install aiosqlite`
See https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaverfor more information.
[trace=e559546b] Task call_b40b2e6c6fa548cb9e45baaf failed: The SqliteSaver does not support async methods. Consider using AsyncSqliteSaver instead.
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
Note: AsyncSqliteSaver requires the aiosqlite package to use.
Install with:
`pip install aiosqlite`
See https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.sqlite.aio.AsyncSqliteSaverfor more information.
[DeerFlowClient] Received tool result: name=task, tool_call_id=call_b40b2e6c6fa548cb9e45baaf

架构对比分析

Gateway 官方实现

# app/gateway/deps.py
app.state.checkpointer = await stack.enter_async_context(
    make_checkpointer(config)  # → AsyncSqliteSaver 或 InMemorySaver
)

• Checkpointer: AsyncSqliteSaver / InMemorySaver (异步)
• 执行方式: 全异步 (agent.astream())
• 结果: ✓ 类型匹配，正常工作

CLI 自定义实现

# cli/engine.py
cm = SqliteSaver.from_conn_string(str(db_path))  # 同步
client = DeerFlowClient(checkpointer=checkpointer)

• Lead agent: SqliteSaver (同步) + agent.stream() (同步) → ✓ 正常
• Subagent: SqliteSaver (同步) + agent.astream() (异步) → ✗ 不匹配

执行流程对比

Gateway (官方):
FastAPI (异步)
    → AsyncSqliteSaver
        → Lead: astream() ✓
            → Subagent: astream() ✓

CLI (自定义):
同步主线程
    → SqliteSaver
        → Lead: stream() ✓
            → Subagent: astream() ✗ NotImplementedError

问题 1：Subagent 与 Checkpointer 类型匹配

当前实现

subagents/executor.py 的 _create_agent() 方法没有传入 checkpointer：

def _create_agent(self, tools: list[BaseTool] | None = None):
    # ... 模型创建和 middleware 构建 ...

    return create_agent(
        model=model,
        tools=tools if tools is not None else self.tools,
        middleware=middlewares,
        system_prompt=None,
        state_schema=ThreadState,
        # checkpointer 未传入 ← 问题所在
    )

建议修复方案

方案 A: Subagent 使用 InMemorySaver

def _create_agent(self, tools: list[BaseTool] | None = None):
    # ... 现有代码 ...

    from langgraph.checkpoint.memory import InMemorySaver
    checkpointer = InMemorySaver()  # Subagent 不需要持久化

    return create_agent(
        model=model,
        tools=tools if tools is not None else self.tools,
        middleware=middlewares,
        system_prompt=None,
        state_schema=ThreadState,
        checkpointer=checkpointer,  # ← 新增
    )

理由:

• Subagent 是短期委托任务，结果返回给 Lead agent 即可
• 不需要跨会话持久化
• InMemorySaver 同时支持同步和异步方法

方案 B: CLI 改为异步架构

使用 AsyncSqliteSaver，但这需要重构整个 CLI 为异步架构，改动较大。

---

问题 2：Subagent Tracing 完整性

当前 Tracing 传递机制

检查 tools/builtins/task_tool.py 发现，Subagent 只继承了部分 tracing 信息：

# 第266-270行
metadata = runtime.config.get("metadata", {})
parent_model = metadata.get("model_name")
trace_id = metadata.get("trace_id") or str(uuid.uuid4())[:8]

# 传递给 SubagentExecutor:
executor = SubagentExecutor(
    ...
    trace_id=trace_id,  # ✓ 用于日志记录
    # ✗ tracing callbacks 未传递
    # ✗ Langfuse metadata 未传递
)

Tracing 功能对比

功能	Lead Agent	Subagent
Langfuse callbacks	✓ (通过 build_tracing_callbacks())	✗
Langfuse metadata	✓ (通过 inject_langfuse_metadata())	✗
Token usage 收集	✓	✓ (通过 SubagentTokenCollector)
Trace ID (日志)	✓	✓

关键发现

Subagent 的 Langfuse tracing 在 Gateway 和 CLI 下都不完整。

• trace_id 仅用于日志记录，不上报 Langfuse
• 没有 Langfuse callbacks，无法在 Langfuse 中看到 Subagent 的 LLM 调用
• 没有 Langfuse metadata (session_id, user_id 等)

讨论问题

1. Subagent 是否需要完整的 Langfuse tracing？

   • 当前只有 Token usage 被收集
   • 缺少 LLM 调用的详细追踪

2. 如果需要完整 tracing，应该如何实现？

   • 在 task_tool.py 中从 parent 继承 callbacks？
   • 在 executor.py 中重新构建 tracing callbacks？
   • 如何处理分布式追踪的父子关系？

3. Subagent 的 checkpointer 设计建议？

   • 官方推荐使用哪种 checkpointer？
   • InMemorySaver 是否足够？

4. 这是有意设计还是未完成功能？

   • tracing 不完整是否有技术原因？
   • 未来是否有计划完善？

环境信息

• DeerFlow 版本: feature/add-production-cli branch (commit: 4855692)
• Python 版本: 3.12
• 操作系统: Linux
• 运行方式: CLI 模式 (自定义实现)

相关代码

• backend/packages/harness/deerflow/subagents/executor.py - Subagent 执行逻辑
• backend/packages/harness/deerflow/tools/builtins/task_tool.py - Task 工具调用入口
• backend/packages/harness/deerflow/tracing/factory.py - Tracing callbacks 构建
• cli/engine.py - CLI 会话管理和 checkpointer

---

期待: 希望了解官方对 Subagent tracing 和 checkpointer 的设计思路，以及是否有推荐的实现方式。