我们是怎么给 AI 学习助手造一条「流水线」的

标签：agentzoo Rust 工作流引擎 AI 智能体 审批门 声明式 LLM 工程

上一篇《你的 AI 刷题导师》把功能列了一遍，读完你大概知道它能做什么，但不知道它为什么长这样、怎么跑起来的。这篇换个写法：不介绍产品，只拆引擎。我会带着你读 agentzoo_assistant 里真正在跑的那几百行 Rust，看一个「阶段 + 审批门 + 工件」的声明式工作流引擎是怎么把一堆 LLM 调用串成一条可控的流水线的。

一、问题：把 LLM 串起来，为什么不能只是一条 prompt

最直觉的做法是把「整理资料 → 梳理知识点 → 出题 → 分析 → 出报告」整段塞进一个 prompt，让模型一口气跑完。我们试过，问题很具体：

1. 中间产物不可见、不可改。 模型把知识图谱、试卷、分析全揉在一个回答里。学习者想改一道题，只能把整段话再发回去重跑，前面已经满意的部分也跟着被重写。
2. 没有「刹车」。 出题这一步如果偏了（比如难度全压在最难档），模型不会停下来等你确认，而是顺着错误一路生成到报告。等看到报告才发现卷子不行，整条链路白跑。
3. 状态无处安放。 一次练习要跨好几轮对话、可能跨几天。把状态藏在对话上下文里，上下文一长就丢，模型开始「忘记」前面出过什么题。
4. 不可重放。 想单独重跑「分析」这一步（比如补了几道错题后重新分析），没有入口。

这四个问题指向同一个结论：别把流程交给一次 LLM 调用，要把它拆成显式的、有状态、可暂停、可重放的步骤。 这就是 agentzoo 工作流引擎要解决的事。下面看它具体怎么做。

二、声明式：流程是一张数据，不是一段代码

整个学习流程不是用 Rust 控制流写出来的，而是声明出来的。应用实现 WorkflowApp trait，把流程描述成一组纯数据：

// server/src/lib.rs —— 应用「贡献」给宿主的全部内容
impl WorkflowApp for AssistantApp {
    fn artifact_types(&self) -> Vec<ArtifactSpec> {
        vec![
            artifact("materials", "Materials", "materials.json"),
            artifact("knowledge", "Knowledge points", "knowledge-map.md"),
            artifact("quiz", "Timed practice", "quiz.json"),
            artifact("learner_state", "Practice progress", "learner-state.json"),
            artifact("review", "Analysis", "review.json"),
            artifact("report", "Report", "report.md"),
        ]
    }

    fn workflows(&self) -> Vec<WorkflowSpec> {
        vec![WorkflowSpec {
            id: "timed_practice_flow".into(),
            roles: vec![
                role("learner", "Learner", RoleKind::Human, vec![], vec!["practice"]),
                role("tutor", "Tutor", RoleKind::Agent,
                     vec!["file_read", "web_search", "file_write"], vec![]),
            ],
            gates: vec![gate("practice", "Practice review", "learner", Some("practice"))],
            stages: vec![
                stage("intake", "Materials", "tutor", vec!["file_read"],
                      vec![], vec!["materials"], "…"),
                stage("map", "Knowledge points", "tutor",
                      vec!["file_read","web_search","file_write"],
                      vec!["intake"], vec!["knowledge"], "…"),
                gated(stage("practice", "Timed practice", "tutor",
                      vec!["file_read","file_write"], vec!["map"], vec!["quiz"], "…"),
                      "practice"),
                stage("review", "Analysis", "tutor",
                      vec!["file_read","file_write"], vec!["practice"], vec!["review"], "…"),
                stage("report", "Report", "tutor",
                      vec!["file_read","file_write"], vec!["review"], vec!["report"], "…"),
            ],
        }]
    }
}

