
.part .art-main pre{background:#1E1B16;color:#E8E3D8;border-radius:10px;padding:16px 18px;overflow-x:auto;font:13px/1.7 ui-monospace,SFMono-Regular,Menlo,Consolas,monospace;margin:14px 0}
.part .art-main pre code{background:none;color:inherit;padding:0;font-size:13px}
.part .art-main code:not(pre code){background:#F1EEE7;color:#B4531B;border-radius:5px;padding:2px 6px;font:0.92em ui-monospace,SFMono-Regular,Menlo,Consolas,monospace}
.part .art-main p code{white-space:nowrap}
凌晨一点,我第 7 次打开生成好的课件:第 3 页图片裂了,第 5 页例题的答案和题干对不上。
生成这活儿 AI 干得挺欢,验收全是我人肉。每跑一次等 5 分钟,我盯一次 5 分钟,纯纯的苦力。
那晚我摔了鼠标:与其天天自己当质检员,不如写个 Agent 自己验收自己。这是这个系列的第 1 篇——从 0 到 1 搭骨架,纯工程复盘,不是什么风口突击套利的故事。
一个 while 循环加几个函数,不就能跑起来了吗?能,demo 阶段还挺爽,上生产第一天就翻车。
裸写调度循环有三个硬伤:
LangGraph 给我的正好是解这三刀的药:StateGraph 把“下一步执行谁”从业务代码里抽走;checkpoint 让状态可持久化、可续跑;条件边(conditional edge)让你声明式地写“不过就打回”。
还有个前提认知:课件生成本质是流水线,不是聊天。多步骤、强依赖、要回退——天然就是一张图的形状,硬塞进对话循环才是别扭的。
说人话:LangGraph 之于 Agent,约等于 git 之于写代码。平时你可能感知不强,翻车那天你会谢谢它。
先上结论,整条链长这样:
<p>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 推给浏览器。
def _emit(event: str, data: dict):</p>
<h2> 往 Redis Stream 丢一条,前端消费后渲染进度条</h2>
<p>redis.xadd(f"agent:{run_id}:events", {"event": event, **data})
事件就三类,够用了:
为什么不用轮询任务表?能跑,但事件流延迟低一个量级,而且天然有序、天然适合回放。
进度条的本质不是技术,是让用户知道“它没死,它在干活”。
我第一版在节点里 state["scenes"].append(scene) 然后 return state,单测全绿。一开并行分支,直接 InvalidUpdateError,排查到后半夜,被坑惨了。
三个真相,踩过才懂:
state.scenes 这种属性访问,就是键值对。你在节点里 mutate 它,并行分支根本看不见。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]} # ✅ 只交增量</p>
<h2> return state # ❌ 全量返回,并行分支写的数据直接被你抹掉</h2>
说人话:节点是交作业只交改过的那道题,不是把整本练习册重抄一遍。
为什么单测全绿、一并行就炸?因为单线程里“覆盖”和“累加”表现一模一样,脏问题只在并发下现形。
调这种 dict 状态,最笨也最好用的招:json.dumps(state) 打出来,丢进本站的 JSON 格式化 里展开看——哪个节点把哪个字段写坏了,层级一折叠,一眼定位。我靠这招省下的调试时间,够多睡好几个整觉。
Q1:LangGraph 必须绑 LangChain 全家桶吗?
不用。pip install langgraph 就能裸跑,模型调用随你挑。我只用它的图和 checkpoint,Chain 一个没沾。
Q2:一次性流水线跑完就结束,checkpoint 是不是多余?
不多余。scenes 这步又耗时又烧钱,中途挂掉从上个 checkpoint 续跑,省的是双倍 token——这行没有躺赚,省钱就是赚钱。
Q3:AI 验收 AI,会不会自己骗自己?
分指标。格式合规、答案重算、链接探活这类机器可判的硬指标,交给代码判,比人稳;让它评“这页好不好看”才是玄学,硬审美别赌运气。
评论列表 (0条)