从 Hermes Kanban 到确定性 DAG：多 Agent 协同和 AI 稳定性的一次系统追问

回路 / The Loop — 提出有价值的问题，比直接获取答案更重要。

这个专题收集和整理我与 Gemini、ChatGPT、Claude 等 AI 的对话记录。每一篇对应一次完整的提问与回答过程。通过留存这些层层追问，还原日常思考的真实轨迹。

概要

起点是一个具体的工程实践：用 Hermes Agent 搭多智能体系统，试过飞书群聊、delegate_task、curl 互调，最后落到”一张 SQLite 表 + Kanban”这个极简架构上，并沉淀出五层架构（身份 / 能力 / IO / 质量 / 协作）。

我想搞清楚两件事：

1. Kanban 在这里到底是”项目管理隐喻”，还是一套具体的技术组件？它是社区自己造的轮子，还是 Hermes 自带的？
2. 顺着这个思路往前推，AI Agent 想在企业级场景里稳定跑复杂任务，主流的技术栈到底长什么样？

追问的路径大致是这样：先把 Hermes Kanban 和最近火起来的开源项目 Multica 的关系搞清楚 → 再回到”Agent 稳定性”这个根问题 → 从 Anthropic 的动态工作流谈到 LangGraph / LlamaIndex Workflows / Mastra 三家主流思路 → 认真辨析 DAG（有向无环图）和状态机的差别 → 落到”Plan-and-Execute + 原生 asyncio 手写 DAG”这套朴素但可控的实现范式 → 最后补上”什么时候才该触发它”这个被大部分教程忽略的问题：意图路由 / 看门狗（Gatekeeper）模式。

收敛下来，企业级 Agent 编排在 2026 年这个时间点上的稳态解法可以浓缩成一句话：

静态骨架 + 运行时动态计划 + 前置意图路由 + 每节点原子化落盘。

四件事分别对应”图结构不飘”、“业务演进不僵”、“简单请求不烧钱”、“任务中断不重跑”。

Q：Hermes Kanban 到底是什么？是那种带便签的项目管理看板吗？

不是。这里的 “Kanban” 借用了看板的隐喻，但底层不是 Trello / Jira，而是一套以数据库表为媒介、面向 Agent 的异步任务分发与通信架构。

传统看板的心智模型很好用：任务像一张卡片，从 Todo 挪到 In Progress，再挪到 Done。Hermes Agent 把这个心智模型直接搬到了多 Agent 协同上——多个具名 Agent 之间不聊天、不互调 API，只对一张共享的 SQLite 表读写，通过”任务状态扭转”来完成协作。

社区文章描述的 Hermes Kanban 底层大致是三层结构：

• 控制面：hermes kanban CLI、Web Dashboard、聊天网关（Telegram / Discord 等）
• 状态面：一个 SQLite 数据库文件（社区文档提到默认路径 ~/.hermes/kanban.db）作为唯一真源，一个 Dispatcher 定时（约 60 秒）扫描待处理任务
• 执行面：隔离的 Worker 进程；每个 Worker 是一个独立的 Agent Profile，认领任务、执行、写回结果

任务的生命周期是一条清晰的状态流水线：

Triage → Todo → Ready → In Progress → (Blocked | Done) → Archived

Agent 侧不需要写 shell 脚本自己 poll 数据库，而是通过一组内置的 kanban_* 工具跟看板交互：

• kanban_show / kanban_list：查看自己的任务
• kanban_heartbeat：跑长任务时报活，防止被判定为 crash
• kanban_block：遇到歧义时挂起，等人类介入
• kanban_complete：原子化写回结果和摘要

这套设计解决的是”多 Agent 直接聊天/互调”的经典痛点：状态不持久化、耦合过重、任何一方掉线都会连锁失败。

Q：那这个 Kanban 是社区自己造的中间件，还是 Hermes 自带的？

是 Hermes Agent 自带的官方功能。社区实践里的原话是：

然后我就发现了 Hermes 自带的一个多 Agent 协作系统——kanban。