注意这里没有任何 if/loop/await。流程的全部结构——谁先谁后、谁产什么、谁在哪个节点审批——都在这几个结构体里。引擎不关心「学习」这件事，它只认这套通用词汇。换个场景（比如把 quiz 换成 script、review 换成 voiceover），引擎一行不改。

这套词汇就五个核心结构体，定义在 server/src/workflow/contribution.rs：

结构体	一句话	关键字段
ArtifactSpec	一种会被持久化的产物	id、path、可选 schema
RoleSpec	一个业务角色（人或 agent）	kind（Human/Agent）、capabilities、can_approve
GateSpec	一道审批门	approver（角色 id）、on_reject（退回哪个阶段）
StageSpec	流水线的一个环节	role、depends_on、outputs、gate、prompt
WorkflowSpec	把上面四样组装成一张图	roles、gates、stages

StageSpec 是核心，把它拆开看就懂了：

pub struct StageSpec {
    pub id: String,
    pub name: String,
    pub role: String,                 // 哪个角色执行
    pub requires: Vec<String>,        // 执行需要哪些能力
    pub depends_on: Vec<String>,      // 等哪些阶段先完成 —— 这就是「图」的边
    pub inputs: Vec<String>,          // 读哪些工件
    pub outputs: Vec<String>,         // 产哪些工件
    pub gate: Option<String>,         // 完成后要不要过一道门
    pub foreach: Option<String>,      // 扇出（如 script.chapters）
    pub prompt: Option<String>,       // 交给执行者的指令
}

depends_on 是关键。阶段之间不是「第 1 步、第 2 步」的线性编号，而是显式依赖。map 依赖 intake，practice 依赖 map……于是整条流程天然是一张 DAG。线性流程只是 DAG 的特例；哪天你想让「出题」和「整理易错点」并行，只要改 depends_on，引擎的调度逻辑一行不用动。

三、工件：状态不藏在对话里，落在文件上

引擎最核心的一个决定：所有跨阶段的共享状态都是文件，不是内存、不是对话上下文。

看目录约定（server/src/runs.rs 顶部注释）：

<project>/.zoo_agent/plugins/<plugin>/runs/<run>/run.json   # 运行时状态
<project>/.zoo_agent/plugins/<plugin>/assets/              # 私有素材
<project>/<plugin>/<artifact path>                          # 被 Git 跟踪的工件

这里有一个刻意的分离，值得停下来想：

• 运行时状态（run.json：当前在哪个阶段、哪些阶段完成了、工件列表、审批备注）放在 .zoo_agent/ 下，是「机器内部账本」，不进版本库。
• 工件（materials.json、knowledge-map.md、quiz.json、report.md）放在项目根目录下、被 Git 跟踪。

为什么工件要进 Git？因为工件就是这次练习的「交付物」。学习者能 git diff 看模型这一轮把知识图谱改了哪几行，能回滚，能当真实文件分享。把 LLM 的产出当成一等公民的源文件来管理，而不是当成一段聊天记录——这是整个设计里我觉得最值钱的取舍。

每个工件有自己的状态机（run.rs）：

pub enum ArtifactStatus {
    Empty,      // 还没生成
    Draft,      // 已生成，待审
    Review,     // 送审中
    Approved,   // 审批通过
    Locked,     // 锁定，后续阶段不许再改
}

阶段执行完，引擎会去磁盘上扫一遍（refresh_artifacts），文件存在就把状态从 Empty 翻成 Draft、记下大小和时间戳。注意：引擎不信模型说「我写好了」，它只看文件在不在。 这就是下一节 check_outputs 的伏笔。

四、引擎：一次只跑一个阶段，绝不贪心

这是整个引擎的灵魂，也是和「一条 prompt 跑到底」最本质的区别。看 runs.rs 里 start_run 的注释，原文就一句：

The run executes exactly one stage and stops for review; it never chains the whole workflow.

一次只执行一个阶段，然后停下来。 引擎永远不自己把整条流水线串起来跑。

