凌晨一点,我第 7 次打开生成好的课件:第 3 页图片裂了,第 5 页例题的答案和题干对不上。
生成这活儿 AI 干得挺欢,验收全是我人肉。每跑一次等 5 分钟,我盯一次 5 分钟,纯纯的苦力。
那晚我摔了鼠标:与其天天自己当质检员,不如写个 Agent 自己验收自己。这是这个系列的第 1 篇——从 0 到 1 搭骨架,纯工程复盘,不是什么风口突击套利的故事。
一个 while 循环加几个函数,不就能跑起来了吗?能,demo 阶段还挺爽,上生产第一天就翻车。
裸写调度循环有三个硬伤:
LangGraph 给我的正好是解这三刀的药:StateGraph 把“下一步执行谁”从业务代码里抽走;checkpoint 让状态可持久化、可续跑;条件边(conditional edge)让你声明式地写“不过就打回”。
还有个前提认知:课件生成本质是流水线,不是聊天。多步骤、强依赖、要回退——天然就是一张图的形状,硬塞进对话循环才是别扭的。
说人话:LangGraph 之于 Agent,约等于 git 之于写代码。平时你可能感知不强,翻车那天你会谢谢它。
先上结论,整条链长这样:
```text
plan → scaffold → scenes → assemble → audit ─┬─ 通过 → play → publish
└─ 不通过 → 打回 scenes(带问题清单,最多 3 次)
```
每个节点只干一件事:
audit 为什么不并进 assemble?因为验收和生成必须解耦——验收规则改了不用动生成逻辑,打回时的上下文也干净。
打回的实现:audit 把问题清单写进 state,scenes 节点读到 `retry_count` 和问题清单后,只修被点名的场景。重试 3 次还不过就停机报人工接盘。
节点签名就一句话:`def node(state) -> dict`。读什么、吐什么,白纸黑字写清楚。
| 节点 | 读什么 | 返回什么(增量) |
|---|---|---|
| plan | topic / 年级 / 课时 | `{"outline": {...}}` 大纲 |
| scaffold | outline | `{"file_tree": {...}}` 文件骨架 |
| scenes | outline + file_tree | `{"scenes": [Scene]}`(配了 reducer 累加) |
| assemble | scenes + file_tree | `{"deck_path": "..."}` 成品路径 |
| audit | deck_path + outline | `{"audit_report": {...}, "passed": bool}` |
| play | deck_path | `{"preview_url": "..."}` |
| publish | preview_url | `{"published_url": "..."}` |
两条纪律:单一职责——一个节点只干一件事,翻车了好定位;增量返回,只 return 你改过的字段。
节点会不会读到半成品 state?会。所以每个节点开头自己判空、判形态,别指望框架帮你校验。
整条链跑 3-5 分钟,前端白屏转圈,用户以为服务挂了,转头就刷新。一刷新,后台任务还在跑,token 照烧,纯浪费。
解法很朴素:节点入口/出口往事件流里丢消息,SSE 推给浏览器。
```python
def _emit(event: str, data: dict):
redis.xadd(f"agent:{run_id}:events", {"event": event, **data})
```
事件就三类,够用了:
为什么不用轮询任务表?能跑,但事件流延迟低一个量级,而且天然有序、天然适合回放。
进度条的本质不是技术,是让用户知道“它没死,它在干活”。
我第一版在节点里 `state["scenes"].append(scene)` 然后 `return state`,单测全绿。一开并行分支,直接 `InvalidUpdateError`,排查到后半夜,被坑惨了。
三个真相,踩过才懂:
```python
from typing import Annotated, TypedDict
import operator
class CourseState(TypedDict):
topic: str
outline: dict
scenes: Annotated[list, operator.add] # 配了 reducer:追加,不是覆盖
passed: bool
retry_count: int
def scenes_node(state: CourseState) -> dict:
scene = generate_one_scene(state["outline"])
return {"scenes": [scene]} # ✅ 只交增量
```
说人话:节点是交作业只交改过的那道题,不是把整本练习册重抄一遍。
为什么单测全绿、一并行就炸?因为单线程里“覆盖”和“累加”表现一模一样,脏问题只在并发下现形。
调这种 dict 状态,最笨也最好用的招:`json.dumps(state)` 打出来,丢进本站的 JSON 格式化 里展开看——哪个节点把哪个字段写坏了,层级一折叠,一眼定位。我靠这招省下的调试时间,够多睡好几个整觉。
Q1:LangGraph 必须绑 LangChain 全家桶吗?
不用。`pip install langgraph` 就能裸跑,模型调用随你挑。我只用它的图和 checkpoint,Chain 一个没沾。
Q2:一次性流水线跑完就结束,checkpoint 是不是多余?
不多余。scenes 这步又耗时又烧钱,中途挂掉从上个 checkpoint 续跑,省的是双倍 token——这行没有躺赚,省钱就是赚钱。
Q3:AI 验收 AI,会不会自己骗自己?
分指标。格式合规、答案重算、链接探活这类机器可判的硬指标,交给代码判,比人稳;让它评“这页好不好看”才是玄学,硬审美别赌运气。
评论列表 (0条)