社区做的是**“怎么用好它”：先在 Hermes 上试了飞书群聊轮询、delegate_task（进程内派生子 Agent）、curl 互调这些通道，都因为平台限速、状态易失、耦合过重而不稳；最后转向官方的 Kanban 组件，在 SQLite 表上包了一层自己的五层架构**：

1. 身份层（SOUL）：定义 Agent 人设与三条铁律
2. 能力层（Skills）：把验证过的模块沉淀下来复用
3. IO 层（API Server）：本地端口服务（例如 systemd + 8645 端口）
4. 质量层（Checker）：一个独立的质检 Agent，支持多种检查模式
5. 协作层（Kanban）：官方 Kanban 组件，作为通信底座

关键洞察是把 Maker 和 Checker 解耦：主 Agent 负责干活，质检 Agent 负责挑刺。两个角色靠 Kanban 上的任务卡片交接，Checker 不通过就打回 Blocked。这跟后面 Anthropic 的 “Adversarial Verify-Fix Loop” 思路是同一件事，只是尺寸不同。

Q：我最近听说有个类似的开源项目，好像叫 Monica 还是 Motika，也做类似的事？

大概率你说的是 Multica（GitHub: multica-ai/multica）。发音和 “Motika” 接近，2026 年在国内开发者圈里火了一波。

它的官方 tagline 很好懂：

The open-source managed agents platform. Turn coding agents into real teammates — assign tasks, track progress, compound skills.

一句话概括：给本地那些命令行 AI Agent（Claude Code、Codex、Hermes、OpenClaw、Cursor Agent 等）配一个开源版的 Linear / Jira，让它们从”每次都要在终端里人肉唤醒的工具”，变成”看板上有工位、能被派单的团队同事”。

技术栈上跟 Hermes Kanban 差别不小：

Multica 干的几件核心事：

• 本地 daemon 自动探知：扫描 PATH，把系统里安装的各种 Agent CLI（Claude Code、Codex、Hermes 等）自动激活成看板上的 Assignee
• 异步生命周期托管：任务的 enqueue → claim → start → complete/fail，全程通过 WebSocket 把进度推到看板
• 技能库（Skills）：某个 Agent 解决完复杂问题后，可以把方案打包成”可复用技能”，让下一个 Agent（不同 CLI 也行）直接调用
• 多 workspace 隔离：不同项目各有独立看板、权限、Agent 池

它的核心哲学跟 Vibe Kanban（偏个人开发者）、Paperclip（偏”无人公司”）都不一样：Multica 假设人类和 Agent 长期在同一个协作空间，人类审批、Agent 执行、看板举牌报告 Blocker，是一个显式的三方系统。

Q：那 Multica 和我本地装的 Claude Code / Hermes / OpenClaw 是什么关系？

一句话：Multica 是项目经理，Claude Code / Hermes / OpenClaw 是干活的工程师。

在没有 Multica 之前，它们是彼此独立的命令行工具：Claude Code 擅长深入代码库改动、Hermes 擅长长路径推理和 MCP 工具调用、OpenClaw 提供本地 Agent 编排和 RAG 管道。你使用它们的方式是”人肉在终端里 copy-paste 上下文”。

Multica 加进来之后：

1. 接单入口统一：所有任务都先落在 Multica 的看板上，你在看板里指派 Assignee
2. 执行时按 CLI 分发：Multica 的本地 daemon 会拉起对应的 Agent 进程（例如任务打给 @claude-code，daemon 就在后台启动 Claude Code 处理这个任务的上下文）
3. 进度双向流转：Agent 的输出通过 WebSocket 变成任务详情里的时间线；遇到歧义时它可以在看板上挂 Blocked 并 @你
4. 技能横向共享：Claude Code 沉淀下来的”部署到测试环境”技能，下一次 Hermes 接到类似任务也能直接调用

Multica 不替代任何一个 Agent CLI，它是把它们粘在一起的胶水层。这跟 Hermes Kanban 的关系正好互补——Hermes Kanban 管的是同一个 Hermes 实例里多个 Profile 之间的协作，Multica 管的是异构 Agent CLI 之间的协作。

Q：好了，回到根问题。AI Agent 的稳定性和确定性到底怎么解决？Anthropic 好像有一套 Dynamic Workflow？