为什么这么保守？因为「停下来」正是把控制权交还给人的地方。流程长这样：

start_run ──► 跑 intake ──► 产出 materials.json ──► 状态回到 Created（停）
                                                          │
                              用户点「继续」────────────────┘
                                                          ▼
                        start_run ──► 跑 map ──► 产出 knowledge-map.md ──► 停
                                                          │
                              用户点「继续」────────────────┘
                                                          ▼
                        start_run ──► 跑 practice ──► 产出 quiz.json ──► 遇到人门 ──► WaitingGate（停）
                                                          │
                              用户审卷子：通过 / 打回──────────────────────┘

具体执行逻辑在 run_one_stage（runs.rs:557），精简后是这样：

async fn run_one_stage(...) -> Result<(), String> {
    // 1. 找下一个该跑的阶段：显式指定的优先，否则按依赖算
    let stage = if let Some(stage_id) = requested_stage {
        workflow.stages.iter().find(|s| s.id == stage_id)?
    } else {
        match next_stage(workflow, &run.completed_stages) {
            Some(s) => s,
            None => { run.status = Completed; return Ok(()); }  // 全跑完了
        }
    };

    // 2. 把阶段指令发给 agent 会话，让 LLM 真正干活
    let prompt = stage_prompt(&run, stage, dir, artifacts);
    sessions::send_message(state, &run.project_id, &session_id,
        &SendMessageRequest { text: prompt, .. }).await?;

    // 3. 扫磁盘，确认声明的产物真的生成了
    refresh_artifacts(&mut run, artifacts).await;
    check_outputs(&run, stage, artifacts)?;   // 缺产物直接报错，不往下走

    // 4. 这个阶段后面挂的是「人」门吗？是就停到 WaitingGate
    let human_gate = stage.gate.as_ref()
        .and_then(|id| workflow.gates.iter().find(|g| &g.id == id))
        .map(|g| is_human(workflow, &g.approver))
        .unwrap_or(false);
    if human_gate {
        run.status = WaitingGate;
        return Ok(());
    }

    // 5. 不是人门：标记完成，状态回 Created，等下一次 start_run
    run.completed_stages.push(stage.id.clone());
    run.status = Created;
    Ok(())
}

几个值得注意的细节：

「下一个阶段」是算出来的，不是写死的。 next_stage 就一行：

fn next_stage<'a>(workflow: &'a WorkflowSpec, completed: &[String]) -> Option<&'a StageSpec> {
    workflow.stages.iter().find(|stage| {
        !completed.iter().any(|id| id == &stage.id)          // 自己还没跑
            && stage.depends_on.iter()
                 .all(|d| completed.iter().any(|id| id == d)) // 依赖的全跑完了
    })
}

这就是 DAG 调度的最小实现：找一个「自己没完成、且依赖全部完成」的阶段。线性流程下它退化成「按顺序取下一个」，但代码本身对并行是开放的。

产物校验是硬约束。 check_outputs 遍历阶段声明的每个 outputs，去磁盘上确认文件（或目录）真的存在，缺一个就返回 Err，整个阶段标 Failed。模型嘴上说「已生成 quiz.json」但实际没写，引擎不会放过它。这把「LLM 的输出」从「一段话」升级成了「可验证的副作用」。

人门是状态机的一个独立状态。 RunStatus 有六个值：Created / Running / WaitingGate / Completed / Failed / Cancelled。WaitingGate 是专门为「等人审批」设的。阶段跑完、产物校验通过，但后面挂了 approver 是 RoleKind::Human 的门，状态就翻到 WaitingGate 然后返回——进程可以就此结束，状态躺在 run.json 里，等用户哪天回来点按钮。 这就是为什么一次练习能跨好几天：状态在磁盘，不依赖任何常驻进程。

五、审批门：把「打回重做」做成一等公民

