LangGraph入门指南：从零构建有状态AI代理与自动化工作流

这次我们来看一个在B站上非常热门的LangGraph教学视频资源。这个系列视频号称“最火最全”，旨在从零开始系统性地讲解LangGraph，帮助开发者入门AI大模型应用开发。对于想要构建复杂、有状态的AI代理（Agent）和自动化工作流的开发者来说，LangGraph是一个至关重要的框架，它补足了LangChain在控制流方面的能力。

本文不是视频的逐字稿，而是基于其核心教学内容，为你整理成一篇结构清晰、可操作性强的技术指南。我们将重点关注LangGraph是什么、它能解决什么问题、以及如何从环境搭建到实际开发一步步上手。无论你是想学习AI Agent开发，还是希望将大模型能力集成到更复杂的业务流程中，这篇文章都能提供直接的路径。

1. 核心能力速览

在深入细节之前，我们先通过一个表格快速了解LangGraph的核心定位和能力边界，这有助于你判断它是否是你当前需要的工具。

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 基础环境配置

1. Python版本：建议使用 Python 3.8 至 3.11 版本。避免使用过新（如3.12的某些早期版本）可能存在的兼容性问题。
2. 包管理工具：使用pip进行安装。强烈建议使用虚拟环境（如venv或conda）来隔离项目依赖。
3. 网络环境：由于需要安装开源包，并可能调用在线大模型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(f"LangChain version: {langchain.__version__}") print(f"LangGraph version: {langgraph.__version__}") print(f"OpenAI 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(model="gpt-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(content=user_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}")

操作步骤与预期结果：

1. 运行上述代码。
2. 程序会提示“用户说：”，输入“你好”。
3. LLM会回复问候语。
4. 程序会再次提示“用户说：”，输入“今天天气怎么样？”。
5. LLM会模拟回答天气。
6. 你可以继续对话，直到你对LLM说“再见”，或者LLM在回复中说出“再见”，对话循环将终止。
7. 控制台会打印出完整的对话历史。

判断成功标准：

• 程序能正确接收用户输入。
• 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(model="gpt-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( llm=llm_with_tools, tools=[calculator], output_parser=ReActSingleInputOutputParser() ) # 调用Agent，传入当前对话历史和中间步骤 agent_response = agent.invoke({ "input": state["messages"][-1].content, "intermediate_steps": state["intermediate_steps"] }) # 解析Agent的响应 if isinstance(agent_response, str): # 如果是最终答案 return { "messages": [AIMessage(content=agent_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(content=f"工具 `{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(content="123乘以456等于多少？")], intermediate_steps=[]) MAX_TURNS = 10 for i in range(MAX_TURNS): print(f"\n--- 第 {i+1} 轮执行 ---") 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(f"AI回复: '{last_msg.content}' \n是否继续？(输入‘继续’或直接输入新问题，输入‘退出’结束): ") if user_feedback == "退出": break else: state['messages'].append(HumanMessage(content=user_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(title="LangGraph 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(content=request.message)]) # 调用图 result_state = GRAPH_APP.invoke(initial_state, config=request.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_code=500, detail=str(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 --reload

7.2 批量任务处理模式

对于批量处理（如处理一个文件中的多个问题），你需要关注状态隔离和错误处理。

import asyncio from concurrent.futures import ThreadPoolExecutor def process_single_item(question: str, app_instance): """处理单个问题的函数。注意：为每个任务创建新的状态。""" try: state = AgentState(messages=[HumanMessage(content=question)]) 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_workers=max_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_exceptions=False) return results # 使用示例 if __name__ == "__main__": question_list = ["你好吗？", "计算一下2+2", "讲个笑话"] final_results = asyncio.run(batch_process(question_list)) for res in final_results: print(res)

关键点：

1. 状态隔离：确保每个批量任务有自己独立的初始状态 (AgentState)。
2. 并发控制：使用ThreadPoolExecutor或ProcessPoolExecutor控制并发数，避免过度消耗资源（尤其是API调用额度）。
3. 错误处理：单个任务失败不应导致整个批量作业崩溃。
4. 资源管理：批量调用大模型API时，注意速率限制和费用。

8. 常见问题与排查方法

在学习和使用LangGraph过程中，你可能会遇到以下典型问题。

9. 最佳实践与使用建议

基于项目开发和社区经验，遵循以下最佳实践可以让你更高效、更稳定地使用LangGraph。

1. 从简单开始，逐步复杂化：不要一开始就设计庞大的图。先构建一个能跑通的最小可行图（如本文的对话循环），然后逐步添加节点、工具和条件逻辑。
2. 状态设计要精简：State应该只包含工作流真正需要共享和更新的数据。避免将临时变量或大型对象放入状态，这会影响序列化和传递效率。
3. 善用可视化调试：LangGraph提供了将图可视化为PNG图像的功能。在开发过程中，定期导出并查看图结构，确保逻辑符合你的设计。

   from IPython.display import Image, display # 假设 `app` 是你的编译后的图 display(Image(app.get_graph().draw_mermaid_png()))

4. 为节点函数添加清晰的日志：在节点函数的开始和结束处打印关键信息（如输入状态、输出结果），这是调试复杂工作流最有效的手段。
5. 隔离副作用：工具调用、数据库读写、API请求等具有副作用的操作，应尽量封装在独立的节点或工具函数中。这使你的图更易于测试和推理。
6. 版本控制你的图定义：图的定义（节点、边、状态）就是你的核心业务逻辑代码。要像对待其他重要代码一样，用Git进行版本管理。
7. 生产环境考虑：
   • 持久化状态：对于长时间运行的Agent，需要将会话状态保存到数据库（如Redis、PostgreSQL）。
   • 错误恢复：设计重试机制和降级策略，特别是对于调用外部API或工具的节点。
   • 监控与指标：为图的执行添加监控，记录节点执行时间、调用次数、错误率等指标。
8. 合规与安全：
   • 工具安全：像calculator例子中的eval是极度危险的，必须替换为安全的表达式解析器。
   • 用户输入过滤：对所有来自用户的输入进行验证和清理，防止注入攻击。
   • 内容审核：如果Agent面向公众，应在最终输出前加入内容安全过滤层，避免生成有害内容。

10. 总结与下一步

LangGraph通过引入“图”这一核心抽象，为构建复杂、有状态的AI应用提供了强大而优雅的范式。它填补了LangChain在控制流编排上的空白，让你能够轻松设计出具备循环、分支和记忆能力的智能体。

通过本文，你应该已经掌握了从环境搭建、核心概念理解，到构建简单和高级Agent，再到将其封装为API服务的完整路径。最值得尝试的下一步是：

1. 复现并改造示例：亲手运行文中的代码，理解每一步。然后尝试修改条件逻辑、添加一个新的工具（如获取天气的API），观察图的行为变化。
2. 应用于实际场景：思考一个你工作或学习中的重复性任务（如信息整理、报告生成、数据查询），尝试用LangGraph将其自动化。从设计状态和节点开始。
3. 探索社区示例：LangChain/LangGraph官方文档和GitHub仓库提供了大量示例，如自主研究Agent、游戏NPC、客服机器人等，这些都是极佳的学习材料。
4. 性能优化：当你的图变得复杂时，研究如何利用LangGraph的异步支持、检查点（Checkpoint）等高级特性来提升性能和可靠性。

最容易踩的坑主要集中在状态管理、条件逻辑设计和工具调用的集成上。多利用日志和可视化工具进行调试，遵循“小步快跑”的迭代开发方式，你将能越来越熟练地驾驭这个强大的框架，构建出真正智能的AI应用。