最近在尝试构建一个能自主调用工具、处理复杂任务的AI智能体时你是否也遇到了这些困扰LangChain、LangGraph、MCP这些概念听起来很酷但官方文档零散代码示例跑不通Agent动不动就报错终止想从零搭建一个可用的系统更是无从下手本文正是为了解决这些问题而生。我将带你从零开始手把手搭建一个基于新版LangChain、LangGraph和MCP协议的本地AI智能体开发环境并通过一个完整的“天气查询与数据分析”实战项目让你彻底掌握从环境配置、核心概念理解到代码实战开发的全流程。无论你是刚接触AI应用开发的新手还是想系统化学习Agent架构的开发者都能从本文获得可直接复用的代码和清晰的工程思路。1. 背景与核心概念为什么需要LangChain LangGraph MCP在深入代码之前我们有必要厘清这几个核心组件分别解决了什么问题以及它们如何协同工作。1.1 LangChainAI应用开发的“脚手架”你可以把LangChain想象成乐高积木的基础连接件。它的核心价值在于标准化和模块化。解决了什么问题早期开发AI应用你需要直接与OpenAI、Anthropic等模型的原始API打交道处理复杂的提示词Prompt工程、上下文管理、输出解析等代码冗长且难以复用。核心能力LangChain提供了一整套高级抽象如LLMChain、PromptTemplate、OutputParser以及对各种工具Tools、向量数据库Vectorstores的集成。它让你能用声明式的方法组合AI能力大大提升了开发效率。类比就像使用Spring框架开发Java Web应用你不用从零处理HTTP连接和Servlet而是专注于业务逻辑。1.2 LangGraph为智能体注入“状态”和“流程”如果说LangChain提供了静态的积木那么LangGraph就是驱动这些积木按照特定规则运转的“发动机”和“流程图”。解决了什么问题传统的链Chain是线性的输入-输出就结束了。但真正的智能体Agent需要根据中间结果做判断、循环执行、处理分支甚至长期保持记忆State。这是简单Chain无法胜任的。核心能力LangGraph基于状态机State Graph的概念。你可以定义一组节点Nodes代表执行步骤和边Edges代表流转条件。智能体根据当前状态决定下一步走到哪个节点。这完美地支持了多轮对话、工具调用循环、复杂任务分解等场景。关键特性持久化状态。这意味着智能体的记忆可以跨越单次调用存在是实现“长期记忆”和复杂会话的基石。1.3 MCPModel Context Protocol打破工具集成的壁垒MCP是2024年由Anthropic等公司推动的一个新兴协议旨在标准化AI模型与外部工具/数据源之间的通信方式。解决了什么问题以前每个AI应用项目都需要自己写代码去连接数据库、调用API、读取文件。这些代码粘合在业务逻辑里难以移植和共享。同时像Claude Desktop、Cursor这类AI原生IDE也无法动态地发现和使用你项目里的工具。核心能力MCP定义了一套标准协议。你可以开发一个MCP Server来暴露你的工具如查询数据库、操作Git和数据源如公司知识库。任何兼容MCP的客户端如Claude Desktop、你的LangChain应用都可以动态发现并调用这些工具无需硬编码。与LangChain的关系LangChain的Tool概念是框架内的。MCP的Tool是协议级的更通用。新版LangChain已经集成了MCP客户端可以让你在LangGraph智能体中直接使用本地或远程MCP Server提供的工具实现了工具生态的共享和解耦。总结一下三者的关系LangChain提供构建AI应用的基础模块LangGraph在这些模块之上增加了有状态、可循环的编排能力用于构建复杂的智能体AgentMCP则从更底层提供了一套标准化的工具接入协议让智能体获取工具的能力更灵活、更强大。我们接下来的实战就是将这三者融合。2. 环境准备与版本说明本教程将在本地Python环境中进行确保你的环境满足以下要求。为了避免依赖冲突强烈建议使用虚拟环境如conda或venv。2.1 基础环境要求操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04)。本文命令以macOS/Linux的bash为例Windows用户可在PowerShell或WSL中运行。Python版本 3.10, 3.13。推荐使用Python 3.11或3.12兼容性最好。包管理工具pip (版本21.0)。2.2 创建虚拟环境与安装依赖首先创建一个项目目录并进入。mkdir langchain-mcp-agent-tutorial cd langchain-mcp-agent-tutorial使用venv创建虚拟环境并激活。# 创建虚拟环境 python -m venv .venv # 激活虚拟环境 # macOS/Linux: source .venv/bin/activate # Windows: # .venv\Scripts\activate安装核心依赖库。这里我们安装新版LangChain、LangGraph以及MCP相关的库。注意LangChain社区发展很快我们安装langchain-core,langchain-community和langchain元包并指定相对较新的版本。pip install langchain0.1.0 pip install langgraph0.0.50 pip install langchain-openai0.0.5 # 用于集成OpenAI模型 # 安装MCP客户端和服务器SDK pip install mcp1.0.0 pip install langchain-mcp0.1.0为了模拟工具调用我们还需要安装一些工具库并安装一个本地开源大模型如果你没有OpenAI API密钥。我们将使用ollama来运行本地模型。# 安装requests用于模拟API调用 pip install requests # 安装ollama请先确保已安装ollama客户端详见其官网 # 本教程假设你已安装并启动了ollama服务并拉取了模型例如 # ollama pull qwen2.5:7b2.3 验证安装与获取API密钥创建一个简单的Python脚本test_env.py来验证基础环境。# test_env.py import sys print(fPython版本: {sys.version}) try: import langchain print(fLangChain版本: {langchain.__version__}) except ImportError as e: print(f导入LangChain失败: {e}) try: import langgraph print(fLangGraph版本: {langgraph.__version__}) except ImportError as e: print(f导入LangGraph失败: {e}) try: import mcp print(fMCP版本: {mcp.__version__}) except ImportError as e: print(f导入MCP失败: {e})运行它python test_env.py如果一切正常你将看到各库的版本号。关于模型API密钥方案一推荐稳定使用OpenAI GPT-4o或GPT-3.5-turbo。你需要一个OpenAI API密钥。在代码中通过环境变量OPENAI_API_KEY设置。export OPENAI_API_KEY你的sk-xxx密钥 # Windows (PowerShell): $env:OPENAI_API_KEY你的sk-xxx密钥方案二本地免费使用Ollama运行本地大模型如Qwen2.5, Llama3.1。确保ollama服务已启动。ollama serve ollama pull qwen2.5:7b本教程后续代码将提供两种方式的示例。3. 核心组件拆解与基础用法在开始构建复杂智能体之前我们先快速过一遍每个核心组件的基础用法建立直观感受。3.1 LangChain 核心LCEL 与 Runnable新版LangChain全面转向了LCELLangChain Expression Language。LCEL使用管道符|将组件连接起来使链的构建像搭积木一样清晰。# 示例一个简单的提示词链 from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.output_parsers import StrOutputParser # 1. 定义提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的翻译官。), (user, 请将以下英文翻译成中文{text}) ]) # 2. 定义模型这里用Ollama本地模型示例 # 使用OpenAI API的写法 # llm ChatOpenAI(modelgpt-3.5-turbo) # 使用Ollama本地模型的写法 from langchain_community.llms import Ollama llm Ollama(modelqwen2.5:7b) # 3. 定义输出解析器 output_parser StrOutputParser() # 4. 使用 LCEL 组合链 chain prompt | llm | output_parser # 5. 调用链 result chain.invoke({text: Hello, LangChain! This is a tutorial about AI Agent.}) print(result) # 预期输出你好LangChain这是一个关于AI智能体的教程。关键点prompt | llm | output_parser构成了一个可执行的Runnable对象。LCEL让链的组装、调试如stream、batch变得非常统一。3.2 LangGraph 核心State 与 GraphLangGraph的核心是定义State和构建Graph。我们来看一个最简单的“聊天机器人”图它只是把用户输入原样返回。from typing import TypedDict, Annotated import operator from langgraph.graph import StateGraph, END # 1. 定义状态State。这是一个类型化的字典描述智能体运行时的所有数据。 class AgentState(TypedDict): # Annotated 用于在LangGraph中声明该字段的合并方式operator.add表示追加列表 messages: Annotated[list, operator.add] # 消息历史 user_input: str # 最新的用户输入 # 2. 定义节点函数。节点是图中的一个步骤它接收状态返回状态更新。 def call_model(state: AgentState): 模拟调用模型这里只是简单返回一个固定回复。 print(f模型节点收到用户输入: {state[user_input]}) # 这里应该调用真正的LLM我们先模拟 ai_message {role: assistant, content: f我收到了你的消息: {state[user_input]}} # 返回要更新到状态中的内容 return {messages: [ai_message]} # 3. 构建图 builder StateGraph(AgentState) # 添加节点命名为“model” builder.add_node(model, call_model) # 设置入口点从哪个节点开始 builder.set_entry_point(model) # 设置出口点执行完“model”节点后图就结束 builder.add_edge(model, END) # 编译图得到一个可执行的对象 graph builder.compile() # 4. 运行图 initial_state AgentState(messages[], user_input你好世界) result graph.invoke(initial_state) print(最终状态中的消息:, result[messages])这个图虽然简单但包含了所有要素定义状态结构、创建节点函数、编排节点顺序。复杂的Agent就是在这些节点中加入工具调用、条件判断等逻辑。3.3 MCP 核心Server 与 ToolMCP涉及Server和Client两端。我们先看一个最简单的MCP Server它只暴露一个工具。创建一个文件simple_mcp_server.py# simple_mcp_server.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.shared.exceptions import McpError # 创建MCP Server实例 server Server(simple-weather-server) # 使用装饰器定义一个工具获取天气 server.list_tools() async def handle_list_tools(): # 返回服务器提供的所有工具列表 return [ { name: get_weather, description: 获取指定城市的当前天气情况。, inputSchema: { type: object, properties: { city: {type: string, description: 城市名称例如Beijing, Shanghai} }, required: [city] } } ] # 使用装饰器处理工具调用 server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_weather: city arguments.get(city, 未知城市) # 这里应该是真实的API调用我们模拟返回 return { content: [ { type: text, text: f模拟天气数据{city}今天晴气温22-28°C微风。 } ] } raise McpError(f未知工具: {name}) async def main(): # 使用标准输入输出与客户端通信这是MCP Server最常见的启动方式 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): session ClientSession(read_stream, write_stream) await server.run( session, InitializationOptions( server_namesimple-weather-server, server_version0.1.0, capabilitiesserver.get_capabilities(NotificationOptions()) ) ) if __name__ __main__: asyncio.run(main())这个Server定义了一个get_weather工具。任何MCP客户端包括我们即将用LangChain构建的Agent都可以连接这个Server并调用该工具。运行这个Server它会阻塞等待客户端连接python simple_mcp_server.py在另一个终端我们可以写一个简单的MCP客户端来测试它。但更常见的是在LangChain中直接集成MCP Client。4. 完整实战构建天气查询与数据分析智能体现在我们将把所有知识融合构建一个功能更丰富的智能体。这个智能体能够理解用户意图判断用户是想查天气还是做数据分析。调用MCP工具如果查天气调用我们自建的MCP Server工具。执行计算如果做数据分析调用Python计算工具。管理对话状态记住历史对话进行多轮交互。4.1 项目结构langchain-mcp-agent-tutorial/ ├── .venv/ # 虚拟环境目录 ├── tools/ # 工具模块目录 │ ├── __init__.py │ └── weather_tools.py # 更完善的天气MCP Server ├── agents/ # 智能体模块目录 │ ├── __init__.py │ └── weather_agent.py # 主智能体图定义 ├── config.py # 配置文件 ├── requirements.txt # 依赖列表 └── main.py # 主程序入口4.2 创建增强版MCP天气工具服务器我们创建一个功能更真实的MCP Server它模拟调用一个天气API。文件tools/weather_tools.py# tools/weather_tools.py import asyncio import random from datetime import datetime from mcp import ClientSession from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.shared.exceptions import McpError server Server(enhanced-weather-server) # 模拟一些城市的天气数据 WEATHER_DATA { beijing: {city: 北京, condition: 晴朗, temp: 25, humidity: 40}, shanghai: {city: 上海, condition: 多云, temp: 28, humidity: 65}, shenzhen: {city: 深圳, condition: 阵雨, temp: 30, humidity: 80}, new york: {city: 纽约, condition: 阴天, temp: 20, humidity: 70}, } server.list_tools() async def handle_list_tools(): return [ { name: get_current_weather, description: 获取指定城市的当前天气详情包括温度、湿度和天气状况。, inputSchema: { type: object, properties: { location: { type: string, description: 城市名称支持中文或英文如北京、Shanghai } }, required: [location] } }, { name: get_weather_forecast, description: 获取指定城市未来几天的天气预报。, inputSchema: { type: object, properties: { location: { type: string, description: 城市名称 }, days: { type: integer, description: 预报天数默认为3天, default: 3 } }, required: [location] } } ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): location arguments.get(location, ).strip().lower() if name get_current_weather: # 查找或生成模拟数据 city_key next((k for k in WEATHER_DATA.keys() if k in location), None) if city_key: data WEATHER_DATA[city_key] else: # 对于未知城市生成随机数据 data { city: location.title(), condition: random.choice([晴朗, 多云, 小雨, 阴天]), temp: random.randint(15, 35), humidity: random.randint(30, 90) } report ( f{data[city]}当前天气{data[condition]}。 f温度{data[temp]}°C湿度{data[humidity]}%。 f数据更新时间{datetime.now().strftime(%Y-%m-%d %H:%M:%S)} ) return {content: [{type: text, text: report}]} elif name get_weather_forecast: days arguments.get(days, 3) forecasts [] for i in range(1, days 1): date (datetime.now() timedelta(daysi)).strftime(%Y-%m-%d) condition random.choice([晴, 多云, 阴, 小雨, 中雨]) high random.randint(20, 35) low high - random.randint(5, 10) forecasts.append(f{date}: {condition}{low}-{high}°C) report f{location.title()}未来{days}天预报\n \n.join(forecasts) return {content: [{type: text, text: report}]} raise McpError(f工具 {name} 未找到或参数错误。) async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): session ClientSession(read_stream, write_stream) await server.run( session, InitializationOptions( server_nameenhanced-weather-server, server_version0.2.0, capabilitiesserver.get_capabilities(NotificationOptions()) ) ) if __name__ __main__: asyncio.run(main())这个Server提供了get_current_weather和get_weather_forecast两个工具模拟了真实的天气查询功能。4.3 定义智能体状态与工具现在我们来构建智能体的核心。文件agents/weather_agent.py# agents/weather_agent.py from typing import TypedDict, Annotated, List, Literal, Union import operator from langchain_core.messages import HumanMessage, AIMessage, ToolMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langgraph.graph import StateGraph, END from langgraph.prebuilt import ToolExecutor, ToolInvocation from langgraph.checkpoint.aiosqlite import AsyncSqliteSaver from langchain_community.tools import DuckDuckGoSearchRun from langchain.agents import load_tools import json # ---------- 1. 定义更丰富的状态 ---------- class AgentState(TypedDict): 智能体的完整状态 # 完整的对话消息历史 messages: Annotated[List[Union[HumanMessage, AIMessage, ToolMessage]], operator.add] # 用户的最新输入 user_input: str # 智能体下一步应该做什么是继续调用工具还是直接回复用户 next_action: Literal[respond_to_user, call_tool] # 记录上一次工具调用的ID用于匹配结果 last_tool_call_id: str # ---------- 2. 加载工具 ---------- # 注意这里我们混合使用三种工具 # 1. MCP工具来自我们启动的Server # 2. LangChain社区工具如DuckDuckGo搜索 # 3. 自定义Python函数工具 # 首先定义一个简单的计算器工具自定义函数工具 from langchain.tools import tool tool def calculate(expression: str) - str: 执行一个数学表达式计算并返回结果。支持加减乘除和括号。 try: # 警告实际生产中应对表达式做严格安全检查这里仅为演示 result eval(expression) return f计算结果: {expression} {result} except Exception as e: return f计算错误: {e} # 加载DuckDuckGo搜索工具需要安装duckduckgo-search包 # pip install duckduckgo-search search_tool DuckDuckGoSearchRun() # 工具列表 tools [calculate, search_tool] # 创建工具执行器 tool_executor ToolExecutor(tools) # ---------- 3. 定义提示词 ---------- system_prompt 你是一个专业的天气与数据分析助手。你可以 1. 使用工具查询世界各地城市的当前天气和天气预报。 2. 使用计算器工具进行数学运算。 3. 使用网络搜索工具回答实时性问题。 4. 进行一般的对话和问答。 请根据用户的问题决定是否需要使用工具以及使用哪个工具。 如果你已经通过工具获得了足够信息请直接给出友好、清晰的回答。 如果用户的问题不明确请礼貌地追问。 保持对话的连贯性记住我们之前的对话历史。 prompt ChatPromptTemplate.from_messages([ (system, system_prompt), MessagesPlaceholder(variable_namemessages), # 自动注入历史消息 (user, {user_input}), ]) # ---------- 4. 初始化模型 ---------- # 方案一使用Ollama本地模型 from langchain_community.llms import Ollama llm Ollama(modelqwen2.5:7b, temperature0.1) # 将工具绑定到模型这样模型才知道它可以调用哪些工具 llm_with_tools llm.bind_tools(tools) # 方案二使用OpenAI API取消注释下面几行并注释掉上面的Ollama部分 # from langchain_openai import ChatOpenAI # llm ChatOpenAI(modelgpt-3.5-turbo, temperature0.1) # llm_with_tools llm.bind_tools(tools) # ---------- 5. 定义图节点 ---------- def should_continue(state: AgentState) - Literal[call_tool, __end__]: 条件判断函数决定下一步是调用工具还是结束。 # 如果上一步设定了要调用工具就继续调用 if state[next_action] call_tool: return call_tool # 否则结束这一轮对话 return __end__ def agent_node(state: AgentState): 智能体节点分析用户输入决定行动。 # 准备输入给模型的消息 formatted_messages prompt.invoke({ messages: state[messages], user_input: state[user_input] }) # 调用模型获取响应。模型可能会返回工具调用的请求。 response llm_with_tools.invoke(formatted_messages) # 将模型的响应添加到消息历史中 new_messages [response] # 检查模型的响应中是否包含工具调用请求 tool_calls [] if hasattr(response, tool_calls) and response.tool_calls: tool_calls response.tool_calls # 如果模型要求调用工具我们设置下一步动作为调用工具 next_action call_tool # 记录工具调用ID用于后续匹配结果 last_tool_call_id response.tool_calls[0][id] if response.tool_calls else else: # 如果模型直接回复了文本则下一步是响应用户本轮结束 next_action respond_to_user last_tool_call_id return { messages: new_messages, next_action: next_action, last_tool_call_id: last_tool_call_id } def tool_node(state: AgentState): 工具执行节点执行模型请求的工具调用。 # 从最新的AI消息中提取工具调用信息 last_message state[messages][-1] if not hasattr(last_message, tool_calls) or not last_message.tool_calls: # 如果没有工具调用直接返回空更新 return {messages: []} # 执行每一个工具调用 tool_messages [] for tool_call in last_message.tool_calls: # 构建工具调用对象 action ToolInvocation( tooltool_call[name], tool_inputtool_call[args] ) # 执行工具 try: output tool_executor.invoke(action) except Exception as e: output f工具调用出错: {e} # 创建工具返回消息 tool_message ToolMessage( contentstr(output), tool_call_idtool_call[id], nametool_call[name] ) tool_messages.append(tool_message) # 工具执行后下一步应该由模型来总结工具结果并回复用户 return { messages: tool_messages, next_action: respond_to_user } # ---------- 6. 构建并编译图 ---------- def create_agent_graph(): 创建并返回编译好的智能体图 builder StateGraph(AgentState) # 添加节点 builder.add_node(agent, agent_node) builder.add_node(tool, tool_node) # 设置入口点 builder.set_entry_point(agent) # 定义边路由逻辑 builder.add_conditional_edges( agent, should_continue, # 条件判断函数 { call_tool: tool, # 如果需要调用工具转到tool节点 __end__: END # 如果直接回复则结束 } ) builder.add_edge(tool, agent) # 工具执行完后回到agent节点分析结果 # 编译图 graph builder.compile() # 可选启用检查点以实现长期记忆 # memory AsyncSqliteSaver.from_conn_string(:memory:) # graph builder.compile(checkpointermemory) return graph # 导出创建好的图 agent_graph create_agent_graph()4.4 主程序与MCP集成现在我们创建主程序main.py它将启动MCP Server在子进程连接MCP工具并运行我们的智能体图进行对话。# main.py import asyncio import subprocess import sys import time from threading import Thread from agents.weather_agent import agent_graph, AgentState from langchain_core.messages import HumanMessage def start_mcp_server(): 在一个子进程中启动我们的MCP天气工具服务器 # 注意这里我们启动的是之前写的增强版Server server_process subprocess.Popen( [sys.executable, tools/weather_tools.py], stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 给服务器一点时间启动 time.sleep(2) if server_process.poll() is not None: stderr server_process.stderr.read() print(fMCP Server启动失败: {stderr}) return None print(MCP天气工具服务器已启动。) return server_process async def run_agent_conversation(): 运行与智能体的对话 print(\n *50) print(AI天气与数据分析助手已启动) print(输入 quit 或 exit 结束对话。) print(*50) # 初始化状态 state AgentState( messages[], # 初始消息历史为空 user_input, next_actionrespond_to_user, last_tool_call_id ) while True: try: # 获取用户输入 user_input input(\n 你: ).strip() if user_input.lower() in [quit, exit, q]: print(对话结束。) break if not user_input: continue # 更新状态中的用户输入 state[user_input] user_input # 调用智能体图 result agent_graph.invoke(state) # 更新状态为最新结果 state result # 打印智能体的最新回复 # 查找最后一条来自AI的、非工具调用的消息 for msg in reversed(result[messages]): if isinstance(msg, AIMessage) and not hasattr(msg, tool_calls): print(f\n 助手: {msg.content}) break elif isinstance(msg, ToolMessage): # 也可以选择显示工具调用的原始结果通常太冗长 # print(f[工具调用结果: {msg.content[:100]}...]) pass except KeyboardInterrupt: print(\n\n对话被中断。) break except Exception as e: print(f\n运行时出错: {e}) import traceback traceback.print_exc() def main(): # 启动MCP Server在后台线程 server_process start_mcp_server() if server_process is None: print(无法启动MCP Server部分工具可能不可用。) try: # 运行主对话循环 asyncio.run(run_agent_conversation()) finally: # 确保清理MCP Server进程 if server_process: server_process.terminate() server_process.wait() print(MCP服务器已关闭。) if __name__ __main__: main()4.5 运行与验证确保所有文件就位并且虚拟环境已激活依赖已安装。运行主程序python main.py进行对话测试测试天气查询“北京今天天气怎么样”或“上海未来三天的天气预报。”测试计算工具“计算一下(1527)*3除以2等于多少”测试搜索工具“搜索一下LangGraph的最新版本是什么”需要网络测试多轮对话先问“深圳的湿度高吗”再问“那温度呢”看智能体是否能记住上下文。一个成功的运行示例如下 AI天气与数据分析助手已启动 输入 quit 或 exit 结束对话。 你: 北京今天天气怎么样 助手: 我将为您查询北京的当前天气。 此时模型决定调用get_current_weather工具工具节点执行后返回结果模型再总结结果回复 助手: 北京当前天气晴朗。温度25°C湿度40%。数据更新时间2024-01-01 10:30:00 你: 计算(1218)*5的值 助手: 计算结果: (1218)*5 1505. 常见问题与排查思路在开发过程中你可能会遇到以下典型问题问题现象可能原因排查思路与解决方案导入错误ModuleNotFoundError1. 虚拟环境未激活或依赖未安装。2. 包名错误如langchainvslangchain-core。1. 运行pip list检查关键包是否存在。2. 确认使用pip install langchain安装元包它会安装核心依赖。3. 对于社区工具如duckduckgo-search需单独安装。运行MCP Server报错或无法连接1. 端口冲突或stdio通信问题。2. MCP协议版本不兼容。3. Server脚本语法错误。1. 确保没有其他进程占用标准输入输出。2. 检查mcp库版本尽量使用较新版本。3. 单独运行python tools/weather_tools.py看是否有Python语法错误。智能体不调用工具总是直接回复1. 模型能力不足特别是小参数本地模型。2. 工具描述description不够清晰。3. 提示词system_prompt未明确指示使用工具。1. 尝试换用更强的模型如GPT-4o、Claude 3.5 Sonnet或更大的本地模型。2. 优化工具描述明确输入输出格式。3. 在system_prompt中更强烈地要求模型“必须使用工具”。4. 使用bind_tools时确保工具列表正确传递。Agent execution terminated due to error1. 工具调用时参数格式错误。2. 工具函数本身抛出异常。3. 状态State结构在节点间传递时出现类型不匹配。1. 在tool_node函数中添加更详细的异常捕获和日志。2. 检查工具函数的输入参数类型是否与模型调用时匹配。3. 使用print或日志打印每个节点前后的state进行调试。Ollama模型加载慢或报错1. 模型未下载ollama pull。2. Ollama服务未启动。3. 内存不足。1. 运行ollama list确认模型存在。2. 运行ollama serve启动服务。3. 在代码中初始化Ollama时指定正确的base_url默认是http://localhost:11434。对话没有记忆多轮对话失效1. 状态中的messages字段没有正确累积。2. 图在每次调用时都被重新初始化状态重置。1. 确认状态定义中messages使用了Annotated[list, operator.add]。2. 考虑使用LangGraph的检查点Checkpointer如AsyncSqliteSaver来实现跨会话的持久化记忆。工具调用结果没有被模型正确处理1.ToolMessage格式不正确。2. 模型在收到工具结果后提示词未引导其总结。1. 确保ToolMessage包含了正确的tool_call_id。2. 在system_prompt中明确要求模型“根据工具返回的结果给出最终答案”。6. 最佳实践与工程建议将原型转化为稳定、可维护的项目需要遵循以下工程实践配置化管理将模型类型、API密钥、服务器地址等抽离到配置文件如config.py或.env文件中使用pydantic-settings进行管理。# config.py from pydantic_settings import BaseSettings class Settings(BaseSettings): openai_api_key: str ollama_base_url: str http://localhost:11434 ollama_model: str qwen2.5:7b mcp_server_command: list [python, tools/weather_tools.py]结构化日志使用logging模块替代print为不同模块设置不同日志级别方便调试和监控。import logging logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) logger.info(智能体开始处理用户输入: %s, user_input)完善的错误处理在工具调用、模型调用、图执行等关键环节添加try-except并给出用户友好的错误提示避免程序崩溃。try: response llm_with_tools.invoke(formatted_messages) except Exception as e: logger.error(模型调用失败: %s, e) # 返回一个兜底的AI消息 return {messages: [AIMessage(content抱歉我暂时无法处理您的请求。)], ...}工具开发的注意事项安全性像calculate工具中使用eval是极度危险的生产环境必须替换为安全的表达式解析库如ast.literal_eval或numexpr。幂等性工具函数应尽可能设计为幂等的即相同输入产生相同输出避免副作用。超时与重试对于网络请求类工具必须设置超时和重试机制。智能体设计模式监督节点Supervisor对于复杂任务可以引入一个“监督”节点将大任务拆解为子任务分发给不同的子智能体Worker处理再汇总结果。LangGraph官方提供了StateGraph和MessagesState来简化这种模式。人工干预Human-in-the-Loop在关键节点如执行删除操作、确认重大决策设置中断等待人工确认后再继续。这可以通过在图中添加一个interrupt节点来实现。性能优化异步化LangGraph和MCP天生支持异步。将节点函数定义为async def并使用ainvoke可以大幅提升I/O密集型工具如网络请求的并发性能。缓存对LLM的响应、工具查询结果进行适当缓存减少重复计算和API调用。流式输出使用stream模式调用模型和图可以实现token-by-token的流式响应提升用户体验。生产环境部署MCP Server应作为独立的守护进程运行并通过进程管理工具如systemd, supervisor进行管理。智能体服务可以封装为FastAPI或Gradio Web应用提供HTTP接口。状态持久化使用AsyncSqliteSaver或RedisSaver将检查点存入数据库实现智能体状态的长期保存和恢复。掌握LangChain、LangGraph和MCP的组合你就掌握了构建下一代AI原生应用的核心范式。从简单的提示词链到有状态、可工具调用的智能体再到通过标准化协议集成无限可能的外部工具这条路径清晰地指向了更强大、更自主的AI应用未来。建议你以本教程的代码为起点尝试添加更多工具如数据库查询、发送邮件、操作文件或实现更复杂的图逻辑如循环审核、多智能体协作在实践中不断深化理解。