审批门（gate）是这套引擎里我最喜欢的部分，因为它把「人对 AI 产物的否决权」做成了一个有数据支撑的循环，而不是一个「不满意你重发一遍」的口头约定。

看 practice 阶段，它是唯一挂了门的：

gates: vec![gate("practice", "Practice review", "learner", Some("practice"))],
//                                                                 ▲ 审批人=学习者  ▲ 打回时退回 practice 自己

GateSpec 的两个字段把语义说清楚了：approver 是「谁有权拍板」（这里指向 learner 这个角色，kind: Human），on_reject 是「打回时退回哪个阶段」（这里退回 practice 自己，意味着重新出题）。

通过和打回是两个不同的状态迁移（runs.rs）：

// 通过：阶段算完成，但下一步不自动启动，等用户再点
pub async fn approve_run(...) {
    run.completed_stages.push(stage_id);   // 关键：记入「已完成」
    run.status = Created;                  // 回 Created，不自动续跑
    run.stage = None;
    run.review_notes = None;
}

// 打回：阶段不算完成，把反馈存下来，下次生成会带着反馈重跑
pub async fn reject_run(...) {
    run.status = Created;
    run.stage = None;
    run.review_notes = Some(body.reason);  // 关键：存下「为什么不行」
}

对比这两段，门的全部语义就出来了：

• 通过 = 把阶段 id 推进 completed_stages。下次 start_run 时 next_stage 会跳过它，走到 review。
• 打回 = 不推进 completed_stages，阶段还「没完成」。同时把用户的反馈意见写进 review_notes。

那 review_notes 怎么作用到下一次生成？看 stage_prompt——每次给 LLM 的指令里，会带上当前运行的全部上下文（输入、所有工件及状态、本阶段输出、阶段说明）。review_notes 作为运行状态的一部分被持久化，重跑同一阶段时模型能看到「上次被打回，原因是 X」，于是带着反馈重新出题。

这就形成了一个带反馈的闭环：

出题 ──► 学习者打回「太难了，多来点基础题」
            │  review_notes 记下原因，阶段保持未完成
            ▼
       重新出题（prompt 里带上了反馈）──► 学习者这次通过 ──► 进入分析

注意这里没有魔法。引擎不知道「太难了」是什么意思，它只负责：记住反馈、保证阶段重跑时反馈在场、保证通过后才放行。语义理解是 LLM 的事，流程控制是引擎的事——职责切得很干净。

还有一个容易忽略的点：approve_run 之后状态是 Created 而不是直接续跑下一阶段。也就是说，通过审批不等于自动进入下一步，用户还得再点一次「继续」。 这是刻意的——引擎把「每一步都经过人的手」当成默认姿态，而不是「能自动就自动」。对一个学习者要逐题确认卷子的场景，这个保守是对的。

六、角色：把「谁执行」和「谁能批」解耦

RoleSpec 里有个设计我觉得比表面看起来重要：

pub struct RoleSpec {
    pub id: String,
    pub kind: RoleKind,          // Human 还是 Agent
    pub capabilities: Vec<String>, // 执行这个角色需要哪些能力
    pub can_approve: Vec<String>,  // 这个角色能审批哪些工件
}

注释里有一句点题的话（contribution.rs）：

Roles are not agent names; the host resolves them against its capability catalog at run time.

角色不是某个具体 agent 的名字，而是「能力槽位」。 tutor 这个角色声明它需要 file_read、web_search、file_write 三种能力，宿主在运行时从自己的能力池里找一个满足这些能力的 agent 来填这个槽。流程描述里永远只写「tutor」，不写「用 GPT 还是用 Claude、用哪个版本的工具」。

这带来两个好处：

1. 换模型不动流程。 今天 tutor 背后是模型 A，明天换成模型 B，只要 B 具备那三种能力，WorkflowSpec 一个字不用改。
2. 人和 agent 用同一套词汇。 learner 是 Human，tutor 是 Agent，但它们在 WorkflowSpec 里都是 RoleSpec，门、阶段、依赖对一视同仁。引擎不需要特殊处理「这一步要叫人」——它只看 kind。

