1. 先搞清楚 LangChain、MCP、LangGraph 和 Agent 到底能帮你做什么
如果你刚开始接触 AI 应用开发,看到 LangChain、MCP、LangGraph、Agent 这些词,第一反应可能是“概念好多,无从下手”。这很正常,因为每个词都代表一个不同的层次和工具。别急着去背定义,我们先从最实际的问题出发:它们合在一起,能帮你解决什么具体问题?
简单来说,这套组合能让你用代码快速搭建一个能“思考”和“行动”的 AI 应用。这里的“思考”指的是让大语言模型(比如 GPT、Claude 或本地部署的模型)理解你的指令、规划步骤;“行动”指的是让模型能调用外部工具,比如查数据库、读文件、调用 API、执行计算。而 LangChain、MCP、LangGraph 就是帮你把“思考”和“行动”组织起来的脚手架。
- LangChain:是基础框架。它提供了和大模型对话、管理对话历史(记忆)、以及连接各种工具(Tools)的标准方法。你可以把它想象成乐高积木的底板和基础连接件。
- MCP(Model Context Protocol):是工具连接协议。它定义了一种标准方式,让你开发的 AI 应用能安全、规范地调用外部工具(比如一个查询天气的 API,或者一个读取本地文件的函数)。MCP 解决了“如何让模型安全地使用工具”这个核心问题。
- LangGraph:是高级流程控制器。当你的 AI 应用逻辑变复杂,需要根据模型输出的结果决定下一步做什么(比如先查天气,再根据天气决定推荐室内还是室外活动),这种带“分支”和“循环”的流程,用基础的 LangChain 链(Chain)写起来会很别扭。LangGraph 允许你用“图”的方式来定义这种有状态的、多步骤的工作流,让复杂 Agent 的逻辑变得清晰可控。
- Agent:是最终呈现的智能体。它是基于以上所有组件构建出来的、能够自主理解目标、规划并执行一系列工具调用以完成任务的 AI 程序。一个强大的 Agent 背后,通常离不开 LangGraph 的流程编排和 MCP 的工具支持。
所以,这个教程的核心价值是:从零开始,手把手教你用这些业界主流工具,搭建一个真正能跑起来的、功能清晰的 AI Agent,而不仅仅是跑通一个“Hello World”的对话示例。你会学到如何组织代码、如何连接工具、如何控制执行流程,以及如何排查那些让新手头疼的典型错误。
2. 环境准备:别在依赖和版本上踩第一个坑
在写第一行业务代码之前,把环境理顺能避免 80% 的莫名报错。我们不追求最新版本,而是追求一个稳定、兼容的起步环境。
2.1 基础 Python 环境
我强烈建议使用Python 3.10 或 3.11。Python 3.12 对一些库的兼容性可能还在完善中,新手先避开。使用conda或venv创建独立的虚拟环境是必须的。
# 使用 conda 创建环境(推荐) conda create -n langchain-demo python=3.11 conda activate langchain-demo # 或者使用 venv python -m venv venv # Windows venv\Scripts\activate # macOS/Linux source venv/bin/activate2.2 核心库安装
通过 pip 安装核心库。注意,langchain是一个元包,我们通常需要安装更具体的子包和社区集成包。
pip install langchain langchain-community langchain-core关键解释:
langchain-core: 包含最核心的抽象基类和运行时。大部分情况下你通过其他包间接使用它。langchain: 包含标准接口、链和基础工具的实现。langchain-community: 这是非常重要的包,包含了大量第三方工具的集成(比如与各种数据库、API 的连接器)。很多教程里提到的工具(Tool)都来自这里。
接下来安装 LangGraph,它是构建复杂 Agent 工作流的关键:
pip install langgraph对于 MCP,目前它更像一个协议标准和一套开发工具集。你可能需要安装mcp客户端库或相关 SDK 来创建或连接 MCP 服务器。由于 MCP 生态在快速演进,一个稳妥的起步方式是关注 LangChain 官方对 MCP 的支持。通常,你可以通过langchain社区工具来调用符合 MCP 协议的工具。
2.3 模型访问准备
你需要一个能够访问的大语言模型。有两种主要路径:
使用云端 API(如 OpenAI, Anthropic):最简单快捷,适合学习和原型开发。
pip install openai langchain-openai然后需要设置环境变量
OPENAI_API_KEY。export OPENAI_API_KEY='你的sk-...密钥' # Windows: set OPENAI_API_KEY=你的sk-...密钥使用本地模型(如通过 Ollama):更注重隐私和成本控制,适合深入研究和生产部署。
# 首先安装并启动 Ollama,从官网下载安装包 # 然后拉取一个模型,例如 Llama 3.1 ollama pull llama3.1:8b # 安装 LangChain 的 Ollama 集成 pip install langchain-ollama
新手建议:为了减少环境变量和网络问题的干扰,我强烈建议初学者先从本地 Ollama 模型开始。它能让你立刻聚焦于 LangChain 和 LangGraph 的代码逻辑本身,而不是卡在 API 密钥配置或网络连通性上。
2.4 初始化一个清晰的项目目录
不要把所有代码扔在一个文件里。建立清晰的目录结构有助于后续管理工具、工作流和配置。
your_agent_project/ ├── tools/ # 存放自定义工具类,或 MCP 服务器文件 │ └── weather_tool.py ├── workflows/ # 存放 LangGraph 工作流定义 │ └── travel_agent.py ├── config.py # 配置文件,存放模型、API密钥等设置 ├── main.py # 主入口文件 └── requirements.txt在requirements.txt中记录依赖:
langchain==0.1.0 langchain-community==0.0.10 langgraph==0.0.17 langchain-ollama==0.1.03. 从核心概念到第一个能跑的 Agent
现在,我们跳过理论深水区,直接通过代码来理解这几个核心组件是如何协作的。我们会构建一个简单的“旅行建议助手”Agent。
3.1 第一步:创建一个简单的工具(Tool)
工具是 Agent 的手和脚。我们先创建一个模拟的“获取天气”工具。
在tools/weather_tool.py中:
from langchain.tools import tool from typing import Optional @tool def get_weather(city: str, date: Optional[str] = None) -> str: """ 根据城市和日期查询天气信息。 如果没有提供日期,则返回当前天气。 Args: city: 城市名称,例如 "北京"。 date: 日期,格式为 YYYY-MM-DD。可选。 Returns: 返回该城市的天气描述字符串。 """ # 这是一个模拟函数,真实场景会调用天气API if date: return f"{city}在{date}的天气是晴朗,温度25°C。" else: return f"{city}当前天气是多云,温度22°C。"关键点:
- 使用
@tool装饰器,LangChain 能自动将其识别为一个可用的工具。 - 文档字符串(
""")非常重要!大语言模型会根据它来决定何时以及如何调用这个工具。 - 输入参数要有明确的类型提示和说明。
3.2 第二步:初始化模型和工具列表
在main.py中,我们开始组装 Agent。
import os from langchain_ollama import ChatOllama from tools.weather_tool import get_weather # 1. 初始化模型(使用本地 Ollama) model = ChatOllama(model="llama3.1:8b", temperature=0) # 2. 准备工具列表 tools = [get_weather] # 3. 将工具绑定到模型,创建一个“具备工具调用能力”的模型 model_with_tools = model.bind_tools(tools) print("模型和工具初始化完成。")运行一下python main.py,如果没有报错,说明基础环境 OK。
3.3 第三步:创建你的第一个简单 Agent(使用 LangChain 内置 AgentExecutor)
在main.py中继续:
from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain.prompts import ChatPromptTemplate # 4. 定义提示词模板,告诉 Agent 它的角色和能力 prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个有用的旅行助手。请根据用户的问题,使用工具来获取信息并给出回答。"), ("placeholder", "{chat_history}"), # 预留对话历史的位置 ("human", "{input}"), # 用户输入 ("placeholder", "{agent_scratchpad}"), # Agent 思考过程暂存处 ]) # 5. 创建 Agent agent = create_tool_calling_agent( llm=model_with_tools, prompt=prompt, tools=tools, ) # 6. 创建 Agent 执行器 agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True) # 7. 运行 Agent try: response = agent_executor.invoke({"input": "北京明天天气怎么样?"}) print("\n--- Agent 回答 ---") print(response["output"]) except Exception as e: print(f"执行出错: {e}")运行并观察: 执行python main.py。如果一切正常,你应该在控制台看到详细的verbose日志。它会展示类似这样的过程:
- Agent 接收到输入:“北京明天天气怎么样?”
- Agent(模型)思考后,决定调用
get_weather工具。 - 日志会显示它准备传入的参数:
city=“北京”, date=“明天对应的日期”。 - 工具被执行,返回模拟的天气结果。
- Agent 将工具结果整合,生成最终回答:“北京在YYYY-MM-DD的天气是晴朗,温度25°C。”
恭喜!你已经创建了一个最基本的、能根据问题自动选择并调用工具的 AI Agent。这个 Agent 的核心是 LangChain 的AgentExecutor,它帮你处理了“模型思考 -> 决定调用工具 -> 执行工具 -> 将结果返回给模型 -> 模型生成最终回答”的循环。
3.4 第四步:当简单链不够用,引入 LangGraph
上面的AgentExecutor对于线性任务很好用。但如果任务复杂呢?比如,用户问:“我想去一个温暖的海边城市度假,预算不高,有什么推荐吗?并告诉我那里下周的天气。”
这个任务需要:1) 查询符合“温暖”、“海边”、“预算低”条件的城市(可能需调用一个“城市推荐”工具)。2) 对推荐出的每个城市,调用“获取天气”工具。这是一个有条件分支和潜在循环的任务。
这时,AgentExecutor的线性控制流就显得力不从心。我们需要LangGraph。
LangGraph 核心思想:将工作流定义为一个“图”(Graph),图中的节点(Node)是执行步骤(可以是调用模型、运行工具、判断条件),边(Edge)决定了步骤之间的流转逻辑。
我们来构建一个简化版的“旅行规划”工作流。
在workflows/travel_agent.py中:
from typing import TypedDict, Annotated, List import operator from langgraph.graph import StateGraph, END from langchain_ollama import ChatOllama from langchain.prompts import ChatPromptTemplate # 1. 定义工作流的“状态”(State)。这是一个全局共享的数据结构。 class AgentState(TypedDict): # 用户原始问题 user_query: str # 模型/工具链产生的消息列表 messages: Annotated[List, operator.add] # 用于存储中间结果,如推荐的城市列表 recommended_cities: List[str] # 最终收集的天气信息 weather_info: List[str] # 2. 初始化模型和工具(这里复用之前的工具,假设我们还有一个 `recommend_city` 工具) model = ChatOllama(model="llama3.1:8b", temperature=0) # ... 假设已定义 get_weather 和 recommend_city 工具 ... # 3. 定义各个节点(Node)函数 def recommend_city_node(state: AgentState): """节点:调用工具推荐城市""" prompt = ChatPromptTemplate.from_messages([ ("system", "你是一个旅行规划助手。根据用户需求推荐城市。"), ("human", "{query}"), ]) chain = prompt | model.bind_tools([recommend_city_tool]) response = chain.invoke({"query": state["user_query"]}) # 这里需要解析 response,提取出推荐的城市列表,放入 state # 为简化,我们模拟结果 state["recommended_cities"] = ["三亚", "厦门"] state["messages"].append(response) return state def fetch_weather_node(state: AgentState): """节点:为每个推荐城市获取天气""" weather_results = [] for city in state["recommended_cities"]: # 调用天气工具 weather = get_weather.invoke({"city": city, "date": "下周"}) # 简化日期 weather_results.append(f"{city}: {weather}") state["weather_info"] = weather_results state["messages"].append(("assistant", f"已获取天气信息: {weather_results}")) return state def generate_final_answer_node(state: AgentState): """节点:整合信息,生成最终回答""" final_prompt = f""" 用户问题:{state['user_query']} 推荐城市:{state['recommended_cities']} 这些城市下周天气:{state['weather_info']} 请生成一份友好的旅行建议总结。 """ chain = ChatPromptTemplate.from_messages([("human", final_prompt)]) | model final_response = chain.invoke({}) state["messages"].append(("assistant", final_response.content)) return state # 4. 构建图(Graph) workflow = StateGraph(AgentState) # 添加节点 workflow.add_node("recommend_city", recommend_city_node) workflow.add_node("fetch_weather", fetch_weather_node) workflow.add_node("generate_answer", generate_final_answer_node) # 设置边的流转逻辑 workflow.set_entry_point("recommend_city") # 从推荐城市开始 workflow.add_edge("recommend_city", "fetch_weather") # 推荐完就去查天气 workflow.add_edge("fetch_weather", "generate_answer") # 查完天气就生成答案 workflow.add_edge("generate_answer", END) # 生成答案后结束 # 编译图 app = workflow.compile() # 5. 运行这个 LangGraph 工作流 if __name__ == "__main__": initial_state = { "user_query": "我想去一个温暖的海边城市度假,预算不高,有什么推荐吗?并告诉我那里下周的天气。", "messages": [], "recommended_cities": [], "weather_info": [] } final_state = app.invoke(initial_state) for message in final_state["messages"]: if isinstance(message, tuple): print(f"{message[0]}: {message[1]}") else: print(message.content if hasattr(message, 'content') else message)这个例子展示了 LangGraph 如何将复杂任务分解为清晰的步骤(节点),并控制执行流。你可以看到,它比单一的AgentExecutor更灵活,可以轻松扩展(例如,增加一个“判断预算是否足够”的条件节点,根据结果决定是继续推荐还是直接结束)。
4. 深入实战:连接 MCP 工具与处理常见错误
4.1 如何理解和使用 MCP?
MCP 的目标是标准化工具调用。在实践中,你可能会遇到两种角色:
- MCP 服务器(Server):提供工具的一方。它将工具的功能通过 MCP 协议暴露出来。
- MCP 客户端(Client):使用工具的一方。你的 LangChain Agent 可以作为客户端去连接 MCP 服务器。
对于初学者,一个更实用的切入点是:许多符合 MCP 协议的工具,已经可以通过langchain-community中的集成来方便地使用。你不需要从零开始搭建 MCP 服务器。
例如,假设有一个公开的“天气 MCP 服务器”,你可能可以这样连接(示例代码,具体取决于工具实现):
# 伪代码,展示概念 from langchain_community.tools.mcp import MCPTool # 配置连接到 MCP 服务器的信息 weather_mcp_tool = MCPTool( server_url="http://weather-mcp-server:8000", tool_name="get_weather" ) tools.append(weather_mcp_tool)然后,这个weather_mcp_tool就可以像我们之前自定义的get_weather工具一样,被bind_tools绑定,并被 Agent 调用。
现阶段建议:先掌握如何创建和使用自定义的@tool,理解工具调用的流程。当需要集成更复杂、更标准化的外部服务时,再去深入研究如何部署或连接特定的 MCP 服务器。
4.2 你必须知道的常见错误与排查清单
在开发过程中,你几乎一定会遇到下面这些错误。别慌,按顺序排查。
错误1:Agent stopped due to iteration limit or time limit.或Agent execution terminated due to error.
- 原因:这是最常见的问题。Agent 陷入了“思考-调用-思考”的循环,或者在某一步出错了。
- 排查:
- 开启
verbose=True:这是最重要的调试手段,查看 Agent 每一步的思考和工具调用输出。 - 检查工具描述:模型的“思考”完全依赖于工具的文档字符串。确保你的
@tool函数下的"""描述清晰、准确地说明了工具的功能、输入和输出。描述不清会导致模型错误调用或反复调用。 - 简化问题:用一个最简单的问题(如“今天天气如何?”)测试,看是否能走通单次工具调用。
- 检查模型输出:在
verbose日志中,看模型是否输出了一个格式正确的tool_calls对象。如果没有,可能是提示词(Prompt)不够清晰,没有“教会”模型使用工具。
- 开启
错误2:Context size exceeded...(上下文过长)
- 原因:对话历史(
chat_history)或中间过程太长,超过了模型的最大上下文长度。 - 排查与解决:
- 使用
verbose确认:看看是不是每次调用都把大量历史信息传给了模型。 - 精简历史:对于 LangGraph,确保你的
State设计是高效的,只保留必要信息。对于AgentExecutor,可以考虑使用ConversationBufferWindowMemory来只保留最近几轮对话。 - 总结历史:对于长对话,可以实现一个“总结”节点(在 LangGraph 中),定期将冗长的历史压缩成摘要。
- 使用
错误3:工具调用失败,返回非预期结果或异常
- 原因:工具函数本身执行出错,或者返回的数据格式让模型无法理解。
- 排查:
- 独立测试工具:在 Agent 之外,直接调用你的工具函数,传入各种参数,看它是否能正确返回。
- 检查输入参数:在
verbose日志中,查看模型传给工具的参数值是否正确。类型错误、格式错误是常见原因。 - 工具返回需为字符串:
@tool装饰的函数必须返回字符串(或可转换为字符串的对象)。如果返回复杂字典或对象,模型可能无法处理。
错误4:ModuleNotFoundError: No module named 'langchain_xxx'
- 原因:包没安装对。LangChain 生态的包名经常变化。
- 解决:
- 使用
pip list | grep langchain查看已安装的包。 - 仔细核对官方文档或教程中使用的包名。
langchain-community,langchain-openai,langchain-ollama等都是独立的包。
- 使用
错误5:LangGraph 工作流卡住或不按预期执行
- 原因:图的边(Edge)逻辑定义有误,或者某个节点函数没有正确修改或返回
state。 - 排查:
- 可视化你的图:LangGraph 支持将工作流导出为图片,这是调试的神器。
from langgraph.graph import StateGraph # ... 构建你的 workflow ... app = workflow.compile() # 导出为 PNG app.get_graph().draw_mermaid_png(output_file_path="my_workflow.png") - 检查节点返回值:每个节点函数都必须返回更新后的
state字典。 - 检查条件边:如果你使用了
add_conditional_edges,确保你的条件函数返回的下一个节点名称是图中存在的。
- 可视化你的图:LangGraph 支持将工作流导出为图片,这是调试的神器。
5. 从 Demo 到项目:架构与进阶思考
当你跑通第一个 Agent 后,下一步就是思考如何把它变成一个可维护、可扩展的项目。
5.1 项目结构优化
回顾第 2.4 节的目录,并进一步细化:
agents/:存放不同功能的 Agent 定义(使用create_tool_calling_agent创建的部分)。chains/:存放一些可复用的简单链(Prompt + Model)。tools/:按领域分类存放工具,如tools/weather/,tools/search/。复杂的工具可以考虑封装成简单的 MCP 服务器。config/:使用pydantic或python-dotenv管理配置,区分开发和生产环境。tests/:为你的工具和关键工作流节点编写单元测试。
5.2 生产环境考量
稳定性:
- 错误处理与重试:在工具调用和模型调用层添加重试逻辑(如使用
tenacity库)。 - 超时控制:为每个工具调用和模型调用设置超时,避免单个步骤卡死整个 Agent。
- 降级方案:当核心工具(如天气 API)失败时,是否有备用数据源或友好的默认回复?
- 错误处理与重试:在工具调用和模型调用层添加重试逻辑(如使用
可观测性:
- 结构化日志:不要只依赖
verbose=True。集成像structlog这样的库,将 Agent 的执行步骤、工具调用参数和结果、耗时等信息以 JSON 格式记录下来,方便后续监控和分析。 - 追踪(Tracing):使用 LangSmith(LangChain 官方平台)或 OpenTelemetry 来可视化整个 Agent 的调用链,这对于调试复杂工作流至关重要。
- 结构化日志:不要只依赖
性能:
- 缓存:对模型响应和工具结果进行适当缓存(例如,相同城市一小时内不再重复查询天气),减少开销和延迟。
- 异步:如果工作流中多个步骤可以并行(例如,为多个城市同时查询天气),考虑使用 LangGraph 的异步支持或
asyncio来提升效率。
5.3 关于 MCP 的深入方向
当你需要让 Agent 使用公司内部系统、特定数据库或复杂 API 时,MCP 的价值就凸显了。
- 开发 MCP 服务器:你可以用任何语言(Python, JavaScript, Go 等)编写一个符合 MCP 协议的服务器,将内部能力“工具化”。
- 安全性:MCP 协议设计时考虑了安全性,比如工具调用的权限控制、输入输出审计。在生产中,这是连接企业内部工具时必须评估的。
- 动态工具发现:高级的 Agent 可以在运行时从 MCP 服务器动态发现可用的新工具,而无需重启应用。
5.4 持续学习路径
- 官方文档是第一位:
docs.langchain.com和langchain-ai.github.io/langgraph/是核心资料。先看概念指南(Concepts),再查 API 参考。 - 从模板和案例入手:LangChain 和 LangGraph 的 GitHub 仓库有大量示例(
cookbook)。克隆下来,运行并修改它们,比从头开始写更快。 - 加入社区:遇到具体错误时,在 LangChain Discord 或相关 GitHub Issues 中搜索,很多坑已经有人踩过。
- 关注演进:这个领域迭代极快。关注核心库的 Release Notes,了解新特性和不兼容的变更。
最后,也是最关键的建议:不要试图一次性构建一个全能的超级 Agent。从一个具体、微小但完整的问题开始(比如“查询天气并建议是否带伞”),把它做透,跑通“用户输入 -> 模型思考 -> 工具调用 -> 结果整合 -> 输出回答”的完整闭环。在这个小闭环中,你会遇到并解决 90% 的基础问题。之后,再逐步增加工具、引入 LangGraph 处理复杂流程、考虑连接 MCP 服务器,你的 AI 应用开发之路就会清晰而扎实。