AI Agent 开发(06):LangGraph 入门

QuibblerAgentQuibblerAgent 2026-10-25 约 17 分钟 1 次阅读

前五篇把地基打完了:第二篇用 create_agent 三十行跑通会调工具的 Agent,第 5 篇又把它的内核拆开——那段不到 20 行的 while 循环。工具调用这层协议吃透之后,该进引擎室了。从本篇起,系列进入第二模块"LangGraph 核心",主角换成 create_agent 背后的编排引擎。代码基于 langgraph 1.2.x:1.0 版于 2025 年 10 月发布,API 宣告稳定;1.2 线补上了节点超时、错误恢复与优雅关停等生产级能力,版本时间线可在 GitHub 查。现在上车,踩的正好是稳定期。

本篇是模块二的第一块砖。全文按"为什么需要图 → 三要素拆解 → 第一个图 → 编译与运行 → 条件边预告 → 两套 API 选型"的顺序展开,看完你会亲手搭出第一张 StateGraph,并弄明白第二篇里 create_agent 返回的那个"编译好的图"究竟从哪来。

1、为什么需要图

链(Chain)是 LCEL(LangChain Expression Language)时代的基本形态:输入顺次流过一串步骤,每步加工一次,最后交出输出。它是一条流水线,擅长翻译、摘要、固定步骤的 RAG 这类"一步接一步"的任务。流水线的铁律是物料只能往前走——想"回上一步重试"或"中途换路线",链就无能为力了。

Agent 偏偏不是流水线。第 5 篇画过它的真身:决策 → 行动 → 观察 → 再推理,是一个圈。单个 Agent 用 while 循环就能跑,可需求一旦长出下面这些形状,循环就开始失控:

  • 多阶段:检索、写稿、自查、重写,环里套环,if-else 堆成面条代码
  • 人工介入:写完要先审批再发,得把循环拦腰斩断
  • 崩溃恢复:状态全在局部变量里,进程一死全部归零
  • 并行执行:两个工具本可同时跑,while 只能让它们排队

举个具体场景:客服 Agent 要"查订单 → 判断金额 → 超限额转人工审批 → 发通知"。用 while 写,四步是四段 if;再来一个"审批不通过重来"的分支,循环体直接翻倍——而且没人能从代码上一眼看出这个流程长什么样。

根因只有一句:while 循环把流程结构藏进了函数体的执行顺序里,结构既看不见、存不下来,也没法断点续跑。LangGraph 的解法是把结构本身升为第一公民:每一步声明成节点(Node),"下一步去哪"声明成边(Edge),节点与边组成一张有向图,引擎照图执行。

链是单行道,图是交通网——环岛、岔路口、起点站与终点站都修在路上,而不是写在函数体里。

交通网修好之后,循环是环、分支是岔路口、审批是路障检查站,它们都成了图上看得见的结构。第 5 篇那段 while 循环,从本篇起正式退役。

2、StateGraph 三要素

StateGraph 把一张图拆成三个正交概念:State 存数据、Node 干活、Edge 定路线。逐一拆开看。

2.1、State 共享内存

State 是全图共享的一块内存,相当于车间白板:每个节点都能读它,也都能往上写。声明就是普通 TypedDict:

from typing import TypedDict

class State(TypedDict):
    messages: list[str]   # 全图共享的消息列表

TypedDict 只声明结构、不含数据:真正的值在 invoke 时传入,节点之间靠这块白板传递中间结果。字段类型随意,本篇用最朴素的 list[str];第 7 篇会换成带 reducer(合并函数)的消息列表,那是 State 设计的主菜。

2.2、Node 执行单元

Node 是干活的最小单元,本体就是一个普通 Python 函数:入参拿到整个 State,返回值是要写回白板的更新。示例如下:

def greet(state: State) -> dict:
    # 读旧值、拼新值;返回的 dict 就是本次写回的更新
    return {"messages": state["messages"] + ["你好"]}

注意函数返回的是部分更新而不是就地修改:引擎拿到 dict 后按键合并进 State,没提到的键保持原样,提到的键直接替换旧值——所以列表拼接要自己做。"按什么规则合并"由 reducer 决定,追加还是替换,第 7 篇展开。节点只算新值、不动入参,这种纯函数风格是 LangGraph 能做快照与重放的前提。

2.3、Edge 路线声明

Edge 声明"这个节点执行完,下一站去哪"。它不是函数调用,而是一句路由声明;START 与 END 是两个虚拟节点,分别代表图的入口与出口:

  • add_edge(START, "greet"):从入口进入 greet
  • add_edge("greet", "farewell"):greet 完去 farewell
  • add_edge("farewell", END):farewell 完收工

三要素各归其位,汇总成一张表:

要素 本体 一句话职责
State TypedDict 或 Pydantic 模型 共享内存,节点间传数据
Node 普通 Python 函数 干一步活,返回状态更新
Edge add_edge 声明 定路线,决定下一站

3、第一个图

三要素凑齐,拼一张最小的图:greet 打招呼、farewell 道别,两个节点一条线。完整代码如下:

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    messages: list[str]

def greet(state: State) -> dict:
    return {"messages": state["messages"] + ["你好,我是第一个节点"]}

def farewell(state: State) -> dict:
    return {"messages": state["messages"] + ["再见,全图跑完了"]}

builder = StateGraph(State)
builder.add_node("greet", greet)          # 节点名,函数本体
builder.add_node("farewell", farewell)
builder.add_edge(START, "greet")          # 入口 → greet
builder.add_edge("greet", "farewell")     # greet → farewell
builder.add_edge("farewell", END)         # farewell → 出口
graph = builder.compile()                 # 图纸 → 可执行单元

result = graph.invoke({"messages": []})   # 传初始状态
print(result["messages"])

逐段看:StateGraph(State) 建图并绑定状态结构;add_node 第一个参数是节点名(图结构里的标识符),第二个是函数本体;三条 add_edge 铺出 START → greet → farewell → END 的路线。invoke 收下初始状态,一路跑到底,返回终态。输出如下:

['你好,我是第一个节点', '再见,全图跑完了']

两个节点各执行一次,各自往 messages 追加一句话。你写的是结构,引擎负责执行——这就是声明式编排的全部含义,后面所有复杂 Agent 都是往这个骨架上加节点、加边。

环境沿用第二篇的 uv 项目(langgraph 当时已随 uv add 装好):把代码存成 first_graph.py,执行 uv run first_graph.py 即可复现。这个例子不花一分钱 API 费——节点是纯 Python 函数、没有模型参与,最适合先把图的肌肉记忆练熟。

4、compile 与 invoke

builder 和 graph 是两种东西:前者是图纸,还能继续 add_node、add_edge;后者是可执行单元,结构冻结,只能 invoke、stream。编译这一步不是仪式感,它至少做三件事:校验结构(没有入口边、节点悬空会直接报错),把声明翻译成可调度单元,再返回一个标准 Runnable。漏写 add_edge(START, ...) 或某个节点没有出边,编译期就会报错并指明问题节点;同样的问题在 while 循环里只会变成运行期的逻辑错误,得靠打印日志人肉排查。把错误从运行期提前到编译期,是声明式编排最实在的一笔收益。

"标准 Runnable"是关键字:编译产物可以 invoke 跑到底、stream 边跑边吐事件(第 13 篇)、batch 并行跑多个输入,还会自动进 LangSmith 追踪。第二篇里 create_agent 返回的正是这种编译产物,现在你知道它从哪来了。compile 还能挂参数,比如 checkpointer,第 12 篇讲持久化时用上。

图还有个被低估的属性:可可视化。get_graph() 拿到结构描述,一行导出 mermaid 文本:

print(graph.get_graph().draw_mermaid())

输出可以直接贴进任意 mermaid 渲染器:

%%{init: {'flowchart': {'curve': 'linear'}}}%%
graph TD;
	__start__([__start__]):::startEnd
	greet([greet])
	farewell([farewell])
	__end__([__end__]):::startEnd
	__start__ --> greet
	greet --> farewell
	farewell --> __end__
	classDef startEnd fill:#bfb,stroke:#333,stroke-width:2px;

要 PNG 用 draw_mermaid_png(output_file_path="graph.png"),默认走 mermaid 在线渲染,内网环境需装 pyppeteer 改本地渲染;想在终端里直接看用 draw_ascii(),先 uv add grandalf。复杂 Agent 动辄十几个节点,把出图贴进 README 或评审材料,比口述结构高效得多——图即文档。可视化与 Graph API 的完整能力清单见 Graph API 概览。

5、从固定边到条件边

add_edge 铺的是固定边,下一站写死。Agent 要的是岔路口:"模型还想调工具就去工具节点,不想调了直接收工"。LangGraph 用 add_conditional_edges 表达:传入一个节点和一个路由函数,路由函数返回下一站的名字。第 9 篇手写 ReAct Agent 的核心骨架,提前看一眼:

def should_continue(state: State) -> str:
    if state["needs_tool"]:   # 示意:状态里带个标记字段
        return "tools"        # 去工具节点
    return END                # 否则收工

builder.add_conditional_edges("agent", should_continue)

对照第 5 篇的 run_agent:那里"没有 tool_calls 就 return"写在循环体里,这里变成路由函数的一行返回值,控制流从代码挪到了图上。更妙的是环:再给 tools 铺一条 add_edge("tools", "agent") 指回上游,第 5 篇手写的 while 循环就变成图里一个肉眼可见的环。引擎照环执行,靠 recursion_limit(默认 25 步)兜底防死循环。条件边、Command 路由与循环控制是第 8 篇的主菜,这里只立个路标。

