如果你已经写过几个 LangChain 的 RAG 应用大概率会有一种感觉链式调用做直线流程很顺手但一旦流程需要“看情况走下一步”或者“做错了要回头重试”代码就开始失控。大模型应用从 Demo 走向生产真正的分水岭往往不是 Prompt 写得好不好而是流程控制能力能不能分支、能不能循环、能不能中途暂停、能不能在任意一步恢复现场。LangGraph 正好是解决这个问题的框架。我的判断是LangGraph 不是给你多一种“写 Agent 的方式”而是把 Agent 从一次性脚本变成可控状态机的关键一步。这篇文章按“零基础实战”的路径来写不堆概念。核心会覆盖五块内容LangGraph 的状态与节点模型、一条最简单的线性流程、conditional_edge 条件路由与分支控制、循环检测与工具调用循环、子图和并行分支。每个部分都有可以直接复制运行的 Python 示例并会明确指出新手最容易踩坑的地方。如果你已经知道 LangChain 但对 LangGraph 还停留在“听说过”的阶段或者你想实现多步工具调用但总觉得不受控这篇文章应该能帮你把缺的那块拼图找回来。1. 这篇文章真正要解决的问题1.1 遇到什么场景LangChain 不够用LangChain 的核心抽象是 Chain也就是把大模型调用、Prompt 模板、文档检索、输出解析这些步骤串成一条固定的流水线。对于“问答、摘要、翻译”这类直线流程Chain 足够用。但真实业务不是直线。一个合格 Agent 至少要面对三种情况大模型决定是否需要调用工具而不是所有问题都走同一个分支工具调用结果需要带回给模型继续推理可能要反复多轮流程中还有人机确认、失败重试、并行检索、子流程复用等复杂控制。如果只用 Chain 硬写最终会得到大量if/else和全局变量逻辑散落在代码各处很难测试也很难恢复现场。LangGraph 把这些问题统一收拢到“图”这个模型里。1.2 LangGraph 是什么LangGraph 是 LangChain 生态中的图编排框架。它把 Agent 工作流建模成一张有向图节点是业务动作边是执行路径。节点函数负责处理状态边负责决定流程走向。这里要强调一个判断LangGraph 的核心不是“图”这个概念本身而是状态管理。图只是骨干State 才是灵魂。理解这一点后面学条件路由、循环、子图都会顺畅很多。1.3 LangChain 和 LangGraph 的区别可以用一个类比LangChain 是“生产线”每个环节顺序固定LangGraph 是“带控制台的调度系统”可以根据当前状态决定下一个环节是继续、跳转、并行还是重试。对比项LangChainLangGraph核心抽象Chain / RunnableStateGraph / Node / Edge流程形态线性串联为主有向图支持分支、循环、并行状态管理依赖外部传参链内共享不够直观全局 State节点函数通过返回值更新可控性适合固定流程适合需要动态决策的 Agent学习成本低略高但换来更强控制力需要说明的是LangGraph 不是替代 LangChain而是互补。你仍然可以用 LangChain 的模型封装、Prompt 模板和文档加载器只是把流程编排交给 LangGraph。2. LangGraph 核心概念State、Node、Edge 与状态合并原理2.1 State整个流程的“共享黑板”State 是 LangGraph 中最重要的概念。它通常是一个TypedDict或 Pydantic 模型保存整个流程共享的数据。可以把它理解成一块公共黑板每个节点都能读黑板上的内容也能往黑板上写新内容。节点之间不直接传参所有交互都通过 State 完成。from typing import TypedDict, Annotated import operator class AgentState(TypedDict): user_input: str # 普通字段默认覆盖旧值 result: str # 普通字段默认覆盖旧值 messages: Annotated[list, operator.add] # 累计字段追加而非覆盖这里藏着 LangGraph 一个关键机制节点函数返回值中出现的字段会按规则合并到全局 State。普通字段新值直接覆盖旧值Annotated[list, operator.add]这类带 reducer 的字段会把返回值追加或按自定义函数合并。这就是“LangGraph 如何在节点函数改变 state 状态值”的底层答案不是直接修改传入的state对象而是返回一个字典由框架负责合并。2.2 Node一个普通函数Node 在 LangGraph 里就是一个普通 Python 函数。它的输入是当前 State输出是一个字典表示你需要更新的字段。def my_node(state: AgentState) - dict: # 读取 state text state.get(user_input, ) # 返回需要更新的字段 return {result: f处理结果{text}}如果节点不需要更新任何字段返回空字典{}即可。一个新手的常见错误是在节点内部直接维护全局变量来记录状态。这样做的问题在于图一旦需要回放、重试或并行执行全局变量会让状态变得不可控。正确做法是所有动态数据都放回 State。2.3 Edge连接节点的路径Edge 决定节点之间的执行顺序。LangGraph 里有两种边普通边A 执行完一定走到 B条件边A 执行完后根据路由函数的结果从多个路径中选一个。另外还有两个特殊节点START表示流程入口END表示流程结束。一个图至少要有一条从START出发的路径最终能到达END。2.4 Checkpointer流程快照LangGraph 还支持 Checkpointer相当于给流程做快照。开启后每次节点执行完都会保存状态可以配合thread_id实现多轮会话记忆、断点恢复、人工审批等能力。从学习路径看先不用急着研究 Checkpointer。把 State、Node、Edge 和条件边跑通再回来看持久化会容易得多。3. 环境准备与前置条件3.1 运行环境LangGraph 是 Python 库需要 Python 3.9 及以上版本。建议使用 3.10 或 3.11语法更舒服第三方库兼容性也更好。操作系统不限Windows、macOS、Linux 都可以。建议在独立目录中创建虚拟环境避免污染系统 Python。mkdir langgraph-demo cd langgraph-demo python -m venv .venv source .venv/bin/activate # Windows 使用 .venv\Scripts\activate3.2 安装依赖LangGraph 主体安装在langgraph包中。如果后面要调用真实大模型还需要langchain-openai推荐一并安装。pip install -U langgraph langchain-core langchain-openai python-dotenv如果你在国内网络环境可以走镜像源加速pip install -U langgraph langchain-core langchain-openai python-dotenv -i https://pypi.tuna.tsinghua.edu.cn/simple版本说明LangGraph 迭代速度很快具体版本号请以实际安装结果为准。本文示例以稳定 API 为主重点演示通用思路。3.3 模型配置准备后续章节中纯逻辑示例不需要 API Key。但如果你要把某个节点替换成真实大模型需要提前准备好模型服务配置。在项目根目录创建.env文件OPENAI_API_KEY你的密钥 # 如果使用国内模型服务商提供的 OpenAI 兼容接口在这里配置 OPENAI_BASE_URLhttps://你的服务商地址/v1注意.env文件包含密钥一定要加入.gitignore不要提交到代码仓库。3.4 验证安装执行下面命令能输出ok就说明基础依赖可用python -c from langgraph.graph import StateGraph, START, END; print(ok)4. 第一个 LangGraph 应用从线性流程开始4.1 目标这一章的目标很简单跑通一个没有任何分支的图理解“节点函数返回值更新 State”这件事。我们定义两个节点analyze和format_reply让数据依次流过它们。文件位置langgraph-demo/state_demo.py4.2 完整代码# langgraph-demo/state_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class AgentState(TypedDict): user_input: str result: str step_count: int def analyze(state: AgentState) - dict: print(--- analyze ---) text state.get(user_input, ) step_count state.get(step_count, 0) 1 return { result: f收到{text}, step_count: step_count, } def format_reply(state: AgentState) - dict: print(--- format_reply ---) return { result: f最终输出{state[result]} | 已执行 {state[step_count]} 次, } builder StateGraph(AgentState) builder.add_node(analyze, analyze) builder.add_node(format_reply, format_reply) builder.add_edge(START, analyze) builder.add_edge(analyze, format_reply) builder.add_edge(format_reply, END) graph builder.compile() out graph.invoke({user_input: LangGraph, step_count: 0}) print(out)4.3 运行与验证python state_demo.py预期输出--- analyze --- --- format_reply --- {user_input: LangGraph, result: 最终输出收到LangGraph | 已执行 1 次, step_count: 1}分析这段输出analyze读取了user_input返回result和step_countformat_reply能读到analyze写入的result说明节点之间的数据传递通过 State 完成step_count从 0 变成 1说明返回字典中的数值覆盖了旧值。这里的关键在于analyze不是直接修改传入的state对象而是把新值放在返回字典中由框架合并进 State。直接修改传入对象不是推荐做法LangGraph 依赖返回值合并这样整个流程才可回放、可追踪。5. 条件路由与分支控制conditional_edge 深度解析5.1 为什么需要条件路由真实 Agent 中最核心的控制点是“根据当前输入决定下一步走哪里”。例如一个客服机器人用户问“今天天气怎么样”时应该调用天气查询工具用户问“你叫什么名字”时直接回答即可。如果所有问题都去调工具成本高且响应慢如果所有问题都不调工具又回答不了实时信息。条件路由正好解决这个问题。5.2 路由函数与路由表add_conditional_edges是条件路由的核心方法常用参数有三个builder.add_conditional_edges( source_node, # 从哪个节点开始做判断 route_function, # 路由函数输入 State返回一个字符串 key { # 路径映射表key - 目标节点 tool: tool_node, normal: normal_node, }, )路由函数本身不复杂def route_after_parse(state: AgentState) - str: if state.get(need_tool): return tool return normal这里最严格的约束是路由函数返回的字符串必须是路径映射表的 key 之一。否则 LangGraph 会在运行时直接报错。5.3 完整示例文件位置langgraph-demo/conditional_demo.py# langgraph-demo/conditional_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class QAState(TypedDict): question: str answer: str need_tool: bool def parse_question(state: QAState) - dict: question state.get(question, ) need_tool (天气 in question) or (查询 in question) print(f解析问题{question}是否需要工具{need_tool}) return {need_tool: need_tool} def tool_fetch_weather(state: QAState) - dict: print(调用天气查询工具...) return {answer: f【工具结果】{state[question]} 的查询结果晴25℃} def normal_reply(state: QAState) - dict: return {answer: f【普通回复】{state[question]}这个问题我可以直接回答。} def route_after_parse(state: QAState) - str: if state.get(need_tool): return tool return normal builder StateGraph(QAState) builder.add_node(parse, parse_question) builder.add_node(tool, tool_fetch_weather) builder.add_node(normal, normal_reply) builder.add_edge(START, parse) builder.add_conditional_edges( parse, route_after_parse, { tool: tool, normal: normal, }, ) builder.add_edge(tool, END) builder.add_edge(normal, END) graph builder.compile() for q in [今天上海天气怎么样, 什么是 LangGraph]: out graph.invoke({question: q}) print(out[answer])5.4 运行输出python conditional_demo.py预期输出解析问题今天上海天气怎么样是否需要工具True 调用天气查询工具... 【工具结果】今天上海天气怎么样 的查询结果晴25℃ 解析问题什么是 LangGraph是否需要工具False 【普通回复】什么是 LangGraph这个问题我可以直接回答。5.5 条件路由的三个踩坑点第一路由函数返回值和映射表的 key 不一致。最容易出现在手写字符串的场景建议用常量或枚举表示路由 key而不是到处写字符串字面量。第二没有兜底分支。如果路由函数遇到未知情况返回了一个没有登记的 key图会直接崩溃。更稳妥的做法是路由函数最后返回一个“兜底分支”比如normal。第三混淆“条件路由”和“并行分支”。条件路由是从多个路径中选一个并行分支是同时走多个路径。两者在add_conditional_edges和add_edge的用法上完全不同。6. 循环控制与“会自我修正”的 Agent 雏形6.1 图如何表达循环图本身是有向的但可以存在环也就是一条边从后一个节点指回前一个节点。让工具调用节点重新指向思考节点就形成了“思考—调用—再思考”的循环。循环必须有终止条件。实践中最常用的手段有两种在 State 中维护一个轮数计数器超过最大值就强制结束在路由函数中判断是否需要继续例如模型已经给出了最终答案就不再进入工具调用。6.2 带最大轮数的循环示例文件位置langgraph-demo/loop_demo.py# langgraph-demo/loop_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END MAX_ITERATIONS 3 class AgentState(TypedDict): question: str tool_result: str final_answer: str iteration: int def llm_think(state: AgentState) - dict: iteration state.get(iteration, 0) 1 print(f第 {iteration} 轮LLM 思考中...) return {iteration: iteration} def tool_call(state: AgentState) - dict: print(f调用工具当前轮次{state[iteration]}) if 天气 in state[question]: return {tool_result: 今天 25℃晴} return {tool_result: 没有找到该结果} def should_reason_again(state: AgentState) - str: if state[iteration] MAX_ITERATIONS and 天气 in state[question]: return continue return finish builder StateGraph(AgentState) builder.add_node(think, llm_think) builder.add_node(tool, tool_call) builder.add_edge(START, think) builder.add_conditional_edges( think, should_reason_again, { continue: tool, finish: END, }, ) builder.add_edge(tool, think) # 关键形成循环 graph builder.compile() out graph.invoke({question: 查一下北京天气, iteration: 0}) print(out)运行命令python loop_demo.py预期输出会看到 3 轮“LLM 思考中”之后结束第 1 轮LLM 思考中... 调用工具当前轮次1 第 2 轮LLM 思考中... 调用工具当前轮次2 第 3 轮LLM 思考中... 调用工具当前轮次3最后进入finish分支流程终止。注意这里iteration的更新方式每次都在节点内读取当前值并加 1再通过返回字典写回 State。因为它是一个普通字段所以默认是覆盖语义。6.3 循环检测的工程意义真实 Agent 场景中模型偶尔会陷入“反复调用同一个工具”的循环。如果不设上限可能产生大量调用费用甚至把下游系统打爆。工程上建议至少做四层保护最大迭代次数从源头控制轮数单节点超时防止某个工具调用卡死工具调用去重同一参数连续调用多次时直接返回缓存成本上限累计 token 或费用超过阈值后强制停止。这些策略都可以通过 State 中的累计字段实现。例如记录已经调用过的工具参数在工具节点里先查缓存再决定是否真的发起调用。6.4 从循环到 ReAct Agent把上面示例中的llm_think替换为真实大模型调用tool_call替换为真实工具就是一个非常接近 ReAct 模式的 Agent 骨架。接入真实模型时可以使用langchain-openaifrom langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini, temperature0) response llm.invoke(你好) print(response.content)如果使用国内模型服务商提供的 OpenAI 兼容接口通过base_url指向对应服务地址即可。实际项目中的关键还是“循环退出条件”不是所有问题都需要调用工具路由函数必须能识别“已经满足用户需求”的时刻。7. 子图与并行分支复杂流程的拆解方式7.1 子图把复杂流程封装成可复用组件当图越来越复杂时直接在一个图里堆节点很难维护。LangGraph 支持在一个节点里调用另一个已经编译好的图也就是子图。子图的优势是职责清晰。比如“文本清洗”这个流程可能在多个业务里都会用到独立成子图后主图只需要调用一次。文件位置langgraph-demo/subgraph_demo.py# langgraph-demo/subgraph_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END # ---- 子图定义 ---- class SubState(TypedDict): text: str formatted: str def sub_lower(state: SubState) - dict: print(子图转小写) return {formatted: state[text].lower()} def sub_strip(state: SubState) - dict: print(子图去空格) return {formatted: state[formatted].strip()} sub_builder StateGraph(SubState) sub_builder.add_node(lower, sub_lower) sub_builder.add_node(strip, sub_strip) sub_builder.add_edge(START, lower) sub_builder.add_edge(lower, strip) sub_builder.add_edge(strip, END) subgraph sub_builder.compile() # ---- 主图定义 ---- class MainState(TypedDict): raw_text: str final_text: str def main_process(state: MainState) - dict: print(主图调用子图) result subgraph.invoke({text: state[raw_text]}) return {final_text: result[formatted]} main_builder StateGraph(MainState) main_builder.add_node(process, main_process) main_builder.add_edge(START, process) main_builder.add_edge(process, END) main_graph main_builder.compile() out main_graph.invoke({raw_text: Hello LangGraph }) print(out)运行输出主图调用子图 子图转小写 子图去空格 {raw_text: Hello LangGraph , final_text: hello langgraph}使用子图时有三个要点子图必须compile()之后才能被调用子图拥有自己的 State主图不会自动共享子图内部字段需要手动把结果写入主图状态子图适合做纯函数式的通用处理例如格式化、校验、检索、权限检查等。7.2 并行分支一个节点触发多个下游并行分支在业务中很常见。例如内容审核需要同时做关键词检查、敏感词检查和长度检查三个检查互不依赖最后再汇总结果。LangGraph 支持多个节点同时连接到同一个上游节点上游执行完后多个下游都会执行。多个下游再连接到同一个汇总节点时框架会等待所有上游都完成再进入汇总节点。文件位置langgraph-demo/parallel_demo.py# langgraph-demo/parallel_demo.py from typing import TypedDict from langgraph.graph import StateGraph, START, END class ReviewState(TypedDict): content: str content_flag: str sensitive_flag: str length_flag: str final_flag: str def prepare(state: ReviewState) - dict: print(准备审核内容) return {} def check_content(state: ReviewState) - dict: return {content_flag: 通过 if len(state[content]) 0 else 失败} def check_sensitive(state: ReviewState) - dict: return {sensitive_flag: 含敏感词 if 违禁 in state[content] else 无敏感词} def check_length(state: ReviewState) - dict: return {length_flag: 长度正常 if len(state[content]) 10 else 长度过长} def merge(state: ReviewState) - dict: print(汇总审核结果) content_ok state.get(content_flag) 通过 sensitive_ok state.get(sensitive_flag) 无敏感词 length_ok state.get(length_flag) 长度正常 return {final_flag: 全部通过 if content_ok and sensitive_ok and length_ok else 需要人工复核} builder StateGraph(ReviewState) builder.add_node(prepare, prepare) builder.add_node(content, check_content) builder.add_node(sensitive, check_sensitive) builder.add_node(length, check_length) builder.add_node(merge, merge) builder.add_edge(START, prepare) # 并行 fan-out builder.add_edge(prepare, content) builder.add_edge(prepare, sensitive) builder.add_edge(prepare, length) # 汇总 fan-in builder.add_edge(content, merge) builder.add_edge(sensitive, merge) builder.add_edge(length, merge) builder.add_edge(merge, END) graph builder.compile() out graph.invoke({content: 这是一段待审核内容}) print(out)运行输出准备审核内容 汇总审核结果 {content: 这是一段待审核内容, content_flag: 通过, sensitive_flag: 无敏感词, length_flag: 长度过长, final_flag: 需要人工复核}7.3 并行分支的注意事项首先要理解 LangGraph 的“并行”是逻辑上的并行调度不是一定开多线程并行执行。如果节点是 IO 密集型任务