capabilities 和 can_approve 的分离也值得说：一个角色能执行什么（capabilities）和能拍板什么（can_approve）是两件事。learner 没有任何执行能力（capabilities: vec![]），但能审批 practice 工件。tutor 有一堆执行能力，但 can_approve 是空的——它干活，但不给自己批条子。执行权和审批权分离，这是任何「人 + AI 协作」流程都该有的一条纪律。

七、入口与对话：流程不是只能「点按钮」启动

前面讲的是「流程化」路径——用户填个表单、点开始，引擎按 DAG 一步步推。但真实使用里，很多人是「聊天」着聊着就想开始刷题。引擎给这两条路留了同一个出口。

WorkflowApp 还有一个 chat_tasks()，声明一组「预设对话任务」：

fn chat_tasks(&self) -> Vec<ChatTaskSpec> {
    vec![
        ChatTaskSpec { id: "analyze_syllabus".into(), label_zh: "分析考纲与真题".into(), ... },
        ChatTaskSpec { id: "timed_practice".into(),   label_zh: "开始限时刷题".into(), ... },
        ChatTaskSpec { id: "review_mistakes".into(),  label_zh: "分析刷题结果".into(), ... },
        // 讲解知识点、刷题报告 …
    ]
}

宿主把这些渲染成对话界面里的建议按钮。用户点「开始限时刷题」，对应的 prompt_zh 被发给主 agent：

请先一次一个问题地询问主题、知识点/考纲/往年真题、题量和限时（能选项的用 ask_user 选项）；确认后调用 start_practice 工具创建并启动限时刷题。

注意最后半句——聊天路径最终也是落到 start_practice 这个工具上，而 start_practice 干的事和表单路径完全一样（lib.rs:394）：

async fn run(&self, args: &Value, ctx: &ToolCtx) -> Result<String, ToolError> {
    // … 解析参数、clamp 题量/限时 …
    let created = crate::runs::create_run(&self.state, ctx.project_id.as_str(),
        "assistant", &CreatePluginRunRequest { entry_id: Some("learning".into()), input }).await?;
    crate::runs::start_run(&self.state, ctx.project_id.as_str(), "assistant",
        &created.id, &StartPluginRunRequest { model: None, stage: Some("intake".into()) }).await?;
    Ok(format!("Started timed practice run {} …", created.id))
}

所以「聊天」和「表单」不是两套系统，而是同一个 run 的两种创建方式。聊天路径多做的事只是「需求澄清」——agent 一次问一个问题，把主题、资料、题量、限时凑齐，再调用工具建 run。建好之后，两条路汇入同一条 DAG，走同一套阶段、门和工件。

这个统一很重要。它意味着：无论用户怎么进来，状态模型都是同一个（run.json + 工件文件），UI 也只需要认这一种东西。不会出现「聊天里聊出来的练习」和「表单里建出来的练习」是两种不同数据、各有一套展示逻辑的尴尬。

八、一次真实的运行：把前面串起来

把前面所有零件拼起来，看一次「复习 Python 装饰器，20 题 30 分钟」到底发生了什么：

1. 用户点「开始限时刷题」
   └─ agent 一次问一个问题：主题? 资料? 题量? 限时?
   └─ 凑齐后调 start_practice(topic="Python 装饰器", question_count=20, time_limit_minutes=30)

2. create_run
   └─ 分配 run id，物化 6 个工件（全 Empty）
   └─ 建一个 agent 会话，写 run.json（status=Created）
   └─ start_run(stage="intake")

3. run_one_stage(intake)
   └─ next_stage → intake（无依赖，第一个）
   └─ 发 prompt 给会话：「读输入，整理成 materials.json」
   └─ LLM 读资料/知识库，写出 materials.json
   └─ check_outputs 确认文件存在
   └─ intake 后面没门 → completed_stages=[intake]，status=Created（停）