这个问题在 The Loop 009 里已经系统聊过一次了，这里挑几个跟这一期主题相关的要点复述一下。

传统单体 Agent 面对复杂长任务时，非确定性问题主要有三个：

1. 上下文漂移：单兵作战、线性反馈，跑到第 50 步早忘了第 1 步的初衷
2. 无限循环 / 死磕：Agent 认死理，反复尝试同一个失败的动作
3. Token 恶性膨胀：每一步都把完整历史塞进上下文，成本指数级上涨

Anthropic 在 Building Effective Agents 和后续的 Dynamic Workflows in Claude Code 里给出的解法可以浓缩成三点：

• Orchestrator-Workers（编排器-工人）架构：主 Agent（Orchestrator）不亲自干活，而是把大目标拆成一张依赖图（DAG），然后 fan-out 出一堆专职 subagent 并发执行，最后 fan-in 汇总
• Code-As-The-Plan（代码即计划）：让 LLM 显式把”思考计划”写成一段可版本化、可静态分析的 JavaScript / Python 脚本，而不是在运行时靠 Prompt 瞎猜下一步。平台提供 agent、parallel、pipeline 这类确定性控制原语
• Adversarial Verify-Fix Loop（对抗性验证修复环）：架构上强制分离 Maker（执行者）和 Checker（质检者），Checker 用客观测试（单测 / 静态分析 / 业务 Lint）把 Maker 挡在门外

再加一条前面 Hermes Kanban 那段其实已经出现过、但值得单独强调的：

• State Checkpointing（状态检查点）：把 Agent 的执行轨迹和中间状态每一步都落盘（Hermes 用 SQLite，LangGraph 用 Postgres）。任何一步 crash，都能从上一个安全点续跑，不需要从头烧 Token 重推

Q：除了 Anthropic 的这套，其他 Agent 框架有类似思路吗？我想看有企业实践、社区口碑的。

有三家值得单拎出来讲，它们在”控噪”的哲学上高度一致，但工程侧重完全不同。

1. LangGraph — 状态机 + 有向图

LangGraph 是 LangChain 团队给自己早期”过度自由 Agent”路线打的一记补丁。核心抽象是把多 Agent 协同建模为有向图（Directed Graph）：

• 每个 Agent / 工具是一个节点（Node），流转是边（Edge）
• 整张图共享一个严格定义的 State 对象，每次节点执行都要原子化地更新这个 State
• 支持节点级检查点（Checkpointing）：State 可以随时序列化落盘到 Postgres / SQLite / Redis，用于故障恢复或人类介入（interrupt_before 就是干这个的）

它在金融、客服、企业内部知识助手这些”重业务流”的场景里落地口碑很好。

2. LlamaIndex Workflows — 事件驱动

LlamaIndex Workflows 走的是另一条路：事件驱动架构（Event-Driven Architecture）。

每个步骤（Step）是一个独立函数，通过接收特定 Event 触发、完成后广播新 Event。没有中心化的”超级大脑”，天然擅长扇出/扇入（fan-out / fan-in）——步骤 A 抛出 DataExtractedEvent，步骤 B 和 C 自动并发监听并执行，聚合节点等所有并行事件到齐才被唤醒。

它特别适合大规模文档并行审计 / RAG pipeline / 微服务批量处理这类数据流为主的场景。

3. Mastra — 硬性把 Agent 和 Workflow 拆开

Mastra（TypeScript 生态里的口碑框架）在 API 层面就把两件事彻底分开：

• Agent = 配置了 system prompt + 工具箱 + 向量库的”纯推理单元”
• Workflow = 纯粹的、确定性的有向图流水线，用代码写死分支和循环

开发者在 Workflow 里硬编码控制流（if/else、while），只在特定节点调用 Agent.execute() 让 LLM 做局部决策。原生内置 Evals 和 Trace，工作流层和 Agent 层的评估互不干扰。

三家的技术共识

不同框架，一致的底层判断：

Q：LangGraph 的图和 Anthropic 的动态图是一回事吗？让 Claude 直接生成 LangGraph 代码可行吗？

目标一致，实现机制完全不同。核心差别在：图的拓扑结构是在什么时候确定的。

