在AI应用开发领域构建能够自主决策、执行复杂任务的智能体Agent已成为技术热点。然而许多开发者在从LangChain等工具链转向更复杂的多步骤、有状态工作流时常常感到无从下手面临状态管理混乱、执行流程不清晰、调试困难等痛点。本文将以LangGraph为核心为你提供一套从零到一的实战指南手把手教你构建一个功能完整的AI Agent。无论你是想入门Agent开发的学生还是需要在项目中集成智能工作流的工程师都能从本文获得可直接复用的代码、清晰的架构思路以及避坑经验。1. LangGraph 核心概念与定位在深入代码之前我们必须厘清LangGraph是什么以及它解决了什么问题。这有助于我们在正确的场景下选择它而不是盲目跟风。1.1 什么是LangGraphLangGraph 是 LangChain 生态系统中的一个库它专为构建有状态、多环节的智能体Agent和工作流Workflow而设计。你可以将它理解为一个用于编排AI调用、工具执行和业务逻辑的“流程图”引擎。它的核心思想是将复杂的AI应用建模为一个图Graph。图中的节点Node代表一个执行单元例如调用一次LLM、执行一个工具函数、进行条件判断边Edge则定义了节点之间的流转逻辑。通过这种方式LangGraph 能够清晰地管理应用的状态State在整个执行过程中的流转和变化。1.2 LangGraph vs. LangChain如何选择这是初学者最困惑的问题。两者并非替代关系而是互补关系。LangChain是一个构建LLM应用的框架。它提供了与各种LLM模型、向量数据库、工具链集成的标准化接口其核心抽象是“链Chain”——一种将多个组件如提示词模板、LLM、输出解析器线性串联的方式。它适合相对简单、线性的任务。LangGraph是构建复杂、有状态工作流的框架。它基于LangChain的组件但引入了“图”的概念专门处理需要循环、分支、并行、持久化状态等复杂逻辑的场景。它是LangChain的上层建筑。简单比喻LangChain 提供了砖块、水泥和工具LLM、工具函数。LangGraph 则是一张建筑设计图和施工流程管理告诉你如何用这些材料按照什么顺序可能包含循环和条件判断来盖一栋复杂的房子智能体。选择指南如果你的任务只是“提问 - 回答”或简单的“提取-总结”用LangChain Chain就够了。如果你的任务类似“分析问题 - 决定使用哪个工具 - 执行工具 - 检查结果 - 如果不满意则重新分析或尝试其他工具”这就是典型的Agent场景应该使用LangGraph。1.3 核心优势与应用场景优势显式状态管理所有步骤共享一个状态字典State状态变更一目了然便于调试和追踪。循环与条件分支原生支持基于执行结果的循环如Agent的思考-行动循环和条件跳转这是构建强大Agent的关键。可视化与可调试性理论上可以可视化整个工作流图并且每一步的状态变化都可以被记录和检查。持久化与容错支持将执行状态持久化对于长时运行的任务可以在中断后从上一个状态恢复。典型应用场景自主智能体Autonomous Agent如AutoGPT风格的AI可以自主规划、执行任务。复杂决策支持系统需要多次调用不同工具或API并根据中间结果调整策略的系统。对话机器人Chatbot需要维护对话历史、用户偏好等状态的复杂对话系统。数据处理流水线涉及多个AI处理步骤如摘要、分类、翻译的自动化流水线。2. 环境准备与项目初始化工欲善其事必先利其器。我们将在一个干净的环境下开始确保所有依赖都可控。2.1 环境与工具要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。本文命令以 macOS/Linux 的 bash 为例Windows 用户建议使用 WSL2 或 Git Bash。Python 版本Python 3.10 或 3.11。LangGraph 对 Python 3.12 的兼容性可能因依赖库而变化3.10/3.11 是最稳定的选择。使用python --version检查。包管理工具推荐使用pip。为了环境隔离强烈建议使用venv或conda。代码编辑器VS Code, PyCharm 等均可。LLM 服务我们需要一个大型语言模型作为Agent的“大脑”。本文将使用OpenAI 的 GPT 模型如 gpt-3.5-turbo作为示例。你需要准备一个有效的 OpenAI API Key。你也可以替换为其他 LangChain 支持的模型如 Anthropic Claude, 本地部署的 Ollama 等。2.2 创建虚拟环境与安装依赖让我们一步步搭建项目环境。# 1. 创建一个新的项目目录并进入 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 2. 创建 Python 虚拟环境 (以 venv 为例) python -m venv venv # 3. 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows (cmd): # venv\Scripts\activate # Windows (PowerShell): # .\venv\Scripts\Activate.ps1 # 激活后命令行提示符前通常会出现 (venv) 字样 # 4. 升级 pip pip install --upgrade pip # 5. 安装核心依赖 pip install langgraph langchain langchain-openai # langgraph: 本教程核心 # langchain: 提供基础组件LLM、工具、记忆等 # langchain-openai: 用于调用 OpenAI 模型的官方集成包 # 可选安装用于可视化和辅助开发的库 pip install jupyterlab # 如果你喜欢用 Notebook pip install python-dotenv # 用于管理环境变量推荐2.3 项目结构初始化创建以下目录和文件这是一个清晰的项目结构。langgraph-agent-tutorial/ ├── .env # 存储敏感信息如 API Key (务必加入 .gitignore) ├── .gitignore # Git 忽略文件 ├── requirements.txt # 项目依赖列表 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── agent/ # Agent 相关代码 │ │ ├── __init__.py │ │ ├── graph.py # 定义 LangGraph │ │ ├── state.py # 定义状态State结构 │ │ └── nodes.py # 定义图中的各个节点函数 │ └── tools/ # 自定义工具函数 │ ├── __init__.py │ └── calculator.py # 示例工具计算器 └── main.py # 主程序入口创建.gitignore文件# .gitignore venv/ .env __pycache__/ *.py[cod] *$py.class .DS_Store创建requirements.txt并冻结当前环境依赖pip freeze requirements.txt在.env文件中添加你的 OpenAI API Key# .env OPENAI_API_KEYsk-your-actual-openai-api-key-here重要安全提示永远不要将.env文件提交到版本控制系统如 Git。requirements.txt可以帮助其他协作者重建环境而敏感信息通过.env本地管理。3. LangGraph 核心组件深度解析理解 LangGraph 的四大核心组件是构建一切的基础。我们将结合代码示例来逐一拆解。3.1 状态State数据的共享容器State 是一个字典或类似字典的对象它随着工作流的执行而更新。它定义了图中所有节点可以读写哪些数据。在 LangGraph 中我们通常使用TypedDict或 PydanticBaseModel来定义 State 的结构这能提供良好的类型提示和验证。示例定义一个简单的 Agent 状态# src/agent/state.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): 定义Agent工作流的状态结构。 # messages: 存储对话历史。add_messages是一个特殊的归约函数用于追加消息。 messages: Annotated[List, add_messages] # 其他自定义字段 question: str # 用户的问题 final_answer: str # 最终答案 intermediate_steps: List[str] # 记录中间步骤用于调试 # 你可以根据需求添加更多字段如 tool_calls, iteration_count 等关键点解释TypedDict从typing模块导入用于定义字典的键和值类型。Annotated用于为字段添加元数据。这里的add_messages是一个归约函数Reducer。归约函数Reducer它定义了当多个节点试图修改同一个字段时如何合并这些修改。add_messages是 LangGraph 内置的专门用于处理消息列表的归约函数它会将新的消息追加到列表末尾而不是覆盖。这对于维护对话历史至关重要。自定义字段如question,final_answer等它们没有特殊的归约函数默认行为是后一个节点的写入会覆盖前一个节点的值。如果你希望累积结果需要自己实现逻辑。3.2 节点Node执行单元Node 是一个普通的 Python 函数或可调用对象它接收当前的State作为输入返回一个包含要更新字段的字典。# src/agent/nodes.py from .state import AgentState def call_llm(state: AgentState) - dict: 节点调用大语言模型。 from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 1. 从状态中获取消息历史 messages state[messages] # 初始状态下可能只有用户的问题。我们可以添加一个系统提示。 if len(messages) 1 and isinstance(messages[0], HumanMessage): system_prompt SystemMessage(content你是一个乐于助人的AI助手。请根据用户问题思考是否需要使用工具并给出最终答案。) full_messages [system_prompt] messages else: full_messages messages # 2. 初始化LLM (这里使用gpt-3.5-turbo) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 调用LLM response llm.invoke(full_messages) # 4. 返回更新后的状态将LLM的回复添加到消息历史中 # 注意由于 messages 字段使用了 add_messages 归约函数 # 我们返回 {messages: [response]} 会自动追加到原列表。 return {messages: [response]} def process_final_answer(state: AgentState) - dict: 节点处理最终答案将其存入特定字段。 # 从消息历史中提取最后一条消息假设是AI的回复 last_message state[messages][-1] final_answer last_message.content if hasattr(last_message, content) else str(last_message) # 更新状态中的 final_answer 字段 return {final_answer: final_answer}3.3 边Edge流程控制器Edge 决定了执行完一个节点后下一步该去哪个节点。有两种主要类型条件边Conditional Edge根据当前State的内容决定下一个节点。普通边直接指向下一个节点。LangGraph 提供了conditional_edge函数来创建条件边。# 在 graph.py 中会用到 from langgraph.graph import END, StateGraph from langgraph.graph import START # 假设我们有一个判断函数 def should_continue(state: AgentState) - str: 根据状态决定下一步。 返回字符串用于匹配条件边的路由键。 last_message state[messages][-1] # 这里是一个简单判断如果消息内容包含“最终答案”则结束否则继续调用LLM。 # 在实际Agent中这里会判断LLM是否调用了工具Tool Call。 if 最终答案 in last_message.content: return end else: return continue # 在构建图时我们会这样添加条件边 # graph.add_conditional_edges( # call_llm_node, # 源节点 # should_continue, # 路由判断函数 # { # continue: call_llm_node, # 如果返回continue则循环回自身 # end: END, # 如果返回end则结束流程。END是LangGraph内置的特殊节点。 # } # )3.4 图Graph组装与编译StateGraph类用于将节点和边组装起来最终编译成一个可执行的CompiledGraph。# src/agent/graph.py 的骨架 from langgraph.graph import StateGraph, END from .state import AgentState from .nodes import call_llm, process_final_answer # 假设从 tools 导入了一些工具节点 def create_agent_graph(): # 1. 创建图并指定状态的结构类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(llm, call_llm) # “llm”是节点在图中的名称 workflow.add_node(process_answer, process_final_answer) # workflow.add_node(use_tool, use_calculator_tool) # 可以添加工具节点 # 3. 设置入口点 workflow.set_entry_point(llm) # 4. 添加边这里先添加一个简单的线性边作为示例 workflow.add_edge(llm, process_answer) workflow.add_edge(process_answer, END) # 连接到结束节点 # 5. 编译图 compiled_graph workflow.compile() return compiled_graph4. 实战构建一个具备工具调用能力的计算器Agent现在我们将综合运用以上知识构建一个能理解自然语言数学问题并调用计算器工具给出答案的智能体。4.1 定义工具Tool首先我们创建一个简单的计算器工具。在LangChain中工具是一个能被LLM识别和调用的函数。# src/tools/calculator.py from langchain.tools import tool from typing import Union tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持加减乘除-*/和括号。 例如calculator((3 5) * 2) 返回 16。 # 警告直接使用 eval 有安全风险仅用于演示。 # 在生产环境中必须使用安全的表达式求值库如 ast.literal_eval 处理有限操作或自己解析。 try: # 这是一个极其简化的示例实际应用需要做严格的输入验证和沙箱化。 result eval(expression, {__builtins__: {}}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e} # 注意安全警告 # eval() 函数会执行传入的任意字符串代码如果 expression 来自不可信的输入将导致严重的代码注入漏洞。 # 此示例仅用于教学展示工具的基本形态。 # 真实项目必须替换为安全的计算逻辑例如 # 1. 使用 ast.literal_eval 但只支持常量表达式。 # 2. 使用第三方库如 simpleeval。 # 3. 自己编写语法解析器。4.2 增强状态与节点我们的Agent需要能处理工具调用。LangChain 的Messages中有专门的AIMessage可以包含tool_calls属性。我们需要调整状态和节点来处理它。# src/agent/state_v2.py from typing import TypedDict, List, Annotated, Optional from langgraph.graph.message import add_messages class AgentStateWithTools(TypedDict): 支持工具调用的Agent状态。 messages: Annotated[List, add_messages] # 我们不再需要显式定义 question, final_answer因为它们都包含在 messages 里。 # 但可以保留一些用于调试或控制流的字段。 iteration_count: int # 记录循环次数防止无限循环# src/agent/nodes_v2.py from .state_v2 import AgentStateWithTools from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage from src.tools.calculator import calculator import json # 将工具绑定到LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) llm_with_tools llm.bind_tools([calculator]) # 关键步骤让LLM知道有这个工具 def call_llm_with_tools(state: AgentStateWithTools) - dict: 节点调用绑定了工具的LLM。 messages state[messages] # 添加系统提示指导AI使用工具 if len(messages) 1 and isinstance(messages[0], HumanMessage): system_prompt SystemMessage(content你是一个数学助手。当用户询问数学计算问题时你必须使用calculator工具来计算。直接给出最终数字答案不要解释过程。如果问题不是数学计算请直接回答。) full_messages [system_prompt] messages else: full_messages messages # 调用LLM response llm_with_tools.invoke(full_messages) return {messages: [response]} # 追加AI的回复可能包含 tool_calls def execute_tools(state: AgentStateWithTools) - dict: 节点执行AI在消息中请求的工具并返回结果。 messages state[messages] last_message messages[-1] tool_messages [] if isinstance(last_message, AIMessage) and last_message.tool_calls: # AI 消息中包含了工具调用请求 for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 根据工具名调用对应的工具函数 if tool_name calculator: result calculator.invoke(tool_args) # 调用工具 else: result f错误未知工具 {tool_name} # 创建 ToolMessage这是 LangChain 约定的工具执行结果消息格式 tool_messages.append(ToolMessage(contentresult, tool_call_idtool_call[id])) else: # AI 没有调用工具可能直接给出了最终答案 # 我们也可以选择在这里添加一个标记或者什么都不做 pass # 将工具执行结果追加到消息历史 return {messages: tool_messages} def should_continue(state: AgentStateWithTools) - str: 路由函数判断下一步是执行工具还是结束。 messages state[messages] last_message messages[-1] # 情况1AI的消息包含了工具调用 - 需要去执行工具 if isinstance(last_message, AIMessage) and last_message.tool_calls: return call_tool # 情况2上一条消息是工具执行结果 - 需要让AI继续思考即再次调用LLM elif isinstance(last_message, ToolMessage): return call_llm # 情况3AI直接给出了最终答案没有工具调用 - 结束 else: return end4.3 构建并运行智能体图现在我们将节点和边组合成一个完整的、支持循环的Agent图。# src/agent/graph_v2.py from langgraph.graph import StateGraph, END from .state_v2 import AgentStateWithTools from .nodes_v2 import call_llm_with_tools, execute_tools, should_continue def create_calculator_agent_graph(): # 1. 初始化图 workflow StateGraph(AgentStateWithTools) # 2. 添加节点 workflow.add_node(llm, call_llm_with_tools) # 调用LLM的节点 workflow.add_node(action, execute_tools) # 执行工具的节点 # 3. 设置入口点从LLM开始 workflow.set_entry_point(llm) # 4. 添加条件边这是实现Agent循环的核心 workflow.add_conditional_edges( llm, # 从 llm 节点出来后 should_continue, # 根据状态决定下一步 { call_tool: action, # 如果需要调用工具则前往 action 节点 end: END, # 如果结束则终止 # 注意call_llm 这个分支在这里不会从 llm 节点直接出现。 # 它会在 action 节点之后被用到。 } ) # 5. 从工具执行节点action出来的边是固定的总是回到 llm 节点让AI思考结果。 workflow.add_edge(action, llm) # 6. 编译图 compiled_graph workflow.compile() return compiled_graph4.4 主程序与运行测试创建一个主程序来运行我们构建的Agent。# main.py import os from dotenv import load_dotenv from src.agent.graph_v2 import create_calculator_agent_graph from langchain_core.messages import HumanMessage # 加载环境变量从 .env 文件读取 OPENAI_API_KEY load_dotenv() def main(): # 1. 创建编译好的图 graph create_calculator_agent_graph() # 2. 准备初始状态 # 用户问题一个数学计算 user_question 请问 (12 8) * 3 除以 5 等于多少 initial_state { messages: [HumanMessage(contentuser_question)], iteration_count: 0 } print(f用户问题: {user_question}) print( * 50) # 3. 运行图流式输出可以看到每一步 final_state None for step, output in enumerate(graph.stream(initial_state)): node_name list(output.keys())[0] # 输出是 {节点名: 状态更新} state_update output[node_name] print(f[步骤 {step}] 节点 {node_name} 执行完毕。) # 打印当前的消息历史最后一条 current_messages state_update.get(messages, []) if current_messages: last_msg current_messages[-1] # 根据消息类型格式化输出 if hasattr(last_msg, content): print(f 输出内容: {last_msg.content[:100]}...) # 截断长输出 elif hasattr(last_msg, tool_calls): print(f 工具调用: {last_msg.tool_calls}) print(- * 30) final_state state_update print( * 50) print(执行结束。) if final_state and messages in final_state: # 提取最终的AI回答最后一条非ToolMessage的消息 for msg in reversed(final_state[messages]): if hasattr(msg, content) and not hasattr(msg, tool_call_id): print(f\n最终答案: {msg.content}) break if __name__ __main__: main()运行程序 在项目根目录下确保虚拟环境已激活且.env文件已配置好 API Key然后运行python main.py预期输出示例用户问题: 请问 (12 8) * 3 除以 5 等于多少 [步骤 0] 节点 llm 执行完毕。 输出内容: 我需要计算这个表达式。我将使用 calculator 工具。 工具调用: [{name: calculator, args: {expression: (12 8) * 3 / 5}, id: call_xxx}] ------------------------------ [步骤 1] 节点 action 执行完毕。 输出内容: 计算结果: 12.0 ------------------------------ [步骤 2] 节点 llm 执行完毕。 输出内容: (12 8) * 3 除以 5 等于 12.0。 ------------------------------ 执行结束。 最终答案: (12 8) * 3 除以 5 等于 12.0。恭喜你已经成功构建并运行了一个具备工具调用和循环推理能力的 LangGraph Agent。5. 常见问题与排查思路在实际开发中你可能会遇到以下问题。这里提供排查思路。问题现象可能原因解决思路ModuleNotFoundError: No module named langgraph1. 虚拟环境未激活。2. 依赖未正确安装。1. 确认命令行提示符前有(venv)。2. 在项目根目录执行pip install -r requirements.txt。AuthenticationError或Invalid API Key1. OpenAI API Key 未设置或错误。2..env文件未加载。1. 检查.env文件中的OPENAI_API_KEY是否正确无误。2. 确认main.py中已调用load_dotenv()。3. 尝试在代码中print(os.getenv(OPENAI_API_KEY))检查是否成功加载。Agent 陷入无限循环1. 路由逻辑should_continue有缺陷。2. LLM 没有按照预期生成tool_calls或最终答案。1. 在should_continue函数中添加print语句打印判断逻辑和返回值。2. 检查系统提示词是否清晰明确要求AI在需要时调用工具否则给出最终答案。3. 在状态中添加iteration_count字段并在每次循环时递增在should_continue中判断如果超过阈值如10次则强制返回end。LLM 不调用工具直接回答1. 工具绑定不正确。2. 提示词未引导AI使用工具。3. 模型能力不足如 gpt-3.5-turbo 在某些复杂场景下。1. 确认llm.bind_tools([tool1, tool2])调用成功。2. 强化系统提示词例如“你必须使用提供的工具来回答问题。如果问题涉及计算请调用 calculator 工具。”3. 尝试使用更强大的模型如gpt-4-turbo-preview。工具调用参数错误1. 工具函数的参数定义与LLM生成的不匹配。2. 工具函数内部异常。1. 确保tool装饰器内的函数文档字符串清晰描述了参数。2. 在execute_tools节点中增加异常捕获和日志打印出tool_name和tool_args。状态更新不符合预期1. 归约函数使用错误。2. 节点返回的字典键与状态定义不匹配。1. 回顾状态定义确认Annotated字段使用的归约函数是否正确。messages通常用add_messages。2. 确保节点函数返回的字典其键名是状态字典的键名。例如要更新messages就返回{messages: [...]}。6. 最佳实践与工程建议将原型转化为健壮的生产级应用需要遵循以下实践。6.1 状态设计原则最小化与清晰化只把需要在节点间传递的数据放入 State。避免将整个应用上下文都塞进去。使用类型提示坚持使用TypedDict或 PydanticBaseModel定义 State这能在开发早期发现类型错误。谨慎选择归约函数理解add_messages追加、operator.setitem覆盖/设置等内置归约函数的区别。对于列表累积操作通常需要自定义归约函数。6.2 节点函数设计单一职责每个节点只做一件事如调用LLM、执行一个工具、验证输入。幂等性与副作用尽可能让节点函数幂等相同输入产生相同输出。如果节点有副作用如写数据库、发邮件要做好错误处理和重试机制。充分的日志记录在节点函数的开始、结束和关键分支处记录日志便于调试和监控。可以使用logging模块并注入state中的唯一请求ID。6.3 错误处理与韧性节点级 Try-Catch在每个节点函数内部进行细致的异常捕获并将错误信息妥善地放入 State供后续节点或路由函数处理而不是让整个图崩溃。设置超时与重试对于调用外部API如LLM、数据库的节点配置请求超时和重试策略。LangChain 的许多组件支持max_retries等参数。验证输入在节点开始处理前验证 State 中所需字段的存在性和有效性。6.4 配置管理与可观测性外部化配置将模型类型、API Base URL、温度等参数放在配置文件如config.yaml或环境变量中不要硬编码在节点函数里。状态持久化对于长时任务利用 LangGraph 的检查点Checkpoint功能将状态持久化到数据库如SQLite、PostgreSQL。这允许工作流暂停和恢复。追踪与可视化利用 LangSmithLangChain 官方平台或自定义日志来追踪每次图的执行记录每个节点的输入/输出和执行时间。这对于理解Agent的决策过程和性能瓶颈至关重要。6.5 安全须知工具安全本文示例中的calculator工具使用了不安全的eval()。在生产环境中绝对禁止。必须替换为安全的表达式解析器或严格限制输入范围。权限控制如果Agent可以调用执行删除、发送消息等敏感操作的工具必须在工具函数内部实现严格的权限校验和操作确认。输入净化对用户输入和LLM生成的、将要传递给工具或数据库的内容进行净化和验证防止注入攻击。通过本教程你不仅学会了如何用 LangGraph 搭建一个简单的计算器 Agent更掌握了构建复杂 AI 工作流的核心方法论。从明确的状态管理、清晰的节点划分到灵活的条件路由这些概念是驾驭更高级 Agent 架构的基石。建议你以此为基础尝试集成更多工具如网络搜索、数据库查询设计更复杂的路由逻辑或者加入人工审核节点逐步构建出能够解决实际业务问题的智能体系统。
在AI应用开发领域构建能够自主决策、执行复杂任务的智能体Agent已成为技术热点。然而许多开发者在从LangChain等工具链转向更复杂的多步骤、有状态工作流时常常感到无从下手面临状态管理混乱、执行流程不清晰、调试困难等痛点。本文将以LangGraph为核心为你提供一套从零到一的实战指南手把手教你构建一个功能完整的AI Agent。无论你是想入门Agent开发的学生还是需要在项目中集成智能工作流的工程师都能从本文获得可直接复用的代码、清晰的架构思路以及避坑经验。1. LangGraph 核心概念与定位在深入代码之前我们必须厘清LangGraph是什么以及它解决了什么问题。这有助于我们在正确的场景下选择它而不是盲目跟风。1.1 什么是LangGraphLangGraph 是 LangChain 生态系统中的一个库它专为构建有状态、多环节的智能体Agent和工作流Workflow而设计。你可以将它理解为一个用于编排AI调用、工具执行和业务逻辑的“流程图”引擎。它的核心思想是将复杂的AI应用建模为一个图Graph。图中的节点Node代表一个执行单元例如调用一次LLM、执行一个工具函数、进行条件判断边Edge则定义了节点之间的流转逻辑。通过这种方式LangGraph 能够清晰地管理应用的状态State在整个执行过程中的流转和变化。1.2 LangGraph vs. LangChain如何选择这是初学者最困惑的问题。两者并非替代关系而是互补关系。LangChain是一个构建LLM应用的框架。它提供了与各种LLM模型、向量数据库、工具链集成的标准化接口其核心抽象是“链Chain”——一种将多个组件如提示词模板、LLM、输出解析器线性串联的方式。它适合相对简单、线性的任务。LangGraph是构建复杂、有状态工作流的框架。它基于LangChain的组件但引入了“图”的概念专门处理需要循环、分支、并行、持久化状态等复杂逻辑的场景。它是LangChain的上层建筑。简单比喻LangChain 提供了砖块、水泥和工具LLM、工具函数。LangGraph 则是一张建筑设计图和施工流程管理告诉你如何用这些材料按照什么顺序可能包含循环和条件判断来盖一栋复杂的房子智能体。选择指南如果你的任务只是“提问 - 回答”或简单的“提取-总结”用LangChain Chain就够了。如果你的任务类似“分析问题 - 决定使用哪个工具 - 执行工具 - 检查结果 - 如果不满意则重新分析或尝试其他工具”这就是典型的Agent场景应该使用LangGraph。1.3 核心优势与应用场景优势显式状态管理所有步骤共享一个状态字典State状态变更一目了然便于调试和追踪。循环与条件分支原生支持基于执行结果的循环如Agent的思考-行动循环和条件跳转这是构建强大Agent的关键。可视化与可调试性理论上可以可视化整个工作流图并且每一步的状态变化都可以被记录和检查。持久化与容错支持将执行状态持久化对于长时运行的任务可以在中断后从上一个状态恢复。典型应用场景自主智能体Autonomous Agent如AutoGPT风格的AI可以自主规划、执行任务。复杂决策支持系统需要多次调用不同工具或API并根据中间结果调整策略的系统。对话机器人Chatbot需要维护对话历史、用户偏好等状态的复杂对话系统。数据处理流水线涉及多个AI处理步骤如摘要、分类、翻译的自动化流水线。2. 环境准备与项目初始化工欲善其事必先利其器。我们将在一个干净的环境下开始确保所有依赖都可控。2.1 环境与工具要求操作系统Windows 10/11, macOS, 或 Linux (如 Ubuntu 20.04)。本文命令以 macOS/Linux 的 bash 为例Windows 用户建议使用 WSL2 或 Git Bash。Python 版本Python 3.10 或 3.11。LangGraph 对 Python 3.12 的兼容性可能因依赖库而变化3.10/3.11 是最稳定的选择。使用python --version检查。包管理工具推荐使用pip。为了环境隔离强烈建议使用venv或conda。代码编辑器VS Code, PyCharm 等均可。LLM 服务我们需要一个大型语言模型作为Agent的“大脑”。本文将使用OpenAI 的 GPT 模型如 gpt-3.5-turbo作为示例。你需要准备一个有效的 OpenAI API Key。你也可以替换为其他 LangChain 支持的模型如 Anthropic Claude, 本地部署的 Ollama 等。2.2 创建虚拟环境与安装依赖让我们一步步搭建项目环境。# 1. 创建一个新的项目目录并进入 mkdir langgraph-agent-tutorial cd langgraph-agent-tutorial # 2. 创建 Python 虚拟环境 (以 venv 为例) python -m venv venv # 3. 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows (cmd): # venv\Scripts\activate # Windows (PowerShell): # .\venv\Scripts\Activate.ps1 # 激活后命令行提示符前通常会出现 (venv) 字样 # 4. 升级 pip pip install --upgrade pip # 5. 安装核心依赖 pip install langgraph langchain langchain-openai # langgraph: 本教程核心 # langchain: 提供基础组件LLM、工具、记忆等 # langchain-openai: 用于调用 OpenAI 模型的官方集成包 # 可选安装用于可视化和辅助开发的库 pip install jupyterlab # 如果你喜欢用 Notebook pip install python-dotenv # 用于管理环境变量推荐2.3 项目结构初始化创建以下目录和文件这是一个清晰的项目结构。langgraph-agent-tutorial/ ├── .env # 存储敏感信息如 API Key (务必加入 .gitignore) ├── .gitignore # Git 忽略文件 ├── requirements.txt # 项目依赖列表 ├── src/ # 源代码目录 │ ├── __init__.py │ ├── agent/ # Agent 相关代码 │ │ ├── __init__.py │ │ ├── graph.py # 定义 LangGraph │ │ ├── state.py # 定义状态State结构 │ │ └── nodes.py # 定义图中的各个节点函数 │ └── tools/ # 自定义工具函数 │ ├── __init__.py │ └── calculator.py # 示例工具计算器 └── main.py # 主程序入口创建.gitignore文件# .gitignore venv/ .env __pycache__/ *.py[cod] *$py.class .DS_Store创建requirements.txt并冻结当前环境依赖pip freeze requirements.txt在.env文件中添加你的 OpenAI API Key# .env OPENAI_API_KEYsk-your-actual-openai-api-key-here重要安全提示永远不要将.env文件提交到版本控制系统如 Git。requirements.txt可以帮助其他协作者重建环境而敏感信息通过.env本地管理。3. LangGraph 核心组件深度解析理解 LangGraph 的四大核心组件是构建一切的基础。我们将结合代码示例来逐一拆解。3.1 状态State数据的共享容器State 是一个字典或类似字典的对象它随着工作流的执行而更新。它定义了图中所有节点可以读写哪些数据。在 LangGraph 中我们通常使用TypedDict或 PydanticBaseModel来定义 State 的结构这能提供良好的类型提示和验证。示例定义一个简单的 Agent 状态# src/agent/state.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages class AgentState(TypedDict): 定义Agent工作流的状态结构。 # messages: 存储对话历史。add_messages是一个特殊的归约函数用于追加消息。 messages: Annotated[List, add_messages] # 其他自定义字段 question: str # 用户的问题 final_answer: str # 最终答案 intermediate_steps: List[str] # 记录中间步骤用于调试 # 你可以根据需求添加更多字段如 tool_calls, iteration_count 等关键点解释TypedDict从typing模块导入用于定义字典的键和值类型。Annotated用于为字段添加元数据。这里的add_messages是一个归约函数Reducer。归约函数Reducer它定义了当多个节点试图修改同一个字段时如何合并这些修改。add_messages是 LangGraph 内置的专门用于处理消息列表的归约函数它会将新的消息追加到列表末尾而不是覆盖。这对于维护对话历史至关重要。自定义字段如question,final_answer等它们没有特殊的归约函数默认行为是后一个节点的写入会覆盖前一个节点的值。如果你希望累积结果需要自己实现逻辑。3.2 节点Node执行单元Node 是一个普通的 Python 函数或可调用对象它接收当前的State作为输入返回一个包含要更新字段的字典。# src/agent/nodes.py from .state import AgentState def call_llm(state: AgentState) - dict: 节点调用大语言模型。 from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage # 1. 从状态中获取消息历史 messages state[messages] # 初始状态下可能只有用户的问题。我们可以添加一个系统提示。 if len(messages) 1 and isinstance(messages[0], HumanMessage): system_prompt SystemMessage(content你是一个乐于助人的AI助手。请根据用户问题思考是否需要使用工具并给出最终答案。) full_messages [system_prompt] messages else: full_messages messages # 2. 初始化LLM (这里使用gpt-3.5-turbo) llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) # 3. 调用LLM response llm.invoke(full_messages) # 4. 返回更新后的状态将LLM的回复添加到消息历史中 # 注意由于 messages 字段使用了 add_messages 归约函数 # 我们返回 {messages: [response]} 会自动追加到原列表。 return {messages: [response]} def process_final_answer(state: AgentState) - dict: 节点处理最终答案将其存入特定字段。 # 从消息历史中提取最后一条消息假设是AI的回复 last_message state[messages][-1] final_answer last_message.content if hasattr(last_message, content) else str(last_message) # 更新状态中的 final_answer 字段 return {final_answer: final_answer}3.3 边Edge流程控制器Edge 决定了执行完一个节点后下一步该去哪个节点。有两种主要类型条件边Conditional Edge根据当前State的内容决定下一个节点。普通边直接指向下一个节点。LangGraph 提供了conditional_edge函数来创建条件边。# 在 graph.py 中会用到 from langgraph.graph import END, StateGraph from langgraph.graph import START # 假设我们有一个判断函数 def should_continue(state: AgentState) - str: 根据状态决定下一步。 返回字符串用于匹配条件边的路由键。 last_message state[messages][-1] # 这里是一个简单判断如果消息内容包含“最终答案”则结束否则继续调用LLM。 # 在实际Agent中这里会判断LLM是否调用了工具Tool Call。 if 最终答案 in last_message.content: return end else: return continue # 在构建图时我们会这样添加条件边 # graph.add_conditional_edges( # call_llm_node, # 源节点 # should_continue, # 路由判断函数 # { # continue: call_llm_node, # 如果返回continue则循环回自身 # end: END, # 如果返回end则结束流程。END是LangGraph内置的特殊节点。 # } # )3.4 图Graph组装与编译StateGraph类用于将节点和边组装起来最终编译成一个可执行的CompiledGraph。# src/agent/graph.py 的骨架 from langgraph.graph import StateGraph, END from .state import AgentState from .nodes import call_llm, process_final_answer # 假设从 tools 导入了一些工具节点 def create_agent_graph(): # 1. 创建图并指定状态的结构类型 workflow StateGraph(AgentState) # 2. 添加节点 workflow.add_node(llm, call_llm) # “llm”是节点在图中的名称 workflow.add_node(process_answer, process_final_answer) # workflow.add_node(use_tool, use_calculator_tool) # 可以添加工具节点 # 3. 设置入口点 workflow.set_entry_point(llm) # 4. 添加边这里先添加一个简单的线性边作为示例 workflow.add_edge(llm, process_answer) workflow.add_edge(process_answer, END) # 连接到结束节点 # 5. 编译图 compiled_graph workflow.compile() return compiled_graph4. 实战构建一个具备工具调用能力的计算器Agent现在我们将综合运用以上知识构建一个能理解自然语言数学问题并调用计算器工具给出答案的智能体。4.1 定义工具Tool首先我们创建一个简单的计算器工具。在LangChain中工具是一个能被LLM识别和调用的函数。# src/tools/calculator.py from langchain.tools import tool from typing import Union tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持加减乘除-*/和括号。 例如calculator((3 5) * 2) 返回 16。 # 警告直接使用 eval 有安全风险仅用于演示。 # 在生产环境中必须使用安全的表达式求值库如 ast.literal_eval 处理有限操作或自己解析。 try: # 这是一个极其简化的示例实际应用需要做严格的输入验证和沙箱化。 result eval(expression, {__builtins__: {}}, {}) return f计算结果: {result} except Exception as e: return f计算错误: {e} # 注意安全警告 # eval() 函数会执行传入的任意字符串代码如果 expression 来自不可信的输入将导致严重的代码注入漏洞。 # 此示例仅用于教学展示工具的基本形态。 # 真实项目必须替换为安全的计算逻辑例如 # 1. 使用 ast.literal_eval 但只支持常量表达式。 # 2. 使用第三方库如 simpleeval。 # 3. 自己编写语法解析器。4.2 增强状态与节点我们的Agent需要能处理工具调用。LangChain 的Messages中有专门的AIMessage可以包含tool_calls属性。我们需要调整状态和节点来处理它。# src/agent/state_v2.py from typing import TypedDict, List, Annotated, Optional from langgraph.graph.message import add_messages class AgentStateWithTools(TypedDict): 支持工具调用的Agent状态。 messages: Annotated[List, add_messages] # 我们不再需要显式定义 question, final_answer因为它们都包含在 messages 里。 # 但可以保留一些用于调试或控制流的字段。 iteration_count: int # 记录循环次数防止无限循环# src/agent/nodes_v2.py from .state_v2 import AgentStateWithTools from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, SystemMessage, AIMessage, ToolMessage from src.tools.calculator import calculator import json # 将工具绑定到LLM llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) llm_with_tools llm.bind_tools([calculator]) # 关键步骤让LLM知道有这个工具 def call_llm_with_tools(state: AgentStateWithTools) - dict: 节点调用绑定了工具的LLM。 messages state[messages] # 添加系统提示指导AI使用工具 if len(messages) 1 and isinstance(messages[0], HumanMessage): system_prompt SystemMessage(content你是一个数学助手。当用户询问数学计算问题时你必须使用calculator工具来计算。直接给出最终数字答案不要解释过程。如果问题不是数学计算请直接回答。) full_messages [system_prompt] messages else: full_messages messages # 调用LLM response llm_with_tools.invoke(full_messages) return {messages: [response]} # 追加AI的回复可能包含 tool_calls def execute_tools(state: AgentStateWithTools) - dict: 节点执行AI在消息中请求的工具并返回结果。 messages state[messages] last_message messages[-1] tool_messages [] if isinstance(last_message, AIMessage) and last_message.tool_calls: # AI 消息中包含了工具调用请求 for tool_call in last_message.tool_calls: tool_name tool_call[name] tool_args tool_call[args] # 根据工具名调用对应的工具函数 if tool_name calculator: result calculator.invoke(tool_args) # 调用工具 else: result f错误未知工具 {tool_name} # 创建 ToolMessage这是 LangChain 约定的工具执行结果消息格式 tool_messages.append(ToolMessage(contentresult, tool_call_idtool_call[id])) else: # AI 没有调用工具可能直接给出了最终答案 # 我们也可以选择在这里添加一个标记或者什么都不做 pass # 将工具执行结果追加到消息历史 return {messages: tool_messages} def should_continue(state: AgentStateWithTools) - str: 路由函数判断下一步是执行工具还是结束。 messages state[messages] last_message messages[-1] # 情况1AI的消息包含了工具调用 - 需要去执行工具 if isinstance(last_message, AIMessage) and last_message.tool_calls: return call_tool # 情况2上一条消息是工具执行结果 - 需要让AI继续思考即再次调用LLM elif isinstance(last_message, ToolMessage): return call_llm # 情况3AI直接给出了最终答案没有工具调用 - 结束 else: return end4.3 构建并运行智能体图现在我们将节点和边组合成一个完整的、支持循环的Agent图。# src/agent/graph_v2.py from langgraph.graph import StateGraph, END from .state_v2 import AgentStateWithTools from .nodes_v2 import call_llm_with_tools, execute_tools, should_continue def create_calculator_agent_graph(): # 1. 初始化图 workflow StateGraph(AgentStateWithTools) # 2. 添加节点 workflow.add_node(llm, call_llm_with_tools) # 调用LLM的节点 workflow.add_node(action, execute_tools) # 执行工具的节点 # 3. 设置入口点从LLM开始 workflow.set_entry_point(llm) # 4. 添加条件边这是实现Agent循环的核心 workflow.add_conditional_edges( llm, # 从 llm 节点出来后 should_continue, # 根据状态决定下一步 { call_tool: action, # 如果需要调用工具则前往 action 节点 end: END, # 如果结束则终止 # 注意call_llm 这个分支在这里不会从 llm 节点直接出现。 # 它会在 action 节点之后被用到。 } ) # 5. 从工具执行节点action出来的边是固定的总是回到 llm 节点让AI思考结果。 workflow.add_edge(action, llm) # 6. 编译图 compiled_graph workflow.compile() return compiled_graph4.4 主程序与运行测试创建一个主程序来运行我们构建的Agent。# main.py import os from dotenv import load_dotenv from src.agent.graph_v2 import create_calculator_agent_graph from langchain_core.messages import HumanMessage # 加载环境变量从 .env 文件读取 OPENAI_API_KEY load_dotenv() def main(): # 1. 创建编译好的图 graph create_calculator_agent_graph() # 2. 准备初始状态 # 用户问题一个数学计算 user_question 请问 (12 8) * 3 除以 5 等于多少 initial_state { messages: [HumanMessage(contentuser_question)], iteration_count: 0 } print(f用户问题: {user_question}) print( * 50) # 3. 运行图流式输出可以看到每一步 final_state None for step, output in enumerate(graph.stream(initial_state)): node_name list(output.keys())[0] # 输出是 {节点名: 状态更新} state_update output[node_name] print(f[步骤 {step}] 节点 {node_name} 执行完毕。) # 打印当前的消息历史最后一条 current_messages state_update.get(messages, []) if current_messages: last_msg current_messages[-1] # 根据消息类型格式化输出 if hasattr(last_msg, content): print(f 输出内容: {last_msg.content[:100]}...) # 截断长输出 elif hasattr(last_msg, tool_calls): print(f 工具调用: {last_msg.tool_calls}) print(- * 30) final_state state_update print( * 50) print(执行结束。) if final_state and messages in final_state: # 提取最终的AI回答最后一条非ToolMessage的消息 for msg in reversed(final_state[messages]): if hasattr(msg, content) and not hasattr(msg, tool_call_id): print(f\n最终答案: {msg.content}) break if __name__ __main__: main()运行程序 在项目根目录下确保虚拟环境已激活且.env文件已配置好 API Key然后运行python main.py预期输出示例用户问题: 请问 (12 8) * 3 除以 5 等于多少 [步骤 0] 节点 llm 执行完毕。 输出内容: 我需要计算这个表达式。我将使用 calculator 工具。 工具调用: [{name: calculator, args: {expression: (12 8) * 3 / 5}, id: call_xxx}] ------------------------------ [步骤 1] 节点 action 执行完毕。 输出内容: 计算结果: 12.0 ------------------------------ [步骤 2] 节点 llm 执行完毕。 输出内容: (12 8) * 3 除以 5 等于 12.0。 ------------------------------ 执行结束。 最终答案: (12 8) * 3 除以 5 等于 12.0。恭喜你已经成功构建并运行了一个具备工具调用和循环推理能力的 LangGraph Agent。5. 常见问题与排查思路在实际开发中你可能会遇到以下问题。这里提供排查思路。问题现象可能原因解决思路ModuleNotFoundError: No module named langgraph1. 虚拟环境未激活。2. 依赖未正确安装。1. 确认命令行提示符前有(venv)。2. 在项目根目录执行pip install -r requirements.txt。AuthenticationError或Invalid API Key1. OpenAI API Key 未设置或错误。2..env文件未加载。1. 检查.env文件中的OPENAI_API_KEY是否正确无误。2. 确认main.py中已调用load_dotenv()。3. 尝试在代码中print(os.getenv(OPENAI_API_KEY))检查是否成功加载。Agent 陷入无限循环1. 路由逻辑should_continue有缺陷。2. LLM 没有按照预期生成tool_calls或最终答案。1. 在should_continue函数中添加print语句打印判断逻辑和返回值。2. 检查系统提示词是否清晰明确要求AI在需要时调用工具否则给出最终答案。3. 在状态中添加iteration_count字段并在每次循环时递增在should_continue中判断如果超过阈值如10次则强制返回end。LLM 不调用工具直接回答1. 工具绑定不正确。2. 提示词未引导AI使用工具。3. 模型能力不足如 gpt-3.5-turbo 在某些复杂场景下。1. 确认llm.bind_tools([tool1, tool2])调用成功。2. 强化系统提示词例如“你必须使用提供的工具来回答问题。如果问题涉及计算请调用 calculator 工具。”3. 尝试使用更强大的模型如gpt-4-turbo-preview。工具调用参数错误1. 工具函数的参数定义与LLM生成的不匹配。2. 工具函数内部异常。1. 确保tool装饰器内的函数文档字符串清晰描述了参数。2. 在execute_tools节点中增加异常捕获和日志打印出tool_name和tool_args。状态更新不符合预期1. 归约函数使用错误。2. 节点返回的字典键与状态定义不匹配。1. 回顾状态定义确认Annotated字段使用的归约函数是否正确。messages通常用add_messages。2. 确保节点函数返回的字典其键名是状态字典的键名。例如要更新messages就返回{messages: [...]}。6. 最佳实践与工程建议将原型转化为健壮的生产级应用需要遵循以下实践。6.1 状态设计原则最小化与清晰化只把需要在节点间传递的数据放入 State。避免将整个应用上下文都塞进去。使用类型提示坚持使用TypedDict或 PydanticBaseModel定义 State这能在开发早期发现类型错误。谨慎选择归约函数理解add_messages追加、operator.setitem覆盖/设置等内置归约函数的区别。对于列表累积操作通常需要自定义归约函数。6.2 节点函数设计单一职责每个节点只做一件事如调用LLM、执行一个工具、验证输入。幂等性与副作用尽可能让节点函数幂等相同输入产生相同输出。如果节点有副作用如写数据库、发邮件要做好错误处理和重试机制。充分的日志记录在节点函数的开始、结束和关键分支处记录日志便于调试和监控。可以使用logging模块并注入state中的唯一请求ID。6.3 错误处理与韧性节点级 Try-Catch在每个节点函数内部进行细致的异常捕获并将错误信息妥善地放入 State供后续节点或路由函数处理而不是让整个图崩溃。设置超时与重试对于调用外部API如LLM、数据库的节点配置请求超时和重试策略。LangChain 的许多组件支持max_retries等参数。验证输入在节点开始处理前验证 State 中所需字段的存在性和有效性。6.4 配置管理与可观测性外部化配置将模型类型、API Base URL、温度等参数放在配置文件如config.yaml或环境变量中不要硬编码在节点函数里。状态持久化对于长时任务利用 LangGraph 的检查点Checkpoint功能将状态持久化到数据库如SQLite、PostgreSQL。这允许工作流暂停和恢复。追踪与可视化利用 LangSmithLangChain 官方平台或自定义日志来追踪每次图的执行记录每个节点的输入/输出和执行时间。这对于理解Agent的决策过程和性能瓶颈至关重要。6.5 安全须知工具安全本文示例中的calculator工具使用了不安全的eval()。在生产环境中绝对禁止。必须替换为安全的表达式解析器或严格限制输入范围。权限控制如果Agent可以调用执行删除、发送消息等敏感操作的工具必须在工具函数内部实现严格的权限校验和操作确认。输入净化对用户输入和LLM生成的、将要传递给工具或数据库的内容进行净化和验证防止注入攻击。通过本教程你不仅学会了如何用 LangGraph 搭建一个简单的计算器 Agent更掌握了构建复杂 AI 工作流的核心方法论。从明确的状态管理、清晰的节点划分到灵活的条件路由这些概念是驾驭更高级 Agent 架构的基石。建议你以此为基础尝试集成更多工具如网络搜索、数据库查询设计更复杂的路由逻辑或者加入人工审核节点逐步构建出能够解决实际业务问题的智能体系统。
