大家好我是专注于AI应用开发的技术博主。在构建复杂的AI工作流时你是否遇到过这样的困境多个LLM调用、工具使用、状态管理、条件分支等逻辑交织在一起代码迅速变得难以维护和调试传统的线性脚本或简单的LangChain链在应对这类复杂场景时往往力不从心。本文将为你系统性地拆解LangGraph——一个专为构建有状态、多步骤的智能体Agent工作流而生的框架。通过本文你将不仅理解其核心概念更能亲手搭建从单智能体到多智能体协作的完整项目掌握构建下一代AI应用的核心架构能力。1. LangGraph核心概念为什么是它在深入代码之前我们首先要理解LangGraph解决了什么问题以及它与我们熟知的LangChain有何不同。LangChain是一个强大的库它将大语言模型LLM与各种工具、数据源连接起来。其核心抽象是“链”Chain它将不同的组件按顺序连接起来执行任务。然而当任务需要循环、条件分支或长期维护一个共享状态时单纯的链就显得有些笨拙。LangGraph应运而生。它建立在LangChain之上引入了图Graph的计算模型。你可以将工作流中的每个步骤如调用LLM、执行工具、检查条件视为图中的一个节点Node步骤之间的流转由边Edge来定义。更重要的是整个图共享一个持久化的状态State对象这使得构建具有记忆、能够根据中间结果动态决定下一步行动的智能体系统变得异常清晰和强大。核心优势对比LangChain Chain 适合线性、确定性的管道式任务。例如问答 - 检索 - 生成答案。LangGraph Graph 适合非线性、有状态、多分支的智能体任务。例如智能体接收问题 - 决定调用工具A - 根据工具A的结果决定是继续调用工具B还是直接给出答案 - 循环直到满足条件。简单来说LangChain提供了丰富的“积木”LLM、工具、记忆等而LangGraph提供了组装这些积木以构建复杂、动态“机器”的蓝图和引擎。2. 环境准备与项目搭建工欲善其事必先利其器。我们先来搭建一个干净的开发环境。2.1 环境与依赖本文示例基于Python 3.9。建议使用虚拟环境如venv或conda来管理依赖。首先安装核心库。我们将使用OpenAI的GPT模型作为LLM引擎因此需要同时安装LangChain、LangGraph和OpenAI的SDK。# 创建并激活虚拟环境以venv为例 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows # 安装依赖 pip install langgraph langchain-openai重要提示你需要一个有效的OpenAI API密钥。请将其设置为环境变量不要在代码中硬编码。# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here2.2 项目结构创建一个清晰的项目结构有助于管理复杂的图定义。建议如下your_project/ ├── agents/ │ ├── __init__.py │ ├── basic_agent.py # 基础单智能体 │ └── multi_agent.py # 多智能体系统 ├── tools/ │ ├── __init__.py │ └── custom_tools.py # 自定义工具 ├── graphs/ │ ├── __init__.py │ └── workflow_graph.py # 图定义 ├── state.py # 自定义状态类定义 └── main.py # 主入口文件3. 核心组件深度解析理解LangGraph关键在于掌握其四个核心组件State状态、Node节点、Edge边和Graph图。3.1 State工作流的记忆核心State是一个Pydantic模型它定义了在整个图执行过程中传递和修改的数据结构。所有节点都读取和更新这个共享状态。# state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史记录用户、AI、工具的对话 messages: Annotated[List, add_messages] # 当前用户输入的问题 question: str # 智能体思考的步骤或中间结果 reasoning: str # 从工具调用中获取的最终答案或数据 final_answer: strTypedDict 用于定义状态的结构。Annotated和add_messages 这是一个LangGraph提供的特殊注解用于自动合并消息列表而不是覆盖它。这对于维护对话历史至关重要。你可以根据需求添加任意字段如intermediate_steps,selected_tool,iteration_count等。3.2 Node执行单元节点是一个普通的Python函数或可调用对象它接收当前State执行一些操作如调用LLM、运行工具然后返回一个包含对State所做更新的字典。# graphs/workflow_graph.py 片段 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) def call_llm_node(state: State): 节点调用LLM分析问题 question state[question] history state[messages] # 构建Prompt包含历史对话和当前问题 prompt f 基于以下对话历史和最新问题请分析用户意图。 历史对话{history[-5:]} # 取最近5条 最新问题{question} 请简要分析用户想做什么。 # 调用LLM response llm.invoke(prompt) # 返回要更新到State中的内容 return {reasoning: response.content}3.3 Edge流程控制器边决定了执行完一个节点后下一步应该去哪个节点。边可以是条件边Conditional Edge或普通边。普通边 直接指向下一个节点。条件边 根据State中的某个值动态决定下一个节点。这通常由一个路由函数Router实现。from langgraph.graph import END def should_use_tool(state: State) - str: 路由函数根据分析结果决定下一步 reasoning state.get(reasoning, ).lower() if 需要计算 in reasoning or 需要查询 in reasoning: return use_tool # 前往工具调用节点 elif 可以直接回答 in reasoning: return generate_answer # 前往答案生成节点 else: return ask_for_clarification # 前往请求澄清节点3.4 Graph组装与编译这是将节点和边组装成可执行工作流的地方。# graphs/workflow_graph.py from langgraph.graph import StateGraph, START # 1. 创建图构建器并指定State的类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(analyze, call_llm_node) # 分析节点 workflow.add_node(use_tool, tool_node) # 工具调用节点 workflow.add_node(generate_answer, answer_node) # 答案生成节点 workflow.add_node(ask_clarify, clarify_node) # 澄清节点 # 3. 设置入口点 workflow.set_entry_point(analyze) # 4. 添加边包括条件边 workflow.add_conditional_edges( analyze, # 源节点 should_use_tool, # 路由函数 { use_tool: use_tool, generate_answer: generate_answer, ask_for_clarification: ask_clarify } ) workflow.add_edge(use_tool, generate_answer) workflow.add_edge(ask_clarify, END) # END是LangGraph内置的终止节点 # 5. 编译图得到可执行对象 app workflow.compile()编译后的app就是一个可以处理状态、运行工作流的强大对象。4. 实战一构建你的第一个单智能体让我们构建一个能进行简单数学计算和知识问答的单一智能体。4.1 定义工具首先我们为智能体装备“武器”——工具。# tools/custom_tools.py from langchain.tools import tool import math tool def calculate(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /, **, sqrt等。 例如3 5 * 2, sqrt(16) try: # 警告在生产环境中直接eval是危险的此处仅用于演示。 # 应使用安全的表达式解析库如 ast.literal_eval 或自定义解析器。 result eval(expression, {__builtins__: None}, {sqrt: math.sqrt, math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def get_word_length(word: str) - str: 返回输入单词的长度。 return f单词 {word} 的长度是 {len(word)} 个字母。 # 将工具放入列表供智能体使用 tools [calculate, get_word_length]4.2 创建智能体执行器我们将使用LangChain提供的create_react_agent来快速创建一个基于ReAct范式的智能体。# agents/basic_agent.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools.custom_tools import tools # 1. 拉取一个预设的ReAct提示词模板 prompt hub.pull(hwchase17/react) # 2. 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 3. 创建智能体 agent create_react_agent(llm, tools, prompt) # 4. 创建执行器负责运行智能体并处理工具调用循环 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 使用示例 if __name__ __main__: result agent_executor.invoke({input: 请问15的平方根加上20除以4等于多少}) print(智能体最终回答:, result[output])运行这个脚本你会看到详细的执行过程verboseTrue包括智能体的思考Thought、行动Action、工具输入Action Input和观察Observation直到最终输出Final Answer。4.3 用LangGraph包装智能体现在我们将这个智能体执行器嵌入到LangGraph图中以获得更强的流程控制能力。# graphs/basic_agent_graph.py from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from agents.basic_agent import agent, agent_executor from state import State import json # 包装工具执行器 tool_executor ToolExecutor(tools) def agent_node(state: State): 节点运行智能体决定下一步是使用工具还是结束 # 调用智能体执行器 agent_response agent_executor.invoke(state) # 假设agent_response包含下一步指令 # 这里简化处理实际应根据ReAct输出解析 return {messages: agent_response[messages]} def tool_node(state: State): 节点执行工具调用 last_message state[messages][-1] # 解析出要调用的工具名和输入 tool_call last_message.additional_kwargs.get(tool_calls)[0] tool_name tool_call[function][name] tool_args json.loads(tool_call[function][arguments]) # 执行工具 result tool_executor.invoke({tool_name: tool_args}) # 将工具结果作为一条新消息添加到状态 return {messages: [{role: tool, content: result}]} # 构建图 builder StateGraph(State) builder.add_node(agent, agent_node) builder.add_node(action, tool_node) builder.set_entry_point(agent) # 定义条件边根据agent输出的内容判断是否调用工具 def route_after_agent(state: State): last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return action return END builder.add_conditional_edges(agent, route_after_agent) builder.add_edge(action, agent) # 工具执行后返回agent继续思考 graph builder.compile()这个图实现了经典的“思考-行动-观察”循环直到智能体决定不再调用工具为止。5. 实战二设计多智能体协作系统单智能体能力有限。复杂任务通常需要多个各司其职的智能体协作完成。例如一个“研究助手”系统可能包含规划者、研究者、写作者、评审者。5.1 定义多智能体状态状态需要能容纳多个智能体的交互信息。# state_multi.py from typing import TypedDict, List, Optional from langchain_core.messages import BaseMessage class MultiAgentState(TypedDict): 多智能体协作状态 task: str # 原始任务 plan: Optional[str] # 规划者生成的计划 research_data: List[str] # 研究者收集的资料 draft: Optional[str] # 写作者生成的草稿 critique: Optional[str] # 评审者提出的意见 final_report: Optional[str] # 最终报告 messages: List[BaseMessage] # 完整的对话记录 current_agent: str # 当前正在执行的智能体角色5.2 实现各角色智能体每个智能体都是一个独立的节点函数拥有特定的提示词和工具集。# agents/multi_agents.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) def planner_agent(state: MultiAgentState): 规划者拆解任务制定步骤 prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深项目规划师。请将用户复杂的任务拆解成一个清晰、可执行的步骤计划。), (human, 任务{task}) ]) chain prompt | llm plan chain.invoke({task: state[task]}) return {plan: plan.content, current_agent: researcher} def researcher_agent(state: MultiAgentState): 研究者根据计划搜索或生成关键信息 # 模拟研究过程实际应接入搜索工具 prompt ChatPromptTemplate.from_messages([ (system, 你是一个研究员。根据以下任务和计划生成关键的研究要点和数据。), (human, f任务{state[task]}\n计划{state[plan]}) ]) chain prompt | llm research chain.invoke({}) # 模拟收集到多条资料 research_data state.get(research_data, []) research_data.append(research.content) return {research_data: research_data, current_agent: writer} def writer_agent(state: MultiAgentState): 写作者整合研究资料撰写草稿 combined_data \n.join(state[research_data]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个技术写作者。请根据以下研究资料撰写一份结构清晰、内容详实的报告草稿。), (human, f研究资料\n{combined_data}) ]) chain prompt | llm draft chain.invoke({}) return {draft: draft.content, current_agent: critic} def critic_agent(state: MultiAgentState): 评审者评审草稿提出修改意见 prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的评审专家。请仔细审阅以下报告草稿指出其在逻辑、事实、结构和语言上的问题并提出具体的修改建议。), (human, f报告草稿\n{state[draft]}) ]) chain prompt | llm critique chain.invoke({}) # 根据评审意见决定是返回修改还是定稿 if 需要重大修改 in critique.content: # 简单模拟判断 return {critique: critique.content, current_agent: writer} # 返回写作者修改 else: # 进入最终定稿环节 return {critique: critique.content, current_agent: finalizer}5.3 构建协作图使用条件边根据current_agent状态字段来路由工作流。# graphs/multi_agent_graph.py from langgraph.graph import StateGraph, START, END from state_multi import MultiAgentState from agents.multi_agents import planner_agent, researcher_agent, writer_agent, critic_agent # 定义最终定稿节点 def finalizer_node(state: MultiAgentState): 最终定稿节点整合草稿和评审意见生成最终报告 final_prompt f 请根据以下草稿和评审意见生成最终版本的报告。 ---草稿--- {state[draft]} ---评审意见--- {state[critique]} ---最终报告--- # 这里可以调用LLM进行最终润色我们简化为直接使用草稿 final_report state[draft] f\n\n【已根据评审意见修正】\n评审意见摘要{state[critique][:100]}... return {final_report: final_report, current_agent: __end__} # 构建图 builder StateGraph(MultiAgentState) # 添加所有节点 builder.add_node(planner, planner_agent) builder.add_node(researcher, researcher_agent) builder.add_node(writer, writer_agent) builder.add_node(critic, critic_agent) builder.add_node(finalizer, finalizer_node) builder.set_entry_point(planner) # 定义路由逻辑根据状态中的current_agent字段决定下一步 def route_agent(state: MultiAgentState): next_agent state.get(current_agent, __end__) if next_agent __end__: return END return next_agent # 返回下一个节点的名称 # 为所有需要路由的节点添加条件边 builder.add_conditional_edges(planner, route_agent) builder.add_conditional_edges(researcher, route_agent) builder.add_conditional_edges(writer, route_agent) builder.add_conditional_edges(critic, route_agent) # finalizer 节点后直接结束 builder.add_edge(finalizer, END) collaboration_graph builder.compile()5.4 运行与可视化现在我们可以运行这个多智能体系统并可视化其执行流程。# main.py from graphs.multi_agent_graph import collaboration_graph from state_multi import MultiAgentState # 初始化状态 initial_state: MultiAgentState { task: 请撰写一篇关于LangGraph在构建多智能体系统中的优势与挑战的短文。, plan: None, research_data: [], draft: None, critique: None, final_report: None, messages: [], current_agent: planner } # 运行图 final_state collaboration_graph.invoke(initial_state) print(*50) print(任务:, final_state[task]) print(*50) print(生成的计划:, final_state[plan]) print(*50) print(收集的研究资料数量:, len(final_state[research_data])) print(*50) print(最终报告预览:, final_state[final_report][:500], ...) print(*50) # 可视化图结构需要安装graphviz try: from IPython.display import Image, display image_data collaboration_graph.get_graph().draw_mermaid_png() display(Image(image_data)) except ImportError: # 非Jupyter环境可以保存为文件 print(如需可视化请在Jupyter环境中运行或安装pygraphviz并调用get_graph().draw_mermaid_png()保存。) # 也可以打印文本结构 print(\n图结构) print(collaboration_graph.get_graph().print_ascii())6. 常见问题与排查思路在开发LangGraph应用时你可能会遇到以下典型问题。问题现象可能原因排查思路与解决方案KeyError访问State字段State的TypedDict定义与实际访问的键不匹配。1. 检查State类的定义确保字段名拼写正确。2. 在节点函数开头打印state.keys()确认当前状态包含哪些键。3. 使用state.get(key, default)提供默认值避免直接访问。图编译失败或执行报错节点函数返回值格式错误边指向不存在的节点。1. 确保每个节点函数返回一个字典其键是State中定义的字段名。2. 检查add_edge和add_conditional_edges中引用的节点名称是否都已通过add_node添加。3. 路由函数should_use_tool必须返回一个字符串且该字符串是条件映射{...}中的一个键。智能体陷入无限循环条件边逻辑有误导致在两个或多个节点间来回跳转无法到达END。1. 在状态中增加iteration_count字段并在节点中递增它超过阈值则强制结束。2. 仔细检查路由函数的逻辑确保存在能通向END的分支。3. 使用verboseTrue运行图观察状态变化和路由决策。工具调用失败或解析错误工具的参数格式与LLM生成的调用不匹配工具本身抛出异常。1. 为AgentExecutor设置handle_parsing_errorsTrue。2. 在工具函数内部做好异常捕获返回清晰的错误信息。3. 确保工具的tool装饰器中的描述清晰帮助LLM正确使用。多智能体协作中信息丢失后一个节点覆盖了前一个节点写入State的数据。1. 确保每个节点只更新自己负责的字段使用return {“field_to_update”: value}而不是return state。2. 对于列表类数据如messages使用Annotated和add_messages等归约器Reducer来追加而非覆盖。图可视化无法显示缺少graphviz或pygraphviz依赖。1. 安装系统级的Graphvizsudo apt-get install graphviz(Linux) 或brew install graphviz(Mac)。2. 安装Python绑定pip install pygraphviz可能较难安装或使用pip install langgraph[vis]安装包含可视化支持的版本。3. 回退方案使用graph.get_graph().print_ascii()在控制台查看文本结构。7. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下实践能大幅提升代码的健壮性和可维护性。7.1 状态设计原则最小化与清晰化 State只存储工作流真正需要共享和传递的数据。避免将临时变量或中间计算结果全部塞入State。使用强类型 坚持使用TypedDict或PydanticBaseModel定义State这能在开发早期借助IDE和mypy发现类型错误。合理使用归约器 对于列表、集合等需要累积操作的数据务必使用Annotated和LangGraph提供的归约器如add_messages这是实现正确状态更新的关键。7.2 节点与图结构设计节点职责单一 每个节点应只完成一件明确的事情如“调用LLM”、“执行工具A”、“验证结果”。这使调试和测试更容易。善用子图 对于复杂的、可复用的逻辑序列可以将其封装成一个子图Subgraph然后作为单个节点加入主图。这能极大提升模块化程度。预编译与复用 图app workflow.compile()的编译有一定开销。在Web服务等场景中应将编译好的图实例作为全局或单例对象复用而不是每次请求都重新编译。7.3 错误处理与可观测性节点级异常捕获 在关键的节点函数内部使用try...except将异常信息捕获并作为状态的一部分传递到后续节点或专门的处理节点而不是让整个图崩溃。添加检查点与超时 对于可能长时间运行或卡住的循环在状态中设置计数器或时间戳并在路由函数中检查超时则导向人工干预或失败处理节点。全面日志记录 除了使用verboseTrue应在关键步骤如状态转换、路由决策、工具调用前后记录结构化的日志便于后期追踪和审计。可以将state的摘要信息记录下来。7.4 生产环境部署考量状态持久化 LangGraph默认在内存中维护状态。对于长时间运行或需要中断恢复的工作流需要实现检查点Checkpoint功能将状态持久化到数据库如Redis、PostgreSQL。LangGraph提供了相关的接口Checkpointer。异步支持 如果节点涉及网络IO如调用外部API、数据库查询应使用异步节点async def和图StateGraph的异步版本以提高并发性能。安全性 尤其注意工具调用的安全性。避免像示例中那样直接使用eval。对于执行代码、访问文件系统或网络的操作必须进行严格的输入验证、权限控制和沙箱隔离。掌握LangGraph意味着你掌握了构建复杂、可靠、可维护的AI智能体系统的核心框架。从定义清晰的状态模型开始通过节点和边组装业务逻辑再利用条件边实现动态流程控制你可以设计出从前端对话机器人到后端自动化流程引擎的各种应用。建议从本文的单智能体示例开始亲手运行并修改然后逐步挑战多智能体协作项目。在实践中你会更深刻地体会到图计算模型给AI应用开发带来的灵活性与强大威力。 LangGraph实战：从零构建有状态多智能体AI工作流
大家好我是专注于AI应用开发的技术博主。在构建复杂的AI工作流时你是否遇到过这样的困境多个LLM调用、工具使用、状态管理、条件分支等逻辑交织在一起代码迅速变得难以维护和调试传统的线性脚本或简单的LangChain链在应对这类复杂场景时往往力不从心。本文将为你系统性地拆解LangGraph——一个专为构建有状态、多步骤的智能体Agent工作流而生的框架。通过本文你将不仅理解其核心概念更能亲手搭建从单智能体到多智能体协作的完整项目掌握构建下一代AI应用的核心架构能力。1. LangGraph核心概念为什么是它在深入代码之前我们首先要理解LangGraph解决了什么问题以及它与我们熟知的LangChain有何不同。LangChain是一个强大的库它将大语言模型LLM与各种工具、数据源连接起来。其核心抽象是“链”Chain它将不同的组件按顺序连接起来执行任务。然而当任务需要循环、条件分支或长期维护一个共享状态时单纯的链就显得有些笨拙。LangGraph应运而生。它建立在LangChain之上引入了图Graph的计算模型。你可以将工作流中的每个步骤如调用LLM、执行工具、检查条件视为图中的一个节点Node步骤之间的流转由边Edge来定义。更重要的是整个图共享一个持久化的状态State对象这使得构建具有记忆、能够根据中间结果动态决定下一步行动的智能体系统变得异常清晰和强大。核心优势对比LangChain Chain 适合线性、确定性的管道式任务。例如问答 - 检索 - 生成答案。LangGraph Graph 适合非线性、有状态、多分支的智能体任务。例如智能体接收问题 - 决定调用工具A - 根据工具A的结果决定是继续调用工具B还是直接给出答案 - 循环直到满足条件。简单来说LangChain提供了丰富的“积木”LLM、工具、记忆等而LangGraph提供了组装这些积木以构建复杂、动态“机器”的蓝图和引擎。2. 环境准备与项目搭建工欲善其事必先利其器。我们先来搭建一个干净的开发环境。2.1 环境与依赖本文示例基于Python 3.9。建议使用虚拟环境如venv或conda来管理依赖。首先安装核心库。我们将使用OpenAI的GPT模型作为LLM引擎因此需要同时安装LangChain、LangGraph和OpenAI的SDK。# 创建并激活虚拟环境以venv为例 python -m venv langgraph-env source langgraph-env/bin/activate # Linux/Mac # langgraph-env\Scripts\activate # Windows # 安装依赖 pip install langgraph langchain-openai重要提示你需要一个有效的OpenAI API密钥。请将其设置为环境变量不要在代码中硬编码。# Linux/Mac export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-api-key-here2.2 项目结构创建一个清晰的项目结构有助于管理复杂的图定义。建议如下your_project/ ├── agents/ │ ├── __init__.py │ ├── basic_agent.py # 基础单智能体 │ └── multi_agent.py # 多智能体系统 ├── tools/ │ ├── __init__.py │ └── custom_tools.py # 自定义工具 ├── graphs/ │ ├── __init__.py │ └── workflow_graph.py # 图定义 ├── state.py # 自定义状态类定义 └── main.py # 主入口文件3. 核心组件深度解析理解LangGraph关键在于掌握其四个核心组件State状态、Node节点、Edge边和Graph图。3.1 State工作流的记忆核心State是一个Pydantic模型它定义了在整个图执行过程中传递和修改的数据结构。所有节点都读取和更新这个共享状态。# state.py from typing import TypedDict, Annotated, List from langgraph.graph.message import add_messages import operator class State(TypedDict): # 消息历史记录用户、AI、工具的对话 messages: Annotated[List, add_messages] # 当前用户输入的问题 question: str # 智能体思考的步骤或中间结果 reasoning: str # 从工具调用中获取的最终答案或数据 final_answer: strTypedDict 用于定义状态的结构。Annotated和add_messages 这是一个LangGraph提供的特殊注解用于自动合并消息列表而不是覆盖它。这对于维护对话历史至关重要。你可以根据需求添加任意字段如intermediate_steps,selected_tool,iteration_count等。3.2 Node执行单元节点是一个普通的Python函数或可调用对象它接收当前State执行一些操作如调用LLM、运行工具然后返回一个包含对State所做更新的字典。# graphs/workflow_graph.py 片段 from langchain_openai import ChatOpenAI llm ChatOpenAI(modelgpt-4o-mini) def call_llm_node(state: State): 节点调用LLM分析问题 question state[question] history state[messages] # 构建Prompt包含历史对话和当前问题 prompt f 基于以下对话历史和最新问题请分析用户意图。 历史对话{history[-5:]} # 取最近5条 最新问题{question} 请简要分析用户想做什么。 # 调用LLM response llm.invoke(prompt) # 返回要更新到State中的内容 return {reasoning: response.content}3.3 Edge流程控制器边决定了执行完一个节点后下一步应该去哪个节点。边可以是条件边Conditional Edge或普通边。普通边 直接指向下一个节点。条件边 根据State中的某个值动态决定下一个节点。这通常由一个路由函数Router实现。from langgraph.graph import END def should_use_tool(state: State) - str: 路由函数根据分析结果决定下一步 reasoning state.get(reasoning, ).lower() if 需要计算 in reasoning or 需要查询 in reasoning: return use_tool # 前往工具调用节点 elif 可以直接回答 in reasoning: return generate_answer # 前往答案生成节点 else: return ask_for_clarification # 前往请求澄清节点3.4 Graph组装与编译这是将节点和边组装成可执行工作流的地方。# graphs/workflow_graph.py from langgraph.graph import StateGraph, START # 1. 创建图构建器并指定State的类型 workflow StateGraph(State) # 2. 添加节点 workflow.add_node(analyze, call_llm_node) # 分析节点 workflow.add_node(use_tool, tool_node) # 工具调用节点 workflow.add_node(generate_answer, answer_node) # 答案生成节点 workflow.add_node(ask_clarify, clarify_node) # 澄清节点 # 3. 设置入口点 workflow.set_entry_point(analyze) # 4. 添加边包括条件边 workflow.add_conditional_edges( analyze, # 源节点 should_use_tool, # 路由函数 { use_tool: use_tool, generate_answer: generate_answer, ask_for_clarification: ask_clarify } ) workflow.add_edge(use_tool, generate_answer) workflow.add_edge(ask_clarify, END) # END是LangGraph内置的终止节点 # 5. 编译图得到可执行对象 app workflow.compile()编译后的app就是一个可以处理状态、运行工作流的强大对象。4. 实战一构建你的第一个单智能体让我们构建一个能进行简单数学计算和知识问答的单一智能体。4.1 定义工具首先我们为智能体装备“武器”——工具。# tools/custom_tools.py from langchain.tools import tool import math tool def calculate(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /, **, sqrt等。 例如3 5 * 2, sqrt(16) try: # 警告在生产环境中直接eval是危险的此处仅用于演示。 # 应使用安全的表达式解析库如 ast.literal_eval 或自定义解析器。 result eval(expression, {__builtins__: None}, {sqrt: math.sqrt, math: math}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def get_word_length(word: str) - str: 返回输入单词的长度。 return f单词 {word} 的长度是 {len(word)} 个字母。 # 将工具放入列表供智能体使用 tools [calculate, get_word_length]4.2 创建智能体执行器我们将使用LangChain提供的create_react_agent来快速创建一个基于ReAct范式的智能体。# agents/basic_agent.py from langchain import hub from langchain.agents import create_react_agent, AgentExecutor from langchain_openai import ChatOpenAI from tools.custom_tools import tools # 1. 拉取一个预设的ReAct提示词模板 prompt hub.pull(hwchase17/react) # 2. 初始化LLM llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 3. 创建智能体 agent create_react_agent(llm, tools, prompt) # 4. 创建执行器负责运行智能体并处理工具调用循环 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 使用示例 if __name__ __main__: result agent_executor.invoke({input: 请问15的平方根加上20除以4等于多少}) print(智能体最终回答:, result[output])运行这个脚本你会看到详细的执行过程verboseTrue包括智能体的思考Thought、行动Action、工具输入Action Input和观察Observation直到最终输出Final Answer。4.3 用LangGraph包装智能体现在我们将这个智能体执行器嵌入到LangGraph图中以获得更强的流程控制能力。# graphs/basic_agent_graph.py from langgraph.graph import StateGraph, START, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from agents.basic_agent import agent, agent_executor from state import State import json # 包装工具执行器 tool_executor ToolExecutor(tools) def agent_node(state: State): 节点运行智能体决定下一步是使用工具还是结束 # 调用智能体执行器 agent_response agent_executor.invoke(state) # 假设agent_response包含下一步指令 # 这里简化处理实际应根据ReAct输出解析 return {messages: agent_response[messages]} def tool_node(state: State): 节点执行工具调用 last_message state[messages][-1] # 解析出要调用的工具名和输入 tool_call last_message.additional_kwargs.get(tool_calls)[0] tool_name tool_call[function][name] tool_args json.loads(tool_call[function][arguments]) # 执行工具 result tool_executor.invoke({tool_name: tool_args}) # 将工具结果作为一条新消息添加到状态 return {messages: [{role: tool, content: result}]} # 构建图 builder StateGraph(State) builder.add_node(agent, agent_node) builder.add_node(action, tool_node) builder.set_entry_point(agent) # 定义条件边根据agent输出的内容判断是否调用工具 def route_after_agent(state: State): last_message state[messages][-1] if hasattr(last_message, tool_calls) and last_message.tool_calls: return action return END builder.add_conditional_edges(agent, route_after_agent) builder.add_edge(action, agent) # 工具执行后返回agent继续思考 graph builder.compile()这个图实现了经典的“思考-行动-观察”循环直到智能体决定不再调用工具为止。5. 实战二设计多智能体协作系统单智能体能力有限。复杂任务通常需要多个各司其职的智能体协作完成。例如一个“研究助手”系统可能包含规划者、研究者、写作者、评审者。5.1 定义多智能体状态状态需要能容纳多个智能体的交互信息。# state_multi.py from typing import TypedDict, List, Optional from langchain_core.messages import BaseMessage class MultiAgentState(TypedDict): 多智能体协作状态 task: str # 原始任务 plan: Optional[str] # 规划者生成的计划 research_data: List[str] # 研究者收集的资料 draft: Optional[str] # 写作者生成的草稿 critique: Optional[str] # 评审者提出的意见 final_report: Optional[str] # 最终报告 messages: List[BaseMessage] # 完整的对话记录 current_agent: str # 当前正在执行的智能体角色5.2 实现各角色智能体每个智能体都是一个独立的节点函数拥有特定的提示词和工具集。# agents/multi_agents.py from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate llm ChatOpenAI(modelgpt-4o-mini, temperature0.7) def planner_agent(state: MultiAgentState): 规划者拆解任务制定步骤 prompt ChatPromptTemplate.from_messages([ (system, 你是一个资深项目规划师。请将用户复杂的任务拆解成一个清晰、可执行的步骤计划。), (human, 任务{task}) ]) chain prompt | llm plan chain.invoke({task: state[task]}) return {plan: plan.content, current_agent: researcher} def researcher_agent(state: MultiAgentState): 研究者根据计划搜索或生成关键信息 # 模拟研究过程实际应接入搜索工具 prompt ChatPromptTemplate.from_messages([ (system, 你是一个研究员。根据以下任务和计划生成关键的研究要点和数据。), (human, f任务{state[task]}\n计划{state[plan]}) ]) chain prompt | llm research chain.invoke({}) # 模拟收集到多条资料 research_data state.get(research_data, []) research_data.append(research.content) return {research_data: research_data, current_agent: writer} def writer_agent(state: MultiAgentState): 写作者整合研究资料撰写草稿 combined_data \n.join(state[research_data]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个技术写作者。请根据以下研究资料撰写一份结构清晰、内容详实的报告草稿。), (human, f研究资料\n{combined_data}) ]) chain prompt | llm draft chain.invoke({}) return {draft: draft.content, current_agent: critic} def critic_agent(state: MultiAgentState): 评审者评审草稿提出修改意见 prompt ChatPromptTemplate.from_messages([ (system, 你是一个严格的评审专家。请仔细审阅以下报告草稿指出其在逻辑、事实、结构和语言上的问题并提出具体的修改建议。), (human, f报告草稿\n{state[draft]}) ]) chain prompt | llm critique chain.invoke({}) # 根据评审意见决定是返回修改还是定稿 if 需要重大修改 in critique.content: # 简单模拟判断 return {critique: critique.content, current_agent: writer} # 返回写作者修改 else: # 进入最终定稿环节 return {critique: critique.content, current_agent: finalizer}5.3 构建协作图使用条件边根据current_agent状态字段来路由工作流。# graphs/multi_agent_graph.py from langgraph.graph import StateGraph, START, END from state_multi import MultiAgentState from agents.multi_agents import planner_agent, researcher_agent, writer_agent, critic_agent # 定义最终定稿节点 def finalizer_node(state: MultiAgentState): 最终定稿节点整合草稿和评审意见生成最终报告 final_prompt f 请根据以下草稿和评审意见生成最终版本的报告。 ---草稿--- {state[draft]} ---评审意见--- {state[critique]} ---最终报告--- # 这里可以调用LLM进行最终润色我们简化为直接使用草稿 final_report state[draft] f\n\n【已根据评审意见修正】\n评审意见摘要{state[critique][:100]}... return {final_report: final_report, current_agent: __end__} # 构建图 builder StateGraph(MultiAgentState) # 添加所有节点 builder.add_node(planner, planner_agent) builder.add_node(researcher, researcher_agent) builder.add_node(writer, writer_agent) builder.add_node(critic, critic_agent) builder.add_node(finalizer, finalizer_node) builder.set_entry_point(planner) # 定义路由逻辑根据状态中的current_agent字段决定下一步 def route_agent(state: MultiAgentState): next_agent state.get(current_agent, __end__) if next_agent __end__: return END return next_agent # 返回下一个节点的名称 # 为所有需要路由的节点添加条件边 builder.add_conditional_edges(planner, route_agent) builder.add_conditional_edges(researcher, route_agent) builder.add_conditional_edges(writer, route_agent) builder.add_conditional_edges(critic, route_agent) # finalizer 节点后直接结束 builder.add_edge(finalizer, END) collaboration_graph builder.compile()5.4 运行与可视化现在我们可以运行这个多智能体系统并可视化其执行流程。# main.py from graphs.multi_agent_graph import collaboration_graph from state_multi import MultiAgentState # 初始化状态 initial_state: MultiAgentState { task: 请撰写一篇关于LangGraph在构建多智能体系统中的优势与挑战的短文。, plan: None, research_data: [], draft: None, critique: None, final_report: None, messages: [], current_agent: planner } # 运行图 final_state collaboration_graph.invoke(initial_state) print(*50) print(任务:, final_state[task]) print(*50) print(生成的计划:, final_state[plan]) print(*50) print(收集的研究资料数量:, len(final_state[research_data])) print(*50) print(最终报告预览:, final_state[final_report][:500], ...) print(*50) # 可视化图结构需要安装graphviz try: from IPython.display import Image, display image_data collaboration_graph.get_graph().draw_mermaid_png() display(Image(image_data)) except ImportError: # 非Jupyter环境可以保存为文件 print(如需可视化请在Jupyter环境中运行或安装pygraphviz并调用get_graph().draw_mermaid_png()保存。) # 也可以打印文本结构 print(\n图结构) print(collaboration_graph.get_graph().print_ascii())6. 常见问题与排查思路在开发LangGraph应用时你可能会遇到以下典型问题。问题现象可能原因排查思路与解决方案KeyError访问State字段State的TypedDict定义与实际访问的键不匹配。1. 检查State类的定义确保字段名拼写正确。2. 在节点函数开头打印state.keys()确认当前状态包含哪些键。3. 使用state.get(key, default)提供默认值避免直接访问。图编译失败或执行报错节点函数返回值格式错误边指向不存在的节点。1. 确保每个节点函数返回一个字典其键是State中定义的字段名。2. 检查add_edge和add_conditional_edges中引用的节点名称是否都已通过add_node添加。3. 路由函数should_use_tool必须返回一个字符串且该字符串是条件映射{...}中的一个键。智能体陷入无限循环条件边逻辑有误导致在两个或多个节点间来回跳转无法到达END。1. 在状态中增加iteration_count字段并在节点中递增它超过阈值则强制结束。2. 仔细检查路由函数的逻辑确保存在能通向END的分支。3. 使用verboseTrue运行图观察状态变化和路由决策。工具调用失败或解析错误工具的参数格式与LLM生成的调用不匹配工具本身抛出异常。1. 为AgentExecutor设置handle_parsing_errorsTrue。2. 在工具函数内部做好异常捕获返回清晰的错误信息。3. 确保工具的tool装饰器中的描述清晰帮助LLM正确使用。多智能体协作中信息丢失后一个节点覆盖了前一个节点写入State的数据。1. 确保每个节点只更新自己负责的字段使用return {“field_to_update”: value}而不是return state。2. 对于列表类数据如messages使用Annotated和add_messages等归约器Reducer来追加而非覆盖。图可视化无法显示缺少graphviz或pygraphviz依赖。1. 安装系统级的Graphvizsudo apt-get install graphviz(Linux) 或brew install graphviz(Mac)。2. 安装Python绑定pip install pygraphviz可能较难安装或使用pip install langgraph[vis]安装包含可视化支持的版本。3. 回退方案使用graph.get_graph().print_ascii()在控制台查看文本结构。7. 最佳实践与工程建议将LangGraph用于实际项目时遵循以下实践能大幅提升代码的健壮性和可维护性。7.1 状态设计原则最小化与清晰化 State只存储工作流真正需要共享和传递的数据。避免将临时变量或中间计算结果全部塞入State。使用强类型 坚持使用TypedDict或PydanticBaseModel定义State这能在开发早期借助IDE和mypy发现类型错误。合理使用归约器 对于列表、集合等需要累积操作的数据务必使用Annotated和LangGraph提供的归约器如add_messages这是实现正确状态更新的关键。7.2 节点与图结构设计节点职责单一 每个节点应只完成一件明确的事情如“调用LLM”、“执行工具A”、“验证结果”。这使调试和测试更容易。善用子图 对于复杂的、可复用的逻辑序列可以将其封装成一个子图Subgraph然后作为单个节点加入主图。这能极大提升模块化程度。预编译与复用 图app workflow.compile()的编译有一定开销。在Web服务等场景中应将编译好的图实例作为全局或单例对象复用而不是每次请求都重新编译。7.3 错误处理与可观测性节点级异常捕获 在关键的节点函数内部使用try...except将异常信息捕获并作为状态的一部分传递到后续节点或专门的处理节点而不是让整个图崩溃。添加检查点与超时 对于可能长时间运行或卡住的循环在状态中设置计数器或时间戳并在路由函数中检查超时则导向人工干预或失败处理节点。全面日志记录 除了使用verboseTrue应在关键步骤如状态转换、路由决策、工具调用前后记录结构化的日志便于后期追踪和审计。可以将state的摘要信息记录下来。7.4 生产环境部署考量状态持久化 LangGraph默认在内存中维护状态。对于长时间运行或需要中断恢复的工作流需要实现检查点Checkpoint功能将状态持久化到数据库如Redis、PostgreSQL。LangGraph提供了相关的接口Checkpointer。异步支持 如果节点涉及网络IO如调用外部API、数据库查询应使用异步节点async def和图StateGraph的异步版本以提高并发性能。安全性 尤其注意工具调用的安全性。避免像示例中那样直接使用eval。对于执行代码、访问文件系统或网络的操作必须进行严格的输入验证、权限控制和沙箱隔离。掌握LangGraph意味着你掌握了构建复杂、可靠、可维护的AI智能体系统的核心框架。从定义清晰的状态模型开始通过节点和边组装业务逻辑再利用条件边实现动态流程控制你可以设计出从前端对话机器人到后端自动化流程引擎的各种应用。建议从本文的单智能体示例开始亲手运行并修改然后逐步挑战多智能体协作项目。在实践中你会更深刻地体会到图计算模型给AI应用开发带来的灵活性与强大威力。