• LangGraph：图在编译期建好，节点和边由代码显式声明。“动态”体现在条件边（Conditional Edges）——LLM 输出一个结果，LangGraph 根据结果决定走 A 分支还是 B 分支。图骨架静态且刚性。
• Anthropic Dynamic Workflows：图在运行期建。Orchestrator 拿到任务，动态决定要 fan-out 几个 Worker、Worker 之间的连线怎么画。图的节点数量和拓扑，完全由 LLM 在执行过程中根据中间结果追加或裁剪。

至于”让 Claude 生成 LangGraph 代码然后 exec 跑起来”——能跑通，但在企业级生产环境里是灾难：

1. 编译开销重：LangGraph 有强类型状态检查、.compile() 预编译、跟 Checkpointer 的序列化绑定。运行时反复动态编译一个复杂图，性能损耗极大
2. 容错率低：代码生成有偶发性。Claude 少写一个节点的 Edge、或者 State 的 key 拼错，整个系统就抛编译异常或死锁
3. 调试和 Evals 是噩梦：动态生成的代码在运行时是黑盒，第 50 步崩溃时没法在 IDE 打断点，也没法做标准化的 Evals

现代企业工程的折中方案有两条：

• 方案 A（刚性骨架 + 动态参数）：LangGraph 里写死一个”万能循环图” [Orchestrator] → [Conditional Routing] → [Worker Node] → [Validator] → [Loop back]，Claude 只生成 JSON 格式的执行计划（子任务列表 + 依赖关系），LangGraph 的节点读 JSON 把参数注入固定 Worker 里
• 方案 B（Code-As-Plan + 轻量 engine）：让 Claude 直接生成一段不依赖任何 Agent 框架的标准 Python asyncio 脚本，或声明式的 DAG JSON，本地用几百行代码的自定义 engine 去执行。因为没有重型框架包袱，模型生成标准 Python 的准确率远高于生成特定框架 DSL

Q：DAG 是什么？跟状态机有什么区别？

DAG = Directed Acyclic Graph（有向无环图）。拆成三个词：

• Graph（图）：由节点（Nodes）和边（Edges）组成的网络
• Directed（有向）：边有方向，A → B 不代表 B → A
• Acyclic（无环）：顺着箭头永远回不到出发点。系统里不存在死循环

一个日常例子：炒青椒肉丝。

1. A：买菜
2. B：洗青椒
3. C：切肉丝
4. D：下锅炒

A 必须在 B/C 之前完成；B 和 C 互不依赖，可以并行做；D 依赖 B 和 C 都做完；你不可能炒完菜（D）又穿越回洗青椒（B）——这就是无环。

DAG vs 状态机的核心区别：

对应到 AI 场景选型：

• DAG 范式：任务是确定性的、复杂的流处理。比如”Query 理解 → 并行搜索谷歌 + 本地向量库 → 合并生成回答”。追求极致并发效率
• 状态机范式：任务包含高频交互和自我修正。比如 Code Agent 写代码 → 跑测试失败 → 修代码 → 再跑测试 → 直到通过。这种”原地打转直到成功”的场景必须靠状态机

Q：我观察到大部分用户 query 其实是”步骤多、但没有循环”的 DAG 场景，只是要稳定地生成结果。怎么做落地？

这个判断很准。在 B 端办公协同这类场景里，绝大多数高价值任务本质上是结构化生成流水线（Structured Generation Pipeline）：群聊摘要 → 待办提取 → 日程冲突校验 → 卡片通知。步骤明确、无需状态回退。

当流程退化为确定性 DAG，最大的风险不再是”迷路”，而是**“数据塌缩”**：上游节点生成的脏数据把下游节点连锁带崩。稳态解法的三根支柱：

1. 强类型 I/O 契约（Strict I/O Contracts）

节点之间不传纯文本，只传被 Schema 严格校验过的结构化对象。用 LLM 的 Structured Outputs / JSON Mode，或者外挂 Pydantic 做校验。节点 A 输出解析失败，立刻在 A 原地 retry，绝不让脏数据流入下游。

2. 节点级断路器 + 兜底降级

