这次我们来看一个关于 LangGraph 实战的 Agent 框架教程。这个教程的核心价值在于它提供了一个从零到一构建、理解和应用 LangGraph 来开发 AI Agent 的完整路径内容覆盖了从基础概念到复杂工作流编排的方方面面。对于想要深入 Agent 开发尤其是希望利用 LangChain 生态构建具备记忆、工具调用和复杂决策能力的智能体的开发者来说这是一个非常系统的学习资源。教程的重点不是空谈理论而是通过大量实战案例带你一步步搭建可运行的 Agent 系统。你会学到如何用 LangGraph 定义状态、编排节点、控制流程循环并最终构建出能处理多轮对话、执行工具链、甚至具备长期记忆的智能体。本文将从实战角度出发为你拆解这套教程的核心内容并提供一个清晰的、可操作的本地学习与验证路径。1. 核心能力速览能力项说明项目类型AI Agent 开发框架实战教程技术栈Python, LangChain, LangGraph, 可能涉及 OpenAI API 或其他 LLM 服务核心目标掌握使用 LangGraph 构建复杂、有状态的 AI Agent 工作流学习门槛需要具备基础的 Python 编程知识对 LangChain 有初步了解更佳硬件要求无特殊 GPU 要求主要依赖 CPU 和网络调用云端 LLM API环境依赖Python 3.8 LangChain/LangGraph 库 必要的 LLM API Key启动方式通过 Python 脚本或 Jupyter Notebook 运行示例代码是否支持 API教程本身是学习材料但学成后可自行构建 API 服务是否支持批量任务Agent 工作流设计本身支持批处理和异步任务适合场景开发者学习 Agent 架构、构建智能客服、自动化工作流、复杂决策系统2. 适用场景与使用边界这套 LangGraph 教程主要面向以下几类开发者LangChain 初学者/进阶者已经了解 LangChain 基础希望深入其最强大的工作流编排组件 LangGraph。AI Agent 爱好者/研究者对构建具备记忆、工具使用和规划能力的智能体感兴趣需要系统的工程化指导。全栈或后端工程师希望将 AI 能力以更可控、更复杂的方式集成到现有业务系统中例如客服机器人、自动化流程引擎、数据分析助手等。它能解决什么问题工作流编排将复杂的 AI 任务分解为多个步骤节点并定义它们之间的流转逻辑边。状态管理在整个 Agent 执行过程中维护和更新对话历史、中间结果、工具调用状态等。循环与条件分支实现类似while循环或if-else的逻辑让 Agent 能够根据当前状态决定下一步行动直到满足特定条件如用户满意、任务完成。与 LangChain 生态无缝集成方便地使用已有的 Chain、Tool、Memory 组件快速构建功能强大的 Agent。需要注意的使用边界非“开箱即用”产品这是一套教程你需要学习并编写代码来构建自己的 Agent而不是下载即用的软件。依赖外部 LLM大多数实战案例需要接入 OpenAI、Anthropic 等云端大语言模型的 API会产生相应费用。本地部署的大模型如通过 Ollama也可集成但可能需要额外配置。开发与调试成本构建复杂的、有状态的图Graph需要清晰的逻辑思维调试可能比线性 Chain 更复杂。合规与安全当 Agent 被赋予工具调用能力如网络搜索、数据库操作时必须严格设计权限和审核机制防止未经授权的操作。3. 环境准备与前置条件在开始跟随教程实践之前你需要准备好以下环境操作系统Windows 10/11, macOS, 或 Linux 发行版均可。本文示例命令以 macOS/Linux 的 bash 和 Windows 的 PowerShell 为主。Python 环境推荐使用 Python 3.8 至 3.11 版本。避免使用 Python 3.12 可能存在的某些库兼容性问题。强烈建议使用虚拟环境如venv或conda进行隔离。代码编辑器或 IDEVisual Studio Code, PyCharm 或 Jupyter Notebook 均可选择你熟悉的即可。版本管理工具Git可选用于克隆示例仓库。网络访问能够访问 Python 包索引 PyPI 以安装依赖以及能够调用你所选的 LLM API 服务如 OpenAI。LLM API 密钥准备一个有效的 OpenAI API Key或其他教程中使用的 LLM 服务商 API Key这是大部分示例运行的前提。4. 安装部署与启动方式由于这是一个教程合集其“部署”实质上是搭建本地学习开发环境。我们假设教程资源以代码仓库形式提供。步骤 1创建并激活虚拟环境# 创建项目目录并进入 mkdir langgraph-tutorial cd langgraph-tutorial # 创建虚拟环境 (venv) python -m venv venv # 激活虚拟环境 # macOS/Linux: source venv/bin/activate # Windows (PowerShell): .\venv\Scripts\Activate.ps1 # Windows (CMD): .\venv\Scripts\activate.bat步骤 2安装核心依赖教程的核心是 LangChain 和 LangGraph。通常还需要openai库来调用模型。pip install langchain langgraph openai根据教程的具体内容可能还需要安装其他工具包例如用于网页搜索的duckduckgo-search用于数学计算的numexpr等。这需要根据你实际获取到的教程代码中的import语句或requirements.txt文件来补充安装。# 示例安装可能需要的额外工具 pip install duckduckgo-search wikipedia numexpr sqlalchemy步骤 3设置 API 密钥在代码中直接硬编码 API Key 是不安全的。推荐使用环境变量。# macOS/Linux export OPENAI_API_KEYyour-openai-api-key-here # Windows (PowerShell) $env:OPENAI_API_KEYyour-openai-api-key-here或者在 Python 脚本中通过os.environ设置仅用于测试import os os.environ[OPENAI_API_KEY] your-openai-api-key-here步骤 4获取教程代码并运行假设教程代码位于一个 Git 仓库中。# 克隆仓库此处为示例URL请替换为实际地址 git clone https://github.com/example/langgraph-tutorial-code.git cd langgraph-tutorial-code # 运行一个简单的示例脚本验证环境 python 01_basic_graph.py如果一切顺利你将看到脚本输出表明你的 LangGraph 基础环境已经就绪。5. 功能测试与效果验证教程通常会由浅入深。我们可以设计几个关键的验证点来检验你是否掌握了核心概念。5.1 验证点一构建一个最简单的链式图测试目的验证 LangGraph 基础环境是否正常工作理解StateGraph、节点和边的概念。操作步骤创建一个新的 Python 文件如test_simple.py。编写一个包含两个节点的简单图第一个节点将输入字符串转为大写第二个节点在字符串后添加感叹号。编译并运行这个图。输入示例# test_simple.py from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态结构 class MyState(TypedDict): message: Annotated[str, operator.add] # 使用 add 操作符来累积消息 # 2. 定义节点函数 def node_to_upper(state: MyState): return {message: state[message].upper()} def node_add_exclamation(state: MyState): return {message: state[message] !} # 3. 构建图 workflow StateGraph(MyState) workflow.add_node(to_upper, node_to_upper) workflow.add_node(add_excl, node_add_exclamation) # 4. 定义边执行顺序 workflow.set_entry_point(to_upper) workflow.add_edge(to_upper, add_excl) workflow.add_edge(add_excl, END) # 5. 编译图 app workflow.compile() # 6. 执行图 initial_state {message: hello langgraph} result app.invoke(initial_state) print(f最终结果: {result[message]})预期输出最终结果: HELLO LANGGRAPH!判断成功程序无报错并输出经过两个节点处理后的最终字符串。常见失败原因TypedDict或Annotated导入错误确保 Python 版本 3.8并正确导入typing。operator.add使用不当在简单示例中如果不需要累积历史可以直接定义message: str。Annotated和operator.add用于更复杂的状态合并。5.2 验证点二实现带条件判断的循环图测试目的掌握 LangGraph 的核心优势——循环和条件路由。构建一个简单的“聊天机器人”直到用户说“再见”才结束。操作步骤定义状态包含对话历史和用户最新输入。创建两个节点一个模拟 AI 回复一个判断是否结束。使用add_conditional_edges来根据判断结果决定是继续循环还是结束。输入示例(简化版未接入真实 LLM)# test_conditional.py from typing import TypedDict, List, Literal from langgraph.graph import StateGraph, END class ChatState(TypedDict): messages: List[str] user_input: str def ai_response_node(state: ChatState): # 模拟AI回复实际应调用LLM ai_reply fAI: 我收到了你的消息{state[user_input]}。 new_messages state[messages] [fUser: {state[user_input]}, ai_reply] return {messages: new_messages, user_input: } def should_continue_node(state: ChatState) - Literal[continue, end]: # 判断条件如果用户输入包含“再见”则结束 if 再见 in state[user_input]: return end return continue workflow StateGraph(ChatState) workflow.add_node(respond, ai_response_node) workflow.add_node(check_end, should_continue_node) workflow.set_entry_point(respond) workflow.add_edge(respond, check_end) # 条件边根据 check_end 节点的返回值决定下一步 workflow.add_conditional_edges( check_end, lambda x: x, # should_continue_node 直接返回 “continue” 或 “end” { continue: respond, # 继续循环 end: END # 结束 } ) app workflow.compile() # 模拟多轮对话 state {messages: [], user_input: 你好} print(f用户: {state[user_input]}) state app.invoke(state) print(f状态: {state}) state[user_input] 今天天气怎么样 print(f\n用户: {state[user_input]}) state app.invoke(state) print(f状态: {state}) state[user_input] 再见 print(f\n用户: {state[user_input]}) final_state app.invoke(state) print(f最终状态: {final_state})预期输出程序应模拟两轮正常对话在第三轮输入“再见”后图执行结束并输出包含所有历史消息的最终状态。判断成功图能够根据条件正确地循环或终止状态中的messages列表正确记录了对话历史。5.3 验证点三集成真实 LLM 与工具调用测试目的这是 LangGraph 实战的关键。验证能否将 LangChain 的 LLM 和 Tools 集成到图中构建一个能自动决定使用工具的 Agent。操作步骤使用langchain_openai创建 ChatOpenAI 实例。定义工具例如一个简单的计算器函数。使用langgraph.prebuilt中的create_react_agent或create_tool_calling_agent来快速构建一个具备 ReAct 推理能力的 Agent 图。输入示例# test_agent_with_tools.py import os from langchain_openai import ChatOpenAI from langchain.agents import tool from langgraph.prebuilt import create_react_agent # 0. 设置API密钥如果未设置环境变量 os.environ[OPENAI_API_KEY] your-api-key # 1. 定义工具 tool def multiply(a: float, b: float) - float: Multiply two numbers. return a * b tool def add(a: float, b: float) - float: Add two numbers. return a b # 2. 准备LLM和工具列表 llm ChatOpenAI(modelgpt-3.5-turbo, temperature0) tools [multiply, add] # 3. 使用预构建函数创建Agent图 agent_executor create_react_agent(llm, tools) # 4. 运行Agent inputs {messages: [(user, 请计算 13 乘以 4 等于多少)]} result agent_executor.invoke(inputs) print(result[messages][-1].content)预期输出Agent 应该能够理解问题选择调用multiply工具并返回计算结果 “13 乘以 4 等于 52。” 或类似内容。判断成功Agent 正确调用了工具并给出了答案。观察控制台输出你应该能看到类似“Action: multiply, Action Input: {a: 13, b: 4}”的日志这表明 LangGraph 在驱动 LLM 进行思考和行动。常见失败原因API 密钥无效或未设置。LLM 模型名称错误或不可用。工具函数定义不符合tool装饰器的要求需有类型提示和文档字符串。6. 接口 API 与批量任务虽然教程本身可能不直接提供 HTTP API 服务但基于 LangGraph 构建的 Agent 可以很容易地封装成服务。6.1 将 Agent 封装为 FastAPI 服务启动方式创建一个main.py文件使用 FastAPI 框架。# main.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from langchain_openai import ChatOpenAI from langgraph.prebuilt import create_react_agent from .tools import your_tools_list # 导入你定义的工具 app FastAPI(titleLangGraph Agent API) llm ChatOpenAI(modelgpt-3.5-turbo) agent create_react_agent(llm, your_tools_list) class QueryRequest(BaseModel): message: str session_id: str None # 用于区分不同会话/状态 class QueryResponse(BaseModel): answer: str session_id: str app.post(/chat, response_modelQueryResponse) async def chat_with_agent(request: QueryRequest): try: # 这里需要实现状态管理。简单起见每次视为新会话。 inputs {messages: [(user, request.message)]} result agent.invoke(inputs) answer result[messages][-1].content return QueryResponse(answeranswer, session_idrequest.session_id or default) except Exception as e: raise HTTPException(status_code500, detailstr(e)) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)运行服务uvicorn main:app --reload --host 0.0.0.0 --port 8000调用示例 (curl)curl -X POST http://127.0.0.1:8000/chat \ -H Content-Type: application/json \ -d {message: 北京今天的天气怎么样, session_id: user_123}6.2 批量任务处理LangGraph 的图是纯函数式的非常适合批量处理。核心思路是循环调用app.invoke()。# batch_processing.py import asyncio from your_agent_graph import app # 导入你编译好的图 async def process_batch(queries: list, session_template: dict): results [] for query in queries: # 为每个查询创建独立的状态副本 initial_state session_template.copy() initial_state[user_input] query try: result app.invoke(initial_state) results.append((query, result[output])) # 假设输出在output字段 except Exception as e: results.append((query, fError: {e})) return results # 示例使用 queries [问题1, 问题2, 问题3] base_state {messages: [], some_context: default} loop asyncio.get_event_loop() batch_results loop.run_until_complete(process_batch(queries, base_state)) for q, a in batch_results: print(fQ: {q}\nA: {a}\n)最佳实践并发控制对于 IO 密集型的 LLM 调用可以使用asyncio.gather进行有限并发避免过量请求导致 API 限制。状态隔离确保批量任务中每个任务的状态是独立的避免交叉污染。错误处理与重试在批量循环中加入健壮的错误捕获和重试机制特别是对于网络请求。日志与监控记录每个任务的开始、结束时间和状态便于排查问题。7. 资源占用与性能观察LangGraph 本身是一个轻量级的编排框架资源占用主要取决于LLM 调用这是最大的开销来源包括网络延迟和 Token 消耗成本。工具执行如果你集成了计算密集型或 IO 密集型的工具如数据库查询、复杂计算这些工具本身会消耗资源。状态内存图运行过程中维护的状态对象。对于长对话或多轮复杂任务状态可能会变大但通常不是主要瓶颈。性能观察方法使用langsmithLangChain 官方提供的监控平台可以跟踪每个链、工具、LLM 调用的耗时、输入输出和 Token 使用情况。这是分析和优化 Agent 性能的最佳工具。代码插桩在关键节点函数的开始和结束处记录时间。import time def my_node(state): start time.time() # ... 节点逻辑 ... end time.time() print(fNode ‘my_node‘ took {end - start:.2f} seconds.) return new_state控制 LLM 成本与延迟模型选择在效果可接受的情况下使用更小、更快的模型如gpt-3.5-turbo而非gpt-4。缓存使用langchain.cache如InMemoryCache或SQLiteCache缓存相同的 LLM 请求结果。优化提示词精简system提示和上下文减少不必要的 Token 消耗。设置超时与重试为 LLM 调用配置合理的超时时间。8. 常见问题与排查方法问题现象可能原因排查方式解决方案导入错误No module named ‘langgraph‘LangGraph 库未安装或不在当前 Python 环境。在终端执行pip list | grep langgraph。在正确的虚拟环境中运行pip install langgraph。运行时报TypedDict或Annotated错误Python 版本过低或typing_extensions版本问题。检查 Python 版本python --version。升级 Python 到 3.8并确保pip install typing_extensions。调用 OpenAI API 超时或报错网络问题、API 密钥错误、额度不足、模型不可用。1. 测试网络连通性。2. 检查环境变量OPENAI_API_KEY。3. 登录 OpenAI 后台查看额度与账单。1. 配置网络代理如需且合规。2. 正确设置 API Key。3. 更换模型或检查服务状态。Agent 不调用工具直接回答1. 工具描述不清LLM 不理解。2. Prompt 未引导 Agent 使用工具。3. LLM 温度 (temperature) 过高随机性太强。1. 检查工具函数的文档字符串 (docstring) 是否清晰。2. 检查创建 Agent 时的system提示词。1. 完善工具描述。2. 使用 LangGraph 预建的create_react_agent它包含了标准的 ReAct 提示。3. 将temperature设为 0 或较低值。图陷入无限循环条件边的逻辑有误未能正确导向END节点。在条件判断节点中添加打印语句查看其返回值。仔细检查add_conditional_edges的条件函数和路由映射。确保存在一条路径能到达END。状态更新不符合预期对Annotated注解和operator的使用理解有误。简化状态先使用普通的str,int,list类型避免使用operator.add等归约器。阅读官方文档理解StateGraph中状态是如何根据节点返回值进行更新的。从简单例子开始测试。批量处理速度慢顺序执行未利用并发或单个 LLM 调用耗时过长。使用langsmith或计时器定位耗时环节。对于 IO 型的 LLM 调用改用异步 (async/await) 并利用asyncio.gather进行并发控制。9. 最佳实践与使用建议从简开始迭代复杂不要一开始就设计庞大的图。从一个线性链开始逐步添加条件分支、循环和更多节点。每步都进行测试。善用langgraph.prebuiltLangGraph 提供了一些预构建的高级组件如create_react_agent、create_tool_calling_agent。在理解其原理前直接使用它们可以快速搭建可用的 Agent之后再考虑自定义。状态设计要清晰使用TypedDict明确定义状态的结构。给每个字段起一个清晰的名字并考虑好哪些数据需要在节点间传递和持久化。为节点函数编写清晰的文档和日志每个节点函数都应该有明确的职责描述。在关键节点添加日志输出便于调试执行流。版本控制与实验管理使用 Git 管理你的图定义代码。复杂的 Agent 工作流可能有很多变体良好的版本控制有助于回溯和对比。测试与评估为你的 Agent 图编写单元测试和集成测试。测试不同的输入和边缘情况。对于非确定性输出LLM 生成可以测试其是否调用了正确的工具或输出是否符合某种模式。生产环境考虑错误处理在图的外层或关键节点内部添加try...except避免单个节点失败导致整个图崩溃。超时控制为整个图的执行或单个 LLM 调用设置超时。速率限制如果你直接调用付费 API务必在代码层面实现速率限制防止意外消费。安全性仔细审查 Agent 可用的工具。特别是文件操作、网络请求、数据库写入等工具必须施加严格的权限控制和输入验证。10. 总结与下一步这套 LangGraph 实战教程的价值在于它系统性地将 Agent 开发的抽象概念转化为了可执行的代码。通过它你不仅能学会 LangGraph 的 API更能理解如何设计一个智能体的“大脑”流程。最值得尝试的起点是亲手运行一个最简单的、带条件循环的图如本文的验证点二这能让你立刻感受到与传统线性链的区别。最容易踩的坑通常集中在状态管理Annotated的使用和条件边的逻辑设计上务必通过小例子反复验证。完成基础学习后你可以沿着以下几个方向深入探索高级模式研究StateGraph的更多特性如并行执行、子图嵌套、人工干预节点等。集成复杂工具将数据库、搜索引擎、内部 API 等封装成工具赋予 Agent 更强大的行动能力。实现长期记忆结合向量数据库让 Agent 能够记住跨会话的历史信息打造更个性化的助手。可视化与调试尝试使用langgraph的可视化功能来展示你的图结构这对于理解和调试复杂工作流至关重要。性能优化与部署将你的 Agent 图打包成 Docker 容器并部署到云服务器提供稳定的 API 服务。建议将本文作为学习地图结合具体的教程代码进行实践。遇到问题时多查阅 LangChain/LangGraph 官方文档和社区讨论。
