LangGraph 单智能体状态机实战复杂任务流的工程级编排ReAct 够用但遇到需要循环、分支、人工干预的复杂任务时就捉襟见肘。LangGraph 把 Agent 的执行流抽象成一个有向图给了工程师掌控感。本文从零开始带你用 LangGraph 构建一个能在生产环境真正运行的单智能体系统。一、为什么需要 LangGraph简单任务用 ReActReason Act 循环就够了但当 Agent 需要处理这些场景时ReAct 就显得力不从心条件分支根据工具执行结果决定下一步循环重试某一步失败后重试有最大重试次数人工审批关键操作需要暂停等待人工确认Human-in-the-loop状态持久化任务中断后可以从断点继续Checkpointing并行执行多个无依赖的工具调用同时发出LangGraph 的核心抽象图Graph 节点Node 边Edge 状态State - Node一个执行步骤LLM 调用、工具执行、人工审批... - Edge节点之间的流转关系可以是条件边 - State在整个图中流转的共享数据结构二、核心概念快速上手2.1 安装pipinstalllanggraph langchain-anthropic langchain-core2.2 最简单的 LangGraph AgentfromtypingimportAnnotated,TypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.prebuiltimportToolNode,tools_conditionfromlangchain_anthropicimportChatAnthropicfromlangchain_core.toolsimporttoolimportoperator# 1. 定义状态结构classAgentState(TypedDict):messages:Annotated[list,operator.add]# 消息列表append 操作# 2. 定义工具tooldefsearch_web(query:str)-str:搜索互联网获取最新信息# 实际实现中调用真实搜索 APIreturnf搜索结果关于{query}的内容...tooldefcalculate(expression:str)-str:计算数学表达式try:resulteval(expression)# 生产环境用更安全的方案returnstr(result)exceptExceptionase:returnf计算错误:{e}tools[search_web,calculate]# 3. 创建 LLM 并绑定工具llmChatAnthropic(modelclaude-3-5-sonnet-20241022)llm_with_toolsllm.bind_tools(tools)# 4. 定义节点函数defcall_model(state:AgentState)-AgentState:调用 LLM 生成响应responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 5. 构建图graph_builderStateGraph(AgentState)# 添加节点graph_builder.add_node(agent,call_model)graph_builder.add_node(tools,ToolNode(tools))# 添加边graph_builder.add_edge(START,agent)graph_builder.add_conditional_edges(agent,tools_condition,# 内置条件有工具调用 → tools否则 → END)graph_builder.add_edge(tools,agent)# 工具执行完回到 agent# 编译graphgraph_builder.compile()三、进阶构建生产级 Agent下面构建一个代码审查 Agent——接收一段代码分析质量、检查安全漏洞、生成改进建议并在发现严重问题时暂停等待人工确认。3.1 定义复杂状态fromtypingimportAnnotated,TypedDict,Optional,Listfromlanggraph.graph.messageimportadd_messagesimportoperatorclassCodeReviewState(TypedDict):# 消息历史messages:Annotated[list,add_messages]# 被审查的代码code:str# 审查结果quality_score:Optional[int]# 0-100security_issues:Optional[List[str]]# 安全问题列表suggestions:Optional[List[str]]# 改进建议# 流程控制severity:Optional[str]# low | medium | high | criticalhuman_approved:Optional[bool]# 人工审批结果# 重试计数retry_count:intmax_retries:int3.2 定义各节点fromlangchain_anthropicimportChatAnthropicfromlangchain_core.messagesimportHumanMessage,SystemMessageimportjson llmChatAnthropic(modelclaude-3-5-sonnet-20241022)defanalyze_quality(state:CodeReviewState)-CodeReviewState:节点1分析代码质量promptf分析以下代码的质量返回 JSON 格式 代码{state[‘code’]}返回格式 {{ quality_score: 0-100的整数, suggestions: [建议1, 建议2, ...] }} response llm.invoke([HumanMessage(contentprompt)]) try: result json.loads(response.content) return { quality_score: result[quality_score], suggestions: result[suggestions], messages: [response] } except Exception: return { quality_score: 50, suggestions: [解析结果失败请手动检查], messages: [response] } def check_security(state: CodeReviewState) - CodeReviewState: 节点2安全漏洞检查 prompt f对以下代码进行安全审查返回 JSON 格式 代码{state[‘code’]}重点检查SQL 注入、XSS、命令注入、敏感信息泄露、不安全的依赖 返回格式 {{ security_issues: [问题1严重度高, 问题2严重度中], severity: low/medium/high/critical }} 如果没有安全问题security_issues 返回空列表severity 返回 low。 response llm.invoke([HumanMessage(contentprompt)]) try: result json.loads(response.content) return { security_issues: result.get(security_issues, []), severity: result.get(severity, low), messages: [response] } except Exception: return { security_issues: [], severity: low, messages: [response] } def generate_report(state: CodeReviewState) - CodeReviewState: 节点4生成最终报告 report f# 代码审查报告 ## 质量评分{state.get(quality_score, N/A)}/100 ## 安全检查 严重度{state.get(severity, N/A)} 发现问题 {chr(10).join(f- {issue} for issue in (state.get(security_issues) or []))} ## 改进建议 {chr(10).join(f- {s} for s in (state.get(suggestions) or []))} {f⚠️ 已通过人工审批 if state.get(human_approved) else } return { messages: [{role: assistant, content: report}] }3.3 Human-in-the-loop 实现fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportinterruptdefhuman_review(state:CodeReviewState)-CodeReviewState:节点3暂停等待人工审批仅在严重问题时触发# interrupt() 会暂停图的执行等待外部输入human_inputinterrupt({question:发现严重安全问题是否继续,security_issues:state.get(security_issues),severity:state.get(severity)})return{human_approved:human_input.get(approved,False)}# 条件函数defshould_review_human(state:CodeReviewState)-str:判断是否需要人工审批severitystate.get(severity,low)ifseverityin(high,critical):returnneed_humanreturnskip_humandefshould_continue_after_human(state:CodeReviewState)-str:人工审批后的流转ifstate.get(human_approved):returnapprovedreturnrejected3.4 组装完整图fromlanggraph.graphimportStateGraph,START,END# 创建带 checkpointing 的图checkpointerMemorySaver()# 生产用 PostgresSaverbuilderStateGraph(CodeReviewState)# 添加节点builder.add_node(analyze_quality,analyze_quality)builder.add_node(check_security,check_security)builder.add_node(human_review,human_review)builder.add_node(generate_report,generate_report)# 添加边builder.add_edge(START,analyze_quality)builder.add_edge(analyze_quality,check_security)# 条件分支是否需要人工审批builder.add_conditional_edges(check_security,should_review_human,{need_human:human_review,skip_human:generate_report})# 人工审批后的分支builder.add_conditional_edges(human_review,should_continue_after_human,{approved:generate_report,rejected:END# 拒绝则直接结束})builder.add_edge(generate_report,END)# 编译带 checkpointinggraphbuilder.compile(checkpointercheckpointer)3.5 运行与断点恢复importasyncioasyncdefrun_code_review(code:str,thread_id:strreview-001):config{configurable:{thread_id:thread_id},recursion_limit:10}initial_state{messages:[],code:code,retry_count:0,max_retries:3}# 第一次运行resultawaitgraph.ainvoke(initial_state,configconfig)# 检查是否在等待人工输入stateawaitgraph.aget_state(config)ifstate.next:# 图被暂停等待 interrupt 恢复print(⚠️ 发现严重问题等待人工审批...)print(f安全问题{state.values.get(security_issues)})# 模拟人工审批实际中从 Web 界面或消息推送获取user_decisioninput(是否批准继续(y/n): )# 恢复执行resultawaitgraph.ainvoke(Command(resume{approved:user_decision.lower()y}),configconfig)returnresult# 测试代码test_code import os import subprocess def execute_user_command(user_input): # 直接执行用户输入 - 安全漏洞 result subprocess.run(user_input, shellTrue, capture_outputTrue) return result.stdout def get_user_data(user_id): # SQL 注入漏洞 query fSELECT * FROM users WHERE id {user_id} return db.execute(query) asyncio.run(run_code_review(test_code))四、状态持久化从内存到 PostgreSQL生产环境必须用持久化存储防止服务重启丢失任务状态fromlanggraph.checkpoint.postgresimportPostgresSaverfrompsycopgimportConnection# PostgreSQL 持久化conn_stringpostgresql://user:passlocalhost/langgraph_dbconnConnection.connect(conn_string)checkpointerPostgresSaver(conn)checkpointer.setup()# 创建必要的表graphbuilder.compile(checkpointercheckpointer)# 任务中断后使用相同的 thread_id 恢复config{configurable:{thread_id:review-001}}stategraph.get_state(config)# 读取之前的状态print(f当前任务状态等待步骤 {state.next})五、错误处理与重试机制defsafe_node_wrapper(node_func,max_retries3):通用的节点包装器添加重试逻辑defwrapper(state:CodeReviewState)-CodeReviewState:retry_countstate.get(retry_count,0)try:resultnode_func(state)return{**result,retry_count:0}# 成功时重置重试计数exceptExceptionase:ifretry_countmax_retries:print(f节点执行失败第{retry_count1}次重试:{e})return{retry_count:retry_count1,messages:[]}else:raiseRuntimeError(f节点重试{max_retries}次后仍然失败:{e})returnwrapper# 使用包装器builder.add_node(analyze_quality,safe_node_wrapper(analyze_quality))六、常见问题与避坑问题原因解决方案图无限循环条件边判断有误设置recursion_limit默认 25状态更新丢失使用而非 Annotated[list, add_messages]消息列表用 add_messages reducerCheckpoint 不生效compile 时没传 checkpointer确认compile(checkpointer...)interrupt 后无法恢复thread_id 不一致确保恢复时用相同 thread_id并发状态冲突多个线程写同一 state用 PostgresSaver 的事务保证七、总结LangGraph 的核心价值不是让 AI 更聪明而是让 AI 的行为可控、可观测、可恢复。有了图结构复杂任务流程可以明确定义而不是黑盒 Loop出错的节点可以单独重试而不是重跑整个任务人工干预点可以精确插入而不是靠外部轮询任务历史可以持久化而不是进程重启后全失对于要在生产环境运行 AI Agent 的工程师LangGraph 是目前最成熟的选择之一。参考文献LangGraph Documentation. https://langchain-ai.github.io/langgraph/LangChain Blog. “LangGraph: Multi-Agent Workflows.” 2024. https://blog.langchain.dev/langgraph/LangGraph. “How to add human-in-the-loop.” https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/LangGraph. “Persistence and Checkpointing.” https://langchain-ai.github.io/langgraph/concepts/persistence/Yao S, et al. “ReAct: Synergizing Reasoning and Acting in Language Models.” ICLR, 2023. https://arxiv.org/abs/2210.03629
LangGraph 单智能体状态机实战：复杂任务流的工程级编排
LangGraph 单智能体状态机实战复杂任务流的工程级编排ReAct 够用但遇到需要循环、分支、人工干预的复杂任务时就捉襟见肘。LangGraph 把 Agent 的执行流抽象成一个有向图给了工程师掌控感。本文从零开始带你用 LangGraph 构建一个能在生产环境真正运行的单智能体系统。一、为什么需要 LangGraph简单任务用 ReActReason Act 循环就够了但当 Agent 需要处理这些场景时ReAct 就显得力不从心条件分支根据工具执行结果决定下一步循环重试某一步失败后重试有最大重试次数人工审批关键操作需要暂停等待人工确认Human-in-the-loop状态持久化任务中断后可以从断点继续Checkpointing并行执行多个无依赖的工具调用同时发出LangGraph 的核心抽象图Graph 节点Node 边Edge 状态State - Node一个执行步骤LLM 调用、工具执行、人工审批... - Edge节点之间的流转关系可以是条件边 - State在整个图中流转的共享数据结构二、核心概念快速上手2.1 安装pipinstalllanggraph langchain-anthropic langchain-core2.2 最简单的 LangGraph AgentfromtypingimportAnnotated,TypedDictfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.prebuiltimportToolNode,tools_conditionfromlangchain_anthropicimportChatAnthropicfromlangchain_core.toolsimporttoolimportoperator# 1. 定义状态结构classAgentState(TypedDict):messages:Annotated[list,operator.add]# 消息列表append 操作# 2. 定义工具tooldefsearch_web(query:str)-str:搜索互联网获取最新信息# 实际实现中调用真实搜索 APIreturnf搜索结果关于{query}的内容...tooldefcalculate(expression:str)-str:计算数学表达式try:resulteval(expression)# 生产环境用更安全的方案returnstr(result)exceptExceptionase:returnf计算错误:{e}tools[search_web,calculate]# 3. 创建 LLM 并绑定工具llmChatAnthropic(modelclaude-3-5-sonnet-20241022)llm_with_toolsllm.bind_tools(tools)# 4. 定义节点函数defcall_model(state:AgentState)-AgentState:调用 LLM 生成响应responsellm_with_tools.invoke(state[messages])return{messages:[response]}# 5. 构建图graph_builderStateGraph(AgentState)# 添加节点graph_builder.add_node(agent,call_model)graph_builder.add_node(tools,ToolNode(tools))# 添加边graph_builder.add_edge(START,agent)graph_builder.add_conditional_edges(agent,tools_condition,# 内置条件有工具调用 → tools否则 → END)graph_builder.add_edge(tools,agent)# 工具执行完回到 agent# 编译graphgraph_builder.compile()三、进阶构建生产级 Agent下面构建一个代码审查 Agent——接收一段代码分析质量、检查安全漏洞、生成改进建议并在发现严重问题时暂停等待人工确认。3.1 定义复杂状态fromtypingimportAnnotated,TypedDict,Optional,Listfromlanggraph.graph.messageimportadd_messagesimportoperatorclassCodeReviewState(TypedDict):# 消息历史messages:Annotated[list,add_messages]# 被审查的代码code:str# 审查结果quality_score:Optional[int]# 0-100security_issues:Optional[List[str]]# 安全问题列表suggestions:Optional[List[str]]# 改进建议# 流程控制severity:Optional[str]# low | medium | high | criticalhuman_approved:Optional[bool]# 人工审批结果# 重试计数retry_count:intmax_retries:int3.2 定义各节点fromlangchain_anthropicimportChatAnthropicfromlangchain_core.messagesimportHumanMessage,SystemMessageimportjson llmChatAnthropic(modelclaude-3-5-sonnet-20241022)defanalyze_quality(state:CodeReviewState)-CodeReviewState:节点1分析代码质量promptf分析以下代码的质量返回 JSON 格式 代码{state[‘code’]}返回格式 {{ quality_score: 0-100的整数, suggestions: [建议1, 建议2, ...] }} response llm.invoke([HumanMessage(contentprompt)]) try: result json.loads(response.content) return { quality_score: result[quality_score], suggestions: result[suggestions], messages: [response] } except Exception: return { quality_score: 50, suggestions: [解析结果失败请手动检查], messages: [response] } def check_security(state: CodeReviewState) - CodeReviewState: 节点2安全漏洞检查 prompt f对以下代码进行安全审查返回 JSON 格式 代码{state[‘code’]}重点检查SQL 注入、XSS、命令注入、敏感信息泄露、不安全的依赖 返回格式 {{ security_issues: [问题1严重度高, 问题2严重度中], severity: low/medium/high/critical }} 如果没有安全问题security_issues 返回空列表severity 返回 low。 response llm.invoke([HumanMessage(contentprompt)]) try: result json.loads(response.content) return { security_issues: result.get(security_issues, []), severity: result.get(severity, low), messages: [response] } except Exception: return { security_issues: [], severity: low, messages: [response] } def generate_report(state: CodeReviewState) - CodeReviewState: 节点4生成最终报告 report f# 代码审查报告 ## 质量评分{state.get(quality_score, N/A)}/100 ## 安全检查 严重度{state.get(severity, N/A)} 发现问题 {chr(10).join(f- {issue} for issue in (state.get(security_issues) or []))} ## 改进建议 {chr(10).join(f- {s} for s in (state.get(suggestions) or []))} {f⚠️ 已通过人工审批 if state.get(human_approved) else } return { messages: [{role: assistant, content: report}] }3.3 Human-in-the-loop 实现fromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.typesimportinterruptdefhuman_review(state:CodeReviewState)-CodeReviewState:节点3暂停等待人工审批仅在严重问题时触发# interrupt() 会暂停图的执行等待外部输入human_inputinterrupt({question:发现严重安全问题是否继续,security_issues:state.get(security_issues),severity:state.get(severity)})return{human_approved:human_input.get(approved,False)}# 条件函数defshould_review_human(state:CodeReviewState)-str:判断是否需要人工审批severitystate.get(severity,low)ifseverityin(high,critical):returnneed_humanreturnskip_humandefshould_continue_after_human(state:CodeReviewState)-str:人工审批后的流转ifstate.get(human_approved):returnapprovedreturnrejected3.4 组装完整图fromlanggraph.graphimportStateGraph,START,END# 创建带 checkpointing 的图checkpointerMemorySaver()# 生产用 PostgresSaverbuilderStateGraph(CodeReviewState)# 添加节点builder.add_node(analyze_quality,analyze_quality)builder.add_node(check_security,check_security)builder.add_node(human_review,human_review)builder.add_node(generate_report,generate_report)# 添加边builder.add_edge(START,analyze_quality)builder.add_edge(analyze_quality,check_security)# 条件分支是否需要人工审批builder.add_conditional_edges(check_security,should_review_human,{need_human:human_review,skip_human:generate_report})# 人工审批后的分支builder.add_conditional_edges(human_review,should_continue_after_human,{approved:generate_report,rejected:END# 拒绝则直接结束})builder.add_edge(generate_report,END)# 编译带 checkpointinggraphbuilder.compile(checkpointercheckpointer)3.5 运行与断点恢复importasyncioasyncdefrun_code_review(code:str,thread_id:strreview-001):config{configurable:{thread_id:thread_id},recursion_limit:10}initial_state{messages:[],code:code,retry_count:0,max_retries:3}# 第一次运行resultawaitgraph.ainvoke(initial_state,configconfig)# 检查是否在等待人工输入stateawaitgraph.aget_state(config)ifstate.next:# 图被暂停等待 interrupt 恢复print(⚠️ 发现严重问题等待人工审批...)print(f安全问题{state.values.get(security_issues)})# 模拟人工审批实际中从 Web 界面或消息推送获取user_decisioninput(是否批准继续(y/n): )# 恢复执行resultawaitgraph.ainvoke(Command(resume{approved:user_decision.lower()y}),configconfig)returnresult# 测试代码test_code import os import subprocess def execute_user_command(user_input): # 直接执行用户输入 - 安全漏洞 result subprocess.run(user_input, shellTrue, capture_outputTrue) return result.stdout def get_user_data(user_id): # SQL 注入漏洞 query fSELECT * FROM users WHERE id {user_id} return db.execute(query) asyncio.run(run_code_review(test_code))四、状态持久化从内存到 PostgreSQL生产环境必须用持久化存储防止服务重启丢失任务状态fromlanggraph.checkpoint.postgresimportPostgresSaverfrompsycopgimportConnection# PostgreSQL 持久化conn_stringpostgresql://user:passlocalhost/langgraph_dbconnConnection.connect(conn_string)checkpointerPostgresSaver(conn)checkpointer.setup()# 创建必要的表graphbuilder.compile(checkpointercheckpointer)# 任务中断后使用相同的 thread_id 恢复config{configurable:{thread_id:review-001}}stategraph.get_state(config)# 读取之前的状态print(f当前任务状态等待步骤 {state.next})五、错误处理与重试机制defsafe_node_wrapper(node_func,max_retries3):通用的节点包装器添加重试逻辑defwrapper(state:CodeReviewState)-CodeReviewState:retry_countstate.get(retry_count,0)try:resultnode_func(state)return{**result,retry_count:0}# 成功时重置重试计数exceptExceptionase:ifretry_countmax_retries:print(f节点执行失败第{retry_count1}次重试:{e})return{retry_count:retry_count1,messages:[]}else:raiseRuntimeError(f节点重试{max_retries}次后仍然失败:{e})returnwrapper# 使用包装器builder.add_node(analyze_quality,safe_node_wrapper(analyze_quality))六、常见问题与避坑问题原因解决方案图无限循环条件边判断有误设置recursion_limit默认 25状态更新丢失使用而非 Annotated[list, add_messages]消息列表用 add_messages reducerCheckpoint 不生效compile 时没传 checkpointer确认compile(checkpointer...)interrupt 后无法恢复thread_id 不一致确保恢复时用相同 thread_id并发状态冲突多个线程写同一 state用 PostgresSaver 的事务保证七、总结LangGraph 的核心价值不是让 AI 更聪明而是让 AI 的行为可控、可观测、可恢复。有了图结构复杂任务流程可以明确定义而不是黑盒 Loop出错的节点可以单独重试而不是重跑整个任务人工干预点可以精确插入而不是靠外部轮询任务历史可以持久化而不是进程重启后全失对于要在生产环境运行 AI Agent 的工程师LangGraph 是目前最成熟的选择之一。参考文献LangGraph Documentation. https://langchain-ai.github.io/langgraph/LangChain Blog. “LangGraph: Multi-Agent Workflows.” 2024. https://blog.langchain.dev/langgraph/LangGraph. “How to add human-in-the-loop.” https://langchain-ai.github.io/langgraph/how-tos/human_in_the_loop/LangGraph. “Persistence and Checkpointing.” https://langchain-ai.github.io/langgraph/concepts/persistence/Yao S, et al. “ReAct: Synergizing Reasoning and Acting in Language Models.” ICLR, 2023. https://arxiv.org/abs/2210.03629