这次我们来看一个在B站上非常热门的LangGraph教学视频资源。这个系列视频号称“最火最全”旨在从零开始系统性地讲解LangGraph帮助开发者入门AI大模型应用开发。对于想要构建复杂、有状态的AI代理Agent和自动化工作流的开发者来说LangGraph是一个至关重要的框架它补足了LangChain在控制流方面的能力。本文不是视频的逐字稿而是基于其核心教学内容为你整理成一篇结构清晰、可操作性强的技术指南。我们将重点关注LangGraph是什么、它能解决什么问题、以及如何从环境搭建到实际开发一步步上手。无论你是想学习AI Agent开发还是希望将大模型能力集成到更复杂的业务流程中这篇文章都能提供直接的路径。1. 核心能力速览在深入细节之前我们先通过一个表格快速了解LangGraph的核心定位和能力边界这有助于你判断它是否是你当前需要的工具。能力项说明项目类型用于构建有状态、多环节AI工作流的框架库是LangChain生态的重要扩展。核心价值解决LangChain在复杂控制流如循环、分支、回溯上的不足使构建的Agent具备“长期记忆”和决策能力。编程语言主要支持 Python。硬件门槛无特殊要求。LangGraph本身是编排框架计算负载取决于其调用的底层大模型如GPT-4、本地LLM。主要功能1. 定义包含节点Node和边Edge的工作流图Graph。2. 支持条件判断和循环实现动态执行路径。3. 维护全局状态State在不同节点间传递和更新信息。4. 与LangChain组件无缝集成轻松调用工具、模型、记忆等。适合场景1. 构建复杂的对话Agent如客服、游戏NPC。2. 实现多步骤任务自动化如研究助手、数据分析流水线。3. 开发需要根据中间结果动态调整流程的应用。不适合场景1. 简单的单次问答直接用LangChain Chain即可。2. 无需状态维护的批量文本处理。2. LangGraph 是什么与 LangChain 有何区别很多初学者会混淆LangGraph和LangChain。简单来说你可以把LangChain看作是一个提供了各种预制零件模型I/O、记忆、工具的“工具箱”而LangGraph则是用来将这些零件按照复杂蓝图组装成自动化“流水线”或“机器人”的“设计图与控制器”。LangChain的核心是“链”Chain它倾向于线性执行。虽然也能处理一些分支但在需要根据运行时的结果反复循环或跳转的场景下设计和实现会变得非常笨拙。LangGraph引入了“图”Graph的概念其核心是一个有向图结构。这个图由节点Nodes代表一个执行单元可以是一个LLM调用、一个工具调用或任何函数。边Edges定义节点之间的执行顺序。边可以是固定的也可以是条件的根据当前状态决定下一个执行哪个节点。状态State一个在所有节点间共享和传递的字典记录了整个工作流的上下文信息。这种设计使得实现“循环直到满足条件”、“尝试方案A失败后自动切换到方案B”这样的逻辑变得直观且容易。因此当你的AI应用需要“长期记忆”和“自主决策”能力时LangGraph是比单纯使用LangChain更强大的选择。3. 环境准备与前置条件开始使用LangGraph前你需要准备好Python开发环境。以下是详细的步骤和注意事项。3.1 基础环境配置Python版本建议使用 Python 3.8 至 3.11 版本。避免使用过新如3.12的某些早期版本可能存在的兼容性问题。包管理工具使用pip进行安装。强烈建议使用虚拟环境如venv或conda来隔离项目依赖。网络环境由于需要安装开源包并可能调用在线大模型API如OpenAI请确保你的开发环境具备稳定的网络连接。3.2 创建并激活虚拟环境这是最佳实践可以避免不同项目间的包冲突。# 1. 创建虚拟环境以venv为例项目目录为my_agent python -m venv my_agent_env # 2. 激活虚拟环境 # 在 Windows 上 my_agent_env\Scripts\activate # 在 macOS/Linux 上 source my_agent_env/bin/activate # 激活后命令行提示符前通常会显示环境名如 (my_agent_env)4. 安装部署与启动方式LangGraph的安装非常简单因为它本质上是一个Python库。我们将同时安装LangChain因为两者需要协同工作。4.1 核心库安装在激活的虚拟环境中执行以下命令pip install langgraph langchain这条命令会安装LangGraph及其核心依赖。langchain包提供了模型、工具等基础组件。4.2 模型提供商SDK安装LangGraph/LangChain本身不提供模型你需要连接一个LLM服务。以最常用的OpenAI为例pip install openai安装后你需要设置OpenAI的API密钥。通常通过环境变量来管理# 在命令行中临时设置仅当前会话有效 export OPENAI_API_KEY你的-api-key-here # Windows (cmd) 使用 # set OPENAI_API_KEY你的-api-key-here # Windows (PowerShell) 使用 # $env:OPENAI_API_KEY你的-api-key-here重要你也可以在代码中直接设置但为了安全更推荐使用环境变量或.env文件。4.3 验证安装创建一个简单的Python脚本test_install.py来验证基础环境是否就绪import langchain import langgraph import openai print(fLangChain version: {langchain.__version__}) print(fLangGraph version: {langgraph.__version__}) print(fOpenAI version: {openai.__version__}) print(所有核心库导入成功)运行该脚本python test_install.py如果成功输出版本号且没有报错说明基础环境配置完成。5. 核心概念与第一个LangGraph应用理解LangGraph的最佳方式就是动手构建一个。我们从最简单的“对话循环”Agent开始它能够持续与用户对话直到用户说再见。5.1 定义状态State状态是图的“记忆中枢”。我们使用TypedDict来定义状态的类型确保结构清晰。from typing import TypedDict, Annotated import operator class AgentState(TypedDict): # 存储当前的对话消息列表 messages: Annotated[list, operator.add] # 可以添加更多字段例如用户偏好、对话轮次等 # turn_count: int这里messages字段使用Annotated[list, operator.add]注解。这是LangGraph的一个关键特性它声明了对该字段的归约器。operator.add意味着当多个节点并发修改此字段时虽然本例没有并发它们的修改会通过“相加”即列表拼接的方式合并。对于列表这通常是我们期望的行为。5.2 创建节点Nodes节点是执行具体工作的函数。它接收当前State执行操作并返回一个包含更新后状态的字典。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage # 初始化大语言模型 llm ChatOpenAI(modelgpt-3.5-turbo) # 定义“调用模型”节点 def call_model(state: AgentState): print(f[节点 call_model] 收到消息历史: {state[messages]}) # 将消息历史发送给LLM获取回复 response llm.invoke(state[messages]) # 将AI的回复添加到消息历史中 return {messages: [response]} # 定义“人类输入”节点模拟用户 def human_input(state: AgentState): user_input input(用户说: ) # 将用户输入转换为消息格式并添加到历史 return {messages: [HumanMessage(contentuser_input)]}5.3 创建条件边Conditional Edges与图编译图需要知道执行完一个节点后接下来该去哪里。我们通过条件函数来实现。from langgraph.graph import StateGraph, END # 判断是否应该继续对话的条件函数 def should_continue(state: AgentState): messages state[messages] last_message messages[-1] # 如果最后一条消息是AI说的且内容包含“再见”则结束 if isinstance(last_message, AIMessage) and 再见 in last_message.content: print([条件判断] 检测到‘再见’结束对话。) return end # 否则继续对话 return continue # 创建图构建器 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(human, human_input) workflow.add_node(model, call_model) # 设置入口点从人类输入开始 workflow.set_entry_point(human) # 添加边人类输入后总是去调用模型 workflow.add_edge(human, model) # 添加条件边调用模型后根据条件决定下一步 workflow.add_conditional_edges( model, should_continue, # 条件判断函数 { continue: human, # 如果返回continue跳转到human节点 end: END # 如果返回end图执行结束 } ) # 编译图得到一个可执行对象 app workflow.compile()5.4 运行与效果验证现在我们可以运行这个简单的对话Agent了。# 初始化状态 initial_state AgentState(messages[]) # 运行图 final_state app.invoke(initial_state) print(\n 对话结束 ) print(最终消息历史) for msg in final_state[messages]: print(f{type(msg).__name__}: {msg.content})操作步骤与预期结果运行上述代码。程序会提示“用户说”输入“你好”。LLM会回复问候语。程序会再次提示“用户说”输入“今天天气怎么样”。LLM会模拟回答天气。你可以继续对话直到你对LLM说“再见”或者LLM在回复中说出“再见”对话循环将终止。控制台会打印出完整的对话历史。判断成功标准程序能正确接收用户输入。LLM能基于对话历史生成连贯回复。当对话内容触发结束条件时图能正确停止。6. 构建高级Agent集成工具与记忆一个真正的智能Agent不仅能聊天还能使用工具如搜索、计算并拥有更丰富的记忆。下面我们构建一个能使用计算器和拥有短期记忆的Agent。6.1 定义更复杂的状态与工具from typing import List from langchain.tools import tool from langchain_core.messages import BaseMessage import json # 定义工具一个简单的计算器 tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /。 try: # 警告实际生产环境应用必须对输入做严格安全检查避免eval的安全风险。 # 此处仅为演示。 result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} # 定义状态包含消息和中间步骤用于记录工具调用 class AdvancedState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 记录Agent思考的中间步骤对调试和观察非常有用 intermediate_steps: Annotated[List[dict], operator.add] # 初始化模型并绑定工具。使用 bind_tools 让模型知道它可以调用哪些工具。 llm_with_tools ChatOpenAI(modelgpt-3.5-turbo).bind_tools([calculator])6.2 创建智能路由节点这个节点是Agent的大脑它决定是直接回复还是调用工具。from langchain.agents import create_react_agent from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import format_tool_to_openai_function def agent_node(state: AdvancedState): print(f\n[Agent节点] 开始思考... 当前消息: {state[messages][-1].content}) # 创建基于ReAct模式的Agent agent create_react_agent( llmllm_with_tools, tools[calculator], output_parserReActSingleInputOutputParser() ) # 调用Agent传入当前对话历史和中间步骤 agent_response agent.invoke({ input: state[messages][-1].content, intermediate_steps: state[intermediate_steps] }) # 解析Agent的响应 if isinstance(agent_response, str): # 如果是最终答案 return { messages: [AIMessage(contentagent_response)], intermediate_steps: [] # 清空步骤 } else: # 如果是要调用工具agent_response 应该是一个 AgentAction 对象 # 这里简化处理假设返回的是工具名和输入 action agent_response return { intermediate_steps: [(action.tool, action.tool_input, action.log)] # 记录步骤 }6.3 创建工具执行节点def tool_node(state: AdvancedState): print(f\n[工具节点] 执行工具... 上一步记录: {state[intermediate_steps][-1]}) # 获取最近一个待执行的动作 last_step state[intermediate_steps][-1] tool_name, tool_input, _ last_step # 根据工具名找到对应的工具函数并执行 if tool_name calculator: result calculator.invoke(tool_input) else: result f未知工具: {tool_name} # 将工具执行结果添加到消息历史以便Agent在下一轮知晓 return { messages: [AIMessage(contentf工具 {tool_name} 返回结果: {result})], # 工具节点执行后不清空 intermediate_steps留给Agent节点判断下一步 }6.4 编译并运行高级Agent图# 创建新图 advanced_workflow StateGraph(AdvancedState) # 添加节点 advanced_workflow.add_node(agent, agent_node) advanced_workflow.add_node(tool, tool_node) # 设置入口点 advanced_workflow.set_entry_point(agent) # 定义路由逻辑根据状态决定下一步 def route_after_agent(state: AdvancedState): # 检查上一步Agent是否产生了新的工具调用步骤 if state.get(intermediate_steps) and len(state[intermediate_steps]) 0: # 有未处理的工具调用去执行工具 return tool else: # 没有工具调用Agent已给出最终答案等待下一轮用户输入或结束 # 这里我们简化直接回到agent等待新输入。实际可能需要一个“等待用户”节点。 return agent def route_after_tool(state: AdvancedState): # 工具执行完毕后总是回到Agent进行下一步思考 return agent # 添加条件边 advanced_workflow.add_conditional_edges( agent, route_after_agent, {tool: tool, agent: agent} ) advanced_workflow.add_edge(tool, agent) # 编译图 advanced_app advanced_workflow.compile()功能测试你可以编写一个简单的循环来测试这个Agent# 模拟交互 state AdvancedState(messages[HumanMessage(content123乘以456等于多少)], intermediate_steps[]) MAX_TURNS 10 for i in range(MAX_TURNS): print(f\n--- 第 {i1} 轮执行 ---) state advanced_app.invoke(state) last_msg state[messages][-1] print(f系统输出: {last_msg.content}) # 简单判断如果最后一条消息是AI的最终回答非工具返回且我们觉得可以了就跳出 if isinstance(last_msg, AIMessage) and 工具 not in last_msg.content: user_feedback input(fAI回复: {last_msg.content} \n是否继续(输入‘继续’或直接输入新问题输入‘退出’结束): ) if user_feedback 退出: break else: state[messages].append(HumanMessage(contentuser_feedback))这个测试中当你问“123乘以456等于多少”Agent会思考决定调用计算器工具执行计算获取结果然后组织语言回复你。这演示了LangGraph如何协调LLM、工具和状态完成一个多步骤的推理任务。7. 接口API与批量任务处理虽然LangGraph本身是一个编程框架但你可以轻松地将其包装成Web API服务以供其他系统调用或处理批量任务。7.1 使用FastAPI创建API服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Any app FastAPI(titleLangGraph Agent API) # 定义请求体模型 class InvokeRequest(BaseModel): message: str # 可以扩展更多参数如session_id, config等 config: dict {} # 全局保存编译好的图应用假设是之前编译的 app # 在实际应用中你需要一个更优雅的方式来管理和加载不同的图 GRAPH_APP app # 这里用之前的简单对话app示例你可以替换成 advanced_app app.post(/invoke) async def invoke_agent(request: InvokeRequest): try: # 初始化状态或从数据库加载会话状态 initial_state AgentState(messages[HumanMessage(contentrequest.message)]) # 调用图 result_state GRAPH_APP.invoke(initial_state, configrequest.config) # 提取最后一条AI消息作为回复 ai_messages [msg for msg in result_state[messages] if isinstance(msg, AIMessage)] response ai_messages[-1].content if ai_messages else 未生成回复 return {response: response, state: result_state} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): return {status: healthy}使用uvicorn运行服务pip install fastapi uvicorn uvicorn your_api_file_name:app --host 0.0.0.0 --port 8000 --reload7.2 批量任务处理模式对于批量处理如处理一个文件中的多个问题你需要关注状态隔离和错误处理。import asyncio from concurrent.futures import ThreadPoolExecutor def process_single_item(question: str, app_instance): 处理单个问题的函数。注意为每个任务创建新的状态。 try: state AgentState(messages[HumanMessage(contentquestion)]) result_state app_instance.invoke(state) ai_messages [msg for msg in result_state[messages] if isinstance(msg, AIMessage)] answer ai_messages[-1].content if ai_messages else ERROR return {question: question, answer: answer, success: True} except Exception as e: return {question: question, error: str(e), success: False} async def batch_process(questions: list, max_workers: int 3): 批量处理问题列表。 # 注意这里共享了同一个app实例。如果图有内部可变状态可能需要深拷贝或为每个任务创建新实例。 # 对于无状态的图依赖传入的State共享是安全的。 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: loop asyncio.get_event_loop() tasks [ loop.run_in_executor(executor, process_single_item, q, GRAPH_APP) for q in questions ] results await asyncio.gather(*tasks, return_exceptionsFalse) return results # 使用示例 if __name__ __main__: question_list [你好吗, 计算一下22, 讲个笑话] final_results asyncio.run(batch_process(question_list)) for res in final_results: print(res)关键点状态隔离确保每个批量任务有自己独立的初始状态 (AgentState)。并发控制使用ThreadPoolExecutor或ProcessPoolExecutor控制并发数避免过度消耗资源尤其是API调用额度。错误处理单个任务失败不应导致整个批量作业崩溃。资源管理批量调用大模型API时注意速率限制和费用。8. 常见问题与排查方法在学习和使用LangGraph过程中你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案导入错误ModuleNotFoundError1. 未安装langgraph或langchain。2. 虚拟环境未激活。3. 包版本冲突。1. 运行pip list | grep lang检查。2. 确认命令行提示符前有(env_name)。3. 查看完整错误信息。1. 在正确的虚拟环境中执行pip install langgraph langchain。2. 尝试安装特定版本pip install langgraph0.0.xx。运行时错误OpenAI API认证失败1. API密钥未设置或错误。2. 环境变量未生效。3. 账户余额不足或区域限制。1. 在Python中import os; print(os.getenv(“OPENAI_API_KEY”))检查。2. 尝试在代码中直接openai.api_key “sk-...”。1. 确保正确设置环境变量并重启IDE/终端。2. 检查OpenAI平台账户状态。图编译错误提示状态字段问题1.State的TypedDict定义错误。2. 节点函数返回的字典键与State定义不匹配。3.Annotated归约器使用不当。1. 仔细检查State类每个字段的类型注解。2. 核对节点函数return的键名。1. 确保节点返回的字典键是State中定义的字段名。2. 对于列表字段使用Annotated[list, operator.add]。图执行陷入无限循环1. 条件边 (add_conditional_edges) 的逻辑有误始终返回同一个节点。2. 没有设置有效的结束条件 (END)。1. 在条件函数中打印日志检查返回值。2. 检查图结构确保存在通向END的路径。1. 重构条件判断逻辑确保在某些情况下能跳出循环。2. 可以设置最大循环次数在State中增加计数器字段。调用工具时LLM不识别或格式错误1. 工具未正确绑定到模型 (bind_tools)。2. 工具的函数描述 (docstring) 不清晰。3. 使用的模型不支持工具调用如某些本地模型。1. 检查llm.bind_tools([...])是否成功。2. 查看模型调用时的原始提示词或响应。1. 使用OpenAI的gpt-3.5-turbo或gpt-4等明确支持工具调用的模型。2. 优化工具的命名和描述使其对LLM更友好。多轮对话中上下文丢失或混乱1.State中的messages列表管理不当。2. 每次调用invoke都使用了全新的初始状态未保留历史。1. 打印每次调用前后的state[‘messages’]。2. 确认你的应用逻辑是持续更新同一个state对象。1. 确保将上一次invoke返回的state作为下一次调用的输入或从中提取所需部分。2. 对于Web应用需要将会话状态保存在服务器内存或数据库中。性能慢响应延迟高1. LLM API调用网络延迟。2. 图逻辑复杂节点过多。3. 未使用异步调用。1. 使用工具监控每个节点的执行时间。2. 检查是否在循环中进行了不必要的重复计算。1. 考虑使用更快的模型或本地模型。2. 优化图结构合并简单节点。3. 对于IO密集型操作如API调用使用async节点和异步调用。9. 最佳实践与使用建议基于项目开发和社区经验遵循以下最佳实践可以让你更高效、更稳定地使用LangGraph。从简单开始逐步复杂化不要一开始就设计庞大的图。先构建一个能跑通的最小可行图如本文的对话循环然后逐步添加节点、工具和条件逻辑。状态设计要精简State应该只包含工作流真正需要共享和更新的数据。避免将临时变量或大型对象放入状态这会影响序列化和传递效率。善用可视化调试LangGraph提供了将图可视化为PNG图像的功能。在开发过程中定期导出并查看图结构确保逻辑符合你的设计。from IPython.display import Image, display # 假设 app 是你的编译后的图 display(Image(app.get_graph().draw_mermaid_png()))为节点函数添加清晰的日志在节点函数的开始和结束处打印关键信息如输入状态、输出结果这是调试复杂工作流最有效的手段。隔离副作用工具调用、数据库读写、API请求等具有副作用的操作应尽量封装在独立的节点或工具函数中。这使你的图更易于测试和推理。版本控制你的图定义图的定义节点、边、状态就是你的核心业务逻辑代码。要像对待其他重要代码一样用Git进行版本管理。生产环境考虑持久化状态对于长时间运行的Agent需要将会话状态保存到数据库如Redis、PostgreSQL。错误恢复设计重试机制和降级策略特别是对于调用外部API或工具的节点。监控与指标为图的执行添加监控记录节点执行时间、调用次数、错误率等指标。合规与安全工具安全像calculator例子中的eval是极度危险的必须替换为安全的表达式解析器。用户输入过滤对所有来自用户的输入进行验证和清理防止注入攻击。内容审核如果Agent面向公众应在最终输出前加入内容安全过滤层避免生成有害内容。10. 总结与下一步LangGraph通过引入“图”这一核心抽象为构建复杂、有状态的AI应用提供了强大而优雅的范式。它填补了LangChain在控制流编排上的空白让你能够轻松设计出具备循环、分支和记忆能力的智能体。通过本文你应该已经掌握了从环境搭建、核心概念理解到构建简单和高级Agent再到将其封装为API服务的完整路径。最值得尝试的下一步是复现并改造示例亲手运行文中的代码理解每一步。然后尝试修改条件逻辑、添加一个新的工具如获取天气的API观察图的行为变化。应用于实际场景思考一个你工作或学习中的重复性任务如信息整理、报告生成、数据查询尝试用LangGraph将其自动化。从设计状态和节点开始。探索社区示例LangChain/LangGraph官方文档和GitHub仓库提供了大量示例如自主研究Agent、游戏NPC、客服机器人等这些都是极佳的学习材料。性能优化当你的图变得复杂时研究如何利用LangGraph的异步支持、检查点Checkpoint等高级特性来提升性能和可靠性。最容易踩的坑主要集中在状态管理、条件逻辑设计和工具调用的集成上。多利用日志和可视化工具进行调试遵循“小步快跑”的迭代开发方式你将能越来越熟练地驾驭这个强大的框架构建出真正智能的AI应用。 LangGraph入门指南：从零构建有状态AI代理与自动化工作流
这次我们来看一个在B站上非常热门的LangGraph教学视频资源。这个系列视频号称“最火最全”旨在从零开始系统性地讲解LangGraph帮助开发者入门AI大模型应用开发。对于想要构建复杂、有状态的AI代理Agent和自动化工作流的开发者来说LangGraph是一个至关重要的框架它补足了LangChain在控制流方面的能力。本文不是视频的逐字稿而是基于其核心教学内容为你整理成一篇结构清晰、可操作性强的技术指南。我们将重点关注LangGraph是什么、它能解决什么问题、以及如何从环境搭建到实际开发一步步上手。无论你是想学习AI Agent开发还是希望将大模型能力集成到更复杂的业务流程中这篇文章都能提供直接的路径。1. 核心能力速览在深入细节之前我们先通过一个表格快速了解LangGraph的核心定位和能力边界这有助于你判断它是否是你当前需要的工具。能力项说明项目类型用于构建有状态、多环节AI工作流的框架库是LangChain生态的重要扩展。核心价值解决LangChain在复杂控制流如循环、分支、回溯上的不足使构建的Agent具备“长期记忆”和决策能力。编程语言主要支持 Python。硬件门槛无特殊要求。LangGraph本身是编排框架计算负载取决于其调用的底层大模型如GPT-4、本地LLM。主要功能1. 定义包含节点Node和边Edge的工作流图Graph。2. 支持条件判断和循环实现动态执行路径。3. 维护全局状态State在不同节点间传递和更新信息。4. 与LangChain组件无缝集成轻松调用工具、模型、记忆等。适合场景1. 构建复杂的对话Agent如客服、游戏NPC。2. 实现多步骤任务自动化如研究助手、数据分析流水线。3. 开发需要根据中间结果动态调整流程的应用。不适合场景1. 简单的单次问答直接用LangChain Chain即可。2. 无需状态维护的批量文本处理。2. LangGraph 是什么与 LangChain 有何区别很多初学者会混淆LangGraph和LangChain。简单来说你可以把LangChain看作是一个提供了各种预制零件模型I/O、记忆、工具的“工具箱”而LangGraph则是用来将这些零件按照复杂蓝图组装成自动化“流水线”或“机器人”的“设计图与控制器”。LangChain的核心是“链”Chain它倾向于线性执行。虽然也能处理一些分支但在需要根据运行时的结果反复循环或跳转的场景下设计和实现会变得非常笨拙。LangGraph引入了“图”Graph的概念其核心是一个有向图结构。这个图由节点Nodes代表一个执行单元可以是一个LLM调用、一个工具调用或任何函数。边Edges定义节点之间的执行顺序。边可以是固定的也可以是条件的根据当前状态决定下一个执行哪个节点。状态State一个在所有节点间共享和传递的字典记录了整个工作流的上下文信息。这种设计使得实现“循环直到满足条件”、“尝试方案A失败后自动切换到方案B”这样的逻辑变得直观且容易。因此当你的AI应用需要“长期记忆”和“自主决策”能力时LangGraph是比单纯使用LangChain更强大的选择。3. 环境准备与前置条件开始使用LangGraph前你需要准备好Python开发环境。以下是详细的步骤和注意事项。3.1 基础环境配置Python版本建议使用 Python 3.8 至 3.11 版本。避免使用过新如3.12的某些早期版本可能存在的兼容性问题。包管理工具使用pip进行安装。强烈建议使用虚拟环境如venv或conda来隔离项目依赖。网络环境由于需要安装开源包并可能调用在线大模型API如OpenAI请确保你的开发环境具备稳定的网络连接。3.2 创建并激活虚拟环境这是最佳实践可以避免不同项目间的包冲突。# 1. 创建虚拟环境以venv为例项目目录为my_agent python -m venv my_agent_env # 2. 激活虚拟环境 # 在 Windows 上 my_agent_env\Scripts\activate # 在 macOS/Linux 上 source my_agent_env/bin/activate # 激活后命令行提示符前通常会显示环境名如 (my_agent_env)4. 安装部署与启动方式LangGraph的安装非常简单因为它本质上是一个Python库。我们将同时安装LangChain因为两者需要协同工作。4.1 核心库安装在激活的虚拟环境中执行以下命令pip install langgraph langchain这条命令会安装LangGraph及其核心依赖。langchain包提供了模型、工具等基础组件。4.2 模型提供商SDK安装LangGraph/LangChain本身不提供模型你需要连接一个LLM服务。以最常用的OpenAI为例pip install openai安装后你需要设置OpenAI的API密钥。通常通过环境变量来管理# 在命令行中临时设置仅当前会话有效 export OPENAI_API_KEY你的-api-key-here # Windows (cmd) 使用 # set OPENAI_API_KEY你的-api-key-here # Windows (PowerShell) 使用 # $env:OPENAI_API_KEY你的-api-key-here重要你也可以在代码中直接设置但为了安全更推荐使用环境变量或.env文件。4.3 验证安装创建一个简单的Python脚本test_install.py来验证基础环境是否就绪import langchain import langgraph import openai print(fLangChain version: {langchain.__version__}) print(fLangGraph version: {langgraph.__version__}) print(fOpenAI version: {openai.__version__}) print(所有核心库导入成功)运行该脚本python test_install.py如果成功输出版本号且没有报错说明基础环境配置完成。5. 核心概念与第一个LangGraph应用理解LangGraph的最佳方式就是动手构建一个。我们从最简单的“对话循环”Agent开始它能够持续与用户对话直到用户说再见。5.1 定义状态State状态是图的“记忆中枢”。我们使用TypedDict来定义状态的类型确保结构清晰。from typing import TypedDict, Annotated import operator class AgentState(TypedDict): # 存储当前的对话消息列表 messages: Annotated[list, operator.add] # 可以添加更多字段例如用户偏好、对话轮次等 # turn_count: int这里messages字段使用Annotated[list, operator.add]注解。这是LangGraph的一个关键特性它声明了对该字段的归约器。operator.add意味着当多个节点并发修改此字段时虽然本例没有并发它们的修改会通过“相加”即列表拼接的方式合并。对于列表这通常是我们期望的行为。5.2 创建节点Nodes节点是执行具体工作的函数。它接收当前State执行操作并返回一个包含更新后状态的字典。from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage # 初始化大语言模型 llm ChatOpenAI(modelgpt-3.5-turbo) # 定义“调用模型”节点 def call_model(state: AgentState): print(f[节点 call_model] 收到消息历史: {state[messages]}) # 将消息历史发送给LLM获取回复 response llm.invoke(state[messages]) # 将AI的回复添加到消息历史中 return {messages: [response]} # 定义“人类输入”节点模拟用户 def human_input(state: AgentState): user_input input(用户说: ) # 将用户输入转换为消息格式并添加到历史 return {messages: [HumanMessage(contentuser_input)]}5.3 创建条件边Conditional Edges与图编译图需要知道执行完一个节点后接下来该去哪里。我们通过条件函数来实现。from langgraph.graph import StateGraph, END # 判断是否应该继续对话的条件函数 def should_continue(state: AgentState): messages state[messages] last_message messages[-1] # 如果最后一条消息是AI说的且内容包含“再见”则结束 if isinstance(last_message, AIMessage) and 再见 in last_message.content: print([条件判断] 检测到‘再见’结束对话。) return end # 否则继续对话 return continue # 创建图构建器 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(human, human_input) workflow.add_node(model, call_model) # 设置入口点从人类输入开始 workflow.set_entry_point(human) # 添加边人类输入后总是去调用模型 workflow.add_edge(human, model) # 添加条件边调用模型后根据条件决定下一步 workflow.add_conditional_edges( model, should_continue, # 条件判断函数 { continue: human, # 如果返回continue跳转到human节点 end: END # 如果返回end图执行结束 } ) # 编译图得到一个可执行对象 app workflow.compile()5.4 运行与效果验证现在我们可以运行这个简单的对话Agent了。# 初始化状态 initial_state AgentState(messages[]) # 运行图 final_state app.invoke(initial_state) print(\n 对话结束 ) print(最终消息历史) for msg in final_state[messages]: print(f{type(msg).__name__}: {msg.content})操作步骤与预期结果运行上述代码。程序会提示“用户说”输入“你好”。LLM会回复问候语。程序会再次提示“用户说”输入“今天天气怎么样”。LLM会模拟回答天气。你可以继续对话直到你对LLM说“再见”或者LLM在回复中说出“再见”对话循环将终止。控制台会打印出完整的对话历史。判断成功标准程序能正确接收用户输入。LLM能基于对话历史生成连贯回复。当对话内容触发结束条件时图能正确停止。6. 构建高级Agent集成工具与记忆一个真正的智能Agent不仅能聊天还能使用工具如搜索、计算并拥有更丰富的记忆。下面我们构建一个能使用计算器和拥有短期记忆的Agent。6.1 定义更复杂的状态与工具from typing import List from langchain.tools import tool from langchain_core.messages import BaseMessage import json # 定义工具一个简单的计算器 tool def calculator(expression: str) - str: 计算一个数学表达式的值。支持 , -, *, /。 try: # 警告实际生产环境应用必须对输入做严格安全检查避免eval的安全风险。 # 此处仅为演示。 result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} # 定义状态包含消息和中间步骤用于记录工具调用 class AdvancedState(TypedDict): messages: Annotated[List[BaseMessage], operator.add] # 记录Agent思考的中间步骤对调试和观察非常有用 intermediate_steps: Annotated[List[dict], operator.add] # 初始化模型并绑定工具。使用 bind_tools 让模型知道它可以调用哪些工具。 llm_with_tools ChatOpenAI(modelgpt-3.5-turbo).bind_tools([calculator])6.2 创建智能路由节点这个节点是Agent的大脑它决定是直接回复还是调用工具。from langchain.agents import create_react_agent from langchain.agents.output_parsers import ReActSingleInputOutputParser from langchain.tools.render import format_tool_to_openai_function def agent_node(state: AdvancedState): print(f\n[Agent节点] 开始思考... 当前消息: {state[messages][-1].content}) # 创建基于ReAct模式的Agent agent create_react_agent( llmllm_with_tools, tools[calculator], output_parserReActSingleInputOutputParser() ) # 调用Agent传入当前对话历史和中间步骤 agent_response agent.invoke({ input: state[messages][-1].content, intermediate_steps: state[intermediate_steps] }) # 解析Agent的响应 if isinstance(agent_response, str): # 如果是最终答案 return { messages: [AIMessage(contentagent_response)], intermediate_steps: [] # 清空步骤 } else: # 如果是要调用工具agent_response 应该是一个 AgentAction 对象 # 这里简化处理假设返回的是工具名和输入 action agent_response return { intermediate_steps: [(action.tool, action.tool_input, action.log)] # 记录步骤 }6.3 创建工具执行节点def tool_node(state: AdvancedState): print(f\n[工具节点] 执行工具... 上一步记录: {state[intermediate_steps][-1]}) # 获取最近一个待执行的动作 last_step state[intermediate_steps][-1] tool_name, tool_input, _ last_step # 根据工具名找到对应的工具函数并执行 if tool_name calculator: result calculator.invoke(tool_input) else: result f未知工具: {tool_name} # 将工具执行结果添加到消息历史以便Agent在下一轮知晓 return { messages: [AIMessage(contentf工具 {tool_name} 返回结果: {result})], # 工具节点执行后不清空 intermediate_steps留给Agent节点判断下一步 }6.4 编译并运行高级Agent图# 创建新图 advanced_workflow StateGraph(AdvancedState) # 添加节点 advanced_workflow.add_node(agent, agent_node) advanced_workflow.add_node(tool, tool_node) # 设置入口点 advanced_workflow.set_entry_point(agent) # 定义路由逻辑根据状态决定下一步 def route_after_agent(state: AdvancedState): # 检查上一步Agent是否产生了新的工具调用步骤 if state.get(intermediate_steps) and len(state[intermediate_steps]) 0: # 有未处理的工具调用去执行工具 return tool else: # 没有工具调用Agent已给出最终答案等待下一轮用户输入或结束 # 这里我们简化直接回到agent等待新输入。实际可能需要一个“等待用户”节点。 return agent def route_after_tool(state: AdvancedState): # 工具执行完毕后总是回到Agent进行下一步思考 return agent # 添加条件边 advanced_workflow.add_conditional_edges( agent, route_after_agent, {tool: tool, agent: agent} ) advanced_workflow.add_edge(tool, agent) # 编译图 advanced_app advanced_workflow.compile()功能测试你可以编写一个简单的循环来测试这个Agent# 模拟交互 state AdvancedState(messages[HumanMessage(content123乘以456等于多少)], intermediate_steps[]) MAX_TURNS 10 for i in range(MAX_TURNS): print(f\n--- 第 {i1} 轮执行 ---) state advanced_app.invoke(state) last_msg state[messages][-1] print(f系统输出: {last_msg.content}) # 简单判断如果最后一条消息是AI的最终回答非工具返回且我们觉得可以了就跳出 if isinstance(last_msg, AIMessage) and 工具 not in last_msg.content: user_feedback input(fAI回复: {last_msg.content} \n是否继续(输入‘继续’或直接输入新问题输入‘退出’结束): ) if user_feedback 退出: break else: state[messages].append(HumanMessage(contentuser_feedback))这个测试中当你问“123乘以456等于多少”Agent会思考决定调用计算器工具执行计算获取结果然后组织语言回复你。这演示了LangGraph如何协调LLM、工具和状态完成一个多步骤的推理任务。7. 接口API与批量任务处理虽然LangGraph本身是一个编程框架但你可以轻松地将其包装成Web API服务以供其他系统调用或处理批量任务。7.1 使用FastAPI创建API服务from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import Any app FastAPI(titleLangGraph Agent API) # 定义请求体模型 class InvokeRequest(BaseModel): message: str # 可以扩展更多参数如session_id, config等 config: dict {} # 全局保存编译好的图应用假设是之前编译的 app # 在实际应用中你需要一个更优雅的方式来管理和加载不同的图 GRAPH_APP app # 这里用之前的简单对话app示例你可以替换成 advanced_app app.post(/invoke) async def invoke_agent(request: InvokeRequest): try: # 初始化状态或从数据库加载会话状态 initial_state AgentState(messages[HumanMessage(contentrequest.message)]) # 调用图 result_state GRAPH_APP.invoke(initial_state, configrequest.config) # 提取最后一条AI消息作为回复 ai_messages [msg for msg in result_state[messages] if isinstance(msg, AIMessage)] response ai_messages[-1].content if ai_messages else 未生成回复 return {response: response, state: result_state} except Exception as e: raise HTTPException(status_code500, detailstr(e)) app.get(/health) async def health_check(): return {status: healthy}使用uvicorn运行服务pip install fastapi uvicorn uvicorn your_api_file_name:app --host 0.0.0.0 --port 8000 --reload7.2 批量任务处理模式对于批量处理如处理一个文件中的多个问题你需要关注状态隔离和错误处理。import asyncio from concurrent.futures import ThreadPoolExecutor def process_single_item(question: str, app_instance): 处理单个问题的函数。注意为每个任务创建新的状态。 try: state AgentState(messages[HumanMessage(contentquestion)]) result_state app_instance.invoke(state) ai_messages [msg for msg in result_state[messages] if isinstance(msg, AIMessage)] answer ai_messages[-1].content if ai_messages else ERROR return {question: question, answer: answer, success: True} except Exception as e: return {question: question, error: str(e), success: False} async def batch_process(questions: list, max_workers: int 3): 批量处理问题列表。 # 注意这里共享了同一个app实例。如果图有内部可变状态可能需要深拷贝或为每个任务创建新实例。 # 对于无状态的图依赖传入的State共享是安全的。 results [] with ThreadPoolExecutor(max_workersmax_workers) as executor: loop asyncio.get_event_loop() tasks [ loop.run_in_executor(executor, process_single_item, q, GRAPH_APP) for q in questions ] results await asyncio.gather(*tasks, return_exceptionsFalse) return results # 使用示例 if __name__ __main__: question_list [你好吗, 计算一下22, 讲个笑话] final_results asyncio.run(batch_process(question_list)) for res in final_results: print(res)关键点状态隔离确保每个批量任务有自己独立的初始状态 (AgentState)。并发控制使用ThreadPoolExecutor或ProcessPoolExecutor控制并发数避免过度消耗资源尤其是API调用额度。错误处理单个任务失败不应导致整个批量作业崩溃。资源管理批量调用大模型API时注意速率限制和费用。8. 常见问题与排查方法在学习和使用LangGraph过程中你可能会遇到以下典型问题。问题现象可能原因排查方式解决方案导入错误ModuleNotFoundError1. 未安装langgraph或langchain。2. 虚拟环境未激活。3. 包版本冲突。1. 运行pip list | grep lang检查。2. 确认命令行提示符前有(env_name)。3. 查看完整错误信息。1. 在正确的虚拟环境中执行pip install langgraph langchain。2. 尝试安装特定版本pip install langgraph0.0.xx。运行时错误OpenAI API认证失败1. API密钥未设置或错误。2. 环境变量未生效。3. 账户余额不足或区域限制。1. 在Python中import os; print(os.getenv(“OPENAI_API_KEY”))检查。2. 尝试在代码中直接openai.api_key “sk-...”。1. 确保正确设置环境变量并重启IDE/终端。2. 检查OpenAI平台账户状态。图编译错误提示状态字段问题1.State的TypedDict定义错误。2. 节点函数返回的字典键与State定义不匹配。3.Annotated归约器使用不当。1. 仔细检查State类每个字段的类型注解。2. 核对节点函数return的键名。1. 确保节点返回的字典键是State中定义的字段名。2. 对于列表字段使用Annotated[list, operator.add]。图执行陷入无限循环1. 条件边 (add_conditional_edges) 的逻辑有误始终返回同一个节点。2. 没有设置有效的结束条件 (END)。1. 在条件函数中打印日志检查返回值。2. 检查图结构确保存在通向END的路径。1. 重构条件判断逻辑确保在某些情况下能跳出循环。2. 可以设置最大循环次数在State中增加计数器字段。调用工具时LLM不识别或格式错误1. 工具未正确绑定到模型 (bind_tools)。2. 工具的函数描述 (docstring) 不清晰。3. 使用的模型不支持工具调用如某些本地模型。1. 检查llm.bind_tools([...])是否成功。2. 查看模型调用时的原始提示词或响应。1. 使用OpenAI的gpt-3.5-turbo或gpt-4等明确支持工具调用的模型。2. 优化工具的命名和描述使其对LLM更友好。多轮对话中上下文丢失或混乱1.State中的messages列表管理不当。2. 每次调用invoke都使用了全新的初始状态未保留历史。1. 打印每次调用前后的state[‘messages’]。2. 确认你的应用逻辑是持续更新同一个state对象。1. 确保将上一次invoke返回的state作为下一次调用的输入或从中提取所需部分。2. 对于Web应用需要将会话状态保存在服务器内存或数据库中。性能慢响应延迟高1. LLM API调用网络延迟。2. 图逻辑复杂节点过多。3. 未使用异步调用。1. 使用工具监控每个节点的执行时间。2. 检查是否在循环中进行了不必要的重复计算。1. 考虑使用更快的模型或本地模型。2. 优化图结构合并简单节点。3. 对于IO密集型操作如API调用使用async节点和异步调用。9. 最佳实践与使用建议基于项目开发和社区经验遵循以下最佳实践可以让你更高效、更稳定地使用LangGraph。从简单开始逐步复杂化不要一开始就设计庞大的图。先构建一个能跑通的最小可行图如本文的对话循环然后逐步添加节点、工具和条件逻辑。状态设计要精简State应该只包含工作流真正需要共享和更新的数据。避免将临时变量或大型对象放入状态这会影响序列化和传递效率。善用可视化调试LangGraph提供了将图可视化为PNG图像的功能。在开发过程中定期导出并查看图结构确保逻辑符合你的设计。from IPython.display import Image, display # 假设 app 是你的编译后的图 display(Image(app.get_graph().draw_mermaid_png()))为节点函数添加清晰的日志在节点函数的开始和结束处打印关键信息如输入状态、输出结果这是调试复杂工作流最有效的手段。隔离副作用工具调用、数据库读写、API请求等具有副作用的操作应尽量封装在独立的节点或工具函数中。这使你的图更易于测试和推理。版本控制你的图定义图的定义节点、边、状态就是你的核心业务逻辑代码。要像对待其他重要代码一样用Git进行版本管理。生产环境考虑持久化状态对于长时间运行的Agent需要将会话状态保存到数据库如Redis、PostgreSQL。错误恢复设计重试机制和降级策略特别是对于调用外部API或工具的节点。监控与指标为图的执行添加监控记录节点执行时间、调用次数、错误率等指标。合规与安全工具安全像calculator例子中的eval是极度危险的必须替换为安全的表达式解析器。用户输入过滤对所有来自用户的输入进行验证和清理防止注入攻击。内容审核如果Agent面向公众应在最终输出前加入内容安全过滤层避免生成有害内容。10. 总结与下一步LangGraph通过引入“图”这一核心抽象为构建复杂、有状态的AI应用提供了强大而优雅的范式。它填补了LangChain在控制流编排上的空白让你能够轻松设计出具备循环、分支和记忆能力的智能体。通过本文你应该已经掌握了从环境搭建、核心概念理解到构建简单和高级Agent再到将其封装为API服务的完整路径。最值得尝试的下一步是复现并改造示例亲手运行文中的代码理解每一步。然后尝试修改条件逻辑、添加一个新的工具如获取天气的API观察图的行为变化。应用于实际场景思考一个你工作或学习中的重复性任务如信息整理、报告生成、数据查询尝试用LangGraph将其自动化。从设计状态和节点开始。探索社区示例LangChain/LangGraph官方文档和GitHub仓库提供了大量示例如自主研究Agent、游戏NPC、客服机器人等这些都是极佳的学习材料。性能优化当你的图变得复杂时研究如何利用LangGraph的异步支持、检查点Checkpoint等高级特性来提升性能和可靠性。最容易踩的坑主要集中在状态管理、条件逻辑设计和工具调用的集成上。多利用日志和可视化工具进行调试遵循“小步快跑”的迭代开发方式你将能越来越熟练地驾驭这个强大的框架构建出真正智能的AI应用。