当前位置: 首页 > news >正文

从零构建AI Agent:LangChain、LangGraph与MCP实战指南

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 对一些库的兼容性可能还在完善中,新手先避开。使用condavenv创建独立的虚拟环境是必须的。

# 使用 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/activate

2.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 模型访问准备

你需要一个能够访问的大语言模型。有两种主要路径:

  1. 使用云端 API(如 OpenAI, Anthropic):最简单快捷,适合学习和原型开发。

    pip install openai langchain-openai

    然后需要设置环境变量OPENAI_API_KEY

    export OPENAI_API_KEY='你的sk-...密钥' # Windows: set OPENAI_API_KEY=你的sk-...密钥
  2. 使用本地模型(如通过 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.0

3. 从核心概念到第一个能跑的 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日志。它会展示类似这样的过程:

  1. Agent 接收到输入:“北京明天天气怎么样?”
  2. Agent(模型)思考后,决定调用get_weather工具。
  3. 日志会显示它准备传入的参数:city=“北京”, date=“明天对应的日期”
  4. 工具被执行,返回模拟的天气结果。
  5. 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 的目标是标准化工具调用。在实践中,你可能会遇到两种角色:

  1. MCP 服务器(Server):提供工具的一方。它将工具的功能通过 MCP 协议暴露出来。
  2. 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 陷入了“思考-调用-思考”的循环,或者在某一步出错了。
  • 排查
    1. 开启verbose=True:这是最重要的调试手段,查看 Agent 每一步的思考和工具调用输出。
    2. 检查工具描述:模型的“思考”完全依赖于工具的文档字符串。确保你的@tool函数下的"""描述清晰、准确地说明了工具的功能、输入和输出。描述不清会导致模型错误调用或反复调用。
    3. 简化问题:用一个最简单的问题(如“今天天气如何?”)测试,看是否能走通单次工具调用。
    4. 检查模型输出:在verbose日志中,看模型是否输出了一个格式正确的tool_calls对象。如果没有,可能是提示词(Prompt)不够清晰,没有“教会”模型使用工具。

错误2:Context size exceeded...(上下文过长)

  • 原因:对话历史(chat_history)或中间过程太长,超过了模型的最大上下文长度。
  • 排查与解决
    1. 使用verbose确认:看看是不是每次调用都把大量历史信息传给了模型。
    2. 精简历史:对于 LangGraph,确保你的State设计是高效的,只保留必要信息。对于AgentExecutor,可以考虑使用ConversationBufferWindowMemory来只保留最近几轮对话。
    3. 总结历史:对于长对话,可以实现一个“总结”节点(在 LangGraph 中),定期将冗长的历史压缩成摘要。

错误3:工具调用失败,返回非预期结果或异常

  • 原因:工具函数本身执行出错,或者返回的数据格式让模型无法理解。
  • 排查
    1. 独立测试工具:在 Agent 之外,直接调用你的工具函数,传入各种参数,看它是否能正确返回。
    2. 检查输入参数:在verbose日志中,查看模型传给工具的参数值是否正确。类型错误、格式错误是常见原因。
    3. 工具返回需为字符串@tool装饰的函数必须返回字符串(或可转换为字符串的对象)。如果返回复杂字典或对象,模型可能无法处理。

错误4:ModuleNotFoundError: No module named 'langchain_xxx'

  • 原因:包没安装对。LangChain 生态的包名经常变化。
  • 解决
    1. 使用pip list | grep langchain查看已安装的包。
    2. 仔细核对官方文档或教程中使用的包名。langchain-communitylangchain-openailangchain-ollama等都是独立的包。

错误5:LangGraph 工作流卡住或不按预期执行

  • 原因:图的边(Edge)逻辑定义有误,或者某个节点函数没有正确修改或返回state
  • 排查
    1. 可视化你的图:LangGraph 支持将工作流导出为图片,这是调试的神器。
      from langgraph.graph import StateGraph # ... 构建你的 workflow ... app = workflow.compile() # 导出为 PNG app.get_graph().draw_mermaid_png(output_file_path="my_workflow.png")
    2. 检查节点返回值:每个节点函数都必须返回更新后的state字典。
    3. 检查条件边:如果你使用了add_conditional_edges,确保你的条件函数返回的下一个节点名称是图中存在的。

5. 从 Demo 到项目:架构与进阶思考

当你跑通第一个 Agent 后,下一步就是思考如何把它变成一个可维护、可扩展的项目。

5.1 项目结构优化

回顾第 2.4 节的目录,并进一步细化:

  • agents/:存放不同功能的 Agent 定义(使用create_tool_calling_agent创建的部分)。
  • chains/:存放一些可复用的简单链(Prompt + Model)。
  • tools/:按领域分类存放工具,如tools/weather/,tools/search/。复杂的工具可以考虑封装成简单的 MCP 服务器。
  • config/:使用pydanticpython-dotenv管理配置,区分开发和生产环境。
  • tests/:为你的工具和关键工作流节点编写单元测试。

5.2 生产环境考量

  1. 稳定性

    • 错误处理与重试:在工具调用和模型调用层添加重试逻辑(如使用tenacity库)。
    • 超时控制:为每个工具调用和模型调用设置超时,避免单个步骤卡死整个 Agent。
    • 降级方案:当核心工具(如天气 API)失败时,是否有备用数据源或友好的默认回复?
  2. 可观测性

    • 结构化日志:不要只依赖verbose=True。集成像structlog这样的库,将 Agent 的执行步骤、工具调用参数和结果、耗时等信息以 JSON 格式记录下来,方便后续监控和分析。
    • 追踪(Tracing):使用 LangSmith(LangChain 官方平台)或 OpenTelemetry 来可视化整个 Agent 的调用链,这对于调试复杂工作流至关重要。
  3. 性能

    • 缓存:对模型响应和工具结果进行适当缓存(例如,相同城市一小时内不再重复查询天气),减少开销和延迟。
    • 异步:如果工作流中多个步骤可以并行(例如,为多个城市同时查询天气),考虑使用 LangGraph 的异步支持或asyncio来提升效率。

5.3 关于 MCP 的深入方向

当你需要让 Agent 使用公司内部系统、特定数据库或复杂 API 时,MCP 的价值就凸显了。

  • 开发 MCP 服务器:你可以用任何语言(Python, JavaScript, Go 等)编写一个符合 MCP 协议的服务器,将内部能力“工具化”。
  • 安全性:MCP 协议设计时考虑了安全性,比如工具调用的权限控制、输入输出审计。在生产中,这是连接企业内部工具时必须评估的。
  • 动态工具发现:高级的 Agent 可以在运行时从 MCP 服务器动态发现可用的新工具,而无需重启应用。

5.4 持续学习路径

  1. 官方文档是第一位docs.langchain.comlangchain-ai.github.io/langgraph/是核心资料。先看概念指南(Concepts),再查 API 参考。
  2. 从模板和案例入手:LangChain 和 LangGraph 的 GitHub 仓库有大量示例(cookbook)。克隆下来,运行并修改它们,比从头开始写更快。
  3. 加入社区:遇到具体错误时,在 LangChain Discord 或相关 GitHub Issues 中搜索,很多坑已经有人踩过。
  4. 关注演进:这个领域迭代极快。关注核心库的 Release Notes,了解新特性和不兼容的变更。

最后,也是最关键的建议:不要试图一次性构建一个全能的超级 Agent。从一个具体、微小但完整的问题开始(比如“查询天气并建议是否带伞”),把它做透,跑通“用户输入 -> 模型思考 -> 工具调用 -> 结果整合 -> 输出回答”的完整闭环。在这个小闭环中,你会遇到并解决 90% 的基础问题。之后,再逐步增加工具、引入 LangGraph 处理复杂流程、考虑连接 MCP 服务器,你的 AI 应用开发之路就会清晰而扎实。

http://www.cnnetsun.cn/news/4132952.html

相关文章:

  • 服务网格治理中的关键链路取舍
  • 求职全流程系统化解决方案:从定位到谈薪
  • Claude导出Excel总“翻车”?技术拆解“AI导出鸭插件”如何让AI输出秒变专业文档
  • 从单片机到系统级工程:宁德时代BMS岗位核心技术栈构建指南
  • 《赛场48小时极简时间表:2026数学建模国赛期间如何分配睡眠与编程时间》
  • 多品类同城派单系统定制开发架构
  • stm32usart通信接口
  • Qt文件浏览器进阶:QFileSystemModel与QTreeView深度定制与性能优化
  • DeepSeek Harness智能体框架实战:从零部署大肥鱼宠物插件
  • GEO行业服务商怎么选?六家代表企业多维度实测!差距明显
  • IOP旗下工程综合刊《Engineering Research Express》,EI检索 3.5月录用,硕博毕业/评职晋升新选择!
  • CachyOS:极致性能的Arch Linux发行版安装与优化指南
  • TensorRT nvinfer配置参数模板:从核心原理到实战调优指南
  • Firth惩罚Logit结果解读:稀有事件的偏差修正估计
  • Java集合框架与数据结构面试全解析
  • macOS把OBS画面变成系统级虚拟摄像头:obs-mac-virtualcam从安装到排障的完整指南
  • springboot湘超足球联赛在线购票系统95656-计算机课程设计、毕业设计
  • 大厂面试中的LLM与RAG技术解析与优化策略
  • 时间序列分析实战:从ARIMA到SARIMA的建模全流程解析
  • LTspice仿真流程八步法:从原理图到可靠结果的工程化实践
  • 基于SpringBoot的二手车交易平台源码+文档
  • VVC仿射运动补偿:从原理到实践,提升视频编码效率的关键技术
  • 场边检下机旅客实时无感定位系统建设方案——基于人脸识别、视频结构化数据与衣着、姿态、微动作特征多模态融合及镜像视界单视频三维重构
  • 罗技鼠标宏LUA脚本编程指南:从原理到实践,探索硬件自动化
  • 大模型时代职业发展指南:从技术栈到求职策略
  • 量化招聘专家:连接顶尖人才与量化机构的核心桥梁
  • 【MATLAB例程分享,车联网14】基于V2X通信的匝道汇入协同速度控制仿真——对比无协同与V2X协同策略下的汇入间隙、延误、最大制动及风险指标。附完整代码的下载链接
  • Gym-V:统一视觉环境接口,加速Agentic Vision与视觉强化学习研究
  • 计算病理学临床整合:基础模型与智能体AI如何重塑诊断范式
  • 从零构建Am29000处理器模拟器:深入机器码与窗口化OS交互