ARTICLE DETAIL

资讯详情

深耕网站建设、视觉设计与SEO优化的一线实战洞察。

FastAPI+LangGraph+MCP实战:从零搭建生产级AI Agent

FastAPI+LangGraph+MCP实战:从零搭建生产级AI Agent 说实话这个项目我前后折腾了将近一个月。一开始我天真地以为AI Agent 就是把大模型 API 接进来写几个工具函数然后让模型在循环里自己调。等真正把 FastAPI、LangGraph、MCP 这三个东西拼在一起做一个能上线用的 Agent 时才发现文档里那些快速上手示例距离生产环境隔着无数个坑。这篇文章不打算讲什么华丽的理论就老老实实记录我用这套技术栈搭一个可用 AI Agent 的完整思路、核心代码、以及我实际踩到的那些坑。内容会涉及 FastAPI 的异步处理和生命周期管理、LangGraph 的状态图编排、MCP 协议的工具接入适合正在做 AI Agent 开发、或者准备从单体脚本折腾成正式服务的同学参考。如果你只会写简单的 prompt 调用看完这篇也能明白一个真正的 Agent 服务是怎么运转起来的。1. 为什么是 FastAPI LangGraph MCP 这套组合1.1 三个组件到底各自解决什么问题先把定位说清楚不然你很容易被各种概念绕晕。FastAPI 负责的是最外层的事接收 HTTP 请求、参数校验、鉴权、返回响应。选它不光是图它开发快更重要的是它原生支持异步和流式响应。Agent 的回复是逐个 token 蹦出来的如果没有 SSEServer-Sent Events流式输出用户就只能干等十几秒体验非常糟糕。FastAPI 的 StreamingResponse 天然支持这种场景。另外它的 Pydantic 校验和 LangChain/LangGraph 生态的 BaseModel 同源数据模型可以共用省去了手工转换参数的过程。LangGraph 负责的是中间层Agent 的执行流程。你可以把它理解为给 Agent 画了一张状态机图图里有节点node和边edgeAgent 在节点之间按条件流转。什么时候调大模型、什么时候调工具、什么时候结束逻辑清清楚楚。相比自己在 while 循环里判断 tool_callsLangGraph 提供了一套可持久化、可中断、可恢复的执行框架。后面我会详细讲它的状态管理机制这是整个 Agent 能不能做复杂业务的地基。MCPModel Context Protocol负责的是最底层把外部工具标准化地接进来。它的大白话解释是以前你每接一个工具就要给 Agent 写一套专用适配代码现在工具方只要实现一个 MCP ServerAgent 端用 MCP Client 就能像插 USB 一样把工具接上。MCP 定义了三类核心能力tools可调用的函数、resources可读取的数据资源、prompts可复用的提示词模板。对我们做 Agent 的人来说日常打交道最多的是 tools。这三层合在一起就是一个标准的生产级 Agent 服务架构FastAPI 对外暴露 APILangGraph 编排大脑MCP 统一工具接入。各层职责单一哪一层出了问题直接看哪一层的日志就行。1.2 LangGraph 和 LangChain 到底什么关系这个问题的出现频率极高所以单独拿出来说一下。我自己的理解是LangChain 是一套积木提供大模型封装、Prompt 模板、文档加载、向量存储等组件LangGraph 是一套施工图纸负责定义积木怎么拼、按照什么顺序执行、条件分支怎么走。LangGraph 不仅不和 LangChain 二选一反而高度依赖 LangChain。你在 LangGraph 的节点里用的仍然是 ChatOpenAI 这类模型封装、Runnable 协议包括工具节点用的也是 LangChain 工具格式。所以更准确的关系是LangGraph 把原来 LangChain 里的 AgentExecutoragent 执行器做成了更灵活的图执行引擎而不是替代 LangChain。那直接用 LangChain 的 AgentExecutor 行不行小玩具可以但复杂业务不行。Agent 一旦涉及多步骤任务、中途需要人工确认、需要多轮会话记忆你需要在任意节点暂停或者回滚AgentExecutor 那种黑盒循环就很难控制。LangGraph 把这些能力全部暴露成图结构每一步都可视化、可追踪。这也是我选 LangGraph 的核心原因不是因为它新潮而是它把 Agent 的失控风险关进了笼子里。1.3 什么场景适合、什么场景建议别硬套这套组合的适用场景我实测后觉得非常聚焦需要一个独立部署的 Agent 服务前端或第三方系统要通过 HTTP API 调用Agent 需要动态接入多个外部工具且会话是多轮的、有上下文记忆的。比如做一个客服机器人、一个带文件读写能力的办公助手、一个能查数据库并生成报表的数据分析 Agent。但如果你的需求只是接个大模型做个智能问答那这套组合就是杀鸡用牛刀。老老实实用 FastAPI 调一下大模型 API 就完事。如果业务量不大、又不想自己运维底层设施也可以考虑现成的 Agent 平台。自己搭这套东西最大的成本是开发调试复杂度——抽象层越多排查链路越长这点要有心理准备。不过换个角度想如果目标是真正理解 AI Agent 的工作原理或者要交付一个高度定制化的 Agent 服务那这套FastAPI LangGraph MCP就是值得投入的技术路线。上面的平台没法让你快速拉出任意工具也没法让你精细控制每一步逻辑。2. 项目整体架构与核心设计2.1 系统分层HTTP 层、编排层、工具层我最终落地的系统分了三层每一层的边界都很清楚这也是整个项目少踩坑的关键。HTTP 层对应 FastAPI 路由。这一层负责接收请求体、鉴权、做参数校验、然后调用编排层。它不应该知道大模型的存在更不应该有业务逻辑。我见过不少人把 Agent 逻辑直接写在路由函数里结果代码越堆越长最后无法维护。HTTP 层就做成一个薄壳把输入校验好、输出格式化好就算完成了使命。编排层对应 LangGraph 定义的状态图。这层是 Agent 的大脑负责决定下一步执行什么。它不关心请求是怎么进来的也不关心工具内部怎么实现只关心当前状态和应该往哪个节点走。这样设计的好处是Agent 的逻辑可以脱离 HTTP 服务单独测试——直接把初始状态喂给图看它怎么流转。工具层对应 MCP Client 和各 MCP Server。这层负责执行具体的事比如查数据库、读文件、调第三方 API。对编排层来说工具就是一个函数给入参数返回结果。MCP 把工具的输入输出格式标准化后编排层接入新工具的成本就变得特别低不用重新写适配逻辑。这三层之间通过明确的接口通信。我画架构图的时候习惯用依赖方向来约束HTTP 层依赖编排层编排层依赖工具层不能反向调用。实际开发里只要守住这条线出了问题基本一下就能定位到层。2.2 LangGraph 状态图设计节点、条件边、检查点LangGraph 的核心概念如果你不先理解看文档会一头雾水。我用大白话拆一遍。首先是 State状态。所有节点共享一个状态对象通常是一个 TypedDict 或 Pydantic 模型。最常见的状态是消息列表 messagesLangGraph 内置了一个 reducers 机制叫 add_messages作用是把新消息追加到旧消息列表里。为什么要加 reducer因为图执行过程中可能有多个节点写同一个字段如果没有合并策略后面的写入会直接覆盖前面的。add_messages 保证了消息是按顺序累积而不是互相覆盖。然后是 Node节点。每个节点就是一个函数输入当前状态输出更新后的状态。比如agent 节点负责调用大模型根据大模型输出决定是否调工具tools 节点负责取出模型生成的 tool_calls 并执行。然后是 Edge边。边分普通边和条件边。普通边就是执行完 A 必然去 B条件边是执行完 A根据条件决定去 B 还是去 C。一个标准的 Agent 循环长这样agent 节点首先调用大模型如果模型决定要调工具就走条件边进 tools 节点tools 节点执行完后无条件边回 agent 节点让模型基于工具结果继续推理如果模型认为不需要调工具了就走条件边到 END结束整个流程。最后是 Checkpointer检查点。这是 LangGraph 最值钱的能力。它会把每一步执行后的状态持久化下来。你可以理解成给 Agent 装了一个存档系统多轮对话中把历史状态存起来下次对话时通过 thread_id 把存档读回来。这样即使用户隔了一天才说下一句话Agent 依然记得上下文。生产环境建议用 Postgres 或 Redis 存储检查点别用默认的内存存储——服务一重启记忆全没了到时候哭都来不及。2.3 MCP 接入方式工具、资源、提示三类原语MCP 协议本身不复杂复杂的是接入方式的选择。先理解它的三类原语。Tools 是最常用的对应真实世界里的动作。比如一个文件系统 MCP Server 可以暴露 read_file、write_file、list_directory 这些工具。Agent 通过大模型的 function calling 能力生成结构化的工具调用指令MCP Client 负责把指令转成真实的函数调用。Resources 对应可读取的数据。它可以是一个文件内容、一张数据库表的查询结果、一套远程配置。与 Tools 的区别是Resources 是读取型不会产生副作用。Agent 根据用户问题决定要不要把某个 resource 内容作为上下文喂给模型。Prompts 对应可复用的提示词模板。比如把中文翻译成英文的 Prompt、生成项目周报的 Prompt。这算是 MCP 对 Agent 能力的一种附加扩展实际用到的人不多但当你需要和团队共享一套交互模板时会很方便。接入方式上MCP Server 有 stdio 和 HTTP/SSE现在官方在推 streamable HTTP两种传输方式。stdio 适合本地开发Agent 启动时拉起一个子进程通过标准输入输出来通信HTTP/SSE 适合生产部署MCP Server 独立跑在一个地址上Agent 通过网络连过去。我建议从一开始就把连接逻辑封装成独立的管理类这样后续从 stdio 切到远程只改一行配置。3. 实操演示从零搭一个带 MCP 工具的 Agent3.1 项目初始化与依赖安装我先说明一下下面这套是能跑起来的最小可复现方案基于我最终调通的版本。环境是 Python 3.11 uv 包管理器。uv 确实比 pip 快很多而且虚拟环境管理一体化热搜词里也有人问这里顺便安利一下。mkdir agent-demo cd agent-demo uv init uv python pin 3.11 uv add fastapi uvicorn[standard] langgraph langchain-openai langchain-mcp-adapters mcp python-dotenv pydantic-settings这里有一个版本相关的忠告不要一上来就全部装最新版一定要看兼容性。我踩过一个大坑最新版 langchain-mcp-adapters 对 langgraph 老版本不兼容装完直接 ImportError。我最后固定下来的版本组合是uv add fastapi0.115.0 langgraph0.2.60 langchain-openai0.2.14 langchain-mcp-adapters0.1.14 mcp1.2.0如果你用了和我不一样的版本组合装完依赖后先做一个冒烟测试写一行代码把 langgraph.graph 和 mcp 都 import 一遍能过再继续。3.2 FastAPI 生命周期里管理 MCP 连接MCP 连接是长连接尤其是 stdio 模式启动一个子进程是有成本的。绝对不能每次请求都去 new 一个 ClientSession那样服务一开压力测试就崩。正确做法是把连接放到 FastAPI 的 lifespan 里应用启动时建立连接关闭时清理。MCP 官方的 stdio_client 是异步上下文管理器直接拿 session 用完就退出的话连接就断了。所以我的做法是写一个连接管理类手动接管上下文的生命周期把它维持在存活状态。# mcp_manager.py from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class MCPManager: def __init__(self, command: str, args: list[str]): self.command command self.args args self._client_ctx None self._session_ctx None self.session: ClientSession | None None async def start(self): params StdioServerParameters( commandself.command, argsself.args, envNone, ) self._client_ctx stdio_client(params) read_stream, write_stream await self._client_ctx.__aenter__() self._session_ctx ClientSession(read_stream, write_stream) self.session await self._session_ctx.__aenter__() await self.session.initialize() async def cleanup(self): if self._session_ctx: await self._session_ctx.__aexit__(None, None, None) if self._client_ctx: await self._client_ctx.__aexit__(None, None, None)然后在 FastAPI 的 lifespan 里注册启动和清理逻辑# main.py from contextlib import asynccontextmanager from fastapi import FastAPI from agent.graph import build_agent_graph from mcp_manager import MCPManager asynccontextmanager async def lifespan(app: FastAPI): manager MCPManager( commandpython, args[local_server.py], # 某个 MCP Server 的启动入口 ) await manager.start() app.state.mcp_manager manager app.state.agent_graph await build_agent_graph(manager.session) yield await manager.cleanup() app FastAPI(lifespanlifespan)这一段是整个项目的基石。连接管理做不好后面所有工具调用都会间歇性失败而且特别难排查。3.3 LangGraph 构建 Agent 工作流有了 MCP session接下来就是把工具加载成 LangChain 能识别的工具列表再构建 LangGraph 状态图。langchain-mcp-adapters 里的 load_mcp_tools 已经帮我们完成了协议转换它会去 MCP Server 上通过 tools/list 拿到工具名、描述和 inputSchema然后包装成 LangChain 的 BaseTool。你就不用再手写 JSON Schema 映射了。# agent/graph.py from typing import Annotated, TypedDict from langchain_mcp_adapters.tools import load_mcp_tools from langchain_openai import ChatOpenAI from langgraph.graph import END, START, StateGraph from langgraph.graph.message import add_messages from langgraph.prebuilt import ToolNode, tools_condition class AgentState(TypedDict): messages: Annotated[list, add_messages] async def build_agent_graph(mcp_session): # 把 MCP Server 上的工具拉下来转成 LangChain 工具 tools await load_mcp_tools(mcp_session) llm ChatOpenAI( modelgpt-4o, temperature0, api_keysk-xxx, # 实际开发用环境变量 ) llm_with_tools llm.bind_tools(tools) def agent_node(state: AgentState): response llm_with_tools.invoke(state[messages]) return {messages: [response]} tool_node ToolNode(tools) graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tool_node) graph.add_edge(START, agent) # 关键的条件边模型说要调工具就进 tools否则结束 graph.add_conditional_edges(agent, tools_condition) graph.add_edge(tools, agent) return graph.compile()这里的灵魂是 tools_condition。它是 LangGraph 预置的条件函数判断大模型返回的 AIMessage 里有没有 tool_calls。有就去 tools 节点没有就走 END。这其实就是 function calling 循环的标准实现。你如果想在上面加人工审核、掐断超时循环之类的逻辑都在这个条件边或者节点中间插入。3.4 流式对话接口SSE 输出FastAPI 一个比较爽的地方是异步栈统一。你可以直接用 graph.astream_events 把 Agent 执行过程中的事件流推给前端中间不需要再起额外的队列或 WebSocket 服务。# main.py from fastapi import Request from fastapi.responses import StreamingResponse from pydantic import BaseModel class ChatRequest(BaseModel): message: str thread_id: str default app.post(/api/chat) async def chat(req: ChatRequest): graph app.state.agent_graph async def event_stream(): async for event in graph.astream_events( { messages: [ { role: user, content: req.message, } ] }, versionv2, config{configurable: {thread_id: req.thread_id}}, ): kind event[event] # 只把模型流式输出的增量文本透传给前端 if kind on_chat_model_stream: chunk event[data][chunk] if chunk.content: yield fdata: {chunk.content}\n\n # 工具调用日志也可以推出去方便前端显示 Agent 思考过程 if kind on_tool_start: yield fevent: tool_start\ndata: {event[name]}\n\n return StreamingResponse( event_stream(), media_typetext/event-stream, headers{Cache-Control: no-cache, X-Accel-Buffering: no}, )需要提示一下versionv2 这个参数很关键。astream_events 在 v1 版本里事件结构不同我最初照着老博客写解析事件类型总是对不上。还有一个小技巧config 里的 thread_id 一定要带上这就是 LangGraph 检查点的索引 key。同一个 thread_id 会共享上下文记忆不同 thread_id 互相隔离相当于会话维度的隔离。3.5 前端怎么接SSE 用浏览器的 EventSource 就能接但 EventSource 只能 GET。如果你业务里必须 POST比如传很长的 message那就得用 fetch ReadableStream 自己解析代码也很短。// sse-client.js async function sendMessage(message) { const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ message, thread_id: user-abc }), }); const reader response.body.getReader(); const decoder new TextDecoder(); let buffer ; while (true) { const { value, done } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); const lines buffer.split(\n); buffer lines.pop(); // 保留不完整的行 for (const line of lines) { if (line.startsWith(data: )) { const content line.slice(6); // 把增量文本拼到界面上 appendToChatBox(content); } } } }这里要注意一个老坑SSE 的一行 data 可能被 TCP 分包切碎所以必须做缓冲处理。我首版前端代码就没处理这个长回复经常漏字害得我以为是后端流有问题查了半天才发现是前端解析的问题。4. 踩坑实录我实际遇到的 7 个问题4.1 MCP stdio 连接被意外关闭现象服务跑着跑着突然一次工具调用就报错日志里出现 BrokenPipeError 或者 Connection closed。原因拆解stdio transport 的本质是 Agent 进程当父进程MCP Server 作为子进程通过标准输入输出通信。任何一种情况导致子进程退出或者管道被占用连接就废了。比如子进程代码里 print 输出了一段调试信息——这会污染 stdout 管道MCP 客户端解析立即失败。还有 FastAPI 在 dev 模式--reload下会启两个进程父子进程都抢着去拉 MCP Server也容易出问题。解决办法第一MCP Server 里所有调试日志必须走 stderr或者写日志文件绝对不能 print 到 stdout。第二开发时关掉 uvicorn reload或者给 reload 单独配一套进程管理。第三我上面写的 MCPManager 里最好加一个探活和重连机制每 30 秒发一个轻量请求探测连接存活断了就自动重建。生产环境强烈建议直接用 HTTP/SSE 远程模式MCP Server 独立部署进程层面的坑直接消失。4.2 消息状态无限增长导致 Token 暴涨现象多轮对话后某次请求突然报超出上下文长度限制或者费用飙升。原因拆解LangGraph 的 add_messages reducer 只会把新消息追加到 state 里它不管消息总量。我的 Agent 在 tools 节点执行完还会把工具结果也放回 messages一轮工具调用就多出好几条消息。对话轮数一多上下文就爆了。解决办法我给图加了一个消息裁剪节点在每次进入 agent 节点之前检查消息长度超出阈值就裁剪。LangChain 提供了 trim_messages 工具按 token 数保留最近 N 条消息。from langchain_core.messages import trim_messages def trim_state(state: AgentState) - dict: trimmed trim_messages( state[messages], max_tokens4000, strategylast, token_counterllm.get_num_tokens, include_systemTrue, ) return {messages: trimmed}注意裁剪策略要保留 system 提示词和最近几轮关键消息别一股脑全删了。如果你希望 Agent 能记得更久以前的事更优雅的方案是用摘要节点把旧消息总结成一段摘要存起来替换原始消息。4.3 异步调用阻塞导致 FastAPI 卡死现象并发一旦上来请求集体卡住直到超时CPU 占用却不高。原因拆解FastAPI 的 async 路由跑在事件循环上如果里面有同步阻塞调用整个进程的事件循环就被卡住了。我最初在 agent_node 里直接调用 llm.invoke()同步版本这就是祸根。同步调用会阻塞事件循环后续所有请求都排队等它。解决办法所有可能会阻塞的调用全部换成异步版本。LangChain 模型封装的 invoke 都对标有一个异步的 ainvokeMCP 的 session 方法本身也是异步的不需要额外处理。如果某个第三方库没有异步版本就用 asyncio.to_thread 把它丢到线程池import asyncio def sync_tool_call(arg: dict) - str: # 某个没有异步实现的同步工具函数 return real_execute(arg) result await asyncio.to_thread(sync_tool_call, arg)核心原则事件循环里不允许出现任何同步阻塞 I/O。这条规则比任何框架技巧都重要。4.4 LangGraph 版本 API 变更现象网上找的示例代码明明逻辑一样复制过来就是报错说某个函数不存在、某个参数不对。原因拆解LangGraph 迭代非常快0.1 到 0.2 再到 0.3API 发生过不少破坏性变更。比如 START、END 的导入路径变过StateGraph 的初始化方式变过conditional_edges 的写法也有调整。我最初基于 0.1 写的代码升级 0.2 后一大批报错。解决办法没有捷径就是先把版本锁定。项目里用 requirements 或 pyproject.toml 固定主版本不要随手升级。每次升级先看 changelog。另外建议直接看官方文档里对应你当前版本的示例别拿最新版本博客的代码往老环境里套。这里我再说个经验遇到诡异的 API 报错先在官方 GitHub 仓库的 examples 目录里搜一下比在搜索引擎里找答案高效得多。4.5 SSE 被代理网关缓冲现象本地跑得好好的流式输出部署到服务器后前端半天不出字然后一次性全部蹦出来。原因拆解这是 Nginx 默认行为导致的。Nginx 默认会缓冲后端响应攒到一定量才发给客户端。对普通 HTTP 接口这是优化但对 SSE 就是灾难。云厂商的 API 网关、负载均衡也可能做类似缓冲。解决办法在 Nginx 站点配置里针对代理路径关掉缓冲location /api/chat { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_set_header Connection ; proxy_http_version 1.1; chunked_transfer_encoding off; }后端响应头也已经加了 X-Accel-Buffering: no 和 Cache-Control: no-cache双保险。用了云 API 网关的话先确认它是否支持 SSE不支持就换一条不经过网关的域名路径。4.6 Pydantic v1/v2 依赖冲突现象启动时报错说某个类没有 model_fields 属性或者反过来没有fields属性。原因拆解Pydantic v2 把内部字段存储从fields换成了 model_fields。FastAPI 现代版本默认用 v2但一些老版本的 LangChain 插件、MCP 适配器还是按 v1 写法来两套并存就冲突了。解决办法把 langchain 全家桶和 langchain-mcp-adapters 都升到支持 Pydantic v2 的版本。我实际用的版本组合在前面给出过基本避开了这个问题。如果某个库实在迁不到 v2pydantic 官方提供一个兼容模块 pydantic.v1可以同时 import但在同一个进程里混用 v1/v2 极易出诡异问题不推荐。4.7 MCP 工具返回结果过大现象Agent 调用一个读文件工具文件是几 MB 的文本工具结果塞进 messages 后模型直接开始胡言乱语甚至报错。原因拆解MCP 工具返回的 content 是字符串LangGraph 会把它作为一个 ToolMessage 放回状态里再喂给大模型。上下文被撑爆后模型注意力被稀释回复质量急剧下降。解决办法在工具设计层面控制返回体量。我后来给文件读取类工具加了参数默认只返回摘要或者前几 KB 内容完整内容写入临时存储返回一个 content_id。Agent 如果需要细节再调用另一个 get_content_by_id 工具按需读取。记住一个原则工具返回给模型的东西应该是最小必要集而不是全量数据。5. 常见问题速查表与调试技巧5.1 问题速查表我把这个项目里遇到的典型问题整理成了表格方便你遇到症状时快速定位方向。症状可能原因排查与解决工具调用偶发失败报 BrokenPipeMCP stdio 子进程崩溃或 stdout 被污染子进程日志走 stderr加探活重连换 HTTP/SSE 模式多轮对话后 token 超限消息状态无限累积加裁剪节点或摘要节点并发一高请求集体卡住事件循环被同步调用阻塞换 ainvokesync 调用丢 asyncio.to_thread代码按博客抄却不断报错LangGraph 版本差异锁定版本查官方对应版本文档线上流式输出变顿挫Nginx/网关缓冲关 proxy_buffering加 X-Accel-Buffering: no启动报fields或 model_fields 错误Pydantic v1/v2 混用升级 LangChain 全家桶适配 v2工具返回内容太大模型开始乱答上下文被撑爆工具返回摘要 content_id按需拉取前端 SSE 漏字没处理 TCP 分包前端做 buffer 累积按行解析5.2 调试工具与经验这里分享几个真正让我少掉头发的调试手段。第一善用 LangGraph 的 debug 模式。开发时给 compile() 加 debugTrue执行过程中每一步的状态变化、节点输入输出都会打印出来。我第一次看到完整的 agent 决策链的时候瞬间就明白之前为什么工具调用失败了——不是代码问题而是模型在某一步返回了不存在的工具名。第二直接绕过 HTTP 层写脚本测图。把 FastAPI 层拆掉直接构造初始状态调用 graph.ainvoke() 跑一遍这样定位问题范围非常快。是 HTTP 层的参数问题还是图逻辑问题一测便知。第三MCP Server 单独调试。MCP 官方提供了 mcp 命令行工具可以单独启动一个 MCP Server 然后跟它对话查看它提供了哪些工具、返回什么格式。你甚至可以写一个十几行的脚本直接连 MCP Server 手动调工具验证工具本身是否正常再回头看 LangGraph 的调用链。工具层的问题就不要跑到编排层去猜。第四把每一次 LLM 调用的输入输出都记录下来。LangChain 的 callbacks 机制可以实现不过更简单的是在 agent_node 里把 messages 和最终 response 打日志。带 tool_calls 的响应务必记录完整 JSON。这个日志就是你排查一切模型行为异常的底牌。实际调试时我发现绝大多数问题可以归结为三类连接生命周期问题、上下文管理问题、异步阻塞问题。你只要能快速判断出眼前的事属于哪一类解决方案基本就是对症下药不会绕远路。最后再补充一个小经验AI Agent 项目的复杂度是逐步暴露的别指望一次架构全想好。我第一版甚至没接 MCP直接用 Pydantic 定义工具后来为了接蓝湖、接设计稿才意识到工具接入必须要标准化。MCP 最大的价值不是新潮而是当你的工具数量超过 5 个之后它会倒逼你采用统一的接入方式。这个好处前期体会不到后期离不开。如果你正准备踏进这个方向我的建议是先跑通最小闭环再逐步上 MCP、上检查点、上流式输出。每一步都验证没问题再往前进。
返回列表