智能体编排框架横评(2):LangGraph
本文总结LangGraph 入门:先跑通一个最小的图,再逐个讲 State、Node、Edge、检查点和 interrupt,最后一步步写出建单前等员工确认的报销图。
智能体编排框架横评 第 2 站 · 上一站:LangChain · 下一站:厂商 Agent SDK · 全程:总览
LangGraph 是什么
LangGraph 是 LangChain 团队做的一个 Python 库(也有 JS 版),用来把 Agent 的执行流程写成一张图。
- 图里的每个节点是一个普通函数,负责做一件事,比如查制度、调模型、建工单。
- 节点之间的边规定先做什么、后做什么、什么条件下走哪条路。
- 所有节点读写同一份状态,状态就是一个字典。
- 配上检查点以后,图每走一步都会把状态存下来,所以可以中途暂停、等人确认、进程重启后接着跑。
它解决的问题是:流程的步骤、分支和审批点由业务决定,不能全交给模型临场发挥,而且流程可能要跑很久、中间要停下来等人。
和第 1 站的关系:LangChain 的 create_agent 给你一个现成的"模型调工具"循环,它自己就是编译成一张 LangGraph 图来跑的。只是模型加几个工具问答,用 create_agent 就够了;要自己决定流程怎么走,才需要直接写 LangGraph。
安装
pip install -U langgraph
langgraph 自带一个内存版检查点。要把检查点存进数据库,另装对应的包:
pip install -U langgraph-checkpoint-sqlite # SQLite
pip install -U langgraph-checkpoint-postgres # PostgreSQL
本文的代码按 LangGraph 1.x 的 Python 官方文档写,是示意代码,没有运行过。
第一个图
先写一个只有一个节点的图:输入一段文字,节点把它变成大写。
hello_graph.py 文件代码:
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
class State(TypedDict): # 1. 定义状态里有哪些字段
text: str
def shout(state: State): # 2. 定义节点:读状态,返回要更新的字段
return {"text": state["text"].upper()}
builder = StateGraph(State) # 3. 用状态类型创建一张图
builder.add_node(shout) # 加节点,节点名默认就是函数名
builder.add_edge(START, "shout")
builder.add_edge("shout", END)
graph = builder.compile() # 4. 编译后才能运行
print(graph.invoke({"text": "hello langgraph"}))
输出结果为:
{'text': 'HELLO LANGGRAPH'}
写任何 LangGraph 程序都是这四步:定义状态,写节点函数,加节点和边,编译后调用 invoke。START 和 END 是两个特殊节点,分别表示图的入口和出口。
基本概念
State:节点共享的数据
State 是一个 TypedDict,列出图里要传递的字段。节点函数收到的是当前完整状态,返回的是要更新的那几个字段,没返回的字段保持不变。
class State(TypedDict):
question: str
answer: str
def answer_node(state: State):
return {"answer": f"收到:{state['question']}"} # 只更新 answer,question 不动
默认情况下,节点返回的新值会覆盖旧值。如果想让多个节点往同一个列表里追加,要给字段加一个 reducer,告诉 LangGraph 怎么合并:
from typing import Annotated
import operator
class State(TypedDict):
evidence: Annotated[list[str], operator.add] # 新值和旧值用 + 拼起来
def a(state): return {"evidence": ["制度:打车标准 80 元"]}
def b(state): return {"evidence": ["发票:INV-102 金额 118.5 元"]}
a 和 b 都跑完以后,evidence 里就有两条。对话消息列表有一个专用的 reducer add_messages,用法相同。
Node:干活的函数
节点就是普通的 Python 函数,参数是状态,返回值是要更新的字段字典。用 add_node 加进图里:
builder.add_node(answer_node) # 节点名是 "answer_node"
builder.add_node("answer", answer_node) # 也可以自己起名
节点里可以调模型、调工具、查数据库,LangGraph 不限制你在里面写什么。
Edge:节点之间怎么走
普通边表示做完 A 一定去 B:
builder.add_edge(START, "a")
builder.add_edge("a", "b")
builder.add_edge("b", END)
条件边根据状态决定下一步去哪。写一个路由函数,返回下一个节点的名字:
def route(state: State):
return "manual_review" if "超标" in state["answer"] else END
builder.add_conditional_edges("judge", route)
并行:从同一个节点连出两条普通边,两个节点会同时执行。想等两路都完成再往下走,把它们写成列表:
builder.add_edge(START, "policy")
builder.add_edge(START, "invoice")
builder.add_edge(["policy", "invoice"], "confirm") # 两路都到了才跑 confirm
并行的两个节点如果写同一个字段,这个字段必须有 reducer,否则运行时会报 INVALID_CONCURRENT_GRAPH_UPDATE。
节点也可以不靠边,直接在返回值里说下一步去哪,用的是 Command:
from typing import Literal
from langgraph.types import Command
def judge(state: State) -> Command[Literal["create", "__end__"]]:
over = state["amount"] > 80
return Command(update={"over": over}, goto="create" if over else END)
返回类型注解 Command[Literal[...]] 要写上,LangGraph 靠它知道这个节点可能去哪些节点。
Checkpointer 和 thread_id:记住跑到哪了
前面的图每次 invoke 都从零开始,跑完什么也不留。编译时传一个 checkpointer,图每走一步都会把状态存一份快照,叫检查点。快照按 thread_id 分组,同一个 thread_id 就是同一次会话。
from langgraph.checkpoint.memory import InMemorySaver
graph = builder.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "t-123"}}
graph.invoke({"text": "hello"}, cfg)
print(graph.get_state(cfg).values) # 读出这个线程最新的状态
InMemorySaver 存在内存里,进程退出就没了,只适合开发调试。上线时换成 SqliteSaver 或 PostgresSaver(第一次用 PostgresSaver 要调一次 .setup() 建表)。
有了检查点,才谈得上暂停、恢复和查看历史,下面的 interrupt 就建立在它上面。
Interrupt:停下来等人
在节点里调用 interrupt(值),图会在这里暂停:当前状态写进检查点,invoke 返回,你的程序可以把这个值展示给用户。等用户做了决定,用 Command(resume=决定) 再调一次 invoke,图接着往下跑,interrupt() 的返回值就是用户的决定。
from langgraph.types import interrupt, Command
def confirm(state: State):
ok = interrupt({"question": "确认提交这张报销单吗?"})
return {"confirmed": ok}
graph = builder.compile(checkpointer=InMemorySaver()) # 必须有 checkpointer
cfg = {"configurable": {"thread_id": "t-123"}}
out = graph.invoke({"text": "..."}, cfg, version="v2")
print(out.interrupts) # 停在 confirm,这里能看到要问用户的内容
graph.invoke(Command(resume=True), cfg, version="v2") # 用户点了确认,接着跑
version="v2" 让 invoke 返回一个对象,中断信息放在 .interrupts 属性里。恢复时要用同一个 thread_id,LangGraph 才知道接着哪一次往下跑。
有一点必须记住:恢复时,调用 interrupt 的那个节点会从头重新执行,不是从 interrupt 那一行继续。所以 interrupt 前面的代码会跑两次。查数据跑两次没关系,建单、发通知这类有副作用的操作不要放在 interrupt 前面。
动手:一步步写报销图
例子:员工问"我上周出差打车超标了,还能报吗?能的话帮我提报销工单"。我们要做的流程是:
- 同时查差旅制度和发票(INV-102 金额 118.5 元,标准 80 元);
- 两路结果都拿到后,停下来请员工确认;
- 员工确认就建单,拒绝就结束。
模型判断的部分先省略,重点看流程、并行和中断怎么写。search_policy、get_invoice、create_ticket 是三个已有的工具函数。
第 1 步:定义状态
from typing import Annotated, TypedDict
import operator
class S(TypedDict):
evidence: Annotated[list[str], operator.add] # 两个节点会同时往里写,所以要 reducer
ticket_id: str
第 2 步:查制度和查发票两个节点
def policy(s: S):
return {"evidence": [search_policy("差旅 打车 标准")]}
def invoice(s: S):
return {"evidence": [get_invoice("INV-102")]}
第 3 步:确认节点
确认节点里只放 interrupt 和路由,不放任何有副作用的代码,因为恢复时它会从头再跑一遍。
from typing import Literal
from langgraph.graph import END
from langgraph.types import interrupt, Command
def confirm(s: S) -> Command[Literal["create", "__end__"]]:
ok = interrupt({"tool": "create_ticket", "invoice": "INV-102", "over": 38.5})
return Command(goto="create" if ok else END)
over: 38.5 是 118.5 减去 80 得到的超标额,这里直接写出来做示意。
第 4 步:建单节点,带幂等键
建单是有副作用的操作。进程如果在调用 create_ticket 之后、检查点写入之前崩溃,恢复时这个节点会再跑一次。给它一个固定的幂等键,工单系统看到同一个键就返回原来那张单,不会建两张。
from langgraph.runtime import Runtime
def create(s: S, runtime: Runtime):
key = f"{runtime.execution_info.thread_id}:INV-102"
return {"ticket_id": create_ticket("INV-102", idempotency_key=key)}
键由 thread_id 和发票号拼成,每次重跑算出来都一样。不要在节点里用 uuid4() 现生成,那样每次重跑都是一个新键。
第 5 步:连边并编译
from langgraph.graph import StateGraph, START
from langgraph.checkpoint.memory import InMemorySaver
g = StateGraph(S)
for fn in (policy, invoice, confirm, create):
g.add_node(fn)
g.add_edge(START, "policy")
g.add_edge(START, "invoice") # policy 和 invoice 并行
g.add_edge(["policy", "invoice"], "confirm") # 两路都到才确认
app = g.compile(checkpointer=InMemorySaver()) # 上线换 PostgresSaver
confirm 去 create 还是 END 由它返回的 Command 决定,所以这里不用再加边。
第 6 步:运行,然后恢复
cfg = {"configurable": {"thread_id": "t-123"}}
out = app.invoke({"evidence": []}, cfg, version="v2")
print(out.interrupts) # 停在 confirm,把要确认的内容给员工看
# 员工点了确认(可以是几分钟后,也可以是另一个进程)
app.invoke(Command(resume=True), cfg, version="v2")
下面这张图按步骤演示了这次运行:第 1 步两个查询并行,第 2 步停在 confirm,恢复后 confirm 从头重跑,第 3 步建单。
- 第 1 超步:policy 和 invoice 同时跑,结果都写进 evidence
- 两路都完成才进 confirm,节点里调用 interrupt 停下
- 员工确认,confirm 跳到 create 建单
- 拿到单号,回复员工
- 员工拒绝时 confirm 直接去 END,不建单
第 1 超步的细节:
- Plan:第一步选订阅输入通道的节点,这里是 policy 和 invoice
- Execution:两个节点并行,互相看不到对方写了什么
- Update:两条结果用 operator.add 合并,没挂 reducer 会报错
- 这一步写一份检查点;有一个节点失败,整步都不生效
第 2 超步:confirm的细节:
- confirm 节点调用 interrupt,把要确认的内容交出去
- 暂停靠抛异常实现,当前状态写进检查点
- invoke 返回,out.interrupts 里是待确认的内容
- 员工确认后,用同一个 thread_id 发 Command(resume=True)
- confirm 从第一行重新执行,interrupt 这次直接返回 True
- 节点返回 Command(goto=create),下一个超步去建单
第 3 超步:create的细节:
- 幂等键用 thread_id 拼发票号,两样都已经落盘
- 调建单接口时带上这个键
- 接口调完进程崩了,恢复时 create 节点会再跑一次
- 重跑时算出来的还是同一个键
- 接口按键去重,最后只有一张单;用 UUID 就要先单独落盘
进阶
上面的内容够写出一个能用的图了。下面几节讲图运行时的细节,遇到问题时再回来看。
图是怎么一步步跑的:超步
LangGraph 把一次运行分成若干个超步(super-step)。每个超步有三个阶段:
- Plan:选出这一步要执行的节点。第一步选连在
START后面的节点,之后选上一步有更新、并且连到它们的节点。 - Execution:并行执行这些节点。执行过程中,各节点的写入互相看不到。
- Update:把这一步所有节点的写入合并到状态里,有 reducer 的字段按 reducer 合并。
报销图里,policy 和 invoice 在第 1 个超步,confirm 在第 2 个,create 在第 3 个。检查点在每个超步结束时写入。
由此可以理解两条规则:
- 并行节点写同一个字段必须有 reducer,因为它们的写入在同一个 Update 阶段合并。并行写入的先后顺序不固定,reducer 最好不依赖顺序。
- 同一个超步里有一个节点抛异常,这一步的所有写入都不生效。已经成功的节点结果会单独存下,恢复时只重跑失败的那几个。
recursion_limit 限制的是超步数,超过会抛 GraphRecursionError。它要放在 config 的顶层,比如 {"recursion_limit": 50, "configurable": {...}},放进 configurable 里不生效。默认值很大,模型陷入循环时会产生大量调用,建议自己设一个上限。
检查点什么时候写:durability
invoke 和 stream 可以传 durability 参数,控制检查点写入的时机:
| 模式 | 什么时候写 | 代价 |
|---|---|---|
"async"(默认) |
执行下一步的同时在后台写 | 进程崩溃可能丢掉最近几步,那几步会重跑 |
"sync" |
下一步开始前写完 | 每步多一次同步写 |
"exit" |
只在图结束、出错或中断时写 | 中途崩溃无法从中间恢复 |
不管哪种模式,它只管 LangGraph 自己的状态,不保证外部系统只被调用一次。节点重试和恢复都会把整个节点再跑一遍,所以工单、支付这类副作用一律要带幂等键,见 可靠执行:状态、幂等与恢复。
查看和分叉历史
检查点按线程串成历史,可以回到过去的某一步:
history = list(app.get_state_history(cfg)) # 新的在前
snap = history[2] # 找到想回去的那一步
app.invoke(None, snap.config) # 回放:从那一步之后重新执行
fork = app.update_state(snap.config, {"evidence": ["改过的证据"]})
app.invoke(None, fork) # 分叉:改了状态再往下跑,原历史不动
回放时模型调用、API 请求、interrupt 都会再来一遍。
Store:跨线程的数据
State 和检查点只属于一个线程。员工的报销偏好、过去的判定这类跨会话的数据,放进 Store:
from langgraph.store.memory import InMemoryStore
app = g.compile(checkpointer=InMemorySaver(), store=InMemoryStore())
def create(s: S, runtime: Runtime):
runtime.store.put(("user", "u-001"), "last_invoice", {"id": "INV-102"})
...
Store 按 (namespace 元组, key) 存字典。线程内的放 State,跨线程的放 Store,这和 记忆 里短期记忆与长期记忆的划分一致。
多个 Agent 怎么组合
LangGraph 提供三种组合手段:
- 子图:一个编译好的图可以直接作为另一个图的节点。
Send:条件边返回一组Send(节点名, 参数),运行时按数量动态扇出,结果用 reducer 汇合。适合"拆成 N 个子任务分别处理再汇总"。Command(goto=..., graph=Command.PARENT):子图里的节点直接跳回父图的另一个节点,handoff 就是这么实现的。
官方现在推荐的多 Agent 写法是把子 Agent 包成工具让主 Agent 调用,只有流程完全自定义时才直接写图。多 Agent 框架的横向比较见 智能体编排框架横评(4):多智能体框架。
流式输出和调试
stream 可以边跑边拿结果,stream_mode 决定拿什么:
for chunk in app.stream({"evidence": []}, cfg, stream_mode="updates"):
print(chunk) # 每个节点跑完,打印它更新了哪些字段
常用的几种:values 是每步完整状态,updates 是每步增量,messages 是模型逐 token 输出,custom 是节点里用 get_stream_writer() 自己发的内容。
调试可以接 LangSmith:设环境变量 LANGSMITH_TRACING=true 和 API key,每次运行的节点、输入输出都能在网页上看到。本地运行 langgraph dev 会起一个开发服务,在 LangSmith Studio 里单步看图。也可以通过 OpenTelemetry 接到自己的链路追踪系统。更多见 可观测性:追踪一条任务的代价。
另一种写法:Functional API
除了画图,LangGraph 还支持用普通函数写流程:
from langgraph.func import entrypoint, task
@task
def check_invoice(no: str):
return get_invoice(no)
@entrypoint(checkpointer=InMemorySaver())
def reimburse(no: str):
inv = check_invoice(no).result()
ok = interrupt({"invoice": inv})
return create_ticket(no) if ok else None
@entrypoint 标记入口函数,里面用 if、for 写流程;有副作用或结果不固定的调用包进 @task。恢复时整个函数从头执行,已经完成的 task 直接从检查点取结果。所以 task 之外不能有副作用,也不要在部署新版本时调换 task 的调用顺序,否则正在等待中的会话恢复时可能对不上。
两种写法底层是同一个引擎,可以混用。流程要画出来给别人看、分支多,用 Graph API;改造已有代码、流程是线性的,用 Functional API 更省事。
和 MCP 的关系
LangGraph 本身不处理 MCP。MCP 工具先通过 LangChain 的适配层变成普通工具,再在节点里调用,这部分在 智能体编排框架横评(1):LangChain 1.x,MCP 本身见 技能、MCP 与集成。图部署到 LangSmith 的 Agent Server 以后,可以通过 /mcp 端点把整个 Agent 暴露成一个 MCP 工具。
常见问题
恢复后建了两张单。 最常见的原因是在同一个节点里先建单再 interrupt。恢复时节点从头跑,建单又执行一次。把建单放到 interrupt 之后的单独节点里,并带幂等键。
加了一条并行边就报 INVALID_CONCURRENT_GRAPH_UPDATE。 两个节点在同一步写了同一个字段,而这个字段没有 reducer。按"State"一节给它加 Annotated[..., reducer]。
interrupt 不生效,图没停下来。 检查两点:编译时有没有传 checkpointer;interrupt 有没有被 try/except 包住。interrupt 是靠抛一个特殊异常实现暂停的,裸的 except 会把它吞掉。
一个节点里调了多次 interrupt,恢复的值对错位了。 多次 interrupt 按出现顺序和 resume 值配对。不要有条件地跳过其中一个,也不要在 while 循环里反复 interrupt。需要多轮审批时,用条件边回到这个节点。
设了 recursion_limit 没用。 它要放在 config 顶层,不能放进 configurable。
状态里的对象存不进检查点。 默认序列化器支持 LangChain 消息、datetime、enum 等常见类型,自定义类和 DataFrame 不行。状态里尽量只放能转成 JSON 的数据。
检查点数据库要当敏感数据保护。 加载检查点时会反序列化,能写检查点库的人就可能让服务执行任意对象构造。LangGraph 出过这类安全公告,升级到修复版本后,还要按官方说明打开严格反序列化模式(环境变量 LANGGRAPH_STRICT_MSGPACK=true)。见 安全边界:让权限落到工具层。
版本说明
LangGraph 1.0 在 2025-10-22 发布,官方承诺 2.0 之前不做破坏性变更。本文按 2026-11 时的 1.2 系列文档写。和旧教程对照时注意三点:
invoke(..., version="v2")是 1.1 加的,返回带.interrupts的对象。旧教程里从结果字典的__interrupt__键取中断,仍然能用,但官方计划以后移除。- 检查点时机以前用
checkpoint_during=True/False控制,现在换成了durability。 langgraph-supervisor包不再积极维护,新项目按官方迁移文档,把子 Agent 包成工具。
和手写循环的对照
跟读 learn-claude-code:17 个 harness 机制怎么挂到一个循环上 系列从零写了一个 harness,跟读 learn-claude-code(1):循环内核 里的 while 循环、工具调用配对、权限检查、上下文压缩都是自己写的。LangGraph 接管的是另一部分:步与步之间的状态怎么保存、怎么恢复。
| 机制 | 手写 harness | LangGraph |
|---|---|---|
| 保存什么 | s10 每个任务一个 JSON 文件,带 blockedBy 依赖 |
每个超步一份检查点,按线程串成历史 |
| 崩溃后从哪继续 | 读任务文件,重新认领没做完的任务 | 读最新检查点,重跑没完成的节点 |
| 暂停等人 | 循环外的权限确认,或把任务状态挂起 | interrupt() 暂停,Command(resume) 恢复 |
| 恢复后怎么重放 | s16 脚本从头重跑,journal 里有的结果直接返回 | 节点从头重跑;Functional API 是入口函数从头重跑,已完成的 task 取缓存 |
| 外部副作用 | 先落盘再执行,承认至少一次 | 同样是至少一次,去重靠你自己的幂等键 |
两边都没有保存调用栈,恢复都是"从头重跑,已完成的部分取缓存"。固定编排与目标闸门:Workflow 运行时和 Goal Loop 的 Workflow 运行时和 LangGraph 的 Functional API 几乎是同一个思路,代价也一样:缓存之外的代码会再执行一次。
任务图、后台与定时:让工作撑过一轮对话 的任务文件是一份工作清单,可以由多个 worker 认领。LangGraph 的检查点是一次执行的快照,绑定在一个线程上。想要多个 worker 抢任务,要用 Agent Server 的任务队列,或者在图外自己维护。
LangGraph 没管的部分:工具调用和结果的配对、消息裁剪和摘要、工具权限,由 LangChain 的 ToolNode、create_agent 和中间件负责;上下文压缩(对应 计划、子代理与上下文预算)要自己写节点或用第 1 站的摘要中间件。
自测
一个 LangGraph 程序最少要写哪几步?
参考:定义 State(TypedDict),写节点函数(读状态,返回要更新的字段),用 StateGraph 加节点和边(至少有一条从 START 出发、一条到 END),compile() 后调用 invoke。
两个并行节点都往 evidence 里写一条结果,为什么默认会报错,怎么修?
参考:两个节点在同一个超步执行,写入在同一个 Update 阶段合并。默认字段一步只接受一个值,第二个写入触发 INVALID_CONCURRENT_GRAPH_UPDATE。给字段加 reducer,例如 Annotated[list[str], operator.add]。
不传 checkpointer 能用 interrupt 吗?恢复时为什么要用同一个 thread_id?
参考:不能。暂停时要把状态写进检查点,恢复时按 thread_id 找到那一份检查点接着跑。换一个 thread_id 就是一次新的运行。
员工在 confirm 中断时点了确认,confirm 里 interrupt 前面有一行发通知给财务的代码,会发生什么?
参考:恢复时整个节点从头重跑,通知会再发一次。把通知挪到 interrupt 之后,或者拆成单独的节点,或者给通知带幂等键。
durability 的三种模式分别在什么时候写检查点?哪种模式下进程中途崩溃没法从中间恢复?
参考:async(默认)在执行下一步的同时后台写;sync 在下一步开始前写完;exit 只在图结束、出错或中断时写,中途崩溃没法从中间恢复。三种都不保证外部系统只被调用一次。
想回到查完发票、还没确认的那一步,改一下证据再试一次,用什么 API?原来的历史会被覆盖吗?
参考:用 get_state_history 找到那一步的快照,update_state(snapshot.config, values) 分叉出新检查点,再 invoke(None, fork_config)。原历史保留,分叉是一条新分支。
参考资料
以下链接均查阅于 2026-11。
- LangGraph 快速入门
- Graph API 概念
- Use the graph API
- Graph API 控制流(并行、defer、Send)
- Persistence
- Interrupts
- Use time travel
- Fault tolerance
- Memory stores
- Subgraphs
- Streaming
- Functional API
- LangGraph runtime(Pregel)
- INVALID_CONCURRENT_GRAPH_UPDATE 错误说明
- LangGraph v1 发布说明
- 从 langgraph-supervisor 迁移
- Agent Server 的 MCP 端点
- GHSA-g48c-2wqr-h844 检查点反序列化公告
连接
- 智能体编排框架横评:同一个报销 Agent 换六种写法:系列总览,本站是第 2 站。
- 智能体编排框架横评(1):LangChain 1.x:上一站,
create_agent本身编译成 LangGraph 图运行,工具与 MCP 接入在那一层。 - 智能体编排框架横评(3):厂商 Agent SDK:下一站,厂商 SDK 走模型驱动的循环,和这里的显式图对照看。
- 智能体编排框架横评(4):多智能体框架:多 Agent 框架横比。
- Agent 核心:面试版的 LangGraph 问答,可以和本站的基本概念对照着复习。
- 可靠执行:状态、幂等与恢复:检查点只保证状态可恢复,副作用的幂等和对账按这里做。
- 人在回路:
interrupt是实现手段,审批的产品设计在这里。 - 规划与状态:图的结构就是显式写下的计划,状态放在模型之外。
- 记忆:State 对应线程内短期记忆,Store 对应跨线程长期记忆。
- 可观测性:追踪一条任务的代价:stream mode 与 tracing 之外的可观测性原则。
- 安全边界:让权限落到工具层:检查点存储会被反序列化,要按安全边界保护。
- 跟读 learn-claude-code:17 个 harness 机制怎么挂到一个循环上:手写 harness 系列,对照 LangGraph 接管与未接管的部分。
- 任务图、后台与定时:让工作撑过一轮对话:任务文件与先落盘后执行,对照检查点。
- 固定编排与目标闸门:Workflow 运行时和 Goal Loop:journal 回放,对照 interrupt 与 Functional API 的从头重放。
- 计划、子代理与上下文预算:上下文压缩,LangGraph 不内置。