一次任务的完整执行

SSE 只是观察 start_run 创建的同一条 Run

主链现在停在 start_run 已经返回 RunRecord 的位置。sse_consumer 从 bridge 读取这条 record 的事件并发送给浏览器，不负责调用模型，也不拥有 Agent 的生命周期。

因此，“前端没收到流”与“后台没执行”不是同一个故障。前者看 SSE 和 StreamBridge，后者看 RunRecord 与 worker。连接断开后是否取消由 on_disconnect 策略决定，不由 TCP 连接替系统猜测。

这条旁支到此结束。控制流回到 record.task 中的 worker，继续进入 run_agent。

Middleware 包在 agent.astream 的调用边界上

主线现在停在一次工具调用。工具真正执行前后，多个 Middleware 会像函数包装器一样套在外面。读取后再写、进度上报和错误处理看似互不相干，但包装顺序会改变谁能看到原始异常、谁能看到规范化结果。

源码在构造 tail 时特意留下顺序说明：ToolProgressMiddleware 要位于 ToolErrorHandlingMiddleware 外面，才能观察后者已经写入 metadata 的结果。

tool_progress_config = app_config.tool_progress
if tool_progress_config.enabled:
    from deerflow.agents.middlewares.tool_progress_middleware import ToolProgressMiddleware

    tail.append(ToolProgressMiddleware.from_config(tool_progress_config))

tail.append(ToolErrorHandlingMiddleware(app_config=app_config))

ReadBeforeWriteMiddleware 先进入列表，ToolProgressMiddleware 随后追加，ToolErrorHandlingMiddleware 最后成为内层。调用向内经过这些 wrapper，返回时反向穿出；所以“列表先后”不只是视觉排序。

对于读者，记住一个判断方法就够了：Middleware 决定某一步怎样被执行和记录，模型仍决定是否调用这一步。插叙结束，工具结果回到 ThreadState，agent.astream 继续下一轮。

消息过长时，摘要发生在下一次模型调用前

报告任务经过多轮搜索与文件读取后，messages 会越来越长。模型上下文有长度和成本限制，不能把所有历史原封不动地带进每一次调用。Summarization Middleware 会在达到条件时压缩较早内容，同时保留继续执行所需的事实。

摘要不是 Memory。它服务当前 thread 的上下文窗口，目标是让这条任务能继续；Memory 则用于跨 thread 保留用户或 Agent 的长期经验。混淆两者，会把“这次对话太长”和“下次还要记得”当成同一个问题。

DeerFlowSummarizationMiddleware 在工厂中接收 hooks、模型名称、app_config 和 extensions。当前注册的 memory_flush_hook 会在旧消息从上下文移除前，把即将压缩的消息送进长期 Memory 队列；它不是一个通用的附件或任务保存器。

摘要完成后，新的 summary 与最近消息继续交给模型，主循环没有换成另一条 Run。插叙结束，我们仍停在同一个 agent.astream 中，只是下一次模型输入更短。

Memory 在模型调用前取回长期经验

假设用户过去明确说过“销售分析优先比较环比，不要把未确认线索写成结论”。这些偏好不一定在当前 thread 消息里，却会影响报告质量。Dynamic Context Middleware 可以在模型调用前按 agent_name 与 user_id 取回相关 Memory。

检索结果不是把整个记忆库倾倒进提示。配置决定是否注入，Memory manager 负责召回，Middleware 再把结果整理成有限长度的上下文块，与当前日期等动态信息一起进入模型。

injection_enabled = self._app_config.memory.injection_enabled if self._app_config else True
memory_context = (
    _get_memory_context(
        self._agent_name,
        app_config=self._app_config,
        user_id=resolve_runtime_user_id(runtime),
    )
    if injection_enabled
    else ""
)
current_date = _format_current_date()
date_reminder = _format_current_date_reminder(current_date)

memory_block = memory_context.strip() if memory_context else None

return date_reminder, memory_block

injection_enabled 控制召回路径；_get_memory_context 返回 memory_context，随后整理为 memory_block。若关闭注入，模型照常运行，只是不会看到长期经验。

这说明 Memory 是可选增强，不是 ThreadState 的替代品。当前报告的消息、附件和 artifacts 仍在图状态里；长期偏好只在需要时成为额外上下文。

task 工具把独立支线交给 Sub-agent

如果主 Agent 同时调查价格、竞品和区域差异，上下文会迅速变乱。它可以调用 task 工具，把“核对华东区竞品降价”作为边界清楚的支线交给 Sub-agent，自己保留整份报告的统筹责任。

这不是偷偷再开一条用户请求。task 本身是一个普通工具调用，有 description、prompt、subagent_type 和 tool_call_id；结果最终仍回到主 Agent 的工具消息。