4. 用户点「继续」→ run_one_stage(map)
   └─ 基于 materials.json 生成 knowledge-map.md
   └─ 停

5. 用户点「继续」→ run_one_stage(practice)
   └─ 生成 quiz.json（20 题，带答案解析，难度由易到难）
   └─ practice 挂了 learner 的人门 → status=WaitingGate（停，等人）

6. 学习者审卷子
   └─ 不满意 → reject_run(reason="第 15 题超纲了")
        └─ 阶段保持未完成，review_notes 记下原因
        └─ 用户点「重新生成」→ 重跑 practice，prompt 带上反馈
   └─ 满意 → approve_run
        └─ completed_stages 加入 practice，status=Created（停）

7. 用户点「继续」→ run_one_stage(review)
   └─ 读 quiz.json + learner-state.json（实际答案、用时）
   └─ 生成 review.json：正确率、薄弱点、错因分类、记忆卡片
   └─ 停

8. 用户点「继续」→ run_one_stage(report)
   └─ 汇总所有工件生成 report.md
   └─ 没有下一个阶段 → status=Completed

全程引擎自己只「跑」了 5 次 run_one_stage，每次干完就停。人介入了 4 次（3 次「继续」+ 1 次审批）。每一个产物都是一个真实文件，躺在项目目录里，可以被 Git 跟踪、被 diff、被分享。

九、这套设计到底买到了什么

回到开头那四个问题，逐条对账：

一开始的痛点	引擎怎么解的
中间产物不可见、不可改	工件是 Git 跟踪的真实文件，每阶段一个，状态机管理
没有「刹车」	一次只跑一个阶段 + 人门 WaitingGate，关键节点强制停下等人
状态无处安放	run.json + 工件文件全在磁盘，进程可退出，跨天可续
不可重放	start_run(stage=...) 可指定任意阶段重跑；打回带反馈重生成

但我觉得更值钱的不是这四点，而是职责的切分：

• 引擎只管流程控制：谁先谁后、产物在不在、门有没有过、状态存哪。它不懂「学习」，不懂「装饰器」，甚至不懂「太难了」是什么意思。
• LLM只管语义：理解资料、出题、分析、把反馈变成更好的卷子。它不管流程，不存状态，不决定下一步是谁。
• 人只管拍板：在门上通过或打回，打回时给一句人话反馈。

三者各管一段，边界清晰。换模型、换场景、加一个并行阶段，都只动数据（WorkflowSpec）或 prompt，不碰引擎。这就是「声明式」真正的回报——流程变成了一张可以被审查、被版本化、被复用的图，而不是一段和具体业务焊死的控制流。

十、还没做完的

诚实地说，这套引擎还有几处是「留了口子但没填」的：

• foreach 扇出：StageSpec 里有 foreach 字段（比如对 script.chapters 逐章生成），但学习助手这条流程没用上。并行/批量阶段目前是「能声明、没实战」。
• on_reject 只退回自己：当前 practice 门打回就是退回 practice。如果哪天想「打回卷子时连知识图谱一起重梳」，on_reject 指向 map 就能做到，但还没验证过跨阶段回退的体验。
• 多 workflow：select_workflow 支持一个插件挂多条流程，学习助手目前只声明了一条 timed_practice_flow。

这些不是缺陷，是这套「数据描述流程」的设计留出来的扩展点。引擎已经把「图 + 门 + 工件」的骨架搭好了，剩下的只是往图里加节点。

---

本文基于 agentzoo_assistant 的真实源码（server/src/workflow/、server/src/runs.rs、server/src/lib.rs）撰写，代码片段为节选，关键逻辑已简化。如果你想自己跑一遍：make build && make run，在对话里点「开始限时刷题」，然后盯着项目目录下 assistant/ 文件夹里那几个文件一个个冒出来。