zhuangang/plan/agent-mini-project.md

转岗计划 · 第 11~14 周：多 Agent 项目（项目三 · 阶段 3 主交付）

所属阶段：第三段“Agent 主线”。总纲见同目录 README.md。前置：第 6 周 Python、第 78 周 RAG 项目、第 910 周盘点和案例卡。

目标

把在公司里见过的多 Agent 产品“削”成一个没有业务外壳、可以公开演示的内核，证明你“会造”，而不只是“会改”。主框架用 LangGraph（招聘里要求 LangChain / LangGraph 实际项目经验的很多），但先花几天手写一遍循环，把原理吃透，面试时既能讲框架也能讲原理。

做完后要能演示：

1. 提一个问题 → 协调者派活给两个专家 → 专家可以调用工具（包括通过 MCP 调用你第 7~8 周做的 RAG 知识库）→ 多轮讨论 → 达成共识或到达最大轮数后结束 → 生成可审计的报告。
2. 中途 kill -9 杀掉服务进程 → 重新启动 → 任务从最近的检查点接着跑完。
3. 浏览器通过 SSE 看进度；关掉页面任务照跑；重新打开能补齐漏掉的事件。
4. 协调者觉得信息不够时停下来问人（LangGraph interrupt），人回答后继续。
5. 记得用户的偏好和以前的结论（长期记忆），下次提问时用上。
6. 有一个小评测集，能用数字说明改提示词或流程之后是变好还是变差。

另外用 1~2 天体验 Dify 或 Coze（扣子），能讲清低代码平台和写代码各自适合什么场景。

项目名：agent-mini，放在 zhuangang/agent-mini/。

范围

必须有	禁止有（本阶段）
至少 2 个专家（提案者 proposer、评审者 critic）+ 1 个协调者 coordinator	完整任务中心、字典、巡检
明确的结束条件：评审通过（共识）或达到最大轮数，规则写在代码里	照搬公司 UI、提示词或代码
工具调用（Function / Tool Calling），至少一个工具来自自己写的 MCP 服务	依赖公司内网才能演示
状态持久化（LangGraph 检查点存 SQLite，可选 PostgreSQL）+ 重启续跑	没有说明的纯聊天 demo 页
编排跑在服务端后台；SSE 推送进度，断线可补齐	一次接太多框架（LangGraph 之外不再引入别的 Agent 框架）
人机协作（中断问人）、长期记忆、评测集
不需要真实模型密钥也能跑通测试和界面演示（假模型）

讨论的题目用通用例子，比如“一个 5 人创业团队做内部工具，该选什么技术栈”“给一个读书 App 设计会员规则”，不要用公司业务题目。

技术选型

用途	选择
语言和依赖	Python 3.12 + uv
手写阶段	asyncio + httpx（复用 RAG 项目的模型客户端思路）
主框架	LangGraph（StateGraph、检查点、interrupt、Store），模型接入用 LangChain 的对话模型封装
工具	Function Calling；MCP 服务用官方 Python SDK（或 FastMCP）写；Agent 侧用 LangChain 的 MCP 适配器加载工具
检查点	SQLite 检查点（langgraph-checkpoint-sqlite），可选 PostgreSQL 检查点
业务表	自己的 runs / events 表（SQLite），存任务状态、租约、事件
服务	FastAPI + SSE
前端	极简 web/index.html，可选用 Vue / React 重写
质量	pytest、ruff、pyright；评测脚本

LangGraph、LangChain 和 MCP 相关库更新很快，包名、导入路径、推荐写法都可能变（比如 MCP 适配器最近就从独立包并入了 LangChain）。动手前以官方文档当前版本为准，本文写的是概念和要建的文件，具体 API 名称如有出入按文档改。

核心设计（先读懂，再动手）

两层状态：检查点 vs 业务表（重点 B）

• LangGraph 检查点：图的内部状态（每个节点执行完的快照），用 thread_id 区分，负责“从哪一步接着跑”。
• 自己的业务表 runs / events：任务的对外状态（排队、运行、等人、完成、失败、取消）、谁在跑（租约）、给前端的事件流、最终报告。列表页、权限、计费、审计都查这里。

为什么要两层：检查点是框架的内部格式，不适合直接给业务查询；业务表是对外的权威来源。这正是你在工作中见过的“草稿 / 作业 / 任务状态谁说了算”的问题，在这里给出你自己的答案，并写进 README。

状态机（业务表 runs.status）

