AI智能体开发实战:基于LangChain与MCP协议构建可运行应用

发布时间:2026/8/22 18:29:36
AI智能体开发实战:基于LangChain与MCP协议构建可运行应用 这次我们来看一个面向AI智能体Agent开发的实战教程。这个教程的核心目标不是空谈概念而是提供一套从零基础到实战落地的完整路径重点解决“学了很多理论但不知道如何动手”的痛点。它整合了当前最热门的Agent、AI大模型、LangChain和MCPModel Context Protocol等技术栈旨在让你快速构建能实际运行的智能体应用。如果你关心如何利用开源框架和协议在本地或云端快速搭建一个具备规划、工具调用和记忆能力的智能体并且希望了解具体的环境搭建、代码编写、调试排错和部署上线全流程那么这篇文章可以直接收藏。我们将重点关注如何避开那些新手常踩的坑比如环境配置冲突、依赖版本不兼容、API调用失败以及智能体逻辑设计误区。本文不会停留在概念介绍而是会带你完成一个可运行的智能体项目实战。我们会从最基础的环境准备开始一步步搭建开发环境编写核心代码集成大模型与工具并通过MCP协议扩展能力最后部署并测试一个完整的智能体任务。整个过程将重点关注工具链的选择、代码的模块化设计以及遇到问题时的排查思路。1. 核心能力速览在深入代码之前我们先快速了解通过本教程你将掌握的核心能力与技术栈。这有助于你判断是否值得投入时间以及需要准备哪些前置知识。能力项说明与目标技术栈覆盖集成LangChain智能体框架、大模型API/本地模型如GPT、DeepSeek等、MCP协议工具扩展标准开发门槛要求具备基础Python编程能力对HTTP API调用有基本了解。无需深厚的机器学习背景。环境依赖主要依赖Python环境建议3.8以及pip包管理。部分工具可能需要Docker用于MCP Server。硬件要求开发阶段普通CPU即可。涉及本地大模型推理需根据模型尺寸准备足够GPU显存如7B模型约需6-8GB。本文以调用云端API为主。核心产出一个具备任务规划、工具调用如搜索、计算、状态记忆能力的可运行智能体Demo。关键技能1. LangChain Agent的构建与调度2. 大模型API的集成与提示词工程3. MCP Server的本地部署与客户端连接4. 智能体的错误处理与循环控制适合场景个人学习、技术验证、自动化流程原型开发、AI应用前端如聊天机器人的后端逻辑核心。2. 适用场景与使用边界在开始动手之前明确智能体能做什么、不能做什么至关重要。这决定了你投入精力后能否获得预期的回报。智能体非常适合以下场景复杂任务自动化需要多步骤决策的任务例如“分析某公司最近财报总结其风险并生成一份简报”。智能体可以自动分解为搜索财报、提取关键数据、分析风险、组织成文。动态工具调用根据用户输入动态选择并使用不同的工具。例如用户问“北京天气怎么样”调用天气API问“123的平方是多少”调用计算器工具。对话式交互与持久会话构建能记住上下文、进行多轮对话的聊天机器人不仅聊天还能在对话中执行操作如订餐、查日程。快速原型验证当你有一个利用AI处理复杂流程的想法时用LangChain Agent可以快速搭建出可交互的原型验证可行性。需要注意的边界与限制并非万能魔法智能体的能力受限于其集成的大模型的认知与推理能力以及其可调用的工具的广度和质量。它无法执行没有对应工具或知识库的任务。存在“幻觉”与错误大模型可能产生错误信息或错误决策导致智能体执行错误步骤。健壮的智能体需要设计验证机制和错误处理回路。成本与延迟频繁调用大模型API会产生费用多步推理也会增加响应延迟。需要对任务进行合理规划避免不必要的循环。安全与权限智能体能够执行工具所赋予的任何操作。必须严格控制工具权限避免执行危险命令如删除文件、访问敏感数据或进行未经授权的网络请求。合规性如果智能体处理用户数据、生成内容或进行商业决策需确保符合数据隐私、版权和行业监管要求。3. 环境准备与前置条件工欲善其事必先利其器。一个干净、隔离的Python环境是成功的第一步能避免90%的依赖冲突问题。1. 基础环境检查操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)。本文命令以 Linux/macOS 为例Windows 用户可在 PowerShell 或 WSL2 中运行。Python确保安装 Python 3.8 或更高版本。在终端中运行python --version或python3 --version确认。包管理工具pip需要是最新版本。更新命令pip install --upgrade pip。代码编辑器VS Code (推荐有优秀的Python和AI插件)、PyCharm 或任何你熟悉的编辑器。2. 创建并激活虚拟环境强烈建议为项目创建独立的虚拟环境。# 创建名为 agent_env 的虚拟环境 python -m venv agent_env # 激活虚拟环境 # Linux/macOS: source agent_env/bin/activate # Windows: # agent_env\Scripts\activate激活后终端提示符前应显示(agent_env)。3. 关键工具准备Docker (可选但推荐)MCP 协议的一些工具服务器Server常以 Docker 镜像形式提供用 Docker 运行最为方便。前往 Docker 官网下载并安装 Desktop 版本。Git用于克隆示例代码库。确保已安装。完成以上步骤你的基础开发环境就准备好了。接下来我们将安装核心的 Python 库。4. 安装核心依赖与启动 MCP 服务器我们将以 LangChain 为核心框架并集成 MCP 协议来扩展工具能力。首先安装必要的 Python 包。1. 安装 LangChain 及相关库在激活的虚拟环境中执行以下命令pip install langchain langchain-community langchain-core pip install openai # 如果你使用 OpenAI API # 或者使用其他模型提供商例如 # pip install langchain-google-genai # 用于 Google Gemini # pip install langchain-groq # 用于 Groq # pip install langchain-anthropic # 用于 Claude2. 安装 MCP 客户端与必要工具MCP 是连接智能体和外部工具的标准协议。我们需要安装客户端库和一些官方工具。# 安装 LangChain 的 MCP 集成包 pip install langchain-mcp-adapters # 安装 MCP 客户端库 (用于直接与 MCP Server 通信) pip install mcp # 安装一些有用的 CLI 工具用于管理 MCP 服务器 pip install mcp-cli3. 启动一个 MCP 服务器以“文件系统”工具为例MCP 服务器是独立进程为智能体提供工具能力。我们先用一个简单的“文件系统”服务器来演示。首先你需要找到一个可用的 MCP 服务器实现。一个流行的选择是modelcontextprotocol/servers仓库中的工具。由于它们是 Node.js 编写用 Docker 运行最简单。# 拉取官方 MCP 服务器镜像示例 docker pull ghcr.io/modelcontextprotocol/servers/filesystem # 运行文件系统 MCP 服务器 # 将 /path/to/allow/list 替换为你允许智能体访问的本地目录路径例如 $(pwd)/workspace docker run -it --rm \ -v /path/to/allow/list:/allowed \ -p 3000:3000 \ ghcr.io/modelcontextprotocol/servers/filesystem这条命令启动了一个文件系统服务器它监听 3000 端口并只能访问你挂载的/allowed目录。注意出于安全考虑务必将其限制在非敏感目录。4. 验证 MCP 服务器打开另一个终端使用mcpCLI 检查服务器是否正常运行。# 列出服务器提供的工具 mcp ls http://localhost:3000如果成功你应该能看到类似read_file,write_file,list_files这样的工具列表。这表明 MCP 服务器已就绪可以被智能体调用。至此我们的基础服务和框架都已准备完毕。接下来进入核心环节编写智能体逻辑。5. 构建你的第一个 LangChain 智能体我们将创建一个能使用“计算器”和“搜索”通过 MCP 文件读取模拟工具的简单智能体。这里我们使用 OpenAI 的 GPT 模型作为“大脑”你也可以替换为其他兼容的模型。1. 设置 API 密钥首先将你的大模型 API 密钥设置为环境变量。如果你用 OpenAI# 在终端中设置临时 export OPENAI_API_KEYyour-api-key-here或者在 Python 代码中通过os.environ设置。2. 编写智能体构建脚本创建一个名为first_agent.py的文件并写入以下代码import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser # 1. 定义工具这里我们先模拟两个简单工具后续替换为MCP工具 from langchain.tools import tool tool def calculate(expression: str) - str: 计算一个数学表达式。例如calculate(\2 3 * 4\) try: # 警告使用eval有安全风险仅用于演示。生产环境应用安全计算库。 result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} tool def read_file_summary(filepath: str) - str: 读取指定文件并返回其内容摘要模拟。 # 此处模拟MCP文件读取工具的行为 allowed_dir ./workspace full_path os.path.join(allowed_dir, filepath.lstrip(/)) if not os.path.commonpath([allowed_dir, os.path.abspath(full_path)]) os.path.abspath(allowed_dir): return 错误无权访问该路径。 try: with open(full_path, r, encodingutf-8) as f: content f.read() # 简单模拟摘要取前200字符 summary content[:200] (... if len(content) 200 else ) return f文件 {filepath} 的内容摘要{summary} except FileNotFoundError: return f错误文件 {filepath} 未找到。 except Exception as e: return f读取文件时出错: {e} tools [calculate, read_file_summary] # 2. 选择大模型 llm ChatOpenAI(modelgpt-3.5-turbo-1106, temperature0) # 使用gpt-3.5温度设为0使输出更确定 # 3. 构建提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个乐于助人的助手可以调用工具来解决问题。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), # 用于存放工具调用历史 ]) # 4. 绑定工具到LLM llm_with_tools llm.bind_tools(tools) # 5. 创建智能体 agent create_openai_tools_agent( llmllm_with_tools, toolstools, promptprompt, ) # 6. 创建执行器 agent_executor AgentExecutor(agentagent, toolstools, verboseTrue, handle_parsing_errorsTrue) # 7. 运行测试 if __name__ __main__: # 准备测试环境在workspace目录下创建一个测试文件 os.makedirs(./workspace, exist_okTrue) with open(./workspace/test.txt, w) as f: f.write(这是一个测试文件内容是关于LangChain和MCP协议的学习笔记。它们共同用于构建强大的AI智能体。) test_queries [ “123乘以456等于多少”, “请总结一下workspace目录下test.txt文件的内容。” ] for query in test_queries: print(f\n用户提问: {query}) print(- * 40) result agent_executor.invoke({input: query}) print(f智能体回答: {result[output]})3. 运行并观察在终端中运行这个脚本python first_agent.py你应该能看到详细的verbose日志显示智能体的思考过程、工具调用和最终结果。例如对于计算问题它会调用calculate工具对于文件读取它会调用read_file_summary工具。这个简单的智能体已经具备了任务理解、工具选择和结果整合的基本能力。下一步我们将把模拟工具替换为真实的 MCP 工具。6. 集成 MCP 协议连接真实的工具服务器现在我们将用上一节启动的真实 MCP 文件系统服务器替换掉模拟的read_file_summary工具。1. 安装并配置 MCP 客户端连接创建一个新脚本agent_with_mcp.py。import os from langchain.agents import AgentExecutor, create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain.agents.format_scratchpad.openai_tools import format_to_openai_tool_messages from langchain.agents.output_parsers.openai_tools import OpenAIToolsAgentOutputParser # 导入 LangChain 的 MCP 集成 from langchain_mcp_adapters.tools import load_mcp_tools async def main(): # 1. 连接到正在运行的 MCP 服务器 # 假设你的文件系统 MCP 服务器运行在 http://localhost:3000 server_uri http://localhost:3000 print(f正在从 {server_uri} 加载 MCP 工具...) try: # load_mcp_tools 是一个异步函数用于加载工具 mcp_tools await load_mcp_tools(server_uri) print(f成功加载 {len(mcp_tools)} 个 MCP 工具。) for tool in mcp_tools: print(f - {tool.name}: {tool.description}) except Exception as e: print(f加载 MCP 工具失败: {e}) print(请确保 MCP 文件系统服务器正在运行 (docker run ...)) return # 2. 定义本地工具如之前的计算器 from langchain.tools import tool tool def calculate(expression: str) - str: 计算一个数学表达式。例如calculate(\2 3 * 4\) try: result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {e} # 3. 合并所有工具 all_tools [calculate] mcp_tools # 4. 初始化 LLM 和提示词 (与之前相同) llm ChatOpenAI(modelgpt-3.5-turbo-1106, temperature0) prompt ChatPromptTemplate.from_messages([ (system, 你是一个可以调用工具来帮助用户的助手。你可以读写允许目录下的文件并进行数学计算。), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) llm_with_tools llm.bind_tools(all_tools) # 5. 创建智能体及执行器 agent create_openai_tools_agent(llmllm_with_tools, toolsall_tools, promptprompt) agent_executor AgentExecutor(agentagent, toolsall_tools, verboseTrue, handle_parsing_errorsTrue) # 6. 测试使用真实的 MCP 文件工具 test_queries [ “先计算 (15 27) * 3 的值。”, “请列出 MCP 服务器允许目录/allowed下的所有文件。”, “请读取 workspace/test.txt 文件的内容。”, “创建一个名为 ‘hello_mcp.txt’ 的新文件并写入内容 ‘Hello from MCP Agent!’。” ] for query in test_queries: print(f\n{*50}) print(f用户提问: {query}) print(f{*50}) try: # 注意AgentExecutor.invoke 在最新版本可能是同步的这里按同步处理。 # 如果使用完全异步的客户端可能需要使用 await agent_executor.ainvoke(...) result agent_executor.invoke({input: query}) print(f智能体回答:\n{result[output]}) except Exception as e: print(f执行过程中出错: {e}) if __name__ __main__: import asyncio asyncio.run(main())2. 运行与验证确保你的 MCP 文件系统 Docker 容器仍在运行。然后在终端执行python agent_with_mcp.py观察日志。智能体现在应该能调用本地的calculate工具进行数学运算。调用 MCP 服务器的list_files工具来列出目录。调用read_file工具读取文件内容。调用write_file工具创建新文件。至此你已经成功构建了一个集成了真实外部工具的智能体。它通过标准协议MCP与工具对话实现了能力的扩展。7. 高级话题智能体记忆、复杂工作流与错误处理一个基础的智能体已经跑通但要投入实用还需要处理更复杂的情况。1. 为智能体添加会话记忆让智能体记住之前的对话实现多轮交互。from langchain.memory import ConversationBufferMemory # 在创建 AgentExecutor 时加入 memory 参数 memory ConversationBufferMemory(memory_keychat_history, return_messagesTrue) agent_executor AgentExecutor( agentagent, toolstools, verboseTrue, memorymemory, handle_parsing_errorsTrue ) # 调用时传入的 input 会自动与历史记录组合 result agent_executor.invoke({input: “我之前让你创建的文件叫什么名字”}) # 它能记得2. 设计复杂工作流与规划对于多步骤任务智能体有时会“迷路”。可以通过更精细的提示词如 Chain of Thought或使用 LangChain 的PlanAndExecute高级模式来改进。# 示例在系统提示词中鼓励分步思考 prompt ChatPromptTemplate.from_messages([ (system, “””你是一个严谨的助手。在回答前请先思考你需要哪些步骤并一步步调用工具。 对于复杂任务先制定一个简单计划。”””), (user, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ])3. 强化错误处理与超时控制智能体调用工具可能失败网络错误、工具异常等。需要增强鲁棒性。from langchain.agents import Tool from langchain.callbacks.manager import CallbackManagerForToolRun class RobustTool(Tool): 一个带有重试和超时机制的工具包装器 def _run(self, query: str, run_manager: CallbackManagerForToolRun | None None) - str: import requests from tenacity import retry, stop_after_attempt, wait_exponential retry(stopstop_after_attempt(3), waitwait_exponential(multiplier1, min4, max10)) def call_with_retry(): # 这里是实际的工具调用逻辑例如调用一个API response requests.post(self.url, json{query: query}, timeout10) response.raise_for_status() return response.text() try: return call_with_retry() except Exception as e: return f“工具调用失败错误信息: {str(e)}。请检查网络或服务状态。” # 在创建工具列表时使用这个包装器8. 部署与 API 服务化开发完成后你可能希望将智能体封装成 API 服务供其他应用调用。使用 FastAPI 是常见选择。1. 创建 FastAPI 应用创建一个app.py文件from fastapi import FastAPI, HTTPException from pydantic import BaseModel from langchain.agents import AgentExecutor # ... 导入你的智能体构建代码 ... app FastAPI(titleAI Agent Service) # 在启动时初始化智能体执行器注意实际生产环境需要考虑并发和状态管理 agent_executor None app.on_event(startup) async def startup_event(): global agent_executor # 这里调用你之前写的函数来初始化 agent_executor # 例如agent_executor await create_my_agent() print(Agent 服务已启动。) class QueryRequest(BaseModel): input: str session_id: str | None None # 用于区分不同会话 class QueryResponse(BaseModel): output: str session_id: str app.post(/query, response_modelQueryResponse) async def query_agent(request: QueryRequest): if agent_executor is None: raise HTTPException(status_code503, detailAgent not initialized) try: # 根据 session_id 获取或创建对应的 memory # 这里简化处理实际需要维护一个 memory 字典 result agent_executor.invoke({input: request.input}) return QueryResponse(outputresult[output], session_idrequest.session_id or default) except Exception as e: raise HTTPException(status_code500, detailfAgent execution failed: {str(e)}) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port8000)2. 使用 curl 测试 APIcurl -X POST http://localhost:8000/query \ -H Content-Type: application/json \ -d {input: “计算圆周率小数点后5位” “session_id”: “user123”}9. 常见问题与排查方法在开发和运行过程中你几乎一定会遇到以下问题。这里提供快速排查思路。问题现象可能原因排查方式解决方案导入 LangChain 失败虚拟环境未激活pip 版本过低依赖冲突。1. 确认终端提示符有(env_name)。2. pip listgrep langchain 查看。3. 查看完整错误信息。智能体不调用工具提示词未引导工具描述不清晰LLM 温度过高。1. 检查verboseTrue日志看 LLM 思考过程。2. 检查工具函数的docstring是否清晰。1. 在系统提示词中明确要求使用工具。2. 优化工具描述包含清晰的输入输出示例。3. 将 LLM 的temperature参数调低如 0。MCP 服务器连接失败服务器未启动端口被占用网络策略限制。1.docker ps查看容器状态。2.curl http://localhost:3000测试连通性。3. 检查防火墙或安全组。1. 确保 Docker 命令正确容器在运行。2. 更换端口检查-p映射是否正确。3. 在服务器日志中查找错误。工具调用返回权限错误MCP 服务器配置的允许目录路径不对。检查 Docker 命令中-v参数挂载的目录路径。确保挂载的目录是存在的并且是你希望智能体访问的目录。使用绝对路径。API 密钥错误环境变量未设置或设置错误。echo $OPENAI_API_KEY(Linux/macOS) 或echo %OPENAI_API_KEY%(Windows) 检查。1. 正确设置环境变量。2. 或在代码中直接os.environ[‘OPENAI_API_KEY’] ‘key’。智能体陷入循环或错误解析工具输出格式混乱LLM 解析失败。查看verbose日志中工具返回的原始文本和 LLM 的下一步决策。1. 确保工具返回纯文本或结构清晰的 JSON。2. 使用handle_parsing_errorsTrue参数。3. 在提示词中要求 LLM 输出特定格式。部署后 API 响应慢冷启动加载模型网络延迟任务过于复杂。1. 检查服务启动日志。2. 测试简单查询的响应时间。3. 监控服务器资源CPU/内存。1. 考虑使用异步加载和缓存。2. 对于复杂任务设置超时并返回任务ID改为异步处理。3. 升级服务器配置。10. 最佳实践与后续方向掌握了基础构建和问题排查后遵循以下最佳实践能让你的智能体项目更稳健、更易维护。开发阶段最佳实践版本锁定使用requirements.txt或pyproject.toml精确锁定所有依赖库的版本避免未来更新导致的不兼容。配置外置将 API 密钥、服务器地址、模型参数等写入配置文件如config.yaml或环境变量不要硬编码在代码中。日志完备为智能体的关键步骤接收输入、调用工具、得到输出、发生错误添加详细日志便于调试。单元测试为每个自定义工具编写单元测试确保其功能正常。模拟测试智能体对典型 query 的响应。逐步复杂化从一个工具、一个简单任务开始验证通过后再逐步添加更多工具和复杂逻辑。安全与合规建议工具沙箱对于文件操作、系统命令等高风险工具必须在严格受限的沙箱环境中运行如 Docker 容器、无权限的用户。输入验证与清理对所有用户输入和工具返回的内容进行验证和清理防止注入攻击或处理恶意内容。访问控制为你的智能体 API 添加认证和授权层确保只有合法用户能访问。内容审核如果智能体生成的内容对外发布应加入审核环节或使用内容安全过滤器。后续深入方向集成更多 MCP 工具探索 MCP 官方仓库和社区集成数据库、搜索引擎、代码解释器、绘图等丰富工具。尝试本地大模型使用Ollama、vLLM或Transformers库在本地部署开源大模型如 Llama、Qwen降低 API 成本并提升隐私性。实现智能体编排使用LangGraph来构建有状态、可循环、多分支的复杂智能体工作流。加入评估与监控设计评估体系定期用测试集评估智能体性能。加入监控告警关注 API 调用失败率、响应时长等指标。前端界面开发使用Gradio、Streamlit或React等框架为你的智能体构建一个直观的聊天式 Web 界面。构建 AI 智能体的旅程是从一个能运行的小 demo 开始的。不要试图一开始就设计一个完美的全能助手。从解决一个具体的小问题出发比如“自动整理我下载文件夹中的文件并分类”或“根据我的会议纪要生成待办列表”选择一个核心工具打通整个流程。在这个过程中你会深刻理解工具描述、提示词设计、错误处理的重要性。当这个核心循环跑通后再像搭积木一样逐步加入记忆、规划、更多工具和漂亮的界面。