长链条最怕”全凭运气”。6 个步骤，每步成功率 90%，整条跑通只剩 0.9^6 ≈ 53%。所以：

• 原地 retry：某一步失败时，用本地缓存的上游数据从这一步续跑，不要重头执行
• Fallback：非核心节点（比如”提取语气幽默的总结”）连续失败时，自动降级为硬编码规则（直接截取原句），保证主流程走完

3. 动态扇出（Dynamic Fan-Out / Fan-In）

单次 Prompt 处理信息量过大时，让 DAG 具备”动态胖瘦”能力：节点 A 先做任务拆解或文本切片，运行时扇出 5 个相同节点 B1~B5 并行处理，最后聚合节点 C 扇入总结。“总-分-总”结构能把上下文窗口压力降到最低，每一路的生成质量因此更高。

Q：有哪些经过企业验证的、可以”先生成 DAG 再执行”的框架？

这个思路在工程界叫 Plan-Then-Execute（预计划-执行） 范式。生产环境里被大量验证过的路径主要有三条：

1. LangGraph Plan-and-Execute 官方生产模板

LangGraph 的 Plan-and-Execute 教程 是这一模式在开源社区里的经典参考。整张图共享一个全局 State，通常包含 input（用户输入）、plan（当前步骤列表）、past_steps（已完成任务和结果）、response（最终产出）。

标准流向由几个节点构成：

• Planner：LLM 拿到用户 input，输出一个结构化的步骤列表
• Executor：接收 plan 的当前第一步，调具体工具去执行，把结果写回 past_steps
• Re-Planner：审查 past_steps，输出两种决策之一——要么修正剩余 plan 继续（把控制权抛回 Executor），要么认为任务完成、写入 response、走向 END

它的美感在于结构静态、内部计划动态：图的连线（Planner → Executor → Re-Planner）是固定的，跑什么任务在运行时演进。既保留了 LangGraph 的持久化和稳定性，又允许 Re-Planner 根据业务情况动态增删步骤。

2. Microsoft Semantic Kernel 的 Planner 体系

Semantic Kernel 是微软推进 Copilot 落地 B 端的底层依赖。做法是把本地能力（企业微信 API、数据库查询、文档解析）封装成 SK 的 Plugin；Planner 在 runtime 看所有可用 Plugin 的 JSON Schema 元数据，自动把它们编排成一个 Plan 对象（本质上是 XML/JSON 描述的 DAG）。

强调可控性：生成的 Plan 在执行前可以序列化打印出来给用户确认，也可以用代码做静态安全合规检查。

3. 终极稳态：LLM Planner + 工业级工作流引擎

如果业务场景对稳定性有 99.99% 的刚性要求（不能因为网络抖动或 API 超时就死掉），最成熟的实践是不用任何 Agent 框架，而是：

1. 让 LLM 根据用户意图生成一段符合特定 DSL 或标准的 JSON DAG 描述文件
2. 把 JSON 提交给 Temporal.io（Uber、Stripe 等大厂的核心计费/协同系统在用）、Prefect 或 Airflow
3. 由这些工业级引擎负责分布式调度、状态持久化、指数退避重试、高并发扇出

LLM 只负责出图纸，搬砖和抗灾全交给最皮实的工业级引擎。

通用工程铁律

不管用哪条路，有三件事必须焊死：

• JSON Schema 强制约束 Planner 输出：绝不允许 LLM 生成自由文本。经典的 DAG Schema：

  {
    "tasks": [
      { "id": "task_1", "type": "fetch_wecom_chat", "params": { "limit": 100 } },
      { "id": "task_2", "type": "extract_todos", "depends_on": ["task_1"] },
      { "id": "task_3", "type": "verify_calendar", "depends_on": ["task_1"] },
      { "id": "task_4", "type": "generate_report", "depends_on": ["task_2", "task_3"] }
    ]
  }

• 执行前静态编译校验：用传统图算法（例如 Kahn 拓扑排序）跑一遍，100% 确认无环；用代码验证 task_4 需要的参数类型能被上游节点提供

