恒美微站
首页
关于我们
建站服务
主题模板
案例展示
资讯中心
联系我们
LangChain与LangGraph实战:构建具备RAG与工具调用能力的AI智能体
首页
资讯中心
/
LangChain与LangGraph实战:构建具备RAG与工具调用能力的AI智能体
LangChain与LangGraph实战:构建具备RAG与工具调用能力的AI智能体
发布时间:2026/8/24 11:32:22
如果你正在学习 LangChain 和 LangGraph并且已经看过了不少教程却依然感觉“学了很多但不知道如何整合成一个能实际工作的项目”那么这篇文章就是为你准备的。很多教程会教你如何调用一个 API或者搭建一个简单的问答机器人。但当你真正想构建一个具备复杂逻辑、能调用工具、能管理状态、甚至能协作的 AI 智能体Agent时你会发现知识点是零散的LangChain 的链Chain怎么用LangGraph 的图Graph又是什么Agent 和 RAG 如何结合MCP 协议听起来很酷但怎么接入你需要的不是一个个孤立的 API 演示而是一条清晰的、从零到一构建一个“可用、可扩展、可理解”的智能体应用的完整路径。本文将基于一套被验证过的实战课程内容为你拆解 LangChain LangGraph 的核心开发逻辑。我们不会停留在概念复述而是聚焦于解决一个真实问题如何用这些框架构建一个具备长期记忆、能使用工具、并能通过 RAG 查询私有知识的智能体系统。文章会包含完整的环境搭建、核心概念对比、代码实现、运行验证以及你一定会遇到的“坑”和解决方案。读完本文你将能清晰地理解 LangChain 和 LangGraph 的分工并拥有一个可以立刻运行和修改的智能体项目原型。1. 这篇文章真正要解决的问题从“知道”到“做到”的鸿沟学习 AI 应用开发框架时最大的障碍往往不是某个 API 的调用方式而是如何将多个组件有机地组合起来形成一个稳定、可维护的应用架构。LangChain 提供了丰富的“积木”组件如模型封装、提示词模板、记忆模块、检索器等。LangGraph 则提供了组装这些“积木”的“图纸”和“流水线”基于图的工作流。但仅仅拥有积木和图纸离建成一座房子还有很远。具体来说开发者常面临以下痛点概念混淆LangChain 的 Agent 和 LangGraph 的 Graph 都能实现多步骤推理它们到底有什么区别我该用哪个状态管理混乱智能体在对话中需要记住历史、维护工具调用结果等状态。如何以清晰、可序列化的方式管理这些状态工具集成不顺畅如何让智能体不仅能聊天还能真正地执行代码、查询数据库、调用外部 APIRAG 与 Agent 结合生硬RAG检索增强生成负责知识查询Agent 负责决策和工具调用。如何让它们协同工作而不是简单拼接缺乏可观测性智能体的决策过程像个黑盒出错时难以调试。本文的目标就是架起这座从“知道”到“做到”的桥梁。我们将通过构建一个“研究助手”智能体作为主线项目它需要完成理解用户复杂的研究问题、利用 RAG 从本地知识库检索资料、根据资料决定是否需要调用网络搜索工具、整合信息并生成报告。这个过程将串联起 LangChain 的核心组件和 LangGraph 的图编排能力。2. 基础概念与核心原理分工明确的“积木”与“图纸”在开始写代码前必须厘清几个核心概念的关系这是避免后续架构混乱的关键。LangChain标准化的“积木工厂”你可以把 LangChain 理解为一个提供了大量预制件、接口标准和通用胶水的工具箱。它的核心价值在于标准化为不同的 LLMOpenAI, Anthropic, 本地模型等提供了统一的调用接口。组件化将 AI 应用常见功能如提示词模板、记忆、检索、输出解析封装成可复用的组件。链Chain这是 LangChain 早期的主要编排方式将多个组件按顺序组合成一个线性的执行流程。它简单易用但对于复杂、有分支、有循环的逻辑显得力不从心。LangGraph基于状态的“流程图纸”LangGraph 是 LangChain 生态系统内一个更强大的编排框架。它用“图Graph”来定义应用的工作流。节点Node代表一个执行单元可以是一个工具调用、一个条件判断、或调用一个 LangChain Chain。边Edge定义了节点之间的流转条件。可以是无条件跳转也可以根据上一个节点的输出结果决定下一步去哪里。状态State这是 LangGraph 的灵魂。整个图共享一个状态字典State Schema每个节点读取和修改这个状态。这使得管理对话历史、工具结果、中间变量变得非常清晰和自然。它擅长处理多轮对话、循环如“思考-行动-观察”循环、条件分支、并行执行等复杂逻辑。Agent具备推理和行动能力的智能体在 LangChain/LangGraph 语境下Agent 通常指一个由 LLM 驱动、能够根据目标自主决定调用哪些工具Tools的系统。LangGraph 是构建复杂、稳定 Agent 的推荐方式因为它能更好地管理 Agent 的思考循环和内部状态。RAG为模型注入外部知识RAG 系统通过将外部文档切片、向量化并存储到向量数据库中在回答问题时先检索相关文档片段再将其作为上下文提供给 LLM从而生成更准确、信息量更丰富的答案。在智能体系统中RAG 可以作为一个强大的“知识工具”被集成。MCP智能体的“工具扩展协议”Model Context Protocol 是一个新兴的开放协议它允许任何应用程序如 IDE、设计工具、数据库客户端以标准化的方式向 AI 智能体暴露其功能和数据。你可以把它想象成智能体的“USB 接口”标准。通过 MCP 服务器智能体可以直接操作 Figma 文件、查询数据库、读写本地文件等极大地扩展了能力边界。它们之间的关系可以概括为用 LangChain 的标准组件模型、提示词、检索器作为“材料”用 LangGraph 的图来设计和组装实现复杂逻辑的“智能体Agent”其中 RAG 和 MCP 工具是增强智能体能力的核心手段。3. 环境准备与前置条件我们将使用 Python 作为开发语言。请确保你的环境满足以下要求。3.1 基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文命令以 Linux/macOS 为例Windows 用户可在 Git Bash 或 WSL 中运行。Python 版本3.10 或 3.11。3.12 可能存在部分依赖兼容性问题建议暂用 3.11。包管理工具使用pip或poetry。本文使用pip。代码编辑器VS Code 或 PyCharm。3.2 创建虚拟环境强烈推荐隔离项目依赖避免版本冲突。# 创建项目目录并进入 mkdir research_assistant_agent cd research_assistant_agent # 创建虚拟环境使用 venv python -m venv venv # 激活虚拟环境 # Linux/macOS source venv/bin/activate # Windows # venv\Scripts\activate激活后命令行提示符前应显示(venv)。3.3 安装核心依赖我们将安装 LangChain、LangGraph 以及用于向量数据库、网络搜索等功能的包。pip install langchain langgraph langchain-openai langchain-community tiktokenlangchain: LangChain 核心框架。langgraph: LangGraph 图编排框架。langchain-openai: OpenAI 模型的官方 LangChain 集成。langchain-community: 社区维护的大量第三方集成工具、检索器等。tiktoken: OpenAI 用于计算 Token 的库。3.4 安装向量数据库与嵌入模型为了构建 RAG 系统我们需要一个向量数据库和嵌入模型。这里使用轻量级的Chroma和开源的sentence-transformers嵌入模型。pip install chromadb sentence-transformerschromadb: 一个轻量级、易于使用的向量数据库。sentence-transformers: 提供高质量的本地文本嵌入模型无需 API 密钥。3.5 可选安装网络搜索工具为了让智能体能获取最新信息我们集成一个网络搜索工具。DuckDuckGo是一个不错的选择。pip install duckduckgo-search3.6 设置 API 密钥如使用 OpenAI如果你计划使用 OpenAI 的 GPT 模型需要设置 API 密钥。我们将其存储在环境变量中。# Linux/macOS export OPENAI_API_KEYyour-api-key-here # Windows (PowerShell) # $env:OPENAI_API_KEYyour-api-key-here重要请勿将 API 密钥硬编码在代码中。对于生产环境应使用.env文件配合python-dotenv等库管理。至此开发环境已准备就绪。4. 核心流程拆解构建“研究助手”智能体的四步走我们的目标是构建一个能理解复杂问题、检索本地知识、决定是否搜索网络、并生成报告的智能体。其核心工作流如下图所示概念图用户提问 | v [理解与规划节点] - 分析问题决定是否需要网络搜索 | v {条件判断} --需要搜索-- [网络搜索节点] | | |--不需要搜索 v | [整合搜索结果到状态] | | v v [RAG检索节点] ----------- | v [生成报告节点] | v 最终答案下面我们分步拆解实现这个工作流的关键环节。4.1 第一步定义智能体的“大脑”和“记忆”智能体的“大脑”是 LLM我们为其定义系统提示词明确其角色和能力。“记忆”则是 LangGraph 的状态我们需要设计一个状态结构来承载整个对话和推理过程的数据。4.2 第二步打造智能体的“工具包”智能体需要工具来与世界交互。我们将创建三个核心工具retrieve_from_knowledge_base: RAG 检索工具从本地向量数据库查询相关资料。search_web: 网络搜索工具获取最新信息。generate_final_answer: 报告生成工具也可以是一个节点整合所有信息。4.3 第三步用 LangGraph 绘制“工作蓝图”这是最核心的一步。我们将把上述工具和 LLM 决策点组织成一个有向图。图包含不同类型的节点工具节点执行具体的工具函数。路由节点由 LLM 驱动根据当前状态决定下一步是调用工具还是结束。条件边根据路由节点的输出决定流程走向。4.4 第四步运行与迭代将图实例化并运行观察智能体的推理过程根据结果调整提示词、工具逻辑或图结构。5. 完整示例与代码实现让我们开始编写代码。项目结构如下research_assistant_agent/ ├── knowledge_base/ # 存放原始文档 ├── data/ # 存放向量数据库 ├── tools.py # 工具函数定义 ├── graph.py # LangGraph 图定义 ├── main.py # 主运行入口 └── requirements.txt # 依赖列表5.1 构建本地知识库RAG 准备首先我们准备一些本地文档并存入向量数据库。# 文件prepare_knowledge_base.py import os from langchain_community.document_loaders import TextLoader from langchain_text_splitters import RecursiveCharacterTextSplitter from langchain_community.embeddings import HuggingFaceEmbeddings from langchain_community.vectorstores import Chroma # 1. 准备文档示例创建一个简单的文档 doc_content LangGraph 是 LangChain 的一个库用于构建由大语言模型驱动的有状态多智能体应用。 它使用图Graph来定义应用的工作流其中节点代表功能单元边代表控制流。 状态State是 LangGraph 的核心概念它是一个共享字典节点可以读取和修改它。 Agent 通常指能够自主规划并执行任务通过调用工具的 AI 系统。 RAG (Retrieval-Augmented Generation) 通过检索外部知识来增强大语言模型的生成能力。 with open(./knowledge_base/langgraph_intro.txt, w, encodingutf-8) as f: f.write(doc_content) # 2. 加载和分割文档 loader TextLoader(./knowledge_base/langgraph_intro.txt, encodingutf-8) documents loader.load() text_splitter RecursiveCharacterTextSplitter(chunk_size500, chunk_overlap50) splits text_splitter.split_documents(documents) print(f文档被分割成 {len(splits)} 个片段) # 3. 创建向量存储使用本地嵌入模型 embeddings HuggingFaceEmbeddings(model_nameall-MiniLM-L6-v2) # 轻量级句子嵌入模型 vectorstore Chroma.from_documents( documentssplits, embeddingembeddings, persist_directory./data/chroma_db # 向量数据库持久化目录 ) print(知识库向量化完成已保存至 ./data/chroma_db)运行此脚本以创建初始知识库python prepare_knowledge_base.py5.2 定义工具接下来在tools.py中定义智能体可用的工具。# 文件tools.py from langchain_community.tools import DuckDuckGoSearchRun from langchain_community.vectorstores import Chroma from langchain_community.embeddings import HuggingFaceEmbeddings from typing import List, Dict, Any # 初始化网络搜索工具 search DuckDuckGoSearchRun() # 初始化 RAG 检索器向量数据库 def get_retriever(): embeddings HuggingFaceEmbeddings(model_nameall-MiniLM-L6-v2) vectorstore Chroma( persist_directory./data/chroma_db, embedding_functionembeddings ) # 返回一个检索器搜索最相关的3个片段 return vectorstore.as_retriever(search_kwargs{k: 3}) retriever get_retriever() def retrieve_from_knowledge_base(query: str) - str: 从本地知识库检索相关文档。 参数: query: 用户查询字符串 返回: 检索到的相关文本拼接成一个字符串 docs retriever.invoke(query) content \n\n---\n\n.join([doc.page_content for doc in docs]) if not content: content 在知识库中未找到相关信息。 return f【来自知识库的参考资料】\n{content} def search_web(query: str) - str: 使用 DuckDuckGo 进行网络搜索。 参数: query: 搜索查询字符串 返回: 搜索结果的摘要文本 try: result search.run(query) return f【来自网络的搜索结果】\n{result} except Exception as e: return f网络搜索失败: {e} # 将所有工具包装成 LangChain Tool 对象方便后续绑定 from langchain.tools import Tool knowledge_tool Tool( namequery_knowledge_base, funcretrieve_from_knowledge_base, description当用户的问题涉及 LangChain, LangGraph, Agent, RAG 等概念时使用此工具从本地知识库查找详细资料和定义。输入应为清晰的问题或关键词。 ) web_search_tool Tool( namesearch_web, funcsearch_web, description当用户的问题需要最新的信息、新闻、实时数据或知识库中没有涵盖的内容时使用此工具进行网络搜索。输入应为搜索关键词。 ) # 导出工具列表 TOOLS [knowledge_tool, web_search_tool]5.3 定义状态与构建 LangGraph这是核心部分我们在graph.py中定义智能体的状态和图结构。# 文件graph.py from typing import TypedDict, Annotated, List, Union from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from langchain_openai import ChatOpenAI from langchain.tools import BaseTool import operator # ---------- 1. 定义状态模式 ---------- class AgentState(TypedDict): 智能体的共享状态。 # 用户原始问题 question: str # 对话历史用于上下文 messages: Annotated[List[Union[str, dict]], operator.add] # 从知识库检索到的内容 knowledge: str # 网络搜索的结果 web_results: str # 由路由节点决定的下一个动作 next_action: str # ---------- 2. 初始化模型和工具执行器 ---------- # 使用 GPT-4 或 GPT-3.5 作为推理核心 llm ChatOpenAI(modelgpt-4, temperature0) # 从 tools.py 导入工具 from tools import TOOLS tool_executor ToolExecutor(TOOLS) # ---------- 3. 定义图节点函数 ---------- def route_question(state: AgentState) - AgentState: 路由节点分析问题决定下一步是检索知识库、搜索网络还是直接生成答案。 这个节点由 LLM 驱动。 from langchain_core.prompts import ChatPromptTemplate # 系统提示词定义智能体的角色和决策逻辑 route_prompt ChatPromptTemplate.from_messages([ (system, 你是一个研究助手AI。你的任务是分析用户的问题并决定如何获取信息来回答它。 请遵循以下规则 1. 如果问题明确关于 LangChain, LangGraph, Agent, RAG, MCP 等技术概念优先使用 query_knowledge_base 工具从本地知识库查找。 2. 如果问题涉及最新新闻、事件、实时数据或知识库中不可能有的信息则使用 search_web 工具。 3. 如果问题非常简单或已包含足够信息可以直接回答则设置 next_action 为 generate_answer。 请只输出一个单词use_knowledge, search_web, 或 generate_answer。 ), (human, 用户问题{question}) ]) # 调用 LLM 进行决策 route_chain route_prompt | llm response route_chain.invoke({question: state[question]}) decision response.content.strip().lower() # 将决策存入状态 state[next_action] decision # 也可以将 LLM 的思考过程存入 messages 以供后续参考 state[messages].append({role: assistant, content: f路由决策{decision}}) return state def retrieve_knowledge_node(state: AgentState) - AgentState: 执行知识库检索的节点 result tool_executor.invoke( ToolInvocation(toolquery_knowledge_base, tool_inputstate[question]) ) state[knowledge] result state[messages].append({role: tool, content: f知识库检索完成。, name: query_knowledge_base}) return state def search_web_node(state: AgentState) - AgentState: 执行网络搜索的节点 result tool_executor.invoke( ToolInvocation(toolsearch_web, tool_inputstate[question]) ) state[web_results] result state[messages].append({role: tool, content: f网络搜索完成。, name: search_web}) return state def generate_answer_node(state: AgentState) - AgentState: 最终答案生成节点。 整合知识库内容、网络搜索结果和对话历史生成最终回答。 from langchain_core.prompts import ChatPromptTemplate # 构建最终生成提示词 generate_prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的研究助手。请根据以下信息为用户的问题提供一个全面、准确、结构清晰的回答。 如果信息不足请诚实说明。尽量引用来源如‘根据知识库资料...’或‘网络搜索显示...’。 ), (human, 用户问题{question} 相关信息 - 本地知识库内容{knowledge} - 网络搜索结果{web_results} 请生成最终答案 ) ]) generate_chain generate_prompt | llm response generate_chain.invoke({ question: state[question], knowledge: state.get(knowledge, 无), web_results: state.get(web_results, 无) }) final_answer response.content state[messages].append({role: assistant, content: final_answer}) return state # ---------- 4. 构建图 ---------- def create_agent_graph(): 创建并返回编译好的 LangGraph 工作流 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(route, route_question) workflow.add_node(retrieve_knowledge, retrieve_knowledge_node) workflow.add_node(search_web, search_web_node) workflow.add_node(generate_answer, generate_answer_node) # 设置入口点 workflow.set_entry_point(route) # 添加条件边根据路由决策跳转到不同节点 def decide_next_step(state: AgentState) - str: 条件判断函数决定下一步去哪里 action state.get(next_action) if action use_knowledge: return retrieve_knowledge elif action search_web: return search_web elif action generate_answer: return generate_answer else: # 默认去生成答案 return generate_answer workflow.add_conditional_edges( route, decide_next_step, { retrieve_knowledge: retrieve_knowledge, search_web: search_web, generate_answer: generate_answer, } ) # 设置从工具节点到生成答案节点的边 workflow.add_edge(retrieve_knowledge, generate_answer) workflow.add_edge(search_web, generate_answer) # 设置终点 workflow.add_edge(generate_answer, END) # 编译图 return workflow.compile() # 创建图实例 agent_graph create_agent_graph()5.4 主运行入口最后我们创建main.py来运行这个智能体。# 文件main.py from graph import agent_graph, AgentState import asyncio async def run_agent(question: str): 运行研究助手智能体。 # 初始化状态 initial_state: AgentState { question: question, messages: [], knowledge: , web_results: , next_action: } print(f\n 用户提问{question}) print(*50) # 以流式方式运行图可以看到每一步的输出 async for event in agent_graph.astream_events(initial_state, versionv1): kind event[event] if kind on_chain_start: # 可以在这里记录链的开始 pass elif kind on_chain_end and event.get(name) route: # 路由决策完成 print(f 路由决策{event[data][output][next_action]}) elif kind on_tool_start: print(f 正在执行工具{event[name]}) elif kind on_tool_end: # 工具执行完成 pass elif kind on_chat_model_stream: # 流式输出模型生成的内容如果需要 pass # 获取最终状态 final_state await agent_graph.ainvoke(initial_state) print(\n *50) print( 最终答案) # 从 messages 中提取最后的助手消息 for msg in final_state[messages]: if isinstance(msg, dict) and msg.get(role) assistant and 路由决策 not in msg.get(content, ): print(msg[content]) break print(*50) # 可选打印完整的状态以供调试 # import json # print(json.dumps(final_state, indent2, ensure_asciiFalse)) if __name__ __main__: # 测试不同的问题 test_questions [ LangGraph 中的状态State是什么, 今天 OpenAI 有什么最新新闻, 请解释一下 RAG 的工作原理。, ] for q in test_questions: asyncio.run(run_agent(q)) input(\n按 Enter 键继续下一个问题...)6. 运行结果与效果验证现在让我们运行这个智能体看看它如何工作。6.1 运行程序在项目根目录下执行python main.py6.2 预期输出示例对于问题“LangGraph 中的状态State是什么”你可能会看到类似以下的输出 用户提问LangGraph 中的状态State是什么 路由决策use_knowledge 正在执行工具query_knowledge_base 最终答案 根据本地知识库的资料LangGraph 中的状态State是其核心概念之一。它是一个共享的字典dictionary结构在整个图Graph工作流执行期间持续存在并可以被各个节点Node读取和修改。 状态用于存储和管理智能体应用运行过程中的所有相关信息例如 - 用户的输入消息 - 对话的历史记录 - 工具Tool调用的结果 - 中间计算产生的变量 - 智能体的内部决策逻辑 通过这种集中式的状态管理LangGraph 能够优雅地处理多轮对话、复杂的循环逻辑如“思考-行动-观察”循环以及条件分支使得构建有状态的、复杂的多智能体应用变得更加清晰和易于维护。 对于问题“今天 OpenAI 有什么最新新闻”输出可能如下 用户提问今天 OpenAI 有什么最新新闻 路由决策search_web 正在执行工具search_web 最终答案 根据网络搜索结果以下是关于 OpenAI 的最新动态截至近期 1. **模型更新**OpenAI 可能发布了 GPT-4 的增量更新或新的文本/多模态模型。 2. **API 与定价**可能存在对 API 访问策略、速率限制或定价结构的调整。 3. **安全与研究**OpenAI 通常会发布关于 AI 安全、对齐技术或前沿研究的新论文或博客。 4. **产品与合作**可能与某些企业宣布了新的产品集成或战略合作。 注意网络搜索结果是实时获取的具体内容会根据搜索时的实际情况而变化。建议访问 OpenAI 官方博客或科技新闻网站获取最准确的信息。 6.3 如何验证成功路由正确性观察路由决策是否符合预期。概念性问题应触发use_knowledge实时性问题应触发search_web。工具执行看到正在执行工具xxx的日志且没有报错。答案质量最终答案应整合了正确的信息来源知识库或网络并且回答结构清晰。状态完整性你可以取消main.py末尾的注释打印出完整的final_state检查knowledge和web_results字段是否被正确填充。7. 常见问题与排查思路在构建和运行此类智能体时你可能会遇到以下问题问题现象可能原因排查方式解决方案导入错误No module named ‘langchain_xxx’依赖未正确安装或虚拟环境未激活。1. 运行pip list | grep langchain检查包是否存在。2. 确认命令行提示符前有(venv)。1. 激活虚拟环境。2. 使用pip install -r requirements.txt重新安装所有依赖。运行时报错OpenAI API相关错误API 密钥未设置或模型名称错误。1. 检查OPENAI_API_KEY环境变量。2. 检查graph.py中ChatOpenAI的model参数。1. 正确设置环境变量。2. 确认模型名称有效如gpt-3.5-turbo。可先使用gpt-3.5-turbo测试。知识库检索返回空结果1. 向量数据库路径错误。2. 文档未正确嵌入或分割。3. 查询与文档语义不匹配。1. 检查./data/chroma_db目录是否存在且包含文件。2. 运行prepare_knowledge_base.py查看分割日志。3. 尝试更具体或更通用的查询词。1. 重新运行知识库准备脚本。2. 调整文本分割器的chunk_size和chunk_overlap。3. 尝试不同的嵌入模型。网络搜索工具超时或无结果网络问题或duckduckgo-search库的临时故障。1. 检查网络连接。2. 在 Python 交互环境中直接调用search.run(“test”)测试。1. 添加超时和重试逻辑到search_web函数。2. 考虑使用备用搜索工具如SerpAPI需 API 密钥。LangGraph 图编译或运行错误1. 状态模式TypedDict定义错误。2. 节点函数返回值与状态模式不匹配。3. 边指向不存在的节点。1. 仔细检查AgentState每个字段的注解类型。2. 确保每个节点函数都返回一个完整的AgentState字典。3. 检查add_edge和add_conditional_edges中的节点名称拼写。1. 使用更简单的状态结构开始测试。2. 在节点函数内打印state以调试。3. 参考 LangGraph 官方文档和示例。智能体决策路由不符合预期系统提示词不够清晰或 LLM 理解有偏差。1. 打印出路由节点的完整提示词和 LLM 响应。2. 用更简单、更指令化的语言重写提示词。1. 在提示词中提供更具体的例子Few-shot。2. 调整 LLM 的temperature参数更低的值更确定。3. 考虑使用更强大的模型如 GPT-4进行路由决策。程序异步运行报错asyncio事件循环问题尤其在 Jupyter 或已有循环的环境中。查看完整的错误堆栈信息。将asyncio.run(run_agent(q))改为使用asyncio.get_event_loop().run_until_complete(run_agent(q))或确保在正确的异步上下文中运行。8. 最佳实践与工程建议当你掌握了基础构建方法后以下实践能让你的智能体更健壮、更易维护8.1 状态设计要精简且明确只存储必要的状态不是数据垃圾场。只存储对工作流决策和最终输出有直接影响的数据。使用清晰的类型提示TypedDict和Annotated能极大提高代码可读性和 IDE 支持度。考虑序列化如果你需要持久化或调试状态确保其中的数据类型如自定义对象是可 JSON 序列化的。8.2 提示词工程是核心系统提示词要具体明确角色、规则和输出格式。模糊的指令会导致不可预测的行为。为不同节点设计专用提示词路由节点、工具调用节点、答案生成节点的目标不同提示词应各司其职。加入示例Few-shot对于复杂决策在提示词中提供1-2个输入输出示例能显著提升 LLM 表现。8.3 工具设计要可靠做好错误处理工具函数内部应有try-except返回明确的错误信息避免整个图崩溃。工具描述要精准description字段是 LLM 选择工具的主要依据应清晰说明工具的用途、输入格式和适用场景。控制工具输出长度工具返回的内容可能很长需合理截断或总结后再放入状态避免超出模型上下文限制。8.4 图的复杂性与可读性平衡开始时保持简单先用线性或简单分支图验证核心逻辑再逐步增加复杂性。使用子图对于复杂功能模块如一个完整的审核流程可以将其封装成一个子图StateGraph然后在主图中引用提高复用性和可读性。利用interrupt实现人工审核在关键节点如执行高风险操作前设置中断等待人工确认后再继续增加安全性。8.5 可观测性与调试利用astream_events如示例所示它可以让你实时观察图的执行步骤是强大的调试工具。记录完整轨迹将每次运行的完整状态流State Stream记录到日志或数据库中便于事后分析和复现问题。为关键节点添加检查点可以在节点函数中验证输入数据的有效性。8.6 向生产环境迈进配置管理将模型参数、API 密钥、文件路径等抽取为配置文件如config.yaml或环境变量。依赖注入考虑使用依赖注入框架来管理工具、模型客户端等资源的生命周期提高可测试性。添加监控与告警对智能体的调用延迟、费用、错误率进行监控。版本化你的图当对图结构或提示词进行重大更新时应有版本控制和平滑升级方案。9. 总结与后续学习方向通过本文的实战演练我们完成了一个基于 LangChain 和 LangGraph 的、具备 RAG 和工具调用能力的“研究助手”智能体从零到一的构建。我们不仅串联了各个核心概念更重要的是你获得了如何将它们组织成一个可运行、可调试的系统的第一手经验。本文的核心收获在于明确了分工LangChain 是组件库LangGraph 是编排器。用 LangChain 的标准接口准备“材料”用 LangGraph 的图来设计“智能体”的复杂逻辑流。掌握了状态驱动开发AgentState是连接图中所有节点的纽带理解了如何设计状态、如何在节点间传递和修改状态就掌握了 LangGraph 的精髓。实践了工具集成将 RAG 检索器和网络搜索封装成标准工具并让 LLM 学会在合适的时机调用它们。构建了条件工作流通过路由节点和条件边实现了智能体根据问题类型自主选择执行路径的能力。接下来你可以从以下几个方向深化探索更复杂的图模式研究 LangGraph 官方文档中的Human-in-the-loop、Multi-Agent Collaboration、Hierarchical Graphs等高级模式。集成 MCP 服务器尝试将 Figma、数据库等外部工具通过 MCP 协议接入你的智能体使其能力突破“搜索”和“检索”实现真正的“操作”。优化 RAG 流水线尝试不同的文本分割策略、嵌入模型、重排序器Re-ranker和向量数据库提升知识检索的准确率。加入记忆管理为智能体引入更长期的记忆例如使用LangGraph的Checkpointer或将对话历史存储到外部数据库实现跨会话的记忆。进行系统评估设计测试用例从回答准确性、工具调用正确率、响应速度等维度评估你的智能体并持续迭代优化。构建 AI 智能体是一个迭代和工程化的过程。最好的学习方式就是动手改造这个示例增加新的工具、设计更复杂的决策逻辑、处理更棘手的边缘情况。当你开始为真实场景定制智能体时你会更深刻地体会到这些框架带来的抽象能力和工程效率。建议你将本文的代码作为起点不断实验和扩展。