queued ──(被某个进程领取)──> running ──> completed
   ^                           │  ├──> failed（模型错误、输出无法解析等，带 error_code）
   │                           │  ├──> cancelled（用户取消）
   │                           │  └──> awaiting_input（图里调用了 interrupt）
   └──(人回答 / 失败后重试 / 进程崩溃后租约过期)──┘

图的结构（LangGraph）

START → load_memory → coordinator ─┬→ proposer ─┐
                         ^         ├→ critic   ─┤
                         │         ├→ ask_human─┤（interrupt）
                         └─────────┴────────────┘
                       coordinator ─→ summarize → save_memory → END

• 路由函数（条件边）里先检查硬规则：最近一次评审是 VERDICT: APPROVE → 去 summarize；轮数达到上限 → 去 summarize；否则按协调者的结构化决定走。
• 专家节点内部可以调用工具（工具调用循环可以用 LangGraph 预置的工具节点，也可以自己写一个小循环）。

防止双跑（重点 A）和续跑（重点 C）

• runs 表上有租约字段 lease_owner、lease_until，带条件的 UPDATE 领取任务，运行中定期续租；进程崩溃后租约过期，看门狗重新领取，用同一个 thread_id 从最近检查点继续。
• 节点内正在进行的模型调用在恢复后会重做（至少一次），已经完成的节点不会重做。工具如果有副作用（比如“创建待办”），要设计成幂等（带请求 ID），README 里写清楚。

---

最终项目结构

agent-mini/
├── README.md
├── pyproject.toml
├── .env.example                    # 默认 LLM_PROVIDER=fake
├── docker-compose.yml              # agent-mini、mcp 服务；可连第 8 周的 rag-qa
├── Dockerfile
├── docs/
│   ├── architecture.md             # 组件图、状态机、图结构、崩溃恢复时序图
│   ├── design-decisions.md         # 关键取舍
│   ├── mapping.md                  # 与生产概念的映射表
│   ├── handwritten-vs-langgraph.md # 手写版和 LangGraph 版对照
│   └── talk.md                     # 3 分钟讲解稿
├── handwritten/                    # 第 11 周：手写版（理解原理，保留作对照）
│   ├── llm.py                      # 最小模型客户端（支持 tools 参数）
│   ├── tool_loop.py                # 手写工具调用循环
│   ├── multi_agent_loop.py         # 手写协调者 + 两个专家循环
│   └── test_handwritten.py
├── mcp_servers/
│   ├── knowledge_server.py         # MCP 服务：知识库检索（调 rag-qa）、读笔记
│   └── todo_server.py              # MCP 服务（可选）：创建待办（调项目一的接口，幂等）
├── src/agent_mini/
│   ├── __init__.py
│   ├── config.py
│   ├── errors.py                   # 错误码
│   ├── models.py                   # Decision、RunStatus、事件结构
│   ├── llm.py                      # 构造对话模型：真实 / 假模型
│   ├── tools/
│   │   ├── local_tools.py          # 本地工具：计算、当前时间
│   │   └── mcp_tools.py            # 通过 MCP 加载外部工具
│   ├── graph/
│   │   ├── state.py                # 图的 State 定义（含 reducer）
│   │   ├── prompts.py              # 各角色提示词
│   │   ├── nodes.py                # coordinator / proposer / critic / ask_human / summarize
│   │   ├── routing.py              # 条件边：硬规则 + 协调者决定
│   │   ├── memory.py               # load_memory / save_memory 节点（长期记忆）
│   │   └── build.py                # 组装并编译图
│   ├── store/
│   │   ├── schema.sql              # runs / events 业务表
│   │   └── run_store.py            # 业务表读写、租约
│   ├── runner.py                   # RunManager：后台运行、租约续期、看门狗、取消
│   ├── report.py                   # 由检查点状态生成可审计报告
│   ├── api/
│   │   ├── app.py
│   │   ├── routes_runs.py
│   │   └── sse.py
│   └── cli.py
├── evals/
│   ├── cases.jsonl                 # 评测用例
│   ├── checks.py                   # 规则检查 + 模型评分
│   ├── run_evals.py
│   └── reports/                    # 每次评测的结果
├── web/index.html
├── scripts/demo_crash_resume.sh
└── tests/
    ├── conftest.py
    ├── test_routing.py
    ├── test_nodes.py
    ├── test_graph_resume.py
    ├── test_run_store.py
    ├── test_mcp_server.py
    └── test_api_sse.py

业务表（store/schema.sql）

