资讯动态

LangGraph与LangChain实战:构建具备长期记忆与复杂推理的本地AI智能体

发布时间:2026/8/21 10:42:52 来源:尧图企业网站定制
这次我们来看一个关于 LangGraph、LangChain 和 AI Agent 的实战教程。这个系列教程长达549集内容覆盖了从基础概念到项目实战的全过程核心目标是帮助你掌握如何利用 LangGraph 和 LangChain 框架构建具备长期记忆和复杂推理能力的本地 AI 智能体。对于想要深入 AI 应用开发特别是希望打造自主决策、能处理多步骤任务的智能系统的开发者来说这是一个非常值得关注的资源。教程的重点非常明确不是空谈理论而是手把手教你如何将 LangGraph 的图计算能力、LangChain 的模块化工具链以及本地大模型如 Ollama结合起来打造一个真正可运行的智能体。你会学到如何设计 Agent 的工作流、如何管理对话记忆Memory、如何处理工具调用错误以及如何将整个系统部署为可用的服务。无论你是想开发一个自动化的数据分析助手还是一个能理解复杂指令的客服机器人这套教程提供的思路和代码都具有很高的参考价值。本文不会复述549集的全部内容而是会提炼出其中最核心、最实用的部分为你梳理出一条清晰的学习和实践路径。我们将重点关注以下几个问题LangGraph 和 LangChain 结合能解决什么痛点构建一个智能体的最小可行步骤是什么如何利用 Ollama 等工具进行本地部署以控制成本和隐私在开发过程中又会遇到哪些典型的“坑”如内存泄漏、Agent 执行错误以及如何解决通过本文的结构化拆解你可以快速判断这个技术栈是否适合你的项目并知道如何开始你的第一个智能体开发。1. 核心能力速览在深入细节之前我们先通过一个表格快速了解基于 LangGraph 和 LangChain 构建的 AI 智能体核心能力与门槛。这能帮助你快速判断是否值得投入时间学习。能力项说明技术栈核心LangChain (工具链与模块化) LangGraph (有状态工作流图) 大语言模型 (如 GPT-4, Claude, 或本地模型如 Llama 3)核心功能构建具备多步骤推理、工具调用、长期记忆和复杂条件分支的 AI Agent智能体。关键特性有状态工作流通过图Graph定义 Agent 的执行步骤和状态流转。长期记忆Memory支持对话历史、知识库的存储与检索实现上下文感知。Human-in-the-Loop允许在关键节点人工干预或确认。错误处理与重试内置机制处理工具调用失败、模型输出格式错误等问题。部署方式可本地部署配合 Ollama 等本地模型框架也可云端部署调用 OpenAI、Anthropic 等 API。硬件门槛云端API模式对本地硬件无要求依赖网络和API费用。本地模型模式需根据所选大模型确定。例如运行 7B 参数的量化模型可能需要 8GB 以上显存纯 CPU 推理则需要足够的内存和耐心。启动与开发基于 Python 编写通过代码定义图和工作流。可启动为 Web 服务如 Flask/FastAPI提供 API 接口。适合场景自动化客服、复杂任务拆解与执行如数据分析报告生成、个性化助手、游戏 NPC、业务流程自动化等需要多轮交互和状态管理的场景。学习资源官方文档、社区教程、以及本文所参考的系列实战视频课程。2. 适用场景与使用边界在投入开发之前明确技术的适用场景和边界至关重要。LangGraphLangChain 的 Agent 架构并非万能但在特定问题上优势明显。它非常适合以下场景复杂任务自动化任务需要分解为多个子步骤且步骤间有依赖关系或条件判断。例如“监控市场新闻发现某公司负面消息时自动收集其财报数据并生成风险分析简报”。需持久化状态的对话系统客服机器人或个性化导师需要记住用户的历史偏好、对话上下文并在多轮对话中保持一致的行为逻辑。集成外部工具与APIAgent 需要调用搜索引擎、数据库、计算工具或企业内部系统来完成工作。LangChain 提供了大量的工具集成LangGraph 能优雅地编排调用流程。需要人工审核的流程在关键决策点如发送邮件、执行交易插入人工确认节点Human-in-the-Loop确保安全可控。它的局限性与不适用场景简单问答如果只是单轮的、无需调用工具和记忆的问答直接使用大模型 API 更简单高效引入 Agent 框架反而增加了复杂度。对延迟极其敏感Agent 的多步推理和工具调用会引入额外延迟不适合实时性要求极高的场景如高频交易。完全确定性的流程如果业务流程每一步都完全固定无分支判断使用传统的脚本或工作流引擎如 Airflow可能更稳定、更易调试。资源极度受限的环境在本地运行大型模型对算力要求高。如果硬件条件不足频繁的推理会导致体验很差。合规与安全边界提醒数据隐私如果处理用户敏感信息确保记忆存储Memory部分加密并遵守相关数据保护法规如 GDPR。本地部署是保障隐私的优选方案。工具调用安全严格限制 Agent 可调用的工具范围和权限。例如删除文件、调用支付接口等高风险操作必须加入权限校验或人工确认节点。内容合规对 Agent 的生成内容施加必要的审查和过滤防止产生有害或违规信息。可以利用 LangChain 的 Output Parsers 或后处理模块进行控制。模型授权确保所使用的语言模型无论是云端 API 还是本地模型拥有合法的使用授权。3. 环境准备与前置条件开始构建你的第一个智能体之前需要准备好开发环境。以下是一个通用的环境清单具体版本可根据项目要求调整。1. 基础软件环境操作系统推荐 Linux (Ubuntu 20.04) 或 macOS。Windows 也可行但需注意某些依赖如某些向量数据库客户端可能兼容性稍复杂。Python版本 3.10 或 3.11。这是 LangChain 和 LangGraph 社区主要支持且稳定的版本。包管理工具pip或poetry。推荐使用venv或conda创建独立的虚拟环境。2. 核心 Python 包你需要安装 LangChain、LangGraph 以及对应大模型的支持包。# 创建并激活虚拟环境以 venv 为例 python -m venv langgraph-agent-env source langgraph-agent-env/bin/activate # Linux/macOS # langgraph-agent-env\Scripts\activate # Windows # 安装核心框架 pip install langchain langgraph # 根据你选择的模型提供商安装对应的集成包 # 例如使用 OpenAI API pip install openai # 例如使用本地模型通过 Ollama pip install ollama # 例如需要用到向量数据库做记忆检索 pip install chromadb # 一个轻量级向量数据库3. 大模型接入准备云端 API 模式准备相应的 API Key如 OpenAI, Anthropic。将其设置为环境变量。export OPENAI_API_KEYyour-api-key-here本地模型模式安装并运行模型服务。以 Ollama 为例从 Ollama 官网下载并安装。拉取一个模型例如 Llama 3.1 8Bollama pull llama3.1:8b确保 Ollama 服务在本地运行默认端口 11434。4. 开发工具可选但推荐代码编辑器VS Code 或 PyCharm。调试工具LangGraph 提供了可视化工具来调试工作流图强烈建议在开发初期使用。API 测试工具如curl或 Postman用于测试部署后的 Agent 服务。4. 安装部署与启动方式安装好环境后我们来创建一个最简单的 LangGraph Agent 并启动它。这里我们以使用本地 Ollama 模型为例构建一个能进行简单对话并拥有记忆的 Agent。第一步编写基础 Agent 脚本创建一个名为simple_agent.py的文件。# simple_agent.py from langchain_community.chat_models import ChatOllama from langchain_core.messages import HumanMessage, AIMessage from langgraph.graph import StateGraph, END from typing import TypedDict, List, Annotated import operator # 1. 定义状态State # 状态图会在这个状态对象上运行它包含了整个工作流需要的数据 class AgentState(TypedDict): messages: Annotated[List, operator.add] # 对话消息列表这是一个累加式状态 # 可以添加更多状态如 context, intermediate_steps 等 # 2. 初始化模型 # 连接到本地运行的 Ollama 服务使用 llama3.1:8b 模型 llm ChatOllama(modelllama3.1:8b, temperature0.7) # 3. 定义节点函数Node Function def call_model(state: AgentState): 调用语言模型生成回复 print(f[DEBUG] 当前消息历史: {state[messages]}) response llm.invoke(state[messages]) # 将模型的回复添加到消息历史中 return {messages: [response]} def human_input(state: AgentState): 模拟人工输入节点实际应用中可替换为真实用户输入接口 user_input input(\n[Human] 请输入你的问题 (或输入 quit 退出): ) if user_input.lower() quit: return {messages: [AIMessage(content对话结束。)], __end__: True} return {messages: [HumanMessage(contentuser_input)]} # 4. 构建图Graph workflow StateGraph(AgentState) # 添加节点 workflow.add_node(human, human_input) workflow.add_node(assistant, call_model) # 设置边的连接关系 workflow.set_entry_point(human) # 从 human 节点开始 workflow.add_edge(human, assistant) # human 输入后流向 assistant workflow.add_edge(assistant, human) # assistant 回复后再次等待 human 输入 # 注意这是一个简单的循环没有终止条件。我们通过 human_input 函数中的 quit 逻辑来结束。 # 编译图 app workflow.compile() # 5. 运行图 if __name__ __main__: print(启动简单对话 Agent (使用 Ollama Llama 3.1)...) initial_state {messages: [AIMessage(content你好我是你的AI助手。)]} # 运行图并传入初始状态 for output in app.stream(initial_state, stream_modevalues): # app.stream 会按照图的定义一步步执行 if __end__ in output and output[__end__]: print(对话结束。) break message output[messages][-1] print(f[{message.type}] {message.content})第二步运行 Agent在终端中确保 Ollama 服务正在运行然后执行你的脚本python simple_agent.py你应该会看到类似以下的交互启动简单对话 Agent (使用 Ollama Llama 3.1)... [assistant] 你好我是你的AI助手。 [Human] 请输入你的问题 (或输入 quit 退出): 今天的天气怎么样 [DEBUG] 当前消息历史: [AIMessage(content你好我是你的AI助手。), HumanMessage(content今天的天气怎么样)] [assistant] 我是一个AI助手无法访问实时数据。要获取最新天气信息建议您查看天气预报网站或应用。 [Human] 请输入你的问题 (或输入 quit 退出): quit 对话结束。第三步进阶部署为 API 服务要将这个 Agent 部署成一个 Web API 供其他程序调用我们可以使用 FastAPI。安装 FastAPI 和 Uvicornpip install fastapi uvicorn创建api_server.py# api_server.py from fastapi import FastAPI, HTTPException from pydantic import BaseModel from simple_agent import app as agent_app # 导入上面编译好的图 from langchain_core.messages import HumanMessage, AIMessage app FastAPI(titleLangGraph Agent API) class ChatRequest(BaseModel): message: str session_id: str default # 用于区分不同对话会话 class ChatResponse(BaseModel): response: str session_id: str # 简单的内存存储实际项目应使用数据库或 Redis memory_store {} app.post(/chat, response_modelChatResponse) async def chat_endpoint(request: ChatRequest): 接收用户消息返回Agent的回复 session_id request.session_id # 获取或初始化该会话的历史消息 if session_id not in memory_store: memory_store[session_id] {messages: [AIMessage(content你好)]} current_state memory_store[session_id] # 将用户新消息添加到状态中 current_state[messages].append(HumanMessage(contentrequest.message)) try: # 调用Agent图进行处理 # 这里我们简化只调用一次 assistant 节点。更复杂的图需要完整执行。 # 注意实际应根据图的逻辑来调用这里仅为示例。 from simple_agent import call_model new_state call_model(current_state) # 更新存储 memory_store[session_id] new_state # 提取最新的AI回复 latest_message new_state[messages][-1] response_text latest_message.content if hasattr(latest_message, content) else str(latest_message) return ChatResponse(responseresponse_text, session_idsession_id) except Exception as e: raise HTTPException(status_code500, detailfAgent执行出错: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)启动 API 服务python api_server.py服务启动后你可以通过http://localhost:8000/docs访问自动生成的 API 文档并使用/chat端点进行测试。5. 功能测试与效果验证构建好 Agent 后需要通过一系列测试来验证其核心功能是否按预期工作。我们将从基础对话、记忆持久化、工具调用和错误处理几个维度进行测试。5.1 基础对话与上下文记忆测试测试目的验证 Agent 能否进行多轮对话并正确引用之前的对话历史。操作步骤运行simple_agent.py或通过 API 发送连续消息。进行如下对话序列用户“我叫小明。”Agent应回复问候如“你好小明”用户“我的名字是什么”预期结果Agent 的第二轮回复应能正确说出“小明”。这证明messages状态被成功累加并传递给了模型。判断成功Agent 的第二次回复中包含“小明”。常见失败原因状态State定义错误messages字段未使用Annotated[List, operator.add]进行累加声明。模型上下文长度不足导致最早的消息被“遗忘”。需要检查模型的最大 token 数或引入摘要式记忆Summary Memory。5.2 工具调用能力测试测试目的验证 Agent 能否根据用户需求正确调用外部工具如计算器、搜索引擎并整合结果。操作步骤为 Agent 增加一个计算工具。修改simple_agent.py引入 LangChain 的Tool和create_react_agent一种经典的 Agent 执行器。向 Agent 提问“计算 125 乘以 38 等于多少”示例代码片段工具定义与调用from langchain.agents import create_react_agent from langchain.tools import Tool def multiplier(a: float, b: float) - float: Multiply two numbers. return a * b tools [ Tool( nameMultiplier, funcmultiplier, descriptionUseful for multiplying two numbers together., ) ] # 使用 create_react_agent 来构建一个能使用工具的 Agent # 注意这里需要将 llm 适配为支持 ReAct 框架的格式具体请参考 LangChain 文档预期结果Agent 应识别出需要计算调用Multiplier工具并返回计算结果“4750”。判断成功最终回复中包含正确的计算结果。常见失败原因工具描述description不够清晰导致模型无法理解何时调用该工具。模型输出格式不符合 ReAct 或所选 Agent 执行器的预期导致解析失败。需要检查输出解析器Output Parser。5.3 长期记忆向量数据库测试测试目的验证 Agent 能否从大量的历史信息或知识库中检索出相关上下文。操作步骤集成向量数据库如 Chroma。将一段文本如产品手册存入向量库。向 Agent 提问一个需要从该文本中寻找答案的问题。预期结果Agent 的回复应基于检索到的相关文本片段而不是凭空生成。判断成功回复内容与存入的文本事实相符。常见失败原因文本切分Chunking策略不当导致检索不到关键信息。检索器Retriever返回的片段数量k值不合适过多或过少。嵌入模型Embedding Model不适合当前语种或领域。5.4 错误处理与重试机制测试测试目的验证当工具调用失败或模型输出不符合预期时Agent 能否优雅地处理。操作步骤模拟一个会随机失败的工具。观察 Agent 的行为。在 LangGraph 中可以通过在图中添加“条件边”来实现重试逻辑。示例在 State 中增加failures计数器并在工具调用节点后添加判断。class AgentState(TypedDict): messages: Annotated[List, operator.add] failures: int # 失败计数器 def tool_node(state: AgentState): try: result some_tool(state) return {result: result, failures: 0} except Exception: # 失败时增加计数器并返回一个错误标志 return {failures: state.get(failures, 0) 1, error: Tool failed} # 在图中可以根据 state[failures] 的值来决定是进入重试节点还是错误处理节点。预期结果当工具失败次数小于阈值时Agent 应尝试重试或选择备用方案超过阈值时应转入错误处理流程如通知人工。判断成功系统没有因未处理的异常而崩溃并按照预设的重试逻辑运行。常见失败原因错误状态未在图中正确传递或条件判断的逻辑有误。6. 接口 API 与批量任务当智能体功能稳定后将其封装成服务供其他系统调用是常见的需求。同时处理批量任务也能极大提升效率。6.1 完善 API 服务前面的api_server.py是一个极简示例。一个生产可用的 API 服务需要考虑更多因素会话管理使用数据库如 SQLite、PostgreSQL或 Redis 来持久化存储会话状态而不是内存字典。异步支持模型推理和工具调用可能是 I/O 密集型操作使用async/await可以提高并发处理能力。确保你的 LangChain 调用支持异步很多组件已支持。认证与限流为 API 添加 API Key 认证、请求速率限制防止滥用。更完整的图执行API 端点应能驱动整个 LangGraph 工作流完成一次完整的执行循环而不仅仅是调用一个节点。改进的 API 端点示例异步驱动完整图app.post(/chat_advanced, response_modelChatResponse) async def chat_advanced(request: ChatRequest): session_id request.session_id # 1. 从数据库加载会话状态 session_state await load_state_from_db(session_id) or initial_state # 2. 将用户输入添加到状态 session_state[messages].append(HumanMessage(contentrequest.message)) # 3. 执行编译好的 LangGraph App final_state None async for event in agent_app.astream(session_state, stream_modevalues): # 这里可以处理流式输出或者等待最终状态 if __end__ in event: final_state event break if not final_state: raise HTTPException(status_code500, detailAgent execution did not reach an end state.) # 4. 保存更新后的状态到数据库 await save_state_to_db(session_id, final_state) # 5. 提取回复 # 假设最终状态里最后一条消息是AI的回复 last_msg final_state[messages][-1] return ChatResponse(responselast_msg.content, session_idsession_id)6.2 批量任务处理对于需要处理大量独立任务的场景如分析100份用户反馈可以构建一个批量处理管道。设计思路任务队列使用 Celery、RQ 或简单的concurrent.futures线程池来管理任务队列。无状态 Agent每个任务使用独立的、初始化的 Agent 实例避免状态交叉污染。或者为每个任务分配唯一的session_id。结果收集将每个任务的结果成功、失败、输出存储到数据库或文件中。进度监控提供接口查询批量任务的整体进度。简化版批量处理脚本示例import asyncio from typing import List from your_agent_module import get_agent_app # 导入你的Agent图 async def process_one_task(task_input: str, task_id: str): 处理单个任务 agent_app get_agent_app() # 获取Agent应用实例 initial_state {messages: [HumanMessage(contenttask_input)], task_id: task_id} try: final_state None async for event in agent_app.astream(initial_state, stream_modevalues): if __end__ in event: final_state event break # 从 final_state 中提取结果 result extract_result(final_state) return {task_id: task_id, status: success, result: result} except Exception as e: return {task_id: task_id, status: failed, error: str(e)} async def process_batch(tasks: List[str]): 并发处理一批任务 # 使用 asyncio.gather 进行并发控制注意不要超过系统负载 semaphore asyncio.Semaphore(5) # 控制最大并发数为5 async def sem_task(task, idx): async with semaphore: return await process_one_task(task, ftask-{idx}) coroutines [sem_task(task, i) for i, task in enumerate(tasks)] results await asyncio.gather(*coroutines, return_exceptionsTrue) # 处理结果 for res in results: if isinstance(res, Exception): print(f任务发生异常: {res}) else: print(f任务 {res[task_id]} 完成状态: {res[status]}) return results # 使用示例 if __name__ __main__: task_list [分析一下这份合同的风险点。, 总结这篇技术文章的核心思想。, 将以下需求转化为用户故事...] asyncio.run(process_batch(task_list))7. 资源占用与性能观察在本地部署模式下资源管理是关键。你需要监控 Agent 运行时的内存、CPU 和 GPU 使用情况。1. 显存与内存占用主要来源加载的大语言模型是内存消耗大户。一个 7B 参数的模型以 FP16 精度加载就需要约 14GB GPU 显存。使用量化技术如 GPTQ, AWQ可将显存需求降低到 4-8GB。观察方法在 Linux 下可以使用nvidia-smi命令实时查看 GPU 显存占用。在 Python 中可以使用psutil库监控进程的内存使用。import psutil import os process psutil.Process(os.getpid()) print(f内存使用: {process.memory_info().rss / 1024 ** 2:.2f} MB)2. 推理速度影响因素模型大小、量化程度、提示词长度、生成 token 数量、硬件性能GPU CPU。优化建议使用量化模型这是平衡速度和精度最有效的手段。调整生成参数减少max_tokens使用更高效的采样策略如greedy而非nucleus。缓存注意力K/V Cache对于多轮对话确保模型能缓存之前的对话历史避免重复计算。批处理Batching在批量任务中如果输入长度相近可以尝试微批处理以提高 GPU 利用率。3. LangGraph 与 LangChain 开销框架本身开销很小主要是 Python 对象的管理和状态传递。但在极端复杂的图中如果状态对象非常庞大例如包含大量检索到的文档可能会增加内存压力。建议定期清理状态中不再需要的中间数据。4. 常见性能瓶颈与排查现象响应缓慢排查使用 Python 的cProfile或line_profiler对agent_app.invoke()或关键节点函数进行性能分析找到耗时最长的部分。可能原因工具调用如网络请求超时、向量检索速度慢、模型推理本身慢。现象内存/显存持续增长Memory Leak排查观察长时间运行或处理大量请求后内存是否被回收。搜索网络热词中提到的kmeans is known to have a memory leak on windows with mkl这类已知问题。可能原因第三方库的内存泄漏、全局变量累积未释放、大对象未及时销毁。确保在长时间运行的服务中会话状态有超时清理机制。8. 常见问题与排查方法在开发 LangGraph Agent 的过程中你几乎一定会遇到下面这些问题。这里提供一个排查清单。问题现象可能原因排查方式解决方案ModuleNotFoundError: No module named langgraph依赖未正确安装。检查虚拟环境是否激活以及pip list中是否有langgraph。在正确的虚拟环境中运行pip install langgraph。Agent execution terminated due to error.图中某个节点函数抛出未捕获的异常。查看完整的错误堆栈跟踪定位到具体的节点和代码行。在节点函数内部添加try...except进行错误处理或将错误信息放入状态由后续节点处理。Process exited with code 3221225477 / 0xc0000005 (memory access violation)通常发生在 Windows 上与底层 C/C 扩展库如 PyTorch、某些向量数据库驱动的内存访问冲突有关。确认 Python、CUDA、PyTorch 版本兼容性。1. 使用 Conda 环境管理依赖。2. 尝试更新或回退 PyTorch 版本。3. 在 Linux 子系统WSL2中运行。java: outofmemoryerror: insufficient memory如果在 JVM 环境中使用 LangChain4jJava版会出现此错误。检查 JVM 堆内存设置。增加 JVM 启动参数中的堆内存例如-Xmx4g。No available shared memory broadcast block found in 60 seconds.多进程/分布式训练或推理时的通信问题。检查分布式后端设置如 PyTorch DDP。对于本地单机 Agent 开发通常不会遇到。如果出现检查是否误开启了多进程模式。Agent 不调用工具1. 工具描述不清。2. 模型能力不足。3. 提示词Prompt未引导模型使用工具。1. 打印出模型接收到的完整提示词检查工具描述是否在其中。2. 测试模型的基础推理能力。1. 优化工具描述使其清晰、具体。2. 使用能力更强的模型。3. 在系统提示词中明确要求模型使用工具。Agent 陷入循环不结束图中缺少终止条件或条件判断逻辑有误。使用 LangGraph 的调试工具可视化执行流程观察状态流转。在图中明确设置END节点并在条件边conditional edge中设置合理的结束逻辑。向量检索返回无关内容1. 嵌入模型不匹配。2. 文本切分不合理。3. 检索 top-k 值不合适。1. 检查检索到的文本片段与查询的相关性。2. 尝试不同的切分策略如按句子、按固定长度。1. 尝试不同的嵌入模型。2. 调整文本切分器chunker的参数。3. 优化检索查询的写法或使用多查询检索Multi-Query Retrieval。API 服务请求超时单次 Agent 执行时间过长超过了 HTTP 服务器的默认超时时间。查看服务端日志确认请求处理时长。1. 优化 Agent 性能见第7节。2. 为 FastAPI/Uvicorn 设置更长的超时时间。3. 改为异步流式响应SSE/WebSocket先快速返回一个任务ID客户端再轮询结果。9. 最佳实践与使用建议基于上述的测试、部署和问题排查经验这里总结一些构建生产级 LangGraph Agent 的最佳实践。从简单开始迭代复杂不要一开始就设计一个包含几十个节点的复杂图。从一个只有“用户输入”-“模型调用”-“输出”的简单循环开始确保基础流程跑通。然后逐步添加工具调用、条件分支、记忆等模块。状态设计要精简State是图中流动的数据。只将必要的、会被多个节点读写的数据放入 State。避免放入大型对象如图片、长文档可以存放它们的引用如文件路径、数据库ID。充分利用可视化调试LangGraph 提供了将图可视化的方法。在开发初期一定要将你的图可视化出来检查节点和边的连接是否符合你的设计预期。这能极大减少逻辑错误。from langgraph.graph import StateGraph # ... 构建图的代码 ... graph workflow.compile() # 将图保存为PNG from IPython.display import Image, display display(Image(graph.get_graph().draw_mermaid_png()))为关键节点添加日志和监控在每个节点函数的开始和结束处记录日志包含当前状态的关键信息。这有助于追踪 Bug 和理解 Agent 的决策过程。可以考虑集成像prometheus-client这样的库来暴露性能指标。实施全面的错误处理图中每个可能失败的环节模型调用、工具调用、API请求都应该有错误处理逻辑。错误可以是状态的一部分由专门的“错误处理节点”来统一决定是重试、降级还是上报人工。记忆管理的策略对话记忆对于长对话定期对历史消息进行摘要Summary防止超出模型上下文长度。知识记忆使用向量数据库存储知识检索时结合元数据过滤提高准确率。缓存对频繁使用的、计算成本高的结果如某些工具调用结果、嵌入向量进行缓存。安全与合规前置输入输出过滤在图的入口和出口节点对用户输入和模型输出进行安全检查如敏感词过滤、Prompt 注入检测。工具权限实现细粒度的工具权限控制不同的用户或会话可能拥有不同的工具调用权限。审计日志记录所有用户交互、工具调用和模型响应便于事后审计和模型优化。10. 总结与下一步通过本文的梳理你应该对如何使用 LangGraph 和 LangChain 构建一个实用的 AI 智能体有了清晰的认识。这套技术栈的核心价值在于它提供了一种结构化、可调试、可扩展的方式来编排大语言模型的能力使其从单纯的文本生成器升级为可以自主执行复杂工作流的智能代理。最值得尝试的起点建议你从本文第4节的“简单对话 Agent”例子开始在本地成功运行起来。然后尝试为它添加一个工具比如一个获取当前时间的函数并观察它如何学习使用这个工具。这个“Hello World”级别的成功会给你带来巨大的信心。最容易踩的坑环境配置Python 版本、包版本冲突是新手第一道坎。务必使用虚拟环境。状态管理不理解State的累加operator.add和覆盖机制会导致数据丢失或混乱。多花时间理解TypedDict和Annotated的用法。图的设计急于实现复杂逻辑导致图结构混乱难以调试。坚持“简单优先逐步复杂”的原则。后续深入方向探索更复杂的图模式研究StateGraph中的分支Conditional Edge和循环Cycle实现更动态的工作流。集成更强大的工具将 Agent 与你的业务系统、数据库、API 连接起来解决实际问题。优化性能与成本实验不同的本地模型和量化方案在效果和资源消耗间找到最佳平衡点。研究多智能体Multi-Agent系统让多个具备不同角色的 Agent 协作完成一项任务这是 LangGraph 非常擅长的领域。这个由549集教程浓缩出的实战指南希望能成为你进入 AI Agent 开发世界的有效地图。真正的掌握来自于动手实践当你亲手调试通第一个属于自己的智能体工作流时所有的概念才会变得具体而深刻。建议收藏本文在开发过程中遇到问题时再回来查阅对应的章节。

读完文章,也想定制专属网站?

尧图设计师 24 小时内与您沟通定制方案

免费获取报价