LangChain、LangGraph与MCP实战：从零构建具备规划与执行能力的AI智能体

最近在尝试将大模型能力集成到业务系统中时你是否也遇到了这样的困境模型调用简单但想让AI“理解”业务逻辑、调用工具、记住对话历史并完成复杂任务却感觉无从下手网上资料要么是零散的API调用示例要么是过于学术化的概念讲解缺乏一个从零到一、能直接落地的完整路径。这正是AI Agent智能体框架要解决的核心问题。本文将围绕当前最热门的LangChain、LangGraph和MCP三大技术栈为你拆解一套可实操的Agent开发教程。无论你是想快速构建一个能自动分析数据并生成报告的智能助手还是希望设计一个多角色协作的复杂决策系统本文都将提供从环境搭建、核心概念到项目实战的完整指南。学完后你将能独立搭建具备规划、执行和记忆能力的AI智能体并将其应用到实际业务场景中。1. AI Agent与智能体框架核心概念在深入代码之前我们必须厘清几个关键概念。这能帮助你理解我们为什么要使用这些框架而不是直接调用大模型API。1.1 什么是AI Agent智能体你可以将AI Agent理解为一个“数字员工”。它不仅仅是一个问答机器而是一个具备一定自主性的系统。其核心能力包括感知Perception接收来自用户、环境或其他系统的输入如文本、文件、API数据。规划Planning根据目标和当前状态分解任务制定一系列行动步骤。行动Action执行具体的操作例如调用一个计算器函数、查询数据库、调用外部API。记忆Memory保留对话历史、工具调用结果和内部状态用于后续决策。一个简单的ChatGPT是反应式的而一个真正的Agent是主动且有状态的。例如当你对Agent说“帮我分析上季度的销售数据并总结成一份PPT大纲”一个合格的Agent会自行规划先调用工具获取数据再分析数据趋势最后按照特定格式生成大纲文本。1.2 为什么需要智能体框架LangChain, LangGraph, MCP直接使用大模型API构建复杂Agent如同用零件手搓汽车而框架提供了底盘、发动机和方向盘。LangChain组件化与标准化。它将构建Agent所需的常见模式抽象成可复用的“组件”如提示词模板、各种记忆存储、文档加载器、输出解析器以及大量的现成工具集成。它解决了“有什么”和“怎么连”的问题让你能像搭积木一样快速组装一个基础Agent。LangGraph复杂流程与状态管理。当Agent的任务涉及多步骤循环、条件分支、多个Agent协作时LangChain的简单链Chain就显得力不从心。LangGraph引入了**图Graph的概念允许你显式地定义Agent的工作流其中节点代表步骤如调用LLM、执行工具边代表步骤间的流转逻辑。它完美管理了Agent的状态State**在整个工作流中的传递和变更是构建复杂、可靠Agent系统的利器。MCP (Model Context Protocol)工具与上下文的安全扩展。这是由Anthropic提出的一种协议旨在安全、标准化地将外部数据和工具暴露给大模型。传统方式中工具函数直接写在Agent代码里耦合度高且不安全。MCP则允许你独立运行一个MCP Server专门管理数据源如数据库、文件系统和工具。Agent通过标准协议与Server通信无需了解底层细节。这带来了工具的可发现性、更好的安全边界和上下文的高效管理。简单比喻LangChain是工具箱和标准零件LangGraph是设计复杂机械蓝图的工具而MCP则是为这台机械建立安全、标准的外部原料供应管道。2. 环境准备与项目初始化我们从一个干净的Python环境开始确保你可以复现所有示例。2.1 基础环境配置推荐使用Python 3.10或以上版本以避免不必要的兼容性问题。同时我们需要一个能访问的大语言模型API。本文示例将使用OpenAI的GPT模型你也可以替换为其他兼容OpenAI API的模型如DeepSeek、Ollama本地模型等。# 1. 创建并进入项目目录 mkdir ai-agent-tutorial cd ai-agent-tutorial # 2. 创建虚拟环境可选但强烈推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate # 3. 安装核心依赖 pip install langchain langchain-openai langgraph langchain-community # 4. 安装MCP相关包用于后续MCP部分 pip install mcp[cli] mcp-client-langchain2.2 配置API密钥在项目根目录创建一个.env文件来管理敏感信息不要将密钥硬编码在代码中。# .env OPENAI_API_KEY你的-openai-api-key # 后续可添加其他API密钥如SERPAPI_KEY、TAVILY_API_KEY等然后安装python-dotenv来读取环境变量。pip install python-dotenv3. LangChain核心构建你的第一个智能体让我们先用LangChain快速构建一个能使用简单工具的Agent感受一下框架的便利。3.1 工具Tools的定义与集成工具是Agent延伸能力的“手”。我们首先定义两个简单的工具一个计算器和一个搜索工具这里用模拟的。# tools/custom_tools.py from langchain.tools import tool from math import sqrt, pow tool def calculate(expression: str) - str: 执行一个数学表达式计算。支持加减乘除(-*/)、乘方(**)、开方(sqrt)和括号。 # 安全警告在生产环境中直接eval是危险的这里仅为演示。 # 应使用更安全的表达式解析库如 ast.literal_eval 或第三方库。 try: # 替换一些数学函数和常量 expression expression.replace(^, **).replace(sqrt, sqrt) # 非常简单的安全过滤不完善 allowed_chars set(0123456789-*/.() sqrt) if not all(c in allowed_chars for c in expression): return 错误表达式中包含不安全字符。 # 在受控环境下使用eval result eval(expression, {__builtins__: {}}, {sqrt: sqrt, pow: pow}) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def search_web(query: str) - str: 模拟网络搜索。在实际应用中应替换为真实的搜索API如Tavily、SerpAPI。 # 这里模拟返回一些静态结果 mock_data { LangChain: LangChain是一个用于开发由语言模型驱动的应用程序的框架。, Python: Python是一种广泛使用的高级编程语言。, 天气: 今天北京晴气温15-25摄氏度。 } for key, value in mock_data.items(): if key.lower() in query.lower(): return f模拟搜索“{query}”的结果{value} return f未找到与“{query}”相关的模拟结果。3.2 创建基础Agent并运行现在我们将工具、模型和提示词组合起来创建一个简单的ReAct推理行动风格的Agent。# basic_agent.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain.agents import create_react_agent, AgentExecutor from langchain.prompts import PromptTemplate from tools.custom_tools import calculate, search_web # 加载环境变量 load_dotenv() # 1. 初始化大语言模型 llm ChatOpenAI(modelgpt-4o-mini, temperature0, api_keyos.getenv(OPENAI_API_KEY)) # 2. 准备工具列表 tools [calculate, search_web] # 3. 定义ReAct风格的提示词模板 prompt_template 你是一个有帮助的AI助手可以访问以下工具 {tools} 请严格按照以下格式回答 问题用户提出的问题 思考你需要一步一步思考。如果需要使用工具请明确说明使用哪个工具以及输入是什么。 行动要执行的动作必须是以下格式之一Action: 工具名称 或 Action: Final Answer 行动输入工具的输入 观察工具返回的结果 ... (这个思考/行动/行动输入/观察的循环可以重复多次) 当你得出最终答案时必须使用以下格式 思考我已经得到所有需要的信息。 行动Final Answer 行动输入给用户的最终答案 开始 问题{input} 思考{agent_scratchpad} prompt PromptTemplate.from_template(prompt_template) # 4. 创建Agent和Executor agent create_react_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 5. 运行Agent if __name__ __main__: queries [ 3的平方加上4的平方等于多少, 告诉我LangChain是什么, 先计算(157)*2再搜索一下北京的天气。 ] for query in queries: print(f\n{*50}) print(f用户问题: {query}) print(f{*50}) try: result agent_executor.invoke({input: query}) print(f\n最终答案: {result[output]}) except Exception as e: print(f执行出错: {e})运行这个脚本(python basic_agent.py)你将看到Agent详细的思考过程verboseTrue它如何选择工具、传递参数并整合结果。这就是LangChain Agent的基本形态。4. LangGraph进阶设计多步骤工作流智能体当任务需要严格的步骤顺序、循环或分支判断时我们就需要LangGraph。我们将构建一个“研究助手”Agent它先规划任务然后执行搜索最后总结。4.1 定义状态State与节点Nodes在LangGraph中State是一个贯穿工作流的字典所有节点都读写它。Nodes是执行具体步骤的函数。# research_agent_graph.py import operator from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from dotenv import load_dotenv import os load_dotenv() llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 1. 定义状态结构 class AgentState(TypedDict): 图的工作流状态。 question: str # 用户原始问题 plan: str # 生成的计划 research_results: list # 收集的研究结果 final_answer: str # 最终答案 # 2. 定义各个节点函数 def planner_node(state: AgentState) - dict: 节点1规划节点。分析问题生成研究计划。 prompt ChatPromptTemplate.from_messages([ (system, 你是一个任务规划专家。请将复杂问题分解为2-3个具体的搜索子问题。), (human, 请为以下问题制定一个研究计划{question}) ]) chain prompt | llm | StrOutputParser() plan chain.invoke({question: state[question]}) print(f[规划节点] 生成的计划{plan}) return {plan: plan} def research_node(state: AgentState) - dict: 节点2研究节点。根据计划执行搜索这里模拟。 # 模拟搜索 - 实际应调用真实的搜索工具 plan state[plan] mock_responses { 什么是LangGraph: LangGraph是用于构建有状态、多参与者应用的库。, LangGraph和LangChain的区别: LangChain是组件化框架LangGraph专注于复杂工作流。, MCP协议的作用: MCP用于安全地将工具和上下文暴露给AI模型。 } results [] # 一个非常简单的关键词匹配模拟 for line in plan.split(\n): if ? in line or in line: for key in mock_responses: if key in line: results.append(f问题{line}\n答案{mock_responses[key]}) break else: results.append(f问题{line}\n答案未找到相关信息。) print(f[研究节点] 收集到 {len(results)} 条结果。) return {research_results: results} def synthesizer_node(state: AgentState) - dict: 节点3合成节点。基于研究结果生成最终答案。 question state[question] research_text \n---\n.join(state[research_results]) prompt ChatPromptTemplate.from_messages([ (system, 你是一个研究助手。请根据以下研究材料清晰、有条理地回答用户的问题。如果材料不足请说明。), (human, f用户问题{question}\n\n研究材料\n{research_text}\n\n请给出最终答案) ]) chain prompt | llm | StrOutputParser() final_answer chain.invoke({}) print(f[合成节点] 生成最终答案。) return {final_answer: final_answer} def router_node(state: AgentState) - str: 路由节点。决定下一步是继续研究还是结束。 # 简单的路由逻辑如果已经有研究结果就进入合成阶段否则结束这里简化。 # 更复杂的逻辑可以检查结果质量等。 if state.get(research_results): return synthesize else: # 如果没有结果我们可以选择结束或重试这里直接结束。 return end4.2 构建并运行图Graph将节点连接起来定义工作流。# 续 research_agent_graph.py # 3. 构建图 workflow StateGraph(AgentState) # 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(researcher, research_node) workflow.add_node(synthesizer, synthesizer_node) # 设置入口点 workflow.set_entry_point(planner) # 添加边连接 workflow.add_edge(planner, researcher) # 从researcher出来需要根据条件路由 workflow.add_conditional_edges( researcher, router_node, # 路由函数 { synthesize: synthesizer, # 如果router返回synthesize去synthesizer节点 end: END # 如果返回end直接结束 } ) workflow.add_edge(synthesizer, END) # 编译图 app workflow.compile() # 4. 运行图 if __name__ __main__: # 初始化状态 initial_state: AgentState { question: 请解释LangGraph和LangChain的区别并说明MCP在其中的作用。, plan: , research_results: [], final_answer: } print(开始执行研究助手工作流...) final_state app.invoke(initial_state) print(f\n{*60}) print(f用户问题{final_state[question]}) print(f\n生成的计划\n{final_state[plan]}) print(f\n最终答案\n{final_state[final_answer]}) print(f{*60})运行此脚本你将看到一个清晰的、分阶段的工作流执行过程。LangGraph的强大之处在于你可以轻松地在这个图中添加循环例如如果结果不充分则返回研究节点、并行分支同时执行多个搜索或更复杂的路由逻辑。5. MCP实战安全高效地扩展工具与上下文MCP将工具和数据源的管理外部化、标准化。我们通过一个简单的例子创建一个提供“当前时间”和“文件列表”工具的MCP Server然后在LangChain Agent中调用它。5.1 创建MCP Server首先我们创建一个独立的MCP Server。这通常是一个长期运行的后台进程。# mcp_server_demo.py import asyncio from datetime import datetime from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio import os # 创建Server实例 server Server(demo-tools-server) # 1. 注册一个“获取当前时间”的工具 server.list_tools() async def handle_list_tools(): return [ { name: get_current_time, description: 获取服务器当前的日期和时间。, inputSchema: { type: object, properties: { format: { type: string, description: 时间格式例如 %Y-%m-%d %H:%M:%S。默认为标准格式。, default: %Y-%m-%d %H:%M:%S } } } }, { name: list_files, description: 列出指定目录下的文件和文件夹。, inputSchema: { type: object, properties: { directory_path: { type: string, description: 要列出的目录路径。默认为当前目录。, default: . } } } } ] # 2. 实现工具的执行逻辑 server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_current_time: fmt arguments.get(format, %Y-%m-%d %H:%M:%S) current_time datetime.now().strftime(fmt) return [ { type: text, text: f当前时间服务器: {current_time} } ] elif name list_files: dir_path arguments.get(directory_path, .) try: items os.listdir(dir_path) # 简单区分文件和文件夹 formatted_items [] for item in items: full_path os.path.join(dir_path, item) if os.path.isdir(full_path): formatted_items.append(f[目录] {item}/) else: formatted_items.append(f[文件] {item}) result_text \n.join(formatted_items) if formatted_items else 目录为空。 return [ { type: text, text: f目录 {dir_path} 下的内容\n{result_text} } ] except FileNotFoundError: return [{type: text, text: f错误目录 {dir_path} 不存在。}] except PermissionError: return [{type: text, text: f错误没有权限访问目录 {dir_path}。}] else: raise ValueError(f未知工具: {name}) async def main(): 运行MCP Server over stdio. async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, InitializationOptions( server_namedemo-server, server_version0.1.0, capabilitiesserver.get_capabilities( notification_optionsNotificationOptions(), experimental_capabilities{}, ), ), ) if __name__ __main__: asyncio.run(main())这个Server通过标准输入输出(stdio)通信这是MCP的常见方式便于与客户端集成。5.2 在LangChain中连接并使用MCP Server我们需要在另一个进程中启动LangChain应用并通过MCP客户端连接到上面的Server。# mcp_client_agent.py import asyncio from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from mcp import ClientSession from mcp.client.stdio import stdio_client from langchain_mcp_tools import MCPTools async def run_agent_with_mcp(): # 1. 连接到MCP Server # 注意这里需要启动上面写的 mcp_server_demo.py 进程并通过stdio连接。 # 为了演示我们假设通过子进程启动。实际部署时Server可能独立运行在某个端口。 print(正在连接MCP Server...) async with stdio_client([python, mcp_server_demo.py]) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 2. 将MCP工具转换为LangChain可用的工具 mcp_tools MCPTools(session) tools await mcp_tools.get_tools() # 获取Server注册的所有工具 print(f从MCP Server加载了 {len(tools)} 个工具。) for tool in tools: print(f - {tool.name}: {tool.description}) # 3. 创建LangChain Agent llm ChatOpenAI(modelgpt-4o-mini, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个助手可以调用MCP Server提供的工具。请根据用户问题决定是否调用工具以及调用哪个工具。), (human, {input}), (placeholder, {agent_scratchpad}), ]) agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor(agentagent, toolstools, verboseTrue) # 4. 运行Agent queries [ 现在几点了, 列出当前目录下有什么文件。, 先告诉我时间再列出上级目录的内容。 ] for query in queries: print(f\n{*50}) print(f查询: {query}) result await agent_executor.ainvoke({input: query}) print(f答案: {result[output]}) if __name__ __main__: asyncio.run(run_agent_with_mcp())运行说明你需要打开两个终端。一个运行python mcp_server_demo.py启动Server另一个运行python mcp_client_agent.py启动客户端Agent。你将看到Agent通过标准协议调用到了Server上的工具。这种方式实现了工具与Agent逻辑的解耦工具可以独立开发、部署和升级。6. 综合实战构建多Agent协作系统我们将融合LangChain、LangGraph和MCP设计一个简单的“多专家评审系统”。场景是用户提交一段代码系统通过两个协作的Agent一个“代码审查员”一个“安全专家”来提供反馈最后由一个“协调员”Agent汇总意见。6.1 系统设计与状态定义# multi_agent_review.py from typing import TypedDict, List, Annotated import operator from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain.prompts import ChatPromptTemplate from langchain.schema.output_parser import StrOutputParser from dotenv import load_dotenv import os load_dotenv() llm ChatOpenAI(modelgpt-4o-mini, temperature0.2) # 稍高的temperature让输出更多样 class ReviewState(TypedDict): 多Agent评审系统的状态。 original_code: str # 用户提交的代码 code_review_feedback: str # 代码审查员的反馈 security_feedback: str # 安全专家的反馈 final_summary: str # 协调员生成的最终总结 all_feedback: List[str] # 收集所有反馈用于汇总6.2 实现专家Agent节点每个专家都是一个独立的LangChain Chain被封装为LangGraph的一个节点。# 续 multi_agent_review.py def create_expert_agent(role: str, expertise: str): 工厂函数创建特定角色的专家Agent Chain。 prompt ChatPromptTemplate.from_messages([ (system, f你是一个{role}擅长{expertise}。请针对用户提供的代码给出专业、具体、可操作的反馈。只反馈与专业领域相关的问题。), (human, 请审查以下代码\n\n{code}\n) ]) return prompt | llm | StrOutputParser() # 实例化专家 code_reviewer_chain create_expert_agent(资深代码审查员, 代码风格、可读性、性能优化和最佳实践) security_expert_chain create_expert_agent(安全专家, 代码安全漏洞、潜在的攻击面、不安全函数的使用) def code_review_node(state: ReviewState) - dict: 节点1代码审查员工作。 print([节点] 代码审查员开始工作...) feedback code_reviewer_chain.invoke({code: state[original_code]}) return {code_review_feedback: feedback, all_feedback: [f【代码审查】\n{feedback}]} def security_review_node(state: ReviewState) - dict: 节点2安全专家工作。 print([节点] 安全专家开始工作...) feedback security_expert_chain.invoke({code: state[original_code]}) return {security_feedback: feedback, all_feedback: [f【安全检查】\n{feedback}]}6.3 实现协调员与构建工作流协调员需要综合所有专家的意见。# 续 multi_agent_review.py def coordinator_node(state: ReviewState) - dict: 节点3协调员汇总反馈。 print([节点] 协调员正在汇总意见...) all_feedback state.get(all_feedback, []) combined_feedback \n\n---\n\n.join(all_feedback) prompt ChatPromptTemplate.from_messages([ (system, 你是一个技术团队协调员。请综合以下多位专家的评审意见生成一份给代码作者的、清晰、友好、结构化的最终总结报告。报告应涵盖所有重要问题并给出修改优先级建议。), (human, f原始代码\n\n{state[original_code]}\n\n\n专家反馈\n{combined_feedback}\n\n请生成最终总结报告) ]) chain prompt | llm | StrOutputParser() final_summary chain.invoke({}) return {final_summary: final_summary} # 构建并行工作流图 workflow StateGraph(ReviewState) # 添加节点 workflow.add_node(code_review, code_review_node) workflow.add_node(security_review, security_review_node) workflow.add_node(coordinator, coordinator_node) # 设置入口点并让两个专家并行工作 workflow.set_entry_point(code_review) workflow.add_edge(code_review, security_review) workflow.add_edge(security_review, coordinator) workflow.add_edge(coordinator, END) # 编译图 app workflow.compile()6.4 运行与测试# 续 multi_agent_review.py if __name__ __main__: # 示例代码一个可能存在安全风险和风格问题的Python代码片段 sample_code import subprocess import os def run_user_input(): user_cmd input(Enter command to run: ) # 直接执行用户输入存在命令注入风险 result subprocess.run(user_cmd, shellTrue, capture_outputTrue, textTrue) print(result.stdout) def process_data(data_list): # 低效的循环和字符串拼接 output for item in data_list: output str(item) , return output initial_state: ReviewState { original_code: sample_code, code_review_feedback: , security_feedback: , final_summary: , all_feedback: [] } print(开始多专家代码评审流程...\n) final_state app.invoke(initial_state) print(f\n{*80}) print(【原始代码】) print(sample_code) print(f\n{*80}) print(【代码审查员反馈】) print(final_state[code_review_feedback]) print(f\n{*80}) print(【安全专家反馈】) print(final_state[security_feedback]) print(f\n{*80}) print(【协调员最终总结报告】) print(final_state[final_summary]) print(f{*80})运行此脚本你将看到一个自动化的流水线两个专家Agent并行/顺序地对代码进行分析然后协调员Agent生成一份综合报告。这个模式可以轻松扩展到更多专家如“性能专家”、“架构专家”并可以通过LangGraph灵活控制他们的协作顺序。7. 常见问题、调试与优化指南在实际开发中你肯定会遇到各种问题。以下是一些高频问题的排查思路和优化建议。7.1 常见错误与解决方案问题现象可能原因排查与解决思路ModuleNotFoundError: No module named langchain_...依赖包未安装或版本不兼容。1. 使用pip list | grep langchain检查已安装版本。2. 确认包名是否正确社区工具包常以langchain-community或langchain-xxx命名。3. 使用pip install langchain-community等命令安装。Agent陷入循环不停调用工具。1. 提示词未明确停止条件。2. 工具输出未提供足够信息导致LLM反复尝试。1. 在提示词中强调“当获得足够信息时必须使用Final Answer格式”。2. 为工具调用设置最大迭代次数max_iterations在AgentExecutor中设置。3. 优化工具的描述和输出使其更精确。LLM不按格式输出导致解析失败。1. 模型温度(temperature)过高输出随机性大。2. 提示词中对输出格式的指令不够清晰。1. 将temperature设为0或一个较低的值如0.1。2. 使用ChatPromptTemplate的SystemMessage明确指定格式或用StructuredOutputParser。3. 在AgentExecutor中设置handle_parsing_errorsTrue以优雅处理错误。MCP连接失败或超时。1. MCP Server进程未启动或已崩溃。2. 客户端连接参数如stdio命令错误。3. 协议版本不匹配。1. 检查Server进程是否在运行查看其日志是否有错误。2. 确保客户端启动命令与Server启动方式匹配。3. 检查mcp相关库的版本尝试升级到最新稳定版。工具调用结果未被Agent正确理解。工具返回的内容过于冗长或格式混乱。1. 工具函数应返回清晰、简洁的文本。2. 可以尝试让工具返回结构化数据如JSON并在提示词中指导LLM如何解读。7.2 性能与成本优化模型选择对于简单的工具调用和路由使用轻量级模型如gpt-4o-mini,gpt-3.5-turbo可以大幅降低成本并提高速度。将复杂的推理任务留给更强的模型如gpt-4o。缓存利用LangChain的缓存功能如InMemoryCache,SQLiteCache缓存频繁且不变的LLM响应和工具调用结果。from langchain.cache import InMemoryCache from langchain.globals import set_llm_cache set_llm_cache(InMemoryCache())流式输出对于长时间运行的任务使用流式输出stream可以提升用户体验让用户感知到进度。异步调用如果工作流中有多个可以并行执行的独立步骤如多个网络请求使用ainvoke等异步方法可以提高整体效率。精简上下文避免在状态或提示词中携带过长的历史信息。使用ConversationSummaryMemory或EntityMemory来压缩关键记忆而非传递全部原始对话。7.3 提示词工程最佳实践提示词的质量直接决定Agent的可靠性。角色明确在System Prompt中给Agent设定清晰的角色和职责边界。格式强制明确指定输出格式并使用StructuredOutputParser或JsonOutputParser来约束LLM的输出。少样本示例Few-Shot在提示词中提供1-2个输入输出的例子能极大地提升模型在复杂任务上的表现。分而治之对于复杂任务像我们使用LangGraph那样将其分解为多个子任务并为每个子任务设计专门的、简单的提示词而不是用一个极其复杂的提示词让LLM一次完成所有事。8. 生产环境部署与工程化建议将原型转化为稳定可用的服务需要考虑更多因素。配置管理将所有配置模型API密钥、温度、最大迭代次数、Server地址等外置到环境变量或配置文件中如config.yaml切勿硬编码。日志与监控为Agent的关键步骤如工具调用、LLM请求、状态转换添加结构化日志。监控Token消耗、请求延迟和错误率。错误处理与重试为网络请求、工具调用和LLM调用添加重试机制使用tenacity等库和全面的异常捕获避免单点失败导致整个工作流崩溃。版本控制对提示词模板、工作流图定义、工具集进行版本控制。它们的微小变动可能导致Agent行为巨大差异。测试为你的Agent工作流编写单元测试和集成测试。测试应包括典型用例、边界用例、错误输入下的行为。安全工具权限严格控制工具能访问的资源如数据库、文件系统、网络。MCP架构在这方面有天然优势。用户输入净化对所有来自用户或外部的输入进行验证和净化防止注入攻击。LLM输出审查对于生成代码、执行命令等高风险场景考虑加入人工审核或自动安全扫描环节。学习AI Agent开发是一个循序渐进的过程。建议从LangChain的单工具Agent开始掌握其组件化思想然后通过LangGraph构建有状态的多步骤工作流理解状态管理和控制流最后在需要解耦工具、管理复杂上下文或追求更高安全性时引入MCP。不要试图一开始就构建一个庞大的多Agent系统从解决一个具体的、小规模的问题入手逐步迭代和扩展是掌握这项技术最有效的方法。