@tool("task", parse_docstring=True)
async def task_tool(
    runtime: Runtime,
    description: str,
    prompt: str,
    subagent_type: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> str | Command:

task_tool 用 subagent_type 选择执行配置，用 prompt 传递明确任务。返回类型可以是字符串或 Command，说明委派结果既可能直接返回文本，也可能带状态更新。

Sub-agent 的价值是上下文隔离与职责分解，而不是数量越多越聪明。任务太小会增加延迟，边界不清则会重复搜索；只有能独立验收的支线才值得委派。

Agent 结束时，Memory 才安排写回

present_files 返回以后，模型确认任务已经满足，Agent 生命周期准备结束。现在才轮到 Memory 的捕获阶段：系统从本次对话中筛选值得长期保存的内容，而不是每产生一个 chunk 就同步修改记忆库。

Memory Middleware 的 after_agent 先解析 thread_id、messages、user_id 与 trace_id；如果当前状态不满足写入条件，就直接返回，不影响主任务结果。注意，此处是 Agent 已经停止选择新工具之后，但 worker 还没有完成 Run 的最终结算。

@override
def after_agent(self, state: MemoryMiddlewareState, runtime: Runtime) -> dict | None:
    """Queue conversation for memory update after agent completes."""
    add_args = self._resolve_add_args(state, runtime)
    if add_args is None:
        return None
    thread_id, messages, user_id, trace_id = add_args

    # Hand raw messages to the manager; the backend filters to user + final-AI
    # turns, validates, detects correction/reinforcement, and enqueues.
    get_memory_manager().add(
        thread_id,
        messages,
        agent_name=self._agent_name,
        user_id=user_id,
        trace_id=trace_id,
    )

    return None

after_agent 调用 get_memory_manager().add 安排更新，随后返回 None。源码中的 Queue conversation 表明长期记忆写入与主响应解耦；它不会再回到模型工具循环，也不应该让次要的 Memory 写入阻塞报告交付。

Memory 写回安排完成后，Agent 图产生最后的状态，worker 才离开 agent.astream。主线接下来进入 Run 结算。

Checkpoint 保存 agent.astream 推进到哪里

主线已经到达 Run 收尾。现在回头解释故障恢复需要的三类事实。第一类是 Checkpoint：LangGraph 在节点边界保存状态快照，使同一 thread 的图能够从已知状态继续，而不必只依赖进程内变量。

Checkpoint 关心 messages、artifacts、todos 等图状态；它不等于事件日志，也不负责把实时 chunk 推给网页。部署可以选择不同持久化实现，因此运行时通过 provider 创建 checkpointer。

make_checkpointer 是 async context manager，返回 AsyncIterator[Checkpointer]。上下文管理器负责建立和关闭后端资源，Agent 工厂只接收统一接口，不必知道它来自内存还是数据库。

有了 Checkpoint，系统知道 Agent 图保存到哪里；但用户还需要查询“这条 Run 发生过什么”，这属于下一类事实。

RunEventStore 保存可查询的运行历史

RunEventStore 按 thread_id 与 run_id 保存模型、工具、进度、错误和终态事件。页面刷新后，它可以重新列出历史；运维排查时，也能把同一 thread 中的多条 Run 分开。

持久事件有自己的顺序 seq。查询可以按事件类型、task_id 和 after_seq 过滤，适合断点续读或只查看某类运行事实。

@abc.abstractmethod
async def list_events(
    self,
    thread_id: str,
    run_id: str,
    *,
    event_types: list[str] | None = None,
    task_id: str | None = None,
    limit: int = 500,
    after_seq: int | None = None,
) -> list[dict]:

list_events 同时接收 thread_id、run_id 与 after_seq，并返回事件记录列表。这个接口说明持久历史的游标是序列号，不应与实时流后端的消息 ID 随意比较。

Checkpoint 回答“图状态是什么”，RunEventStore 回答“执行过程发生了什么”。两者都能帮助恢复，却保存不同语义；最后还剩实时交付这第三类事实。

StreamBridge 只负责把正在发生的事件送出去

agent.astream 产生 chunk 后，worker 通过 StreamBridge 发布；SSE consumer 再从 Bridge 消费。Bridge 追求低延迟，不承担全部长期历史。单进程开发可用内存队列，多实例部署则需要共享后端。

选择内存还是 Redis 不改变 worker 的 publish 接口，却会改变跨进程可见性。worker 与 Gateway 若不在同一进程，MemoryStreamBridge 无法让另一个进程读到事件。

if config is None or config.type == "memory":
    from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge

    maxsize = config.queue_maxsize if config is not None else 256
    bridge = MemoryStreamBridge(queue_maxsize=maxsize)
    logger.info("Stream bridge initialised: memory (queue_maxsize=%d)", maxsize)
    try:
        yield bridge
    finally:
        await bridge.close()
    return

if config.type == "redis":
    from deerflow.runtime.stream_bridge.redis import RedisStreamBridge

    redis_url = _resolve_redis_url(config)
    bridge = RedisStreamBridge(
        redis_url=redis_url,
        queue_maxsize=config.queue_maxsize,
        max_connections=config.max_connections,
        stream_ttl_seconds=config.stream_ttl_seconds,
    )

MemoryStreamBridge 在上下文结束时 close；RedisStreamBridge 则接收 redis_url、queue_maxsize、连接数与 TTL。两种实现共享抽象，但可靠性边界不同。

现在三类恢复事实齐了：Checkpoint 保存图状态，RunEventStore 保存可查询历史，StreamBridge 传递正在发生的事件；再加上 RunRecord 的管理终态，页面刷新后才能既恢复结果，又不把旧任务误判为仍在执行。