6、Functional API 与选型

讲完 Graph API,必须交代它的孪生兄弟。LangGraph 1.x 有两套构建入口:Graph API(本篇的 StateGraph)与 Functional API(langgraph.func 的 @entrypoint 与 @task)。后者完全不碰图结构,用普通函数写工作流:

from langgraph.func import entrypoint, task

@task
def greet(name: str) -> str:
    return f"你好,{name}"

@entrypoint()
def workflow(name: str) -> str:
    first = greet(name).result()   # 任务完成即落检查点
    return first + ",欢迎来到 Functional API"

@task 标记可检查点的执行单元,@entrypoint 标记工作流入口,函数体顺序就是执行顺序,.result() 取任务结果——workflow.invoke("开发者") 会输出"你好,开发者,欢迎来到 Functional API"。底层与 Graph API 共享同一个引擎:持久化、中断恢复、流式输出两套写法全都支持,Functional API 不是低配版,只是另一种表达。断点恢复的行为也值得知道:恢复执行时入口函数从头重放,已完成任务直接从检查点取回结果、不再重算——第 5 篇循环里"崩溃归零"的老问题,在这里被检查点悄悄治好。

两套怎么选?按场景对号入座:

场景 更合适 理由
逻辑线性、函数已存在 Functional API 加两个装饰器即得持久化
多分支、多入口、要出图评审 Graph API 结构显式,可视化友好
复杂 Agent 与多 Agent 系统 Graph API 节点边界清晰,路由一目了然
给存量代码补检查点能力 Functional API 改造量最小

官方的 选型指南结论同向:清楚整张图长什么样、要精确控制路由,用 Graph API;流程本质是顺序代码、只想要 LangGraph 的持久化与人机协同,用 Functional API。本系列主线走 Graph API——create_agent 等预构建组件的底层就是图,理解图,才能理解它们暴露的每个参数。Functional API 的深入用法见 Functional API 文档。

7、总结

把本篇收束成一句话:图把流程结构从函数体的执行顺序里解放出来。State 是共享白板,Node 是干活函数,Edge 是路线声明;compile 把图纸变成可校验、可执行、可视化的 Runnable;条件边和环让 while 循环退役;Graph 与 Functional 两套 API 共享同一引擎,按"结构要不要显式"选型。第 5 篇循环里那些失控场景——多阶段、审批、恢复、并行——从图的角度看,都只是往交通网上加节点和边。

练习:给第 3 节的两节点图加一个节点。

  • 在 greet 与 farewell 之间加 praise 节点,往 messages 追加一句夸奖
  • 给 State 加 name: str 字段,让 greet 拼出个性化问候
  • 用 draw_mermaid() 导出新结构,确认中间真的多了一个节点
  • 进阶:把 greet 之后改成条件边,按 name 是否为空决定跳不跳过夸奖

下一篇讲 State 设计详解:Annotated reducer、add_messages、不可变更新与输入输出裁剪——白板怎么定义,直接决定图能长多复杂。对于刚接触图的读者,建议今天就照着第 3 节敲一遍、跑一遍、导一张图,骨架搭熟了,后面每一篇都只是在上面加东西。

相关推荐

Hermes Agent 详解:自进化学习型 AI 代理
Agent

Hermes Agent 详解:自进化学习型 AI 代理

Hermes Agent 详解:自进化学习型 AI 代理Hermes Agent 是由 Nous Research 于 2026 年 2 月开源的自主 AI 代理框架,其核心定位是"与你共同成长的智能体"。与传统的 IDE 编程助手或单一 API 聊天机器人不同,Hermes 运行在你的服务器上,能够记住所学内容,运行时间越长能力越强。它采用 MIT 开源许可,支持 Linux、macOS 和 W

393
Hermes Agent:与你一起成长的开源 AI 代理
Agent

Hermes Agent:与你一起成长的开源 AI 代理

Hermes Agent:与你一起成长的开源 AI 代理大多数 AI 助手是"健忘的租客"——换个窗口就失忆、换个平台就重学、干过的活不留下任何沉淀。Nous Research 出品的 Hermes Agent 反其道而行,口号是"The Agent That Grows With You"(与你一起成长的代理):它是一个开源(MIT)、自托管的 AI 代理,具备持久记忆(学习你的项目、自动生成技

322
speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线
AI

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线

speech-to-speech 详解:HuggingFace 的开源语音 Agent 流水线语音 Agent 的"全开源"路线一直有个尴尬:框架开源了,模型却离不开云端 API。huggingface/speech-to-speech 把这条路走通了:VAD → STT → LLM → TTS 四段低延迟流水线,每个组件都可换成开源模型本地跑——Apple Silicon 一台 Mac 就能撑起

264