CREATE TABLE IF NOT EXISTS runs (
  id            TEXT PRIMARY KEY,          -- 同时作为 LangGraph 的 thread_id
  user_id       TEXT NOT NULL DEFAULT 'demo',
  question      TEXT NOT NULL,
  status        TEXT NOT NULL,             -- queued/running/awaiting_input/completed/failed/cancelled
  model         TEXT NOT NULL,
  max_rounds    INTEGER NOT NULL,
  human_question TEXT,
  final_answer  TEXT,
  end_reason    TEXT,
  error_code    TEXT,
  error_message TEXT,
  lease_owner   TEXT,
  lease_until   REAL,
  created_at    REAL NOT NULL,
  updated_at    REAL NOT NULL
);

CREATE TABLE IF NOT EXISTS events (
  seq        INTEGER PRIMARY KEY AUTOINCREMENT,
  run_id     TEXT NOT NULL REFERENCES runs(id),
  type       TEXT NOT NULL,     -- run_created/node_finished/tool_called/awaiting_input/input_received/run_completed/run_failed/run_cancelled
  data       TEXT NOT NULL,     -- JSON
  created_at REAL NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_events_run_seq ON events(run_id, seq);
CREATE INDEX IF NOT EXISTS idx_runs_status_lease ON runs(status, lease_until);

接口设计

方法	路径	说明
POST	/runs	创建任务 {question, max_rounds?}，立即返回 {id, status: "queued"}
GET	/runs / /runs/{id}	列表 / 详情（状态 + 讨论记录）
GET	/runs/{id}/events	SSE，支持 Last-Event-ID 或 ?after=seq 补发
POST	/runs/{id}/input	回答问题 {answer}，仅 awaiting_input 可用
POST	/runs/{id}/retry	失败后重试，可换模型 {model?}
POST	/runs/{id}/cancel	取消
GET	/runs/{id}/report	Markdown 报告
GET	/memories	查看当前用户的长期记忆（可删除）

---

第 11 周 · M1：手写循环 + Tool Calling + MCP

本周不用 LangGraph，目的是搞懂原理。代码放在 handwritten/ 和 mcp_servers/。

第 1 天：项目初始化和最小模型客户端

学习内容

• 复习：消息列表（system / user / assistant / tool）、温度、最大输出长度、流式输出。
• 错误码沿用 RAG 项目（超时、限流、额度、鉴权、输出异常），哪些重试、哪些不重试。

要建的文件

• uv init --package agent-mini；先只加 httpx pydantic pydantic-settings，开发依赖 pytest pytest-asyncio ruff pyright。
• src/agent_mini/config.py、src/agent_mini/errors.py、.env.example、.gitignore。
• handwritten/llm.py：async def chat(messages, tools=None) -> AssistantMessage，AssistantMessage 含 content 和 tool_calls（id、函数名、参数 JSON 字符串）。

第 2 天：手写工具调用循环（Function / Tool Calling）

学习内容

• 工具调用的完整过程：把工具的名字、说明、参数 JSON Schema 发给模型 → 模型返回“想调用哪个工具、参数是什么” → 你的代码执行工具 → 把结果作为 tool 消息追加回去 → 再问模型 → 直到模型给出最终回答。模型本身从不直接执行任何东西。
• 安全和可靠性：参数要校验（pydantic）；工具出错时把错误信息返回给模型，让它自己纠正，而不是整个流程崩溃；限制最大循环次数；有副作用的工具要幂等、要能审计。

要建的文件

• handwritten/tool_loop.py：
  • @dataclass class Tool：name、description、args_model: type[BaseModel]、func。
  • def to_schema(tool) -> dict：用 args_model.model_json_schema() 生成工具描述。
  • 三个工具：calculator(expression)（只允许数字和四则运算，不能用 eval 执行任意代码）、now()、search_knowledge(query)（调用第 8 周 rag-qa 的 /ask 接口；没启动时返回错误信息）。
  • async def run_tool_loop(question, tools, max_steps=6) -> LoopResult：记录每一步调用了什么工具、参数、结果、耗时。
• handwritten/test_handwritten.py：用假模型（按脚本返回 tool_calls）测：正常调用后得出答案；参数不合法时错误被返回给模型；超过最大步数时停止。

完成标准：用真实模型问“距离 2027 年元旦还有多少天，再乘以 3”，日志里能看到先调 now 再调 calculator。

第 3 天：手写多 Agent 循环

学习内容

• 多 Agent 的本质：几个“角色 = 系统提示词 + 能看到的上下文 + 可用工具”，加上一个决定“下一步谁来做、什么时候结束”的调度逻辑。
• 结束条件写在代码里，不只靠模型自觉。

要建的文件

• handwritten/multi_agent_loop.py（目标 150~250 行）：
  • 协调者输出 JSON 决定 {"action": "dispatch|ask_human|finish", "target", "instruction", "reason"}，用 pydantic 校验，格式错误时带上错误重试一次。
  • 提案者、评审者两个角色；评审者最后一行必须是 VERDICT: APPROVE 或 VERDICT: REVISE。
  • 循环：硬规则（评审通过 / 达到最大轮数）→ 协调者决定 → 执行 → 记录。
  • 每一步把状态写到一个 JSON 文件（先体会“持久化 + 续跑”要解决什么问题：进程中途退出后，从文件里读出状态继续）。
• 测试补充到 handwritten/test_handwritten.py：评审第二次通过时提前结束；一直不通过时在最大轮数结束。

完成标准：能不看代码讲清这个循环的每一步；能说出“写 JSON 文件”这种持久化方式有什么问题（并发、原子性、查询），为后面用检查点和数据库做铺垫。

第 4 天：MCP 概念 + 写一个 MCP 服务

学习内容

• MCP（Model Context Protocol）是什么：让 AI 应用以统一的方式接入外部工具和数据的协议。角色分为宿主（比如 Cursor、Claude 桌面版）、客户端、服务端；服务端可以提供工具（tools）、资源（resources）、提示词模板（prompts）。
• 传输方式：本地进程用 stdio，远程服务用 HTTP（Streamable HTTP）。
• 和 Function Calling 的关系：Function Calling 是“模型怎么表达要调工具”；MCP 是“工具怎么被统一发现和调用”，写一次 MCP 服务，各种支持 MCP 的应用都能用。

要建的文件

• 加依赖：MCP 官方 Python SDK（mcp）或 FastMCP，按官方文档选一个。
• mcp_servers/knowledge_server.py：
  • 工具 search_knowledge(query: str, top_k: int = 5)：调用 rag-qa 的检索接口，返回片段和出处。
  • 工具 list_documents()。
  • 资源（可选）：notes://cards 列出你的案例卡标题（只读脱敏后的 notes/cards/）。
  • 每个工具都写清楚说明和参数说明（模型靠这些判断什么时候调用）。
• mcp_servers/todo_server.py（可选）：工具 create_todo(board_id, title, request_id)，调用项目一“团队任务看板”的接口；request_id 用来保证重复调用不会建出两条（体会“有副作用的工具要幂等”）。
• tests/test_mcp_server.py：直接调用工具函数，测试正常和 rag-qa 不可用时的错误返回。

运行方式：用官方调试工具 MCP Inspector（npx @modelcontextprotocol/inspector 加上启动你服务的命令）连上去，手动调用每个工具。

完成标准：在 Inspector 里能看到并调用你的工具。

第 5 天：在 Cursor 和自己的 Agent 里用 MCP

要建的文件

• 在 Cursor 的 MCP 配置里加上你的 knowledge_server，在 Cursor 对话里让它“查一下我的知识库里关于 X 的内容”，确认能调用。把配置示例写进 docs/mcp-usage.md（不要提交真实路径里的个人信息和密钥）。
• src/agent_mini/tools/mcp_tools.py：用 LangChain 的 MCP 适配器（当前版本的推荐方式以官方文档为准）把 MCP 工具加载成 LangChain 工具，下周给 LangGraph 用。
• handwritten/tool_loop.py：加一个选项，从 MCP 服务读取工具列表来替代本地的 search_knowledge，体会“工具可以动态发现”。

完成标准：同一个 MCP 服务，Cursor 和你自己的 Agent 都能用。

第 6~7 天：对照文档和 M1 验收

• docs/handwritten-vs-langgraph.md：先写“手写版”这半边：循环、状态、持久化、工具调用、结束条件分别是怎么做的，哪里最容易出错。下周学完 LangGraph 再补另一半。
• 余下时间补测试、整理代码，提交推送。

M1 验收

• 手写工具调用循环和多 Agent 循环都能跑，测试通过
• 能讲清 Function Calling 的完整过程，以及工具出错、参数不合法、死循环怎么处理
• 写了至少一个 MCP 服务，Inspector 和 Cursor 都能调用
• 能讲清 MCP 和 Function Calling 的关系
• 在总纲进度记录里勾掉第 11 周

---

第 12 周 · M2：用 LangGraph 实现多 Agent 编排

第 1 天：LangGraph 基础

学习内容（跟官方入门教程走，边看边在 scratch/ 里敲）

• StateGraph：状态（State）、节点（node，普通函数，读状态、返回要更新的字段）、边、条件边、START / END。
• reducer：字段怎么合并（比如消息列表是追加而不是覆盖）。
• 编译图、运行（一次性 / 流式）、画出图结构（导出 mermaid）。
• 预置的工具调用节点 / 预置的 ReAct Agent：知道有，什么时候用。

完成标准：照着教程做出一个带条件边和工具调用的小图，能说出它和你上周手写版的对应关系。

第 2 天：设计 State 和路由

要建的文件

• 加依赖：langgraph、langchain-core、langchain-openai（或你用的模型服务对应的集成包），以官方文档为准。
• src/agent_mini/models.py：Decision（pydantic，同上周）、RunStatus。
• src/agent_mini/graph/state.py：
  • question、turns（追加型 reducer，每条含角色、指令、内容、调用过的工具）、round、max_rounds、last_verdict、decision、human_answers、final_answer、end_reason、memories。
• src/agent_mini/graph/routing.py：
  • def route_after_coordinator(state) -> str：按顺序检查：评审已通过 → "summarize"；round >= max_rounds → "summarize"；decision.action 是 dispatch → 目标专家；ask_human → "ask_human"；finish → "summarize"。
• tests/test_routing.py：每条规则至少一个用例（纯函数测试，不需要模型）。

第 3 天：节点和提示词

要建的文件

• src/agent_mini/graph/prompts.py：协调者、提案者、评审者、总结者的提示词，全部自己写。
• src/agent_mini/llm.py：def build_chat_model(settings)：真实模型（OpenAI 兼容接口，配置 base_url、超时、重试）或假模型（测试用：按角色返回预设回答；工具调用的假回答如果麻烦，测试只覆盖不调工具的路径）。
• src/agent_mini/graph/nodes.py：
  • coordinator：用结构化输出得到 Decision（模型不支持时退回“提取 JSON + pydantic 校验 + 重试一次”）。
  • proposer：可以调用工具（search_knowledge 来自 MCP、calculator、now），内部是一个有最大次数限制的工具调用循环。
  • critic：输出评审意见和 VERDICT，解析后写入 last_verdict。
  • summarize：生成最终结论，写 end_reason。
  • 每个节点控制传给模型的上下文：只给问题、最近 N 条发言摘要和本次指令（上下文工程）。
• tests/test_nodes.py：用假模型测协调者输出格式错误时的重试、评审结论解析。

第 4 天：组装图，命令行跑通

要建的文件

• src/agent_mini/graph/build.py：def build_graph(model, tools, checkpointer=None, store=None)：加节点、边、条件边，编译。
• src/agent_mini/cli.py：
  • agent-mini graph：导出图结构的 mermaid，贴进 docs/architecture.md。
  • agent-mini run "问题" --memory-only：不带持久化，流式打印每个节点的更新（谁、做了什么、调了什么工具）。

运行方式：先启动 rag-qa 和 MCP 服务，再 uv run agent-mini run "5 人创业团队做内部工具该选什么技术栈，结合我的知识库里的笔记"。

第 5 天：真实模型试跑和调提示词

• 用真实模型跑 5 个不同题目，输出存到 docs/samples/（检查没有敏感信息）。
• 观察：协调者是否按格式输出、评审是否太容易通过、专家是否乱调工具、轮数和耗时。调整提示词，每次改动记一行到 docs/prompt-changelog.md（第 14 周评测时会用到）。

第 6~7 天：对照文档、测试和 M2 验收

• 补完 docs/handwritten-vs-langgraph.md 的另一半：手写版的每个部分在 LangGraph 里对应什么，框架帮你做了什么，又有哪些事（租约、业务状态、幂等）框架不管、还得自己做。
• 测试、ruff、pyright 通过，提交推送。

M2 验收

• LangGraph 版能跑完一次带工具调用的多轮讨论，结束条件正确
• 路由规则有单元测试
• 能讲清 State、节点、条件边、reducer，以及和手写版的对应关系
• 在总纲进度记录里勾掉第 12 周

---

第 13 周 · M3：持久化续跑 + 人机协作 + 后台作业 + SSE + 记忆

第 1 天：检查点和续跑

学习内容

• 检查点（checkpointer）：每个节点执行完自动保存状态；同一个 thread_id 再次运行就从最近的检查点继续；还能查看历史状态。
• 用 SQLite 检查点（需要 aiosqlite），注意用完关闭连接，否则程序可能退不出去。官方说明 SQLite 检查点不建议用于生产，生产用 PostgreSQL 检查点——这一点写进 README 的“取舍”里。

要建的文件

• build.py：接上 SQLite 检查点。
• tests/test_graph_resume.py：用假模型（每次调用加一点延迟）运行，跑完 2 个节点后取消任务（模拟崩溃）；再用同一个 thread_id 继续运行，断言最终完成、已完成的节点没有重复执行（统计假模型被调用的次数）。

完成标准：能讲清检查点保存了什么、恢复时从哪里继续、正在执行的节点恢复后会怎样。

第 2 天：业务表和后台作业

要建的文件

• src/agent_mini/store/schema.sql、run_store.py：
  • create_run、get_run、list_runs、set_status、append_event、list_events(run_id, after_seq)。
  • 租约：try_acquire(run_id, worker_id, lease_seconds) -> bool（带条件的 UPDATE，影响行数为 1 才算成功）、renew_lease、release、find_resumable(now)。
• src/agent_mini/runner.py：class RunManager：
  • submit(run_id)：领取成功才启动后台任务，任务对象保存起来。
  • _drive(run_id)：用 thread_id = run_id 流式运行图；每收到一个节点更新就写一条 events；定期续租；正常结束、中断、出错时更新 runs 并释放租约。
  • watchdog()：定期找“该有人跑但没人跑”的任务重新提交。
  • cancel(run_id)、shutdown()。
• tests/test_run_store.py：两个 worker 抢同一任务只有一个成功；租约过期后另一个能抢到。

第 3 天：人机协作（interrupt）和失败重试

学习内容

• LangGraph 的 interrupt：在节点里暂停并把问题交给调用方，状态由检查点保存；之后用“恢复命令”带上人的回答继续，interrupt 的返回值就是人的回答。
• 注意：恢复时这个节点会从头重新执行，所以 interrupt 之前不要放有副作用的操作。
• 什么时候必须停下来问人（重点 D）：信息不足、要执行有副作用或高成本的操作、多次失败。

要建的文件

• nodes.py：ask_human 节点调用 interrupt，把问题写进状态。
• runner.py：检测到中断时 runs.status = awaiting_input，写 human_question，追加 awaiting_input 事件，释放租约。
• POST /runs/{id}/input 的处理逻辑（下面第 4 天接口里实现）：写 input_received 事件，状态改回 queued，后台用恢复命令继续。
• 失败重试：AgentError 时 runs.status = failed 并写错误码；POST /retry 可以换模型后从检查点继续（对应工作中“额度不足 → 换模型恢复”的真实问题）。
• 测试：中断 → 回答 → 完成；模型报错 → 失败 → 换模型重试 → 完成。

第 4 天：FastAPI 接口和 SSE

要建的文件

• src/agent_mini/api/app.py：lifespan 里打开检查点和业务库、构建图、启动 RunManager 和看门狗；关闭时依次清理。
• src/agent_mini/api/routes_runs.py：接口表里除 SSE 外的接口。
• src/agent_mini/api/sse.py：从 events 表按 seq 补发，再等新事件（进程内通知加速，超时就轮询数据库兜底）；15 秒无事件发心跳；任务到终态后发 end 并结束；客户端断开不影响任务。
• src/agent_mini/report.py：从检查点状态和 events 生成报告：问题、每一步谁做了什么、调用了哪些工具和结果摘要、结束原因、最终结论、使用的模型（重点 E：结果可审计）。
• tests/test_api_sse.py：创建 → 读 SSE 到结束；带 Last-Event-ID 只收到之后的事件；不存在的任务 404。

第 5 天：长期记忆（Memory）

学习内容

• 短期记忆：当前任务（线程）内的状态和对话，检查点已经管了。
• 长期记忆：跨任务记住的东西，比如用户偏好（“我们团队只用 Go 和 Vue”）、以前讨论得出的结论。LangGraph 提供 Store（按命名空间存取，可以配合向量做语义搜索）。
• 记忆要可查看、可删除；写什么进记忆要有规则，不能把所有对话都塞进去（成本、隐私、干扰）。

要建的文件

• src/agent_mini/graph/memory.py：
  • load_memory 节点：按 (user_id, "memories") 命名空间取出与问题相关的若干条，放进状态，协调者和专家的提示词里引用。
  • save_memory 节点：任务完成后，让模型从讨论中提炼 0~3 条“值得长期记住的事实或偏好”（结构化输出），写入 Store。
• Store 的持久化：本项目可以先用 SQLite 实现一个简单的 Store，或者使用官方提供的持久化 Store（以文档为准）；在 design-decisions.md 里说明选择。
• GET /memories、DELETE /memories/{key}。
• 测试：第一次任务里说明偏好 → 第二次任务的提示词里出现这条记忆。

第 6 天：页面和崩溃演示

要建的文件

• web/index.html（原生 JS，不超过 250 行，可选之后用 Vue / React 重写）：提问、时间线（每个节点、工具调用）、等待输入时出现输入框、失败时显示错误码和“换模型重试”、结论和报告链接、记忆列表、连接状态。
• scripts/demo_crash_resume.sh：用假模型加延迟启动服务 → 创建任务 → 等几秒 kill -9 → 重启 → 轮询到完成 → 打印报告。

第 7 天：走查和 M3 验收

手动走查：正常跑完；中途关页再打开；中途停服务再启动；触发问人并回答；用错误的密钥制造失败后修好重试；第二次提问用上了记忆。

M3 验收

• 测试全部通过，包括续跑、租约、SSE、中断恢复
• 崩溃恢复演示脚本成功
• 6 个走查场景全部符合预期
• 能讲清：两层状态（检查点 vs 业务表）、为什么编排不能放在浏览器里、什么时候停下来问人、怎么防双跑、长期记忆写什么不写什么
• 在总纲进度记录里勾掉第 13 周

---

第 14 周 · M4：评测 + 文档 + 低代码平台体验

第 1 天：评测集

学习内容

• Agent 评测比 RAG 更难，因为过程有多步。从三个层面看：
  • 结果：最终结论是否回答了问题、是否符合约束。
  • 过程：有没有该调的工具没调、不该调的乱调；轮数是否合理；该问人时有没有问。
  • 成本和稳定性：耗时、token 数、失败率。
• 检查方式：能用规则判断的用规则（调了哪些工具、轮数、是否包含关键词）；不能的用“模型按评分标准打分”，并抽查一部分人工复核，看两者是否一致。

要建的文件

• evals/cases.jsonl：20 条用例，每条包含 id、question、expect（例如 must_call_tools、must_not_call_tools、should_ask_human、max_rounds、must_mention）、rubric（给模型打分用的标准）。覆盖：普通问题、需要查知识库的问题、信息不足需要问人的问题、诱导乱调工具的问题。
• evals/checks.py：规则检查函数 + 模型评分函数（评分输出用结构化格式：分数 + 理由）。
• evals/run_evals.py：逐条运行（需要问人的用例自动给一个预设回答），汇总通过率、平均轮数、工具调用正确率、平均耗时和 token 数，结果写到 evals/reports/<日期>.md，记录当时的模型和提示词版本。

第 2 天：用评测驱动改进

• 跑基线，挑失败最多的一类问题，改提示词或路由规则，再跑，对比结果（数字以实际跑出来的为准）。改动和结果记到 docs/prompt-changelog.md。
• 用假模型做一个不需要密钥的小评测子集，放进测试，防止以后改代码把流程改坏（回归测试）。

第 3 天：文档

要建的文件

• docs/architecture.md：组件图、业务状态机（stateDiagram-v2）、LangGraph 图结构（第 12 周导出的）、崩溃恢复时序图（sequenceDiagram）。对照第 9 周画的公司系统脱敏图，写一段“我简化了什么”。
• docs/mapping.md（同时摘进 README，生产一栏只写通用概念，不写公司接口名、表名）：

生产系统里的概念（通用叫法）	本项目里的实现	对应重点
席位 / 专家角色	proposer、critic 节点	A
主持人派活	coordinator 节点 + 条件边	A
共识 / 结束条件	VERDICT: APPROVE + 最大轮数，写在 routing.py	A
草稿（编排中间状态）	LangGraph 检查点（按 thread_id）	B
任务状态（对外权威）	runs 表	B
后台作业	RunManager + 租约 + 看门狗	A / C
流式输出	GET /runs/{id}/events，按 seq 补发	C
意图澄清 / 等待用户输入	interrupt → awaiting_input → POST /input	D
换模型恢复	错误码 + POST /retry {model}	D
过程报告	report.py	E
外部能力接入	MCP 服务 knowledge_server	E

• docs/design-decisions.md：每条写“选择 / 理由 / 代价”：两层状态；编排在后台；租约防双跑；节点内调用至少一次、副作用工具要幂等；结束条件写在代码里；SQLite 检查点只用于演示，生产换 PostgreSQL；长期记忆的写入规则；没做的事（鉴权、多实例下的实时通知、更完整的评测）。

第 4 天：README 和从零验证

• README.md：一句话介绍 + 动图；功能列表；10 分钟快速开始（假模型模式，不需要密钥）；切换真实模型；连接 rag-qa 和 MCP；接口表；评测结果表；崩溃恢复演示；映射表；项目结构。
• Dockerfile、docker-compose.yml：agent-mini + MCP 服务，可选连接 rag-qa。
• 在新目录（最好换一台电脑或请朋友）git clone 后按 README 操作并计时，目标 10 分钟内跑通；卡住的地方都改进。
• 脱敏检查：git grep 搜公司名、内网域名、IP、密钥前缀。

第 5 天：讲解稿

• 录一段 2~3 分钟的演示：提问 → 工具调用 → 杀进程 → 重启续跑 → 问人 → 报告 → 评测结果。
• docs/talk.md：3 分钟讲解稿，“问题（为什么多 Agent 难做可靠）→ 设计（两层状态、后台作业、租约、中断、记忆）→ 演示 → 评测和取舍”，引用 1~2 张案例卡说明“这些设计来自我在工作中踩过的坑”。

第 6~7 天：Dify / Coze 体验（低代码平台）

为什么：招聘里经常出现 Dify、Coze、RAGFlow、FastGPT 这类平台。不需要精通，但要能说清楚它们能做什么、和写代码比各有什么优缺点，最好亲手搭过一个。

做什么（二选一为主，另一个看看文档即可）

• Dify（开源，可自己部署）：在你的 Ubuntu 服务器上用官方的 Docker Compose 部署（注意服务器内存是否够用，以官方要求为准），或用官方云版。
• Coze（扣子）：直接用在线版。

要搭的东西

1. 知识库问答：把 rag-qa 用的同一批公开文档导入平台的知识库，做一个问答应用，用你第 8 周的评测集里挑 10 个问题对比效果。
2. 工作流：做一个带条件分支的工作流，包含一次 HTTP 请求节点调用你的 agent-mini 接口或 rag-qa 接口（平台支持 MCP 的话也可以接你的 MCP 服务）。
3. 导出应用配置（平台支持导出时），放到 zhuangang/lowcode/（去掉密钥）。

要建的文件

• notes/lowcode-dify-coze.md：
  • 搭建步骤和截图。
  • 和你自己的代码实现对比：开发速度、可控性（提示词、检索参数、状态、错误处理）、可观测性、部署和成本、团队里谁来维护。
  • 结论：什么场景用平台（快速验证、业务人员自己搭、标准问答），什么场景写代码（复杂编排、强可靠性要求、深度集成业务系统）。

M4 验收（也是阶段 3 出门条件）

• 别人按 README 10 分钟内跑通（不需要真实模型密钥）
• 演示：启动 →（中途杀进程）→ 重启 → 最终结果可查
• 映射表至少 5 个概念对齐生产
• 有评测集和至少两次对比结果
• 架构图和 README 一致；测试、ruff、pyright 通过；脱敏检查完成
• Dify 或 Coze 至少搭过一个知识库应用和一个工作流，写了对比笔记
• 在总纲进度记录里勾掉第 14 周和“阶段 3”，填上“迷你项目路径：zhuangang/agent-mini/”

---

和案例卡的关系

第 10 周每张案例卡都有一栏“若重做的设计”。动手前把这些结论列出来，逐条对应到本项目的某个模块或某条设计取舍；对应不上的，在 docs/design-decisions.md 里说明为什么这次没做。面试时就能讲成：“我在工作中遇到 X 问题，根因是 Y，所以在自己的项目里用 Z 解决，这是代码和演示。”

可选扩展（第 15 周有余力再做，二选一）

• Go 网关：zhuangang/agent-gateway/，Gin 服务，复用项目一的 JWT 鉴权；/api/runs/* 反向代理到 Python 服务（SSE 要关闭缓冲）；加用户级限流（Redis）；转发时带上用户 ID，Python 端按用户隔离任务和记忆。讲点：“Go 做接入层，Python 做 Agent 编排”。
• Vue / React 前端：用你熟悉的框架重写 web/index.html，体现前端功底。