用 LangGraph + MCP 构建 Agent 应用：从实例认识 TaoToken 统一 Key 通道
发布时间：2026/10/4 16:07:43 作者：尧图编辑部 阅读量：1,286
1. 从零认识 LangGraph MCP 构建 Agent 应用如果你最近在折腾 Agent 应用大概率会遇到两个绕不开的词LangGraph 和 MCP。前者负责把「思考-行动-观察」这套循环编排成一张可控的状态图后者负责把外部工具视觉识别、文档解析、检索服务以标准协议接进来。把两者拼起来你就能得到一个既能推理、又能真正「动手」的智能体而不是一个只会聊天的对话框。这篇文章面向的是已经写过一点 Python、想动手搭一个完整 Agent 实例的开发者。我会用一个「论文精读助手」作为贯穿案例用户上传一篇 PDFAgent 自动解析结构、识别图表、做深度分析、检索相似论文、生成对比结论。整个链路里模型 endpoint 和 Key 我统一收口到 TaoToken 的 OpenAI 兼容通道这样你换模型、加节点、跑批量任务时不用到处改配置。先说清楚这套架构里每个角色干什么。LangGraph 是「大脑的骨架」它用 State状态字典在节点之间传递数据用 Edge边决定下一步走哪。MCP 是「感官和手脚」它把视觉模型、OCR、文档解析器封装成独立的 Server 进程通过 stdio 或 HTTP 通信大脑不需要知道工具内部怎么实现只要按协议调用就行。RAG 和知识图谱则是「记忆」前者做局部语义检索后者做全局关系推理。我试过把这三层揉在一个脚本里结果就是改一处崩三处。后来拆成「状态层 / 工具层 / 编排层」之后调试成本直线下降。下面我会按这个分层思路把可复制的代码片段一段段给你。在开始写代码之前先明确一个前提所有 LLM 调用都走统一的 Base URL 和 Key。这样你在 LangGraph 的各个节点里创建模型实例时只需要从环境变量读一次配置后面加多少节点都不用重复填。这也是我把 TaoToken 放在前置章节的原因——它是整条链路的入口配错了后面全白搭。2. TaoToken 统一 Key 通道前置配置在写 LangGraph 节点之前先把模型通道配好。这一步看起来简单但 90% 的「跑不通」都出在这里。核心思路是把 Base URL、API Key、Model ID 三件套集中到环境变量让 LangChain 的 ChatOpenAI 直接读。TaoToken 提供的是 OpenAI 兼容接口所以你可以继续用langchain_openai里的ChatOpenAI只需要把base_url指向 TaoToken 的 API 地址。这样做的好处是你现有的 LangChain 代码几乎不用改只换 endpoint 和 Key。先建一个.env文件放在项目根目录# .env TAOTOKEN_API_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api TAOTOKEN_MODEL_FASTgpt-4o-mini TAOTOKEN_MODEL_PRECISIONgpt-4o注意 Base URL 结尾不要带/v1LangChain 的 OpenAI 客户端会自动补全路径。如果你手动拼/v1/chat/completions反而会 404这是最常见的坑之一。然后写一个统一的模型工厂所有节点都从这里拿实例# src/core/llm_factory.py import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI load_dotenv() class LLMFactory: staticmethod def get_model(model_type: str fast) - ChatOpenAI: base_url os.getenv(TAOTOKEN_BASE_URL) api_key os.getenv(TAOTOKEN_API_KEY) if not base_url or not api_key: raise RuntimeError(缺少 TAOTOKEN_BASE_URL 或 TAOTOKEN_API_KEY) if model_type precision: model_id os.getenv(TAOTOKEN_MODEL_PRECISION, gpt-4o) temperature 0 else: model_id os.getenv(TAOTOKEN_MODEL_FAST, gpt-4o-mini) temperature 0 return ChatOpenAI( modelmodel_id, api_keyapi_key, base_urlbase_url, temperaturetemperature, timeout60, max_retries2, )这里有几个细节值得说。第一temperature0是为了让结构化输出稳定Agent 场景里随机性越小越好。第二max_retries2能扛住偶发的网络抖动但别设太大否则一个坏请求会拖慢整条图。第三timeout60对长文本分析够用如果你跑的是超长论文可以调到 120。如果你用的是 Claude Code 或者 Cline 这类工具配置方式类似只是字段名不同。以 Cline 的 MCP 配置为例你需要写全三件套{ mcpServers: { taotoken-llm: { command: npx, args: [-y, modelcontextprotocol/server-openai], env: { OPENAI_BASE_URL: https://taotoken.net/api, OPENAI_API_KEY: sk-你的key, OPENAI_MODEL: gpt-4o } } } }Codex 的auth.json也是同样的逻辑把base_url和api_key填进去即可。关键点永远是Base URL、Key、Model ID 三个都要对缺一个就会报 401 或者 model not found。配好之后先别急着写 LangGraph用一段最小代码验证通道是否通# test_channel.py from src.core.llm_factory import LLMFactory llm LLMFactory.get_model(fast) resp llm.invoke(用一句话说明什么是状态机) print(resp.content)如果这段能打印出内容说明通道没问题可以进入下一步。如果报错先看错误类型401 是 Key 错404 是 Base URL 或路径错model not found 是 Model ID 错。这三种错误后面排障章节会详细讲。3. 可复制的 LangGraph 节点与 MCP 配置片段现在进入核心部分。我会把「论文精读助手」拆成几个节点每个节点职责单一通过 State 传递数据。先定义状态再写节点最后连成图。3.1 定义 AgentState 状态结构状态是 LangGraph 的灵魂它决定了信息如何在节点间流动。我的经验是不要把原始文本全塞进 messages那样 Token 会爆炸。把结构化结果单独放字段messages 只留对话历史。# src/agents/state.py from typing import Annotated, List, TypedDict, Optional, Dict, Any from langgraph.graph.message import add_messages from pydantic import BaseModel, Field class PaperAnalysis(BaseModel): motivation: str Field(description研究动机) core_problem: str Field(description核心问题) innovations: List[str] Field(description创新点列表) limitations: List[str] Field(description局限性) class AgentState(TypedDict): messages: Annotated[List[Any], add_messages] pdf_path: str structured_content: Optional[Dict[str, Any]] vision_results: List[Dict[str, Any]] analysis: Optional[Dict[str, Any]] current_step: stradd_messages是 LangGraph 提供的 reducer它会把新消息追加到列表而不是覆盖。这个细节很重要如果你直接写messages: List每次节点返回都会把历史冲掉。3.2 封装 MCP 客户端为 LangGraph 工具MCP 的价值在于解耦。视觉识别、文档解析这些重活放在独立进程里LangGraph 只负责调用。下面是一个通用的 MCP 客户端管理器# src/core/mcp_client.py from typing import Optional, Dict, Any from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client class MCPClientManager: def __init__(self, server_params: StdioServerParameters): self.server_params server_params self.session: Optional[ClientSession] None async def __aenter__(self): self._read, self._write await stdio_client(self.server_params) self.session ClientSession(self._read, self._write) await self.session.initialize() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.__aexit__(exc_type, exc_val, exc_tb) async def call_tool(self, tool_name: str, arguments: Dict[str, Any]): if not self.session: raise RuntimeError(MCP Session 未初始化) result await self.session.call_tool(tool_name, arguments) if getattr(result, isError, False): raise RuntimeError(fMCP 工具执行错误: {result.content}) return result.content然后把它包装成 LangChain 工具这样 LLM 就能通过 function calling 触发# src/tools/vision_tool.py from langchain_core.tools import tool from mcp import StdioServerParameters from src.core.mcp_client import MCPClientManager VISION_SERVER StdioServerParameters( commandpython, args[src/mcp_servers/vision_service/server.py], ) tool async def analyze_paper_figure(image_path: str, context_hint: str ): 分析论文中的图表返回结构化解读。 async with MCPClientManager(VISION_SERVER) as client: result await client.call_tool( analyze_paper_image, {image_path: image_path, context_hint: context_hint}, ) return result[0].text这里有个坑要注意MCP Server 的路径必须是相对于项目根目录的如果你在子目录里跑脚本args里的路径会找不到。我一般用绝对路径或者os.path.join(os.getcwd(), ...)来拼。3.3 编写解析节点和视觉节点解析节点负责把 PDF 变成结构化文本视觉节点负责识别图表。两者都通过 MCP 调用外部服务# src/agents/nodes/parser.py from mcp import StdioServerParameters from src.core.mcp_client import MCPClientManager from src.agents.state import AgentState PARSER_SERVER StdioServerParameters( commandpython, args[src/mcp_servers/doc_parser/server.py], ) async def paper_parser_node(state: AgentState): print(f--- 解析论文: {state[pdf_path]} ---) try: async with MCPClientManager(PARSER_SERVER) as client: response await client.call_tool(parse_pdf, {path: state[pdf_path]}) raw_md response[0].text return { structured_content: {raw_md: raw_md}, current_step: vision, messages: [f解析完成共 {len(raw_md)} 字符], } except Exception as e: return {messages: [f解析失败: {e}]}视觉节点会遍历解析出的图片列表并发调用视觉 MCP# src/agents/nodes/vision_processor.py import asyncio from src.core.mcp_client import MCPClientManager from src.agents.state import AgentState from src.tools.vision_tool import VISION_SERVER async def vision_processor_node(state: AgentState): images state.get(structured_content, {}).get(extracted_images, []) if not images: return {messages: [无图片需要分析]} results [] async with MCPClientManager(VISION_SERVER) as client: tasks [ client.call_tool(analyze_paper_image, { image_path: img[path], context_hint: img.get(caption, ), }) for img in images ] outputs await asyncio.gather(*tasks, return_exceptionsTrue) for img, out in zip(images, outputs): if isinstance(out, Exception): continue results.append({image_id: img[path], analysis: out[0].text}) return { vision_results: results, current_step: analysis, messages: [f完成 {len(results)} 张图表分析], }3.4 编排成图并接入 TaoToken 模型最后把所有节点连起来。注意这里用LLMFactory拿模型Base URL 和 Key 都从环境变量走# src/agents/graph.py from langgraph.graph import StateGraph, END from src.agents.state import AgentState from src.agents.nodes.parser import paper_parser_node from src.agents.nodes.vision_processor import vision_processor_node from src.agents.nodes.analyzer import research_analyzer_node def create_research_graph(): workflow StateGraph(AgentState) workflow.add_node(parser, paper_parser_node) workflow.add_node(vision, vision_processor_node) workflow.add_node(analyzer, research_analyzer_node) workflow.set_entry_point(parser) workflow.add_edge(parser, vision) workflow.add_edge(vision, analyzer) workflow.add_edge(analyzer, END) return workflow.compile() app create_research_graph()分析节点里创建模型时直接调工厂# src/agents/nodes/analyzer.py from src.core.llm_factory import LLMFactory from src.agents.state import AgentState, PaperAnalysis async def research_analyzer_node(state: AgentState): llm LLMFactory.get_model(precision) structured_llm llm.with_structured_output(PaperAnalysis) content state.get(structured_content, {}).get(raw_md, )[:20000] result await structured_llm.ainvoke(f分析以下论文内容\n{content}) return { analysis: result.dict(), current_step: done, messages: [深度分析完成], }到这里一条完整的链路就搭好了解析 → 视觉 → 分析每一步都通过 MCP 或统一模型通道完成。4. 验证请求与成功结果配置写完必须验证链路真的跑通。我一般分两步先单独测模型通道再跑整张图。4.1 最小验证模型通道是否通# verify_llm.py import asyncio from src.core.llm_factory import LLMFactory async def main(): llm LLMFactory.get_model(fast) resp await llm.ainvoke(用一句话解释 LangGraph 的 State 是什么) print(模型返回:, resp.content) asyncio.run(main())预期输出类似模型返回: State 是 LangGraph 中在节点间传递的共享数据结构用来保存对话历史和中间结果。如果这一步失败先别往下走回到第 2 章检查 Base URL 和 Key。4.2 完整验证跑通整张图# verify_graph.py import asyncio from src.agents.graph import app async def main(): initial_state { pdf_path: data/sample_paper.pdf, messages: [], vision_results: [], } async for output in app.astream(initial_state): for node, state in output.items(): print(f[节点 {node}] 完成当前步骤: {state.get(current_step)}) final await app.ainvoke(initial_state) print(分析结果:, final.get(analysis)) asyncio.run(main())成功时你会看到类似输出[节点 parser] 完成当前步骤: vision [节点 vision] 完成当前步骤: analysis [节点 analyzer] 完成当前步骤: done 分析结果: {motivation: ..., core_problem: ..., innovations: [...], limitations: [...]}4.3 用一次问答请求验证端到端如果你想更直观地验证可以加一个问答节点让 Agent 基于分析结果回答用户问题# verify_qa.py import asyncio from src.core.llm_factory import LLMFactory async def ask(question: str, analysis: dict): llm LLMFactory.get_model(precision) prompt f基于以下论文分析结果回答问题 分析{analysis} 问题{question} resp await llm.ainvoke(prompt) return resp.content async def main(): analysis {core_problem: 长文本检索效率低, innovations: [分层索引]} answer await ask(这篇论文的核心创新是什么, analysis) print(回答:, answer) asyncio.run(main())如果模型能基于结构化分析给出合理回答说明整条链路——从 MCP 工具调用到 TaoToken 模型通道——全部打通。5. 本篇常见错误排查这一章是我踩过的坑合集按报错类型分类你对照着查。5.1 401 Unauthorized最常见。原因通常是 Key 没读到或者格式不对。检查.env里TAOTOKEN_API_KEY是否以sk-开头以及load_dotenv()是否在读取环境变量之前调用。如果你在 Docker 里跑记得把.env挂载进去或者用-e传环境变量。5.2 local proxy failed / connection refused这个报错说明请求根本没发出去。检查TAOTOKEN_BASE_URL是否写成了https://taotoken.net/api不要带/v1也不要带尾部斜杠。另外确认你的网络能正常访问该域名公司内网可能需要配置出口。5.3 reading choices 报错这个错误通常出现在模型返回格式不符合预期时。比如你用了with_structured_output但模型返回的不是合法 JSON。解决办法有两个一是把temperature设为 0二是换用支持结构化输出的模型。如果还不行在 prompt 里明确要求「只输出 JSON不要任何解释」。5.4 OAuth / model not foundmodel not found说明 Model ID 写错了。检查.env里的TAOTOKEN_MODEL_FAST和TAOTOKEN_MODEL_PRECISION是否是有效模型名。不同通道支持的模型列表可能不同建议先用一个确定可用的模型名测试。5.5 MCP Server 启动失败如果报FileNotFoundError或ModuleNotFoundError检查StdioServerParameters里的args路径是否正确。我建议用绝对路径import os PARSER_SERVER StdioServerParameters( commandpython, args[os.path.join(os.getcwd(), src/mcp_servers/doc_parser/server.py)], )另外确认 MCP Server 脚本本身能独立运行先python server.py测一下能启动再接入 LangGraph。5.6 状态被覆盖如果你发现messages每次都被清空检查是否用了Annotated[List, add_messages]。如果只写ListLangGraph 会用新值覆盖旧值。这个坑我踩过两次排查了半天。6. 把通道固定下来继续扩展你的 Agent走到这里你已经有了一个能跑通的 LangGraph MCP Agent 骨架。接下来能做的事很多加 RAG 节点做相似论文检索加知识图谱节点做关系推理加对比节点生成多论文分析表。但无论加多少节点模型通道始终是那一个 Base URL 和 Key不用改。我建议你先把当前这套配置固化成一个模板项目把.env、llm_factory.py、mcp_client.py这三个文件当成基础设施后面每加一个新 Agent直接复制这三个文件就行。这样你就不用每次重新配通道把精力放在业务逻辑上。如果你还没拿到 Key可以去 TaoToken 的 API Keys 页面创建一个然后在接入文档里对照 OpenAI 兼容格式确认参数。想先试试模型对话效果可以直接在模型对话页面发一条请求验证。如果你打算长期跑编码类或 Agent 类任务Coding Plan 会更划算适合高频调用场景。最后留一个实用技巧在 LangGraph 的每个节点里加一行print打印当前current_step这样跑批量任务时你能从日志里一眼看出卡在哪个环节。比事后翻 trace 快得多。
尧图网站开发 