Python实战：用LangGraph和MCP打造你的第一个AI代理（附完整代码）

Python实战用LangGraph和MCP构建智能代理的完整指南在当今快速发展的AI领域构建能够理解和执行复杂任务的智能代理已成为开发者关注的焦点。本文将带您深入了解如何利用LangGraph框架和模型上下文协议(MCP)构建一个功能完备的AI代理从基础概念到实际代码实现一步步掌握这一强大技术组合。1. LangGraph与MCP技术概览LangGraph是一个专为构建和编排AI代理而设计的框架它提供了一种灵活的方式来定义代理的行为和工作流程。与传统的线性处理不同LangGraph采用图结构来表示代理的决策流程使得复杂任务的分解和执行变得更加直观和高效。模型上下文协议(MCP)则是一个开放协议它标准化了应用程序如何向语言模型提供工具和上下文。可以将MCP想象为AI应用的USB接口——它为不同工具和数据源提供了一种标准化的连接方式。通过MCP我们可以轻松地将各种功能(如数学计算、天气查询等)封装为标准化工具供语言模型调用。这两项技术的结合为开发者提供了强大而灵活的工具集LangGraph负责代理的高层逻辑和流程控制MCP处理底层工具调用和上下文管理组合使用可实现复杂任务的分解与执行# 基础架构示例 from langgraph.prebuilt import create_react_agent from langchain_mcp_adapters.client import MultiServerMCPClient # 初始化MCP客户端连接多个工具服务器 client MultiServerMCPClient({ math: {command: python, args: [math_server.py], transport: stdio}, weather: {url: http://localhost:8000/mcp, transport: http} }) # 创建LangGraph代理 tools await client.get_tools() agent create_react_agent(gpt-4, tools)2. 环境准备与工具服务器搭建在开始构建代理前我们需要设置开发环境并准备必要的工具服务器。以下是完整的准备工作流程2.1 安装依赖包首先确保已安装Python 3.8然后通过pip安装所需依赖pip install langchain-mcp-adapters langgraph mcp dashscope提示建议使用虚拟环境来管理项目依赖避免与其他项目产生冲突。2.2 创建数学计算服务器数学服务器将提供基本的算术运算功能。创建一个math_server.py文件from mcp.server.fastmcp import FastMCP mcp FastMCP(Math) mcp.tool() def add(a: int, b: int) - int: 两个整数相加 return a b mcp.tool() def multiply(a: int, b: int) - int: 两个整数相乘 return a * b mcp.tool() def divide(a: int, b: int) - float: 两个整数相除 if b 0: raise ValueError(除数不能为零) return a / b mcp.tool() def subtract(a: int, b: int) - int: 两个整数相减 return a - b if __name__ __main__: mcp.run(transportstdio)2.3 创建天气查询服务器天气服务器模拟天气查询功能。创建weather_server.py文件from mcp.server.fastmcp import FastMCP mcp FastMCP(Weather) mcp.tool() async def get_weather(location: str) - str: 获取指定城市的天气信息 # 实际应用中这里会调用天气API return f{location}今天天气晴气温25℃ if __name__ __main__: mcp.run(transporthttp, port8000)2.4 服务器管理为了同时运行多个服务器可以使用supervisord或简单的终端多窗口方式# 窗口1运行数学服务器 python math_server.py # 窗口2运行天气服务器 python weather_server.py3. 构建核心代理逻辑有了工具服务器后我们可以开始构建主代理逻辑。创建一个main.py文件作为入口点import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient from langgraph.prebuilt import create_react_agent from langchain_community.chat_models.tongyi import ChatTongyi async def main(): # 初始化语言模型(这里使用通义千问) llm ChatTongyi(temperature0) # 配置MCP客户端连接多个工具服务器 client MultiServerMCPClient({ math: { command: python, args: [math_server.py], transport: stdio }, weather: { url: http://localhost:8000/mcp, transport: http } }) # 创建LangGraph代理 tools await client.get_tools() agent create_react_agent(llm, tools) # 测试数学计算 math_response await agent.ainvoke({ messages: [{role: user, content: (3 7) × 12等于多少?}] }) print(数学计算结果:, math_response[messages][-1][content]) # 测试天气查询 weather_response await agent.ainvoke({ messages: [{role: user, content: 杭州天气怎么样?}] }) print(天气查询结果:, weather_response[messages][-1][content]) if __name__ __main__: asyncio.run(main())4. 高级功能与优化基础代理运行后我们可以进一步扩展其功能和性能4.1 添加工具拦截器工具拦截器允许我们在工具调用前后插入自定义逻辑例如日志记录、参数验证等from langchain_mcp_adapters.interceptors import MCPToolCallRequest async def logging_interceptor(request: MCPToolCallRequest, handler): 记录工具调用日志的拦截器 print(f调用工具 {request.name}参数: {request.args}) try: result await handler(request) print(f工具 {request.name} 调用成功结果: {result}) return result except Exception as e: print(f工具 {request.name} 调用失败: {str(e)}) raise # 在创建MultiServerMCPClient时添加拦截器 client MultiServerMCPClient( {...}, # 服务器配置 tool_interceptors[logging_interceptor] )4.2 多服务器负载均衡当有多个同类工具服务器时可以实现简单的负载均衡from random import choice math_servers [ {command: python, args: [math_server1.py], transport: stdio}, {command: python, args: [math_server2.py], transport: stdio} ] client MultiServerMCPClient({ math: choice(math_servers), # 随机选择一个数学服务器 weather: {...} })4.3 持久化与状态管理LangGraph支持将代理状态持久化实现长时间运行的对话from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(:memory:) agent create_react_agent( llm, tools, checkpointmemory ) # 恢复特定会话 agent_response await agent.ainvoke( {messages: [...]}, {configurable: {thread_id: user123}} )5. 实际应用案例与调试技巧让我们通过几个实际案例来展示代理的能力并分享一些调试技巧5.1 复杂数学表达式求解代理可以处理嵌套的数学表达式response await agent.ainvoke({ messages: [{ role: user, content: 计算(15 - 3) ÷ 4 7 × 2的值 }] })5.2 组合查询代理可以组合使用多个工具完成复杂查询response await agent.ainvoke({ messages: [{ role: user, content: 如果杭州气温是25度比北京高7度那么北京气温是多少? }] })5.3 调试技巧当代理表现不符合预期时可以检查工具调用日志确认工具是否被正确调用查看中间推理过程有些模型支持返回推理步骤简化问题从最基本的功能测试开始逐步复杂化调整温度参数降低temperature值减少随机性# 启用详细日志 import logging logging.basicConfig(levellogging.DEBUG)6. 性能优化与最佳实践为了确保代理的高效运行以下是一些性能优化建议6.1 服务器配置优化配置项推荐值说明线程数4-8根据CPU核心数调整超时时间30s避免长时间等待连接池大小10-20平衡资源与并发6.2 缓存常用结果对于不常变化的数据(如天气)可以添加缓存层from functools import lru_cache mcp.tool() lru_cache(maxsize100) async def get_weather(location: str) - str: 带缓存的天气查询 return await fetch_weather_api(location)6.3 异步处理充分利用Python的异步特性提高并发性能async def batch_queries(queries): tasks [agent.ainvoke({messages: [{role: user, content: q}]}) for q in queries] return await asyncio.gather(*tasks)构建基于LangGraph和MCP的AI代理是一个循序渐进的过程从简单的工具集成开始逐步添加复杂性和优化性能。本文介绍的方法和技巧可以帮助您快速上手这一强大技术组合为您的应用添加智能代理能力。