• 每节点原子化落盘：每跑完一个 task_id，输出数据和状态原子化写入 SQLite / Redis。修复后可以直接从失败节点续跑，上游节点永远不重新烧 Token

Q：如果不想上重型框架，手写一个原生 asyncio 的确定性 DAG 引擎大概长什么样？

不复杂。核心组件三个：Pydantic 强类型契约 + SQLite 持久化 + asyncio.gather 天然并行。

1. 契约与状态定义（schemas.py）

from pydantic import BaseModel, Field
from typing import List, Optional
from enum import Enum

class TaskStatus(str, Enum):
    PENDING = "PENDING"
    RUNNING = "RUNNING"
    COMPLETED = "COMPLETED"
    FAILED = "FAILED"

class ChatHistoryOutput(BaseModel):
    chat_id: str
    messages: List[str]

class TodoExtractionOutput(BaseModel):
    todos: List[str] = Field(default_factory=list)

class CalendarCheckOutput(BaseModel):
    has_conflict: bool
    conflicting_events: List[str] = Field(default_factory=list)

class FinalReportOutput(BaseModel):
    summary: str
    action_items: List[str]
    alert: Optional[str] = None

2. SQLite 持久化层（checkpoint.py）

import sqlite3
import json

class SQLiteCheckpointer:
    def __init__(self, db_path: str = "dag_kanban.db"):
        self.db_path = db_path
        self._init_db()

    def _init_db(self):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS task_runs (
                    task_id TEXT PRIMARY KEY,
                    status TEXT,
                    output_data TEXT,
                    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
                )
            """)
            conn.commit()

    def get_task(self, task_id: str):
        with sqlite3.connect(self.db_path) as conn:
            cursor = conn.cursor()
            cursor.execute(
                "SELECT status, output_data FROM task_runs WHERE task_id = ?",
                (task_id,),
            )
            row = cursor.fetchone()
            if row:
                return row[0], json.loads(row[1]) if row[1] else None
            return None, None

    def save_task(self, task_id: str, status: str, output_data: dict = None):
        with sqlite3.connect(self.db_path) as conn:
            conn.execute("""
                INSERT INTO task_runs (task_id, status, output_data, updated_at)
                VALUES (?, ?, ?, CURRENT_TIMESTAMP)
                ON CONFLICT(task_id) DO UPDATE SET
                    status = excluded.status,
                    output_data = excluded.output_data,
                    updated_at = CURRENT_TIMESTAMP
            """, (task_id, status, json.dumps(output_data) if output_data else None))
            conn.commit()

3. 执行引擎（engine.py）

import asyncio
import logging
from typing import Any, Callable, Coroutine

logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
logger = logging.getLogger("DAGEngine")

class DeterministicDAGWrapper:
    def __init__(self, checkpointer):
        self.db = checkpointer
        self._memory_cache = {}

    async def run_node(
        self,
        task_id: str,
        output_schema: Any,
        func: Callable[..., Coroutine[Any, Any, Any]],
        *args, **kwargs,
    ) -> Any:
        # 1. 断点续跑：已完成的节点直接加载历史数据
        status, cached_data = self.db.get_task(task_id)
        if status == TaskStatus.COMPLETED and cached_data is not None:
            logger.info(f"⏭️  [{task_id}] 已完成，加载历史数据。")
            validated = output_schema.model_validate(cached_data)
            self._memory_cache[task_id] = validated
            return validated

        # 2. 原地重试（指数退避）
        retries = 3
        self.db.save_task(task_id, TaskStatus.RUNNING)
        for attempt in range(1, retries + 1):
            try:
                logger.info(f"🚀 [{task_id}] 执行中（{attempt}/{retries}）")
                raw = await func(*args, **kwargs)
                # 3. 强类型校验：Pydantic 拒绝脏数据
                validated = output_schema.model_validate(raw)
                # 4. 落盘
                self.db.save_task(task_id, TaskStatus.COMPLETED, validated.model_dump())
                self._memory_cache[task_id] = validated
                return validated
            except Exception as e:
                logger.error(f"❌ [{task_id}] 第 {attempt} 次失败: {e}")
                if attempt == retries:
                    self.db.save_task(task_id, TaskStatus.FAILED)
                    raise
                await asyncio.sleep(1 * attempt)

4. 拓扑组装（main.py）

无依赖节点通过 asyncio.gather 自动并行：

async def main():
    engine = DeterministicDAGWrapper(SQLiteCheckpointer())

    # 根节点
    chat = await engine.run_node("fetch_chat_01", ChatHistoryOutput, mock_fetch_chat, chat_id="wecom_001")

    # 扇出：待办提取 || 日历校验（并行）
    todo, calendar = await asyncio.gather(
        engine.run_node("extract_todo_01", TodoExtractionOutput, mock_extract_todos, chat_history=chat),
        engine.run_node("check_calendar_01", CalendarCheckOutput, mock_check_calendar, chat_history=chat),
    )

    # 扇入：聚合报告
    report = await engine.run_node(
        "generate_report_01", FinalReportOutput, mock_generate_report,
        todos=todo, calendar=calendar,
    )
    print(report.model_dump_json(indent=2))

if __name__ == "__main__":
    asyncio.run(main())

为什么这套方案在企业场景下确定性够用：

• 类型安全逃生舱：Pydantic 校验挡在每个节点入口，脏数据不流入下游
• 零内存状态漂移：DAG 是无状态的（Stateless），所有中间结果落 SQLite，进程重启也能续跑
• 轻量高透明：没有 Agent 框架的黑盒路由层，拓扑关系完全由 await 和 asyncio.gather 锁死，节点级 tracing 和 evals 直接就能做

Q：LangGraph 的 Plan-and-Execute 跟我们手写的静态 DAG 有什么本质区别？

前面已经提过，这里做个明确对比：

如果你的场景是”用户需求非常动态、经常需要根据中间结果动态增减步骤”，Plan-and-Execute 更合适。反之，如果流程高度确定，手写 asyncio DAG 就够了——更轻、更透明、更好评估。

企业级落地 Plan-and-Execute 有三个官方文档不会写的踩坑点：

1. Re-Planner 的 Token 膨胀陷阱：Re-Planner 每次都要读 past_steps，历史数据积多了 Token 消耗指数级上涨。避坑：Executor 返回的结果必须强行摘要成核心 JSON meta，绝不能把万字原始日志塞进 State
2. 死循环断路器：Re-Planner 遇到无法解决的外部错误时容易陷入执念，反复”重新尝试”疯狂烧钱。避坑：在 State 里埋一个硬编码计数器，超过 5 轮循环强行中断转 Blocked
3. 人类介入：办公场景里 Planner 可能生成敏感操作（发消息给老板、删日程）。避坑：用 LangGraph 的 interrupt_before 特性，图在敏感节点前自动挂起，等用户确认后再续跑

Q：这套流程什么时候应该触发？总不能用户随便说句话就拉起 Planner 吧？

对，这是被大部分教程忽略的关键问题。答案是：必须前置一个意图路由 / 看门狗（Gatekeeper）层。

为什么必须有路由层

不管用户输入什么都直接进 Planner，代价是三个：

1. 延迟崩溃：Planner 生成一个包含多节点的计划 JSON 要几秒甚至更长。用户说”谢谢”或问一个单转问答时，这种卡顿完全不可接受
2. 确定性劣化：Planner 面对简单/模糊 query 容易”想太多”，画出莫名其妙的步骤
3. 成本失控：高规格模型的 Token 昂贵，必须用在刀刃上

主流路由方案（分层级联）

第一层：语义向量路由器（毫秒级、零 Token）

代表作是开源的 semantic-router（Aurelio Labs）。原理不动用大模型推理：

• 预先准备一组代表各类意图的”黄金样本句”（"帮我把这些消息整理成待办"、"查一下明天下午的会议冲突"）
• 用户 query 实时 embedding，跟样本库做 Top-K 向量相似度（cosine similarity）
• 分值超阈值（例如 > 0.85）则精准触发对应 DAG；否则降级为常规单转回答

延迟通常在 10ms - 50ms，本地 CPU/GPU 完成，零 Token。

第二层：结构化小模型路由（针对复杂长句）

当 query 包含复合意图、口语化严重、夹杂大量实体时，向量匹配会失效。这时动用一个专门做 tool calling 优化过的本地小模型（7B / 14B 级别），通过 JSON Mode 锁定输出格式，只做分类选择。

级联结构：

User Query
  → [Level 1: 语义向量路由]
     ├─ 分值极高 → 直达目标 DAG
     └─ 分值模糊 → [Level 2: 小模型强类型判决]
                    ├─ 触发 Planner
                    └─ 降级为 Chat / RAG

路由层的输出契约

Router 的输出必须是严格的结构化对象，例子：

{
  "is_complex_task": true,
  "intent_target": "dynamic_schedule_planner",
  "extracted_entities": {
    "project": "AI 评审",
    "time_boundary": "下周一前"
  },
  "confidence": 0.92
}

关键优化：Router 在做意图判断的同时，顺手把实体也抽出来了。Planner 拿到这些现成的 Key-Value 后，不需要重新通读并提取用户输入，直接根据实体去生成 DAG 节点参数，冷启动速度成倍提升。

收尾：一张图看懂 2026 年企业级 Agent 编排的稳态解法

回到开头那句浓缩：

静态骨架 + 运行时动态计划 + 前置意图路由 + 每节点原子化落盘。

翻译成一个完整的请求流：

User Query
  ↓
[意图路由 / Gatekeeper]   ← 挡住简单请求，避免烧 Planner
  ↓ (确认是复杂任务)
[Planner / LLM]           ← 生成 JSON DAG 或结构化 Plan
  ↓
[静态编译校验]             ← Kahn 拓扑排序 + 类型契约
  ↓
[Executor + Checkpointer] ← asyncio.gather 并行 + SQLite 落盘
  ↓
[Re-Planner (可选)]        ← 根据 past_steps 决定 END 或增补步骤
  ↓
Final Response

Hermes Kanban 和 Multica 解决的是这套流水线上层的”多 Agent 协作”问题——如何让多个 Agent 在异步、持久化的看板上像团队一样接单交付；LangGraph / LlamaIndex Workflows / Mastra 和手写 asyncio DAG 解决的是中层的”单任务确定性”问题——如何让一次编排稳定跑完；语义路由和小模型 gatekeeper 解决的是入口的”什么时候触发”问题——如何避免每个请求都碾过重型 Planner。

三层各司其职，加起来才是当下企业级 Agent 稳定性问题的完整解法。

参考资料

产品与框架（本期 Q&A 中出现的开源/商用项目）

• Hermes Kanban 官方介绍 — Hermes Agent 内置的多 Agent 协作组件，基于 SQLite 的持久化任务看板
• Multica GitHub — 开源 managed-agents 平台，把本地 Agent CLI（Claude Code、Codex、Hermes 等）统一到看板上协作
• LangGraph — LangChain 推出的状态机 + 有向图编排框架，支持节点级 checkpointing
• LangGraph Plan-and-Execute 官方博客 — Plan-and-Execute 模式的开源社区经典参考
• LlamaIndex Workflows — 事件驱动的 Agent 编排框架，天然支持 fan-out / fan-in
• Mastra — TypeScript 生态的 Agent 框架，强制分离 Agent 和 Workflow
• Microsoft Semantic Kernel — 微软 Copilot 底层依赖之一，Planner + Plugin 体系
• Temporal.io — 分布式工作流引擎，Uber / Stripe 等大厂核心系统在用
• Prefect — 现代化数据管道编排工具
• Airflow — Apache 老牌工作流管理平台
• semantic-router — Aurelio Labs 开源的意图路由库，基于 embedding 相似度实现毫秒级零 Token 分类

Anthropic 相关文献（Agent 稳定性思路的源头）

• Building Effective Agents — Anthropic 2024-12-19 发布的 Agent 架构入门经典，划分 workflow vs agent，列出 5 种基础模式
• A harness for every task: dynamic workflows in Claude Code — Claude Code 团队 2026-06 关于动态工作流的深度拆解

相关阅读

• The Loop 009 — 从《Building Effective Agents》到 Claude Code 动态工作流 — 系统聊过 Anthropic 的 5+6 种模式，本期的”上一集”