最近在尝试把一些零散的 AI 任务串起来比如先让模型分析数据再根据结果生成报告最后自动发个通知。一开始用简单的脚本硬编码流程很快就遇到了麻烦状态管理混乱、错误处理麻烦、想加个“人工确认”环节更是无从下手。这让我意识到单靠大模型的对话能力远不足以构建一个稳定、可控的自动化工作流。就在这个节点上LangGraph 进入了视野。它不像 LangChain 那样主要解决“如何调用模型和工具”而是直指一个更核心的工程问题如何用“图”的思维来设计和运行那些带有状态、分支、循环和人工干预的复杂 AI 应用流程。很多人把它当作 LangChain 的进阶版这其实是个误解。LangChain 帮你组装零件模型、工具、记忆而 LangGraph 帮你设计并驱动整条生产线。恰好DeepLearning.AI 推出了由吴恩达老师主讲的 LangGraph 课程。这门课的价值不在于介绍又一个新工具而在于它系统性地传授了用“状态图”构建智能体的工程心法。它没有停留在 API 调用而是深入到了工作流的设计模式、状态管理、错误恢复以及“人在环路”这些真正决定项目能否落地的细节。如果你已经厌倦了写一堆脆弱的胶水代码或者对如何将 AI 能力工程化感到困惑那么理解 LangGraph 及其背后的理念可能是一个关键的转折点。1. 从“链”到“图”为什么你的智能体需要状态机在 LangChain 的“链”式思维里流程通常是线性的用户输入 - 模型思考 - 调用工具 - 模型输出。这对于简单任务很有效。但现实中的业务逻辑很少是一条直线。比如一个客服机器人它可能需要1) 理解问题2) 查询知识库3) 如果答案不确定则询问更多信息循环4) 如果涉及订单则调用查询接口分支5) 最终生成回答并记录对话历史状态。用传统的“链”来硬套这种流程代码会迅速变得难以维护。而 LangGraph 引入了“状态图”的概念。你可以把整个应用流程画成一张图节点代表处理步骤如“理解意图”、“调用工具”边代表步骤间的流转条件。更重要的是它维护了一个共享的“状态”对象所有节点都读写这个状态这使得管理对话历史、中间结果、循环计数等变得异常清晰。1.1 核心抽象StateGraph 与持久化状态LangGraph 的核心是定义一个状态State和一个基于该状态流转的图Graph。状态通常是一个 TypedDict明确了每个步骤需要和产生什么数据。from typing import TypedDict, List from langgraph.graph import StateGraph # 1. 定义状态 class AgentState(TypedDict): # 用户输入的问题 question: str # 模型或工具产生的中间思考 reasoning: str # 调用工具得到的结果 tool_result: str # 最终给用户的回答 answer: str # 历史步骤记录用于调试或回溯 steps: List[str] # 2. 初始化图 graph_builder StateGraph(AgentState)这个AgentState就是整个工作流的“共享内存”。每个节点函数都接收这个状态修改其中部分字段然后返回更新后的状态。LangGraph 负责在节点间传递这个状态并决定下一个该执行哪个节点。1.2 与 LangChain 的关系互补而非替代这是最容易混淆的点。LangChain 是一个庞大的工具集它包含了 Models, Prompts, Indexes, Chains, Agents 等大量组件。它的AgentExecutor本质上也是一个循环执行的工作流。那么为什么还需要 LangGraphLangChain AgentExecutor它是一个高度封装、针对特定智能体模式如 ReAct的解决方案。它很快能搭建起来但如果你想自定义工作流逻辑、改变决策节点、或者插入复杂的校验步骤就需要去继承和重写它的内部类侵入性较强。LangGraph它是一个更底层、更灵活的工作流编排框架。它不关心你用的是哪个模型、哪个工具库当然它与 LangChain 生态无缝集成。它只关心两件事状态和状态之间的流转规则。你可以用 LangGraph 重新实现一个 ReAct 智能体也可以实现一个完全不同的、带有并行审核节点的内容生成流水线。简单来说LangChain 提供了丰富的“建筑材料”LLM 调用、工具封装、记忆存储而 LangGraph 提供了设计并建造复杂“建筑结构”工作流的蓝图和脚手架。你可以只用 LangGraph 来编排流程然后用任何你喜欢的库包括 LangChain去实现图中的每个节点。2. 构建你的第一个 LangGraph 智能体从零到一理论之后我们通过一个经典场景——让 AI 使用搜索工具回答问题——来感受 LangGraph 的构建过程。我们将创建一个简单的 ReAct 风格智能体。2.1 环境搭建与基础配置首先确保安装必要的包。建议使用虚拟环境。pip install langgraph langchain-openai langchain-community这里我们使用 LangChain 的 OpenAI 集成和 DuckDuckGo 搜索工具。你需要准备一个 OpenAI API Key。import os from langchain_openai import ChatOpenAI from langchain_community.tools import DuckDuckGoSearchRun os.environ[OPENAI_API_KEY] your-api-key-here # 初始化大模型和工具 llm ChatOpenAI(modelgpt-4o-mini) search_tool DuckDuckGoSearchRun()2.2 定义节点与边把工作流画出来现在我们来定义图中的节点。一个典型的 ReAct 循环包含reason思考、action执行工具、update_state整合结果。from typing import Dict, Any # 定义状态继承自TypedDict更清晰 class ReActState(TypedDict): input: str # 用户原始问题 thought: str # 模型的思考过程 observation: str # 工具返回的观察结果 answer: str # 最终答案 steps: int # 已执行的步骤数 # 1. 思考节点 def reason_node(state: ReActState) - Dict[str, Any]: 分析问题决定是否需要搜索。 prompt f 你是一个助手。当前问题{state[input]} 已有的历史信息{state.get(observation, 无)} 请思考要回答这个问题我需要搜索网络信息吗如果需要请规划搜索查询词如果不需要请直接给出答案。 你的思考过程 response llm.invoke(prompt) # 这里简化处理实际应解析response内容 new_thought response.content # 判断是否需要行动这里用简单关键词判断生产环境应更严谨 if 搜索 in new_thought or 查询 in new_thought: action search else: action answer return {thought: new_thought, next_action: action} # 2. 行动节点搜索 def action_node(state: ReActState) - Dict[str, Any]: 执行搜索工具。 # 从思考中提取搜索词这里简化实际应用需要解析thought字段 search_query state[input] # 简单用原问题搜索 result search_tool.run(search_query) return {observation: result} # 3. 回答节点 def answer_node(state: ReActState) - Dict[str, Any]: 基于所有信息生成最终答案。 prompt f 问题{state[input]} 思考过程{state[thought]} 观察到的信息{state[observation]} 请给出最终答案。 response llm.invoke(prompt) return {answer: response.content} # 4. 构建图 from langgraph.graph import StateGraph, END builder StateGraph(ReActState) # 添加节点 builder.add_node(reason, reason_node) builder.add_node(action, action_node) builder.add_node(answer, answer_node) # 设置入口点 builder.set_entry_point(reason) # 添加边定义节点执行后的流向 def decide_next_step(state: ReActState) - str: 根据reason节点产生的next_action决定下一步。 # 这个逻辑应该由reason节点写入state这里为演示简化 if state.get(next_action) search: return action else: return answer builder.add_conditional_edges( reason, decide_next_step, { action: action, answer: answer } ) builder.add_edge(action, reason) # 搜索后继续思考 builder.add_edge(answer, END) # 给出答案后结束 # 编译图 graph builder.compile()2.3 运行与调试可视化你的流程运行这个图非常简单# 准备初始状态 initial_state ReActState(input2024年奥运会的主办城市是哪里, thought, observation, answer, steps0) # 执行图 final_state graph.invoke(initial_state) print(final_state[answer])LangGraph 的一个强大特性是可视化。你可以将图导出为图片直观看到工作流。from IPython.display import Image, display try: display(Image(graph.get_graph().draw_mermaid_png())) except: # 如果环境不支持可以输出Mermaid文本 print(graph.get_graph().draw_mermaid())这张图会让你清晰地看到reason- (action-reason循环) -answer的路径这正是 ReAct 智能体的核心循环。调试时你可以检查每次调用后的state对象精确追踪数据在每一步的变化。3. 超越基础掌握 LangGraph 的高级模式单一路径的工作流价值有限。LangGraph 的真正威力在于处理复杂逻辑。3.1 循环与中断实现可控的 ReAct 循环上面的例子中action后无条件地回到reason可能造成无限循环。我们需要增加中断条件。class ReActStateWithControl(TypedDict): input: str thought: str observation: str answer: str steps: int max_steps: int 3 # 最大循环次数 def reason_node_with_control(state: ReActStateWithControl) - Dict[str, Any]: # ... 原有的思考逻辑 ... # 增加步骤检查 if state[steps] state[max_steps]: return {thought: 已达到最大思考步骤准备生成最终答案。, next_action: force_answer} # ... 其余逻辑 ... def decide_next_step_with_control(state: ReActStateWithControl) - str: next_action state.get(next_action) if next_action force_answer: return answer elif state[steps] state[max_steps]: return answer elif next_action search: return action else: return answer # 在状态更新节点中记得递增 steps def update_steps_node(state: ReActStateWithControl): return {steps: state[steps] 1}通过在状态中维护计数器并在条件边conditional_edges中检查我们可以实现精确的循环控制防止智能体“陷入沉思”。3.2 “人在环路”关键决策的人工干预这是 LangGraph 相较于其他框架的杀手级特性。你可以轻松地将一个节点设置为“暂停”等待外部输入比如人工审核。from langgraph.graph import MessagesState from langgraph.checkpoint import MemorySaver from langgraph.prebuilt import tools_condition # 使用预定义的MessagesState它内置了messages列表 builder StateGraph(MessagesState) # 定义一个需要人工批准的节点 def human_review_node(state: MessagesState): # 这里工作流会在此暂停。 # 在实际部署中这可能会触发一个通知到管理界面或等待一个API回调。 # 为了演示我们模拟人工输入“批准” human_feedback 批准 # 实际应从外部获取 if human_feedback 批准: # 在消息中添加批准记录 new_messages state[messages] [{role: user, content: 管理员已批准}] return {messages: new_messages, approved: True} else: return {messages: state[messages], approved: False} builder.add_node(human_review, human_review_node) # ... 添加其他节点 ... # 配置检查点存储器这是实现“暂停”和“恢复”的基础 memory MemorySaver() app builder.compile(checkpointermemory) # 首次运行会在 human_review 节点暂停状态为“挂起” config {configurable: {thread_id: thread-1}} initial_state {messages: [{role: user, content: 生成一份季度财报摘要。}]} try: result app.invoke(initial_state, configconfig) except Exception as e: print(f工作流已暂停等待人工干预。) # 模拟人工处理后从检查点恢复执行 # 假设我们通过其他方式获得了批准信号然后更新状态并继续 resume_state app.invoke( {messages: [{role: user, content: 管理员已批准}]}, configconfig )通过结合CheckpointerLangGraph 可以将工作流状态持久化并在特定节点等待外部事件如人工审批、API 回调后再继续。这对于构建合规、安全的生产系统至关重要。3.3 并行与分支处理多任务工作流有些任务可以并行执行。例如在分析一篇新闻时可以同时进行情感分析和关键实体提取。from langgraph.graph import StateGraph from typing import List import asyncio class ParallelState(TypedDict): text: str sentiment: str entities: List[str] summary: str def sentiment_analysis_node(state: ParallelState): # 调用情感分析模型或函数 analysis 积极 # 模拟结果 return {sentiment: analysis} def entity_extraction_node(state: ParallelState): # 调用实体识别模型或函数 entities [实体A, 实体B] # 模拟结果 return {entities: entities} builder StateGraph(ParallelState) builder.add_node(analyze_sentiment, sentiment_analysis_node) builder.add_node(extract_entities, entity_extraction_node) # 设置入口点并让两个节点并行执行 builder.set_entry_point(analyze_sentiment) builder.add_edge(analyze_sentiment, extract_entities) # 注意简单的add_edge是顺序执行。要实现真并行需要更复杂的编排或使用异步。 # LangGraph 支持在单个节点内进行异步并发调用然后将结果汇总。 # 一个更常见的模式是一个分发节点然后多个并行节点最后一个汇总节点。对于复杂的并行-汇聚模式你需要设计一个协调节点来管理多个并行任务的执行和结果收集。LangGraph 提供了灵活性但并行逻辑需要开发者自己实现。4. 从 Demo 到生产LangGraph 项目实战要点在本地跑通一个 LangGraph 智能体是一回事把它部署为一个稳定服务是另一回事。以下是几个关键的实战考量点。4.1 状态持久化与检查点内存中的状态在服务重启后会丢失。LangGraph 的Checkpointer抽象支持将状态保存到数据库如 PostgreSQL、Redis。from langgraph.checkpoint import PostgresSaver import asyncpg # 创建 PostgreSQL 检查点存储器 async def get_postgres_checkpointer(): conn await asyncpg.connect(databaselanggraph, userpostgres) return PostgresSaver(conn) # 在编译图时传入 # app builder.compile(checkpointerpostgres_checkpointer)生产环境中持久化检查点不仅能实现“人在环路”的暂停/恢复还能提供工作流的历史追溯和断点续跑能力对于调试和审计极其重要。4.2 错误处理与韧性节点中的代码可能出错网络超时、模型异常、工具失败。一个健壮的图需要错误处理策略。节点级重试在节点函数内部使用tenacity等库进行重试。图级容错通过add_edge定义错误发生后的流转路径。例如可以设置一个fallback_node当某个业务节点失败时流转到该节点记录错误并给出友好提示而不是让整个图崩溃。超时控制对于每个节点的执行可以设置超时限制防止某个步骤卡死整个工作流。from langgraph.graph import StateGraph, END import asyncio from asyncio import TimeoutError def unreliable_node(state): # 模拟一个可能失败或超时的操作 raise ConnectionError(API调用失败) def fallback_node(state): return {result: f主节点处理失败已启用降级方案。原始输入{state[input]}} builder StateGraph(...) builder.add_node(process, unreliable_node) builder.add_node(fallback, fallback_node) builder.set_entry_point(process) # 理想情况下从 process 到 END # 但我们需要处理异常。LangGraph本身不自动处理异常需要在节点内捕获或使用更高级的模式。 # 一种模式是使用“Supervisor”节点来包装和监控其他节点的执行。对于复杂的错误处理可以考虑使用langgraph.prebuilt中的ToolNode或create_react_agent它们内置了一些错误处理逻辑。对于自定义图需要更精细的设计。4.3 监控、日志与可观测性你需要知道你的智能体在干什么尤其是在生产环境。结构化日志在每个节点的开始和结束将关键状态信息如state的摘要、节点名、耗时记录到结构化日志系统如 JSON Logger。追踪集成 OpenTelemetry 等追踪框架为每次图的执行生成一个 Trace可以看到所有节点的调用链和耗时。状态快照利用检查点机制定期或按需保存状态快照便于问题复现和调试。import logging import time logger logging.getLogger(__name__) def logged_node(state): node_name my_node start_time time.time() logger.info(f开始执行节点 {node_name}, extra{state_input: state.get(input)}) try: # ... 业务逻辑 ... result do_something(state) duration time.time() - start_time logger.info(f节点 {node_name} 执行成功耗时{duration:.2f}s, extra{result: result}) return result except Exception as e: logger.error(f节点 {node_name} 执行失败, exc_infoTrue, extra{state: state}) raise # 或返回一个错误状态4.4 性能优化与成本控制当工作流复杂或调用昂贵模型时需关注性能与成本。异步执行将节点函数定义为async并在可能的情况下使用asyncio.gather并行执行独立操作。缓存对于确定性操作如工具查询结果、特定输入的模型响应可以考虑在状态中或外部缓存如 Redis中缓存结果避免重复计算和调用。流式输出如果最终答案是文本生成考虑使用模型的流式响应并将state中的answer字段设计为可增量更新以提升用户体验。预算控制在状态中维护一个token_used或cost字段在每个调用模型的节点后累加。可以设置一个条件边在成本超预算时提前结束工作流跳转到总结节点。将 LangGraph 应用于实际项目意味着从“让流程跑起来”转向“让流程稳定、高效、可控地跑下去”。这需要我们在状态设计、错误处理、持久化和可观测性上下足功夫。它不再是一个简单的脚本而是一个需要精心运维的分布式状态机——尽管它可能只运行在一台服务器上。理解这一点是能否用好 LangGraph 的关键分水岭。