1. 从LangChain到LangGraph为什么我们需要“图”来编排AI应用如果你在过去一年里折腾过基于大语言模型的应用开发大概率听说过LangChain。它像是一套乐高积木把提示词模板、记忆、工具调用这些组件都给你准备好了让你能快速搭出一个能聊天的AI应用。但当你真的想做一个稍微复杂点的东西比如一个能根据用户问题自动决定是查数据库、调用API还是生成代码的智能客服或者一个多轮对话的决策分析助手时你可能会发现用LangChain那一套链式调用写着写着代码就变成了一团乱麻。状态在各个函数间传来传去逻辑分支像蜘蛛网一样交织调试起来让人头皮发麻。这就是LangGraph要解决的核心问题。它不是来替代LangChain的而是LangChain生态中的一个专门用于构建有状态、多环节、可循环的复杂AI应用编排框架。你可以把它理解为一个专为AI工作流设计的“可视化编程”引擎只不过它的底层抽象是“图”。为什么是图因为现实世界中的复杂任务尤其是涉及决策、循环和状态维护的对话或业务流程其本质就是一个由节点和边构成的有向图。每个节点执行一个操作比如调用LLM、执行工具、查询状态边则定义了操作的流转逻辑比如根据LLM的输出决定下一步去哪。LangGraph把这种思想变成了代码。它让你能清晰地定义工作流中的每一步以及步骤之间的流转条件从而将复杂的、容易出错的流程控制转化为可维护、可调试的图形化结构。这对于构建智能体、复杂对话系统、审批流程自动化等场景来说是一个游戏规则的改变者。它解决的痛点正是传统链式结构在复杂逻辑面前的无力感将开发者的心智负担从“如何用代码控制流程”转移到“如何设计工作流本身”。2. 核心基石深入理解LangGraph中的“状态机”思想要玩转LangGraph必须吃透它的核心设计模式状态机。这不是计算机科学课本里那个冷冰冰的概念而是LangGraph赋予AI应用“记忆”和“逻辑”的活生生骨架。2.1 什么是LangGraph语境下的状态在LangGraph中状态是一个中心化的、可变的字典对象。它贯穿整个工作流的生命周期是所有节点共享的“工作记忆”。这个状态字典里可以存放任何东西用户的输入、LLM的回复、中间计算结果、工具的执行结果、甚至是循环的计数器。举个例子你构建一个旅行规划助手状态里可能包含messages: 对话历史列表。user_preferences: 从对话中提取的用户偏好如预算、目的地。search_results: 调用航班/酒店API返回的原始数据。current_step: 当前进行到的步骤如“收集信息”、“搜索”、“推荐”。关键在于这个状态是随着工作流执行而不断演化的。每个节点读取状态的某些部分执行操作然后更新状态。这种设计使得工作流具备了“记忆”能够进行多轮交互和基于上下文的决策。2.2 节点与边构建工作流的乐高积木有了状态就需要定义谁来操作它以及操作完后该去哪。这就是节点和边。节点一个可调用的函数。它接收整个状态字典作为输入执行一些操作如调用LLM、运行代码然后返回一个包含对状态更新内容的字典。LangGraph会将这个返回的字典与原有状态进行合并更新。节点通常职责单一比如call_llm,search_web,validate_input。边连接节点的规则。它决定了在当前节点执行完毕后下一个该执行哪个节点。边可以分为两种起始边定义工作流从哪个节点开始。条件边这是状态机灵活性的关键。它不是一个简单的“跳转到节点A”而是一个根据当前状态来决定下一跳的函数。例如在旅行助手决定下一步时条件边函数会检查状态中的“current_step”值如果是“collect_info”就跳转到信息收集节点如果是“recommend”就跳转到推荐生成节点。2.3 “编译”与执行从蓝图到运行实例在LangGraph中你并不是直接“运行”代码而是先“编译”一个图。这个过程就像把设计好的电路图你的节点和边定义生成为一个可执行的程序。from langgraph.graph import StateGraph, END # 1. 定义状态结构通常使用TypedDict或Pydantic Model进行类型提示 from typing import TypedDict, List, Annotated import operator class State(TypedDict): messages: Annotated[List[str], operator.add] # 关键使用注解定义如何合并该字段 user_query: str answer: str # 2. 定义节点函数 def retrieve_node(state: State): # 模拟检索过程 retrieved_info f根据查询‘{state[user_query]}’检索到的信息。 return {messages: [f检索节点{retrieved_info}], answer: retrieved_info} def generate_node(state: State): # 基于检索结果生成回答 response f综合信息答案是{state[answer]} return {messages: [f生成节点{response}]} # 3. 构建图 workflow StateGraph(State) workflow.add_node(retrieve, retrieve_node) workflow.add_node(generate, generate_node) # 4. 设置边 workflow.set_entry_point(retrieve) # 起始边从retrieve节点开始 workflow.add_edge(retrieve, generate) # 无条件边retrieve完后一定去generate workflow.add_edge(generate, END) # 结束边generate完后工作流结束 # 5. 编译图 app workflow.compile()现在app就是一个编译好的、可执行的工作流对象。你可以通过一个初始状态来运行它# 6. 执行图 initial_state {messages: [], user_query: LangGraph是什么, answer: } final_state app.invoke(initial_state) print(final_state[messages])这个简单的例子展示了一个线性链检索 - 生成。但真正的威力在于引入条件边构建非线性工作流。注意Annotated[List[str], operator.add]这个类型注解是LangGraph处理列表状态合并的秘诀。它告诉框架当多个节点都返回对“messages”字段的更新时不要覆盖而是用operator.add即列表的操作来合并它们。这是实现对话历史累积的关键。3. 超越链式调用实战构建一个条件分支工作流让我们构建一个更贴近现实的例子一个智能路由助手。它需要根据用户问题的类型决定调用不同的处理节点。场景用户输入一个问题。工作流需要判断问题类型是“技术概念解释”、“代码生成”还是“闲聊”。根据类型路由到不同的专家节点处理。所有节点处理完后汇总到一个统一格式化的节点。3.1 定义状态与节点首先我们定义更丰富的状态和几个节点函数from typing import Literal from langgraph.graph import StateGraph, END class RouterState(TypedDict): messages: Annotated[List[dict], operator.add] # 存储消息对象 user_input: str problem_type: Literal[concept, code, chitchat, None] # 问题类型 concept_answer: str code_answer: str chitchat_answer: str final_output: str # 节点1分类路由节点 def classify_node(state: RouterState): input_text state[user_input].lower() if 什么是 in input_text or 原理 in input_text: p_type concept elif 代码 in input_text or 写一个 in input_text or 实现 in input_text: p_type code else: p_type chitchat return {problem_type: p_type, messages: [{role: system, content: f分类为{p_type}}]} # 节点2-4三个专家处理节点这里用模拟逻辑 def explain_concept_node(state: RouterState): # 模拟概念解释 answer f关于‘{state[user_input]}’的概念解释这是一个重要的架构模式。 return {concept_answer: answer, messages: [{role: assistant, content: f[概念专家] {answer}}]} def write_code_node(state: RouterState): # 模拟代码生成 answer f# 示例代码\nprint(Hello, {state[\user_input\]}) return {code_answer: answer, messages: [{role: assistant, content: f[代码专家] {answer}}]} def make_chitchat_node(state: RouterState): # 模拟闲聊 answer f您问‘{state[user_input]}’今天天气真不错 return {chitchat_answer: answer, messages: [{role: assistant, content: f[闲聊专家] {answer}}]} # 节点5汇总节点 def format_output_node(state: RouterState): # 根据类型选取对应的答案作为最终输出 if state[problem_type] concept: final state[concept_answer] elif state[problem_type] code: final state[code_answer] else: final state[chitchat_answer] return {final_output: final, messages: [{role: system, content: f最终输出已就绪{final}}]}3.2 构建带条件边的图这是关键步骤。我们需要让classify_node之后根据state[‘problem_type’]的值动态决定下一个节点。# 构建图 workflow StateGraph(RouterState) # 添加所有节点 workflow.add_node(classify, classify_node) workflow.add_node(explain_concept, explain_concept_node) workflow.add_node(write_code, write_code_node) workflow.add_node(make_chitchat, make_chitchat_node) workflow.add_node(format_output, format_output_node) # 设置入口 workflow.set_entry_point(classify) # 关键从classify节点出发定义条件边 def route_after_classify(state: RouterState): # 这个函数返回下一个节点的 *名称* if state[problem_type] concept: return explain_concept elif state[problem_type] code: return write_code elif state[problem_type] chitchat: return make_chitchat else: return END # 如果未分类直接结束 workflow.add_conditional_edges( classify, # 源节点 route_after_classify, # 路由函数 # 可选列出所有可能的目标节点有助于可视化 [explain_concept, write_code, make_chitchat, END] ) # 将三个专家节点都连接到汇总节点 workflow.add_edge(explain_concept, format_output) workflow.add_edge(write_code, format_output) workflow.add_edge(make_chitchat, format_output) # 汇总节点后结束 workflow.add_edge(format_output, END) # 编译 app workflow.compile()3.3 执行与可视化现在我们可以运行并观察状态如何流转# 执行1询问概念 state1 app.invoke({user_input: 什么是状态机, messages: []}) print(f最终输出: {state1[final_output]}) print(f消息历史: {state1[messages]}) # 执行2请求代码 state2 app.invoke({user_input: 写一个Python的hello world代码, messages: []}) print(f\n最终输出: {state2[final_output]})更强大的是LangGraph内置了可视化功能让你能直观看到你构建的“状态机”from IPython.display import Image, display try: display(Image(app.get_graph().draw_mermaid_png())) except: # 如果无法显示图片可以输出文本表示 print(app.get_graph().print_ascii())这张图会清晰地显示classify节点如何分叉到三个不同的专家节点然后汇聚到format_output。这种可视化对于理解复杂工作流和调试至关重要。实操心得在定义条件边函数route_after_classify时务必确保它对所有可能的状态分支都有明确的返回值并且返回值必须是图中已添加的节点名称或END。一个常见的坑是路由逻辑遗漏了某些边界情况导致运行时抛出KeyError。建议在开发初期用简单的if-elif-else覆盖所有枚举值并在最后加一个return END或return “fallback_node”作为兜底。4. 高级模式循环、持久化与多智能体协作当你的应用需要多轮对话、长期记忆或多个AI智能体分工合作时LangGraph的高级特性就派上用场了。4.1 实现循环构建多轮对话智能体循环是状态机的核心能力。在LangGraph中通过让边指向之前的节点可以轻松实现循环。一个典型的模式是“人类在环”智能体执行 - 等待用户输入 - 继续执行。from typing import Optional class ConversationState(TypedDict): messages: Annotated[List[dict], operator.add] needs_human_input: bool def assistant_node(state: ConversationState): # 模拟助理处理逻辑 last_msg state[messages][-1][content] if state[messages] else if 帮我订票 in last_msg: response 我需要您提供出行日期和目的地。 next_action need_human else: response f我收到了您的消息‘{last_msg}’。 next_action wait return { messages: [{role: assistant, content: response}], needs_human_input: (next_action need_human) } def human_input_node(state: ConversationState): # 在实际应用中这里会是一个等待外部输入如API调用、前端事件的节点 # 此处模拟用户回复 simulated_input 明天去上海 # 这通常来自外部 return { messages: [{role: user, content: simulated_input}], needs_human_input: False } # 构建循环图 workflow StateGraph(ConversationState) workflow.add_node(assistant, assistant_node) workflow.add_node(human_input, human_input_node) workflow.set_entry_point(assistant) def decide_next(state: ConversationState): if state.get(needs_human_input): return human_input else: # 可以设置一个结束条件比如对话轮次 if len(state[messages]) 10: return END return assistant # 循环回助理节点 workflow.add_conditional_edges( assistant, decide_next, [human_input, assistant, END] ) workflow.add_edge(human_input, assistant) # 人类输入后回到助理 app workflow.compile() # 运行这个app它会根据needs_human_input标志在assistant和human_input间循环4.2 状态持久化让应用记住“上一次”对于需要长期运行的智能体如聊天机器人状态必须能够持久化到数据库并在下次调用时恢复。LangGraph通过Checkpointer抽象支持这一点。from langgraph.checkpoint.sqlite import SqliteSaver import tempfile import os # 1. 创建一个SQLite检查点存储生产环境可用Postgres等 dir tempfile.mkdtemp() conn_str fsqlite:///{os.path.join(dir, checkpoints.db)} checkpointer SqliteSaver.from_conn_string(conn_str) # 2. 在编译图时传入检查点器 app workflow.compile(checkpointercheckpointer) # 3. 运行并保存检查点通常以线程ID或用户ID作为配置标识 config {configurable: {thread_id: user_123}} initial_state {messages: [{role: user, content: 你好}], needs_human_input: False} # 第一次调用生成检查点 result1, checkpoint_info1 app.invoke(initial_state, configconfig) print(f第一次回复: {result1[messages][-1][content]}) # 模拟应用重启后从检查点恢复 # 我们使用相同的configthread_id来调用LangGraph会自动加载上次的状态 new_initial_state {messages: [{role: user, content: 我昨天问过什么}]} result2, checkpoint_info2 app.invoke(new_initial_state, configconfig) # result2中的messages会包含上一次对话的历史 print(f历史消息: {[m[content] for m in result2[messages]]})注意事项状态持久化是生产级应用的关键。SqliteSaver适合轻量级或演示使用线上环境强烈建议使用PostgresSaver或RedisSaver这类更健壮的后端。另外要注意状态字典中存储的数据必须是可序列化的如基本类型、列表、字典自定义类对象可能需要特殊处理。4.3 子图与多智能体协作模块化复杂系统对于极其复杂的工作流你可以将其分解为多个子图每个子图负责一个特定的子任务然后通过主图进行编排。这对应着多智能体系统中的“主管-工作者”模式。from langgraph.graph import StateGraph, START, END # 子图A研究智能体 class ResearchState(TypedDict): topic: str findings: list def research_subgraph(state: ResearchState): # 模拟研究过程 return {findings: [f关于{state[topic]}的发现1, f关于{state[topic]}的发现2]} research_graph StateGraph(ResearchState) research_graph.add_node(research, research_subgraph) research_graph.add_edge(START, research) research_graph.add_edge(research, END) compiled_research research_graph.compile() # 子图B写作智能体 class WritingState(TypedDict): findings: list report: str def write_subgraph(state: WritingState): # 模拟写作过程 report 报告\n \n.join(state[findings]) return {report: report} writing_graph StateGraph(WritingState) writing_graph.add_node(write, write_subgraph) writing_graph.add_edge(START, write) writing_graph.add_edge(write, END) compiled_write writing_graph.compile() # 主图协调智能体 class ManagerState(TypedDict): query: str research_results: list final_report: str def manager_node(state: ManagerState): # 主节点调用子图 # 1. 调用研究子图 research_state compiled_research.invoke({topic: state[query]}) # 2. 调用写作子图 write_state compiled_write.invoke({findings: research_state[findings]}) return {research_results: research_state[findings], final_report: write_state[report]} main_graph StateGraph(ManagerState) main_graph.add_node(manager, manager_node) main_graph.add_edge(START, manager) main_graph.add_edge(manager, END) main_app main_graph.compile() result main_app.invoke({query: LangGraph的优势}) print(result[final_report])这种方式实现了极致的模块化和复用。每个子图可以独立开发、测试和优化主图则像一个 orchestrator负责流程控制和数据传递。5. 避坑指南与最佳实践来自实战的经验在项目中大规模使用LangGraph后我积累了一些关键的经验和教训能帮你节省大量调试时间。5.1 状态合并的陷阱与正确姿势LangGraph默认使用“浅合并”来更新状态。这意味着如果两个节点都修改了状态字典中同一个嵌套对象比如一个列表里的某个字典元素可能会发生意外覆盖。错误示例class ProblematicState(TypedDict): data: dict # 这是一个嵌套字典 def node_a(state): return {data: {step: a, value: 1}} # 设置整个data def node_b(state): # 本意是想在data中添加一个字段但... new_data state[data] new_data[status] processed return {data: new_data} # 这实际上返回了整个data会覆盖node_a的value吗这里的结果取决于合并顺序行为不确定。正确做法对于嵌套结构的更新使用operator.add对列表是有效的但对字典不直接支持。推荐两种策略扁平化状态尽量避免深层嵌套。将data_step,data_value,data_status作为状态的同级键。使用Pydantic模型与langgraph的add_messages范式对于消息列表LangGraph有内置最佳实践。对于其他复杂结构可以在节点函数中返回一个只包含增量更新的字典并确保你的合并逻辑能处理。from typing import List from pydantic import BaseModel from langgraph.graph.message import add_messages class ProperState(BaseModel): messages: List[dict] [] metadata: dict {} # 复杂对象 def safe_node(state: ProperState): # 更新messages的标准方式 new_messages add_messages(state.messages, [{role: user, content: hi}]) # 更新metadata的增量方式 new_metadata {**state.metadata, last_node: safe_node} return {messages: new_messages, metadata: new_metadata}5.2 调试如何追踪工作流的每一步当工作流没有按预期运行时调试可能很困难。以下是几种有效方法可视化图形首先用app.get_graph().draw_mermaid_png()或.print_ascii()检查你的图结构是否正确边是否连接无误。启用详细日志在invoke时设置debugTrue。result app.invoke(initial_state, config{debug: True})这会在返回结果中包含每个节点的执行输入和输出对于追踪状态变化非常有用。使用检查点即使不需要持久化也可以使用内存检查点器来记录执行历史。from langgraph.checkpoint.memory import MemorySaver checkpointer MemorySaver() app workflow.compile(checkpointercheckpointer) result, checkpoint app.invoke(..., configconfig) # 可以通过checkpointer查看历史单元测试单个节点将节点函数当作纯函数进行测试传入模拟状态断言其输出。5.3 性能考量与生产部署冷启动延迟编译图workflow.compile()有一定开销。在生产环境中应在服务启动时预编译好所有工作流而不是每次请求都编译。状态大小持久化的状态会存储在数据库中。要避免在状态中存储过大的对象如原始文件内容、大型数据集。只存储必要的引用或摘要。并发与锁当多个请求使用相同的thread_id并发访问时需要数据库检查点器支持行级锁或乐观锁以防止状态损坏。SqliteSaver在并发写入时可能有问题生产环境务必使用PostgresSaver。错误处理在图定义中目前没有内置的全局错误处理节点。一个实用的模式是定义一个error_handler节点并在可能出错的节点后通过条件边连接过去。在error_handler节点中你可以记录错误、更新状态以提示用户并决定是重试、跳过还是结束工作流。5.4 与LangChain的整合强强联合LangGraph和LangChain是绝配。你可以轻松地将LangChain的LCEL链、工具、记忆组件作为LangGraph的一个节点。from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langgraph.prebuilt import ToolExecutor, tools_condition from langgraph.graph import MessagesState # 创建一个LangChain链 prompt ChatPromptTemplate.from_template(回答关于{topic}的问题。) llm ChatOpenAI(modelgpt-4) chain prompt | llm | StrOutputParser() # 将链包装成LangGraph节点 def langchain_node(state: MessagesState): # 从状态中提取输入 last_message state[messages][-1].content if state[messages] else # 调用链 response chain.invoke({topic: last_message}) # 更新状态 return {messages: [{role: assistant, content: response}]} # 同样可以集成LangChain Tools from langchain_community.tools import DuckDuckGoSearchRun search_tool DuckDuckGoSearchRun() tool_executor ToolExecutor([search_tool]) def tool_node(state): # 根据消息决定调用哪个工具... result tool_executor.invoke({query: some query}) return {messages: [{role: tool, content: result}]}这种整合让你既能享受LangChain丰富的生态和组件又能利用LangGraph强大的流程控制能力。从我自己的使用体验来看LangGraph最大的价值在于它提供了一种符合直觉的方式来设计和推理复杂的AI应用逻辑。它将混乱的控制流代码转化为了清晰的、可视化的图结构。虽然学习曲线比直接写脚本要陡一些但一旦掌握在开发复杂、可维护的AI应用时其带来的效率和可靠性的提升是巨大的。尤其是在需要多轮交互、分支决策和状态保持的场景下它几乎是目前最优雅的解决方案。开始使用的最佳方式是从一个你之前用链式调用感觉有点别扭的小项目开始重构亲自体会一下“图”思维带来的不同。
