LangGraph 入门:给传统工程师的 Agent 状态与编排指南
- 约 10 分钟读完 - 全文 3619 字LangGraph 入门:给传统工程师的 Agent 状态与编排指南
我做传统软件工程很多年,习惯了状态机、工作流引擎、消息队列这些确定性系统。第一次接触 LangGraph 时,我觉得它像是一个"有状态的 DAG 调度器",只不过节点不再是跑批任务,而是调用 LLM、工具或业务逻辑。这篇文章结合我在方糖猫 AI 短视频生成平台和 GeekCast Engine 口播剪辑项目里的经验,聊聊 LangGraph 最核心的三个概念:State、任务编排、长期记忆,以及几个让调试更轻松的小技巧。
一、LangGraph 是什么?
LangGraph 是 LangChain 生态里的一个 Agent 编排框架。它和传统工作流引擎最大的区别是:它把 LLM 调用、工具调用、业务判断都看成图里的节点,节点之间通过共享的 State 对象传递信息。
你可以把它理解成:
状态机 + LLM + 工具调用 + 持久化 = LangGraph
在方糖猫里,一个短视频生成任务的流程大致是:
- 接收用户输入(主题、风格、素材偏好);
- ReAct Agent 规划分镜和素材需求;
- 检索 RAG 本地素材库;
- 生成脚本初稿;
- 检查脚本长度与平台规范;
- 组装渲染任务并推送到 RocketMQ;
- Python Worker 拉取任务,用 Remotion + FFmpeg 渲染成片。
这个流程如果写成 LangGraph,就是一张图:每个步骤是一个节点,节点之间根据 State 里的数据决定下一步去哪。
二、State:Agent 的"工作记忆"
State 是 LangGraph 里最重要的概念。它是一个 Python 字典(或 TypedDict/Pydantic 模型),在整个图的执行过程中被所有节点共享。
一个简单的 State 定义
from typing import TypedDict, List, Optional
class VideoGenState(TypedDict):
topic: str # 用户输入的主题
style: str # 风格偏好
script: Optional[str] # 生成的脚本
materials: List[str] # 检索到的素材 ID
render_task: Optional[dict] # 渲染任务描述
error: Optional[str] # 错误信息每个节点接收 State,修改 State,然后返回更新后的 State。LangGraph 会自动合并这些更新。
为什么 State 如此重要?
在传统工作流里,我们通常把中间结果写到数据库或 Redis。LangGraph 的 State 相当于一个内存中的上下文容器,让 Agent 在单轮执行里记住:
- 用户原始需求;
- 已经做过什么;
- 下一步该做什么;
- 是否有错误需要处理。
在 GeekCast Engine 里,长视频口播稿会被切分成多个 segment。每个 segment 的处理可以是一个子图,父图的 State 里保存全局信息(比如视频总时长、字幕样式),子图的 State 里保存 segment 级别的信息(比如台词、B-roll、标题卡)。这种"全局 + 局部"的 State 设计,和我们做分布式系统时的"全局配置 + 局部任务"思想是一致的。
三、任务编排:图、节点、边
LangGraph 的编排模型很直观:
- 节点(Node):执行一个函数,通常是调用 LLM、工具或业务逻辑;
- 边(Edge):连接两个节点,控制执行流向;
- 条件边(Conditional Edge):根据 State 的内容决定下一步去哪个节点。
一个简化版的视频生成图
from langgraph.graph import StateGraph, END
builder = StateGraph(VideoGenState)
builder.add_node("plan", plan_node) # Agent 规划
builder.add_node("retrieve", retrieve_node) # RAG 检索素材
builder.add_node("write_script", script_node) # 写脚本
builder.add_node("check", check_node) # 检查规范
builder.add_node("render", render_node) # 推送渲染任务
builder.set_entry_point("plan")
builder.add_edge("plan", "retrieve")
builder.add_edge("retrieve", "write_script")
builder.add_edge("write_script", "check")
# 条件边:检查不通过则重写脚本
builder.add_conditional_edges(
"check",
lambda state: "rewrite" if state.get("error") else "render",
{"rewrite": "write_script", "render": "render"}
)
builder.add_edge("render", END)
graph = builder.compile()这个结构和传统状态机几乎一样:每个节点做完一件事,根据结果决定下一个状态。对于习惯画状态图的工程师来说,上手很快。
条件边的价值
没有条件边,LangGraph 就只是一个固定的 Pipeline。有了条件边,Agent 才能:
- 根据 LLM 判断决定调用哪个工具;
- 根据执行结果决定重试还是继续;
- 根据用户反馈决定是否需要人工介入。
在方糖猫里,素材检索节点之后,如果召回的素材数量不足,我们会走"放宽检索条件"的分支;如果素材足够,才进入脚本生成。这种动态决策能力,是 LangGraph 比简单 Pipeline 更适合 Agent 场景的原因。
四、长期记忆:从单线程到跨会话
LangGraph 的持久化分为两层,这是很多人刚开始会混淆的地方:
| 能力 | Checkpointer | Store(Memory Store) |
|---|---|---|
| 作用范围 | 单个线程(thread_id) |
跨线程、跨会话 |
| 保存内容 | 执行过程中的 State 快照 | 用户事实、交互经验、行为偏好 |
| 典型用途 | 断点续跑、人在回路、time-travel | 长期记忆、个性化 |
| 存储示例 | InMemorySaver、PostgresSaver |
InMemoryStore、PostgresStore |
Checkpointer:线程级的"短期记忆"
Checkpointer 让 LangGraph 可以在任意节点暂停、恢复、回溯。这对于"人在回路"(Human-in-the-loop)场景特别重要。
比如视频生成任务里,脚本生成节点完成后,我们可以让系统停下来,等用户确认脚本内容,确认后再继续渲染。没有 Checkpointer,这种中断-恢复就很难优雅实现。
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())Store:真正的长期记忆
Store 是 LangGraph 为长期记忆设计的存储层。吴恩达和 Harrison Chase 在《Long-Term Agentic Memory with LangGraph》课程里,把长期记忆分为三类:
-
语义记忆(Semantic Memory)
存储关于用户的稳定事实,比如"用户偏好科技感风格"、“用户通常发布横屏视频”。
-
情景记忆(Episodic Memory)
存储具体交互案例,比如"上次用户拒绝了这个 BGM,后来换成了另一首"。这些案例可以作为 few-shot 示例,帮助 Agent 在未来做出类似判断。
-
程序记忆(Procedural Memory)
存储 Agent 自身应该遵循的行为模式,也就是系统提示词或判断标准的优化版本。比如"对于这类用户,生成脚本前要先确认时长"。
在项目中怎么落地?
在方糖猫里,我们目前主要用了语义记忆:把用户确认过的风格偏好、常用素材、发布平台规格等写入 Store,下次用户再创建视频时,Agent 会先读取这些记忆,减少重复确认。
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()
# 在某个节点里写入记忆
store.put(
("user_123", "preferences"),
"style",
{"preferred_style": "tech", "aspect_ratio": "16:9"}
)
# 后续会话中读取
memories = store.search(("user_123", "preferences"), query="风格偏好")生产环境建议用 PostgresStore,并把记忆更新放到后台节点执行,避免阻塞主流程。
五、实战小技巧
1. 用图片/Mermaid 输出流程图
LangGraph 编译后的图可以用 Mermaid 或图片方式可视化。这在设计阶段特别有用,尤其是在团队沟通时。
from IPython.display import Image, display
display(Image(graph.get_graph().draw_mermaid_png()))如果你没有 Jupyter 环境,也可以输出 Mermaid 文本:
print(graph.get_graph().draw_mermaid())然后贴到支持 Mermaid 的编辑器(比如 Notion、GitHub、Draw.io)里查看。这个习惯能帮你在写代码前先确认流程没有死循环或遗漏分支。
2. 打印提示词调试
LangGraph 节点里经常会拼接复杂的 Prompt。调试时,把最终发给 LLM 的完整 Prompt 打印出来,比看代码直观得多。
def plan_node(state: VideoGenState):
prompt = build_plan_prompt(state)
print("\n===== PLAN PROMPT =====\n")
print(prompt)
print("\n===== END PROMPT =====\n")
response = llm.invoke(prompt)
return {"plan": response.content}相信我,80% 的"模型表现不好"问题,打印完 Prompt 就会发现:上下文缺失、格式混乱、或者约束互相矛盾。
3. 把节点拆小,便于追踪
一个大节点里做太多事,调试和复用都困难。我建议每个节点只负责一个明确职责:
- 一个节点做 LLM 调用;
- 一个节点做工具调用;
- 一个节点做 State 校验;
- 一个节点做错误处理。
在 GeekCast Engine 里,我们把"口播精剪"拆成了:Whisper 转写 → 时间轴分析 → 删除气口 → 重新对齐 → 输出剪辑点。每个节点都可以单独测试、单独复用。
4. 用 State 的字段做日志锚点
给 State 加上 logs 或 trace 字段,在每个节点记录关键信息。这样出问题后,可以直接看 State 快照定位到哪个节点、哪个决策点出错。
class VideoGenState(TypedDict):
# ... 其他字段
logs: List[str]
def retrieve_node(state):
materials = search_materials(state["topic"])
logs = state.get("logs", [])
logs.append(f"retrieve: found {len(materials)} materials")
return {"materials": materials, "logs": logs}5. 不要忽视错误分支
LangGraph 的图很容易画成"理想路径",但生产环境必须考虑异常:LLM 返回格式错误、工具调用超时、检索结果为空、State 校验失败。
我的做法是:每个可能失败的节点后面都跟一个错误处理节点,或者把错误信息写入 State,由条件边统一路由到重试或人工审核。
六、LangGraph 适合什么场景?
不是每个 LLM 应用都需要 LangGraph。如果你只是做一个简单的聊天机器人或单次文本生成,直接调 API 可能更轻量。
但当你遇到以下情况时,LangGraph 会很有价值:
- 任务需要多步骤决策,且步骤之间有依赖;
- 需要调用多个工具,并根据中间结果动态选择;
- 需要支持中断、人工确认、断点续跑;
- 需要长期记忆来提供个性化体验;
- 团队需要把 Agent 流程可视化、可维护。
方糖猫和 GeekCast Engine 都属于这类场景。LangGraph 让我们把原本散落在代码各处的"if-else + LLM 调用 + 工具调用"统一成了可看的、可调的、可持久化的图。
七、总结
LangGraph 不是另一个"让代码更复杂"的框架,而是把 Agent 执行过程显性化的工具。它的核心概念其实和传统工程很像:
- State 就是工作记忆或上下文容器;
- 图编排 就是状态机加上动态分支;
- Checkpointer 提供会话内的断点续跑;
- Store 提供跨会话的长期记忆。
对于传统工程师来说,最值得培养的能力不是背 API,而是:
- 如何把业务问题拆成清晰的节点和状态;
- 如何设计 State,让它既能表达业务语义,又便于调试;
- 什么时候用短期记忆,什么时候用长期记忆;
- 如何通过可视化和小技巧降低系统的认知复杂度。
如果你正在用 LangGraph 做项目,我的建议是:先画图,再写代码;先打印 Prompt,再调模型;先把主流程跑通,再逐步加入记忆和异常处理。
我在 ftmao.com/blog 持续记录 AI Agent、LangGraph 和多模态生成相关的工程实践,欢迎关注和交流。