ARTICLE DETAIL

资讯详情

深耕郑州网站建设与运营推广的一线实战洞察。

LangGraph + MCP 构建智能体:从人工智障到任务执行引擎的工程实践

LangGraph + MCP 构建智能体:从人工智障到任务执行引擎的工程实践 如果你正在尝试用 LangChain、LangGraph 和 MCP 来构建一个真正能用的 AI Agent大概率会遇到这几个问题代码跑通了但 Agent 像个“人工智障”只会来回问问题就是干不了实事或者好不容易接入了几个工具却发现上下文瞬间爆炸Agent 直接“失忆”又或者你看着网上各种“五分钟搭建 Agent”的教程依葫芦画瓢结果连最基本的工具调用都报错根本找不到原因。这背后的核心矛盾在于大多数教程只教了你“零件”怎么拼却没告诉你“系统”怎么设计。你学会了 LangChain 的链知道了 MCP 的协议也跑通了 LangGraph 的图但把它们组合成一个稳定、高效、可用的智能体时却处处是坑。Agent 开发不是简单的 API 调用它是一套涉及状态管理、工具调度、记忆保持和错误处理的系统工程。本文将以一个本地文件分析 Agent为例手把手带你从零搭建一个具备长期记忆、能自动调用工具、并能处理复杂工作流的智能体。我们将彻底拆解 LangChain MCP LangGraph 的组合拳重点不是复述文档而是揭示那些官方指南里没写、但实践中一定会遇到的“暗坑”。读完本文你将能清晰地掌握MCP 协议的核心价值它如何从根本上解决工具集成的混乱问题LangGraph 的状态图为什么说它是构建复杂 Agent 的“骨架”工程化实践如何设计提示词、管理上下文、处理错误让 Agent 真正可用我们直接从最棘手的部分开始。1. 为什么你的第一个 Agent 总是“人工智障”很多开发者的第一个 LangChain Agent 体验并不美好。你按照教程用initialize_agent快速搭了一个让它帮你查天气或者算数学。它可能成功了一两次但当你给它一个稍微复杂的任务比如“分析一下项目根目录下的README.md和requirements.txt然后给我总结依赖和项目目标”它就开始陷入循环要么不停地问你“要分析哪个文件”要么直接报错Agent stopped due to iteration limit or time limit。这通常不是代码 bug而是设计缺失。一个基础的、基于简单循环的 Agent 架构也是很多快速入门教程用的存在几个致命缺陷状态管理缺失Agent 不知道自己已经做了什么下一步该做什么。它就像一个有短期失忆症的人每个回合都从头开始思考。工具调用混乱当有多个工具时Agent 缺乏有效的策略来决定先用哪个后用哪个甚至会在几个工具间来回横跳。上下文窗口浪费每次交互都把完整的对话历史和工具输出塞进提示词几次来回后宝贵的上下文窗口就被无关信息占满导致模型“失忆”核心任务。错误处理真空工具调用失败后Agent 往往直接崩溃或陷入死循环没有恢复机制。LangGraph 和 MCP 正是为了解决这些系统性问题而出现的。LangGraph 提供了基于状态图的、有状态的、可循环的执行引擎让你能像设计工作流一样设计 Agent 的决策逻辑。MCPModel Context Protocol则定义了一套标准协议让任何工具都能以统一的方式被 AI 模型发现和调用彻底告别为每个工具写适配层的痛苦。在开始实战前我们必须统一认知我们现在要构建的不是一个“聊天机器人”而是一个“自动化的任务执行引擎”。它的核心是可靠地完成一个多步骤的目标。2. 核心概念拆解LangChain, MCP, LangGraph 各自扮演什么角色在组合使用这些技术前必须厘清它们的边界和职责。很多混淆都源于对它们定位的误解。组件核心职责类比解决了什么问题LangChain应用框架与集成层建筑的钢筋水泥和预制件提供了连接大模型、工具、记忆模块的基础组件和高级接口如 Chain, AgentExecutor让开发者能快速组装 AI 应用。它是生态的基石。MCP (Model Context Protocol)工具接入标准协议建筑的标准化电源插座和接口定义了一套工具如何向模型“自我介绍”名称、描述、参数以及模型如何调用工具的通用协议。任何符合 MCP 的服务都可以被无缝接入无需为每个工具定制开发。LangGraph有状态的工作流编排引擎建筑的智能中央控制系统和管线图用“图”Graph的概念来建模复杂的、多步骤的、有状态的 AI 工作流。节点是执行步骤调用 LLM、运行工具边是控制流根据条件跳转。它管理整个 Agent 的“状态”使其具备记忆和逻辑推进能力。它们之间的关系是你用LangChain提供的底层能力模型调用、文本处理来构建节点功能用MCP来接入和管理你的工具集最后用LangGraph作为顶层控制器将这些功能和工具按照你设计的业务流程图组装起来并管理整个执行过程的状态。一个常见的误解是LangChain 的AgentExecutor和 LangGraph 是二选一的关系。实际上AgentExecutor是一个简单的、内置的 Agent 运行器适合快速验证和简单场景。而LangGraph 是更强大、更灵活、更适合生产环境的替代方案你可以用它来构建比AgentExecutor复杂得多的逻辑。3. 环境准备别在依赖版本上栽跟头Agent 技术栈迭代极快版本不匹配是新手的第一大杀手。以下配置是经过验证的稳定组合能避免大多数兼容性问题。系统与 Python 环境操作系统macOS / Linux (推荐) 或 Windows (WSL2 环境下)Python 版本Python 3.10 或 3.11。强烈建议使用 3.10这是当前多数 AI 库兼容性最好的版本。避免使用 3.12 等过新版本。包管理工具使用uv或pip。uv速度更快依赖解析更优。本文示例使用pip。核心依赖安装创建一个新的虚拟环境并安装依赖。langchain-cli不是必须的但它提供了有用的项目脚手架工具。# 创建并激活虚拟环境 (以 conda 为例) conda create -n langgraph-agent python3.10 conda activate langgraph-agent # 安装核心库 pip install langchain langgraph langchain-cli # 安装 OpenAI 模型接口 (我们将使用 GPT-4 作为推理核心) pip install langchain-openai # 安装 MCP 相关库。注意MCP 生态正在快速发展这里安装官方客户端和基础工具。 pip install mcp langchain-mcp-tools # 可选但推荐安装本地模型接口如 Ollama用于离线测试或降本 # pip install langchain-ollama关键版本检查2026年初参考运行pip list | grep -E langchain|langgraph|mcp确保你安装的是较新且兼容的版本。例如langchain 0.2.0langgraph 0.0.50langchain-mcp-tools 0.0.3获取 API 密钥本文使用 OpenAI GPT-4 作为核心 LLM。你需要准备一个OPENAI_API_KEY。# 在 Linux/macOS 的 shell 配置文件中设置或在运行时设置环境变量 export OPENAI_API_KEY你的-api-key # Windows (PowerShell) $env:OPENAI_API_KEY你的-api-key环境准备好后我们首先来理解最关键的架构部分LangGraph 的状态图。4. LangGraph 核心用状态图StateGraph重塑 Agent 思维LangGraph 的核心抽象是StateGraph。它不像传统代码那样是线性的而是定义了一组节点Nodes和连接它们的边Edges。节点执行具体操作如调用LLM、运行工具边决定下一步走哪条路。整个图共享一个状态State对象所有节点都可以读写这个状态。为什么这比传统 Agent 循环强大显式状态管理所有中间结果、历史记录、决策依据都放在一个明确定义的状态对象里清晰可控。复杂逻辑编排你可以轻松实现“如果工具A成功则执行B否则执行C”、“循环执行直到条件满足”等逻辑。模块化与调试每个节点功能独立你可以单独测试节点也更容易定位问题发生在哪个环节。让我们为“文件分析 Agent”设计一个简单的状态图。这个 Agent 的目标是接收用户一个关于文件系统的自然语言指令如“总结src目录下所有 Python 文件的主要函数”然后自动调用文件读写工具来完成。首先定义我们的状态。这是一个TypedDict规定了我们的 Agent 在运行过程中需要记住哪些信息。# 文件agent_state.py from typing import TypedDict, List, Annotated import operator from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class AgentState(TypedDict): # 消息历史记录用户、AI、系统的所有对话 messages: Annotated[List[BaseMessage], add_messages] # 用户最新的原始输入 user_input: str # 模型生成的下一步动作如调用什么工具 next_action: str # 工具调用的结果列表 tool_outputs: List[str] # 最终给用户的答案 final_answer: str关键点Annotated[List[BaseMessage], add_messages]这是 LangGraph 的魔法。add_messages是一个归约器reducer它会自动将新消息追加到messages列表而不是覆盖。这完美解决了消息历史管理的问题。其他字段如next_action,tool_outputs是自定义的用于在节点间传递特定数据。接下来我们创建图并定义节点。想象一下我们的 Agent 工作流理解用户意图orchestrator节点分析用户输入决定是否需要调用工具以及调用哪个。执行工具调用call_tool节点如果决定调用工具就在这里执行。生成最终回答final_answer节点根据工具结果和对话历史生成给用户的回复。# 文件agent_graph.py (第一部分) from langgraph.graph import StateGraph, END from .agent_state import AgentState from langchain_openai import ChatOpenAI # 初始化大模型 llm ChatOpenAI(modelgpt-4-turbo-preview, temperature0) # 创建图构建器 workflow StateGraph(AgentState) # 定义节点 1: 协调器理解意图决定行动 def orchestrator_node(state: AgentState): from langchain_core.messages import HumanMessage, SystemMessage # 构建系统提示词指导模型如何决策 system_prompt 你是一个文件系统分析助手。你可以调用工具来读取文件、列出目录。 用户会给你一个任务。你需要决定 1. 如果任务需要读取文件或目录信息且你手头没有这些信息则调用相应工具。 2. 如果你已经拥有完成任务所需的信息或者用户只是在闲聊则直接给出回答。 请用以下格式回应 行动: [CALL_TOOL 或 RESPOND] 工具名: [如果行动是 CALL_TOOL填写工具名否则留空] 工具输入: [如果行动是 CALL_TOOL填写JSON格式的输入参数否则留空] messages [ SystemMessage(contentsystem_prompt), *state[messages] # 包含历史对话 ] # 调用模型进行决策 response llm.invoke(messages) # 解析模型的响应这里简化处理实际需要更健壮的解析 content response.content if CALL_TOOL in content: # 解析出工具名和输入这里需要更复杂的解析逻辑例如使用Pydantic # 为简化示例我们假设调用一个叫 list_directory 的工具 return {next_action: CALL_TOOL:list_directory, tool_outputs: []} else: # 直接响应跳转到最终答案节点 return {next_action: RESPOND, final_answer: content} # 将函数注册为节点 workflow.add_node(orchestrator, orchestrator_node)这只是第一个节点。我们已经可以看到orchestrator_node接收整个state根据其中的对话历史state[messages]和系统提示让 LLM 做出决策并更新state中的next_action字段。这个决策逻辑是否调用工具是 Agent 智能的核心。5. 接入 MCP 工具告别“适配器地狱”传统方式为每个工具写 LangChain Tool 封装很繁琐。MCP 协议的魅力在于任何实现了 MCP 服务器的工具都可以被 LangChain 通过langchain-mcp-tools自动识别和加载。假设我们有一个本地的“文件系统 MCP 服务器”。实际上你可以很容易地用mcp库创建自己的服务器或者使用社区已有的。这里我们模拟两个基础工具list_directory列出目录和read_file读取文件。首先看看如何用 LangChain 加载 MCP 工具# 文件mcp_tools.py import asyncio from typing import List from langchain_mcp_tools import MCPClientSession, MCPServer # 示例模拟一个简单的文件系统 MCP 服务器 # 在实际项目中你可能会连接到一个独立的 MCP 服务器进程 class SimpleFileSystemServer(MCPServer): async def list_tools(self) - List[dict]: # 向客户端宣告本服务器提供的工具 return [ { name: list_directory, description: 列出指定目录下的文件和子目录, inputSchema: { type: object, properties: { path: {type: string, description: 目录路径} }, required: [path] } }, { name: read_file, description: 读取指定文件的内容, inputSchema: { type: object, properties: { path: {type: string, description: 文件路径} }, required: [path] } } ] async def call_tool(self, tool_name: str, arguments: dict) - str: # 实际执行工具调用 import os if tool_name list_directory: path arguments.get(path, .) try: items os.listdir(path) return f目录 {path} 下的内容\n \n.join(items) except FileNotFoundError: return f错误路径 {path} 不存在。 except PermissionError: return f错误没有权限访问路径 {path}。 elif tool_name read_file: path arguments.get(path) try: with open(path, r, encodingutf-8) as f: content f.read(2000) # 限制读取长度避免上下文爆炸 return f文件 {path} 的内容 (前2000字符)\n\n{content}\n except FileNotFoundError: return f错误文件 {path} 不存在。 except Exception as e: return f读取文件时出错{str(e)} else: return f错误未知工具 {tool_name}。 # 创建客户端会话并加载工具 async def get_mcp_tools(): server SimpleFileSystemServer() # 在实际中MCPClientSession 会通过进程间通信(IPC)或SSE连接服务器 # 这里我们直接使用模拟的服务器对象 session MCPClientSession(server) tools await session.get_tools() return tools, session # 同步函数包装方便在普通函数中调用 def load_tools_sync(): return asyncio.run(get_mcp_tools())关键优势自动工具发现session.get_tools()会自动获取服务器提供的所有工具及其完整的描述和参数模式。标准化调用无论工具本身是 Python 函数、命令行程序还是远程 API通过 MCP 调用方式都是统一的。无缝集成获取到的tools列表可以直接传递给 LangChain 的 Agent 或 LangGraph 节点使用。现在我们修改之前的orchestrator_node和创建call_tool_node让它们能真正使用这些 MCP 工具。# 文件agent_graph.py (第二部分续接之前代码) from .mcp_tools import load_tools_sync from langchain_core.tools import BaseTool # 在文件顶部或合适的位置加载工具 mcp_tools_list, mcp_session load_tools_sync() # 将 MCP 工具转换为 LangChain Tool 对象langchain-mcp-tools 可能已处理好 # 假设 mcp_tools_list 已经是 BaseTool 列表 tools: List[BaseTool] mcp_tools_list # 创建一个工具名到工具对象的映射方便调用 tools_by_name {tool.name: tool for tool in tools} def call_tool_node(state: AgentState): 执行工具调用的节点 # 从前一个节点orchestrator的设置中解析出要调用的工具和参数 # 这里是一个简化示例。在实际中orchestrator 节点应该输出结构化的动作对象。 action_str state.get(next_action, ) tool_outputs state.get(tool_outputs, []) if action_str.startswith(CALL_TOOL:): # 解析工具名这里简化处理实际应从LLM的结构化输出中解析 tool_name action_str.split(:)[1] # 解析参数这里需要更复杂的逻辑例如从state中获取或让LLM输出 # 假设参数是固定的或从用户输入中提取这是一个需要深入设计的部分。 # 例如我们可以简单地从最新的用户消息中猜测路径 last_user_msg state[messages][-1].content if state[messages] else # 非常简单的启发式规则寻找类似路径的单词 import re potential_paths re.findall(r[\w\/\.\-], last_user_msg) tool_args {path: potential_paths[0] if potential_paths else .} # 获取工具并调用 tool tools_by_name.get(tool_name) if tool: try: # 同步调用工具。如果是异步工具需要适配。 output tool.invoke(tool_args) tool_outputs.append(f工具 {tool_name} 调用结果{output}) except Exception as e: output f调用工具 {tool_name} 时出错{str(e)} tool_outputs.append(output) else: output f错误未找到名为 {tool_name} 的工具。 tool_outputs.append(output) # 更新状态 return {tool_outputs: tool_outputs, next_action: TOOL_EXECUTED} # 如果没有工具调用直接返回原状态 return state # 注册工具调用节点 workflow.add_node(call_tool, call_tool_node)这个call_tool_node展示了如何从状态中提取决策、查找工具、执行调用并记录结果。这里的参数解析tool_args是极其简化的也是实际项目中最容易出错的地方之一。生产环境中你需要让 LLM 输出严格的 JSON 参数并使用 Pydantic 模型进行验证。6. 组装完整工作流定义边与循环有了节点我们需要定义它们之间的流转关系即边Edges。LangGraph 支持条件边让工作流具备判断能力。# 文件agent_graph.py (第三部分完成图构建) def should_continue(state: AgentState) - str: 判断在 orchestration 节点之后下一步应该去哪。 # 根据 orchestator 节点设置的 next_action 决定 action state.get(next_action, ) if action.startswith(CALL_TOOL): # 需要调用工具前往 call_tool 节点 return call_tool elif action RESPOND: # 直接生成回答前往 final_answer 节点 return final_answer elif action TOOL_EXECUTED: # 工具执行完毕需要再次让 orchestrator 判断后续动作例如是否需要继续调用其他工具 return orchestrator else: # 默认情况结束 return END def final_answer_node(state: AgentState): 生成最终回答的节点。这里可以整合所有工具结果和历史让LLM生成友好回复。 # 收集所有信息 history \n.join([msg.content for msg in state[messages] if hasattr(msg, content)]) tool_results \n.join(state.get(tool_outputs, [])) user_query state.get(user_input, ) prompt f 基于以下对话历史和工具执行结果请对用户的问题给出最终、完整的回答。 用户问题{user_query} 对话历史 {history} 工具执行结果 {tool_results} 请给出清晰、有帮助的最终答案 response llm.invoke(prompt) return {final_answer: response.content, messages: state[messages] [response]} # 注册最终答案节点 workflow.add_node(final_answer, final_answer_node) # 设置图的入口点 workflow.set_entry_point(orchestrator) # 添加条件边 workflow.add_conditional_edges( orchestrator, # 源节点 should_continue, # 判断函数返回下一个节点的名字 { call_tool: call_tool, # 如果返回”call_tool“则跳转到 call_tool 节点 final_answer: final_answer, # 如果返回”final_answer“则跳转到 final_answer 节点 END: END # 如果返回 END则结束 } ) # 添加固定边 workflow.add_edge(call_tool, orchestrator) # 工具调用后总是回到协调器重新判断 workflow.add_edge(final_answer, END) # 生成最终答案后图执行结束 # 编译图得到可执行的对象 app workflow.compile()图逻辑解读图从orchestrator节点开始。orchestrator节点运行后由should_continue函数根据其输出的next_action决定下一步。如果是CALL_TOOL前往call_tool节点。如果是RESPOND前往final_answer节点。call_tool节点执行完毕后固定地返回orchestrator节点add_edge。这让协调器可以基于工具执行结果决定下一步是继续调用工具还是生成回答。这就形成了一个循环直到任务完成。final_answer节点运行后图执行结束。这个设计实现了基本的“思考-行动-观察”循环是 Agent 的核心模式。现在我们的 Agent 骨架已经完整了。7. 运行与调试观察你的 Agent 如何思考编译好的app就是一个可执行的智能体。我们创建一个主函数来运行它。# 文件main.py from agent_graph import app from langchain_core.messages import HumanMessage from .agent_state import AgentState def run_agent(query: str): # 初始化状态 initial_state: AgentState { messages: [HumanMessage(contentquery)], user_input: query, next_action: , tool_outputs: [], final_answer: } print(f用户提问{query}) print(*50) # 执行图。stream 方法可以让我们看到每一步的中间状态便于调试。 final_state None for step, output in app.stream(initial_state, stream_modevalues): node_name list(step.keys())[0] # 获取当前执行的节点名 print(f[步骤] 执行节点{node_name}) if node_name orchestrator: print(f 决策结果{output.get(next_action)}) elif node_name call_tool: last_output output.get(tool_outputs, [])[-1] # 避免打印过长的文件内容 if len(last_output) 300: print(f 工具调用结果{last_output[:300]}...) else: print(f 工具调用结果{last_output}) elif node_name final_answer: print(f 生成最终答案{output.get(final_answer, )[:200]}...) print(-*30) final_state output print(*50) print(执行结束。) if final_state: print(f最终答案\n{final_state.get(final_answer)}) if __name__ __main__: # 测试几个查询 test_queries [ 列出当前目录下有什么文件, 请读取 README.md 文件的内容。, 帮我总结一下 src 目录下 main.py 文件是做什么的。 ] for q in test_queries: run_agent(q) input(\n按回车键测试下一个查询...)运行python main.py你将在控制台看到 Agent 的完整思考和执行过程。这是调试和理解 Agent 行为最有效的方式。你会看到它在orchestrator、call_tool和final_answer节点之间流转并打印出每个节点的输入输出。预期你会遇到的挑战与现象工具参数解析错误LLM 输出的next_action可能不是标准格式导致call_tool_node解析失败。这是提示词工程和输出解析需要优化的地方。无限循环如果orchestrator在工具调用后依然判断需要调用工具而条件没有改变就会死循环。需要在状态中增加“最大工具调用次数”或更复杂的终止逻辑。上下文过长如果工具返回的内容如大文件全部塞入历史很快就会超出模型上下文。需要在final_answer_node或状态管理中引入摘要或选择性记忆。8. 进阶实战解决“上下文爆炸”与“长期记忆”问题一个实用的 Agent 必须能处理长对话和大量工具输出。直接堆砌所有历史到提示词是不可行的。LangGraph 的add_messages归约器帮我们管理了列表但我们需要智能地压缩或筛选历史。方案一在状态中维护摘要修改AgentState增加一个conversation_summary字段。在orchestrator_node或一个专门的summarize_node中定期将冗长的对话历史压缩成一个摘要。# 修改后的 agent_state.py class AgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] user_input: str next_action: str tool_outputs: List[str] final_answer: str # 新增对话摘要用于在长对话中替代完整历史 conversation_summary: str # 新增一个摘要节点 def summarize_node(state: AgentState): 当历史消息过长时生成摘要 if len(state[messages]) 10: # 设定一个阈值 history_text \n.join([msg.content for msg in state[messages][:-3]]) # 摘要旧消息 summary_prompt f请将以下对话历史压缩成一个简洁的摘要保留关键事实和用户意图\n{history_text} summary llm.invoke(summary_prompt).content # 用摘要替换旧消息这里简化处理实际可能保留最近几条完整消息 # 一种策略是state[conversation_summary] “\n” summary # 然后清空或截断 state[messages] return {conversation_summary: summary, messages: state[messages][-3:]} # 只保留最近3条 return state方案二使用 LangGraph 的持久化检查点Checkpoint这是更强大、更生产级的方案。Checkpoint 允许你将图的状态包括所有变量持久化到数据库如 SQLite, Postgres并为每个会话thread_id保存。这样你可以处理远超模型上下文窗口的极长工作流并在应用重启后恢复状态。# 使用内存存储的检查点示例 from langgraph.checkpoint import MemorySaver memory MemorySaver() # 在编译图时传入存储 app workflow.compile(checkpointermemory) # 运行图时指定 thread_id config {configurable: {thread_id: user_123_session_1}} initial_state {...} # 使用 stream 或 invoke 时传入 config状态会自动保存和加载 for step in app.stream(initial_state, configconfig, stream_modevalues): ... # 下次可以从上次中断的地方继续 app.get_state(config) # 获取上次的状态对于文件分析 Agent如果用户要求分析一个包含几十个文件的目录Checkpoint 机制可以确保任务不会因为进程重启或意外中断而丢失。9. 生产环境最佳实践与避坑指南将上述示例转化为一个稳定、可维护的生产系统需要注意以下关键点1. 提示词工程是核心系统提示词System Prompt必须清晰定义 Agent 的角色、能力边界、工具使用规范和输出格式。示例中的提示词非常简陋需要细化。结构化输出Structured Output强制 LLM 以 JSON 等固定格式输出next_action使用 LangChain 的PydanticOutputParser或StructuredOutputParser来解析这是避免解析错误的最有效方法。少样本示例Few-Shot在提示词中提供几个“用户输入 - 正确行动”的示例能极大提升模型决策的准确性。2. 健壮的错误处理工具调用容错call_tool_node必须有完善的 try-catch并将错误信息以模型能理解的方式返回状态让协调器能采取补救措施如重试、换工具、向用户求助。图执行超时与循环限制在app.compile()或运行时配置中设置max_turns或timeout防止恶意或错误输入导致无限循环。状态验证对AgentState的字段进行类型和有效性验证避免脏数据导致图执行崩溃。3. 可观测性与监控全面日志记录记录每个节点的输入/输出、工具调用详情、LLM 的请求和响应。这不仅是调试的需要也是分析 Agent 表现、优化提示词的依据。链路追踪Tracing使用 LangSmith 或 OpenTelemetry 集成可视化整个工作流的执行路径、耗时和成本这是生产部署的标配。关键指标监控工具调用成功率、平均完成步数、用户满意度等。4. 安全与权限工具沙箱像文件读写、网络访问这类高风险工具必须在严格的沙箱环境中运行限制其可访问的路径和资源。用户输入净化对用户输入中可能用于路径遍历../或命令注入的字符进行过滤和校验。权限分级为不同的用户或会话配置不同的工具访问权限。5. 性能优化异步执行将llm.invoke和tool.invoke改为异步版本ainvoke并使用async节点函数可以大幅提升高并发下的吞吐量。缓存对昂贵的 LLM 调用或工具查询结果进行缓存尤其是那些频繁且结果不变的操作。流式响应对于生成最终答案等耗时步骤使用流式输出Streaming提升用户体验。构建 AI Agent 不是一个一蹴而就的过程而是一个需要持续迭代的工程。从本文这个具备基本“思考-行动”循环的文件分析 Agent 出发你可以逐步接入更复杂的工具数据库、API、代码解释器设计更精细的状态和决策逻辑最终打造出能够解决实际业务问题的智能助手。记住清晰的架构LangGraph和标准的工具协议MCP是应对复杂性的基石而深入的提示词工程和健壮的异常处理则是系统稳定性的保障。
返回列表