企业官网建设流程全解析

1. 从零认识 LangGraph + MCP 构建 Agent 应用

如果你最近在折腾 Agent 应用，大概率会遇到两个绕不开的词：LangGraph 和 MCP。前者负责把「思考-行动-观察」这套循环编排成一张可控的状态图，后者负责把外部工具（视觉识别、文档解析、检索服务）以标准协议接进来。把两者拼起来，你就能得到一个既能推理、又能真正「动手」的智能体，而不是一个只会聊天的对话框。

这篇文章面向的是已经写过一点 Python、想动手搭一个完整 Agent 实例的开发者。我会用一个「论文精读助手」作为贯穿案例：用户上传一篇 PDF，Agent 自动解析结构、识别图表、做深度分析、检索相似论文、生成对比结论。整个链路里，模型 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_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL_FAST=gpt-4o-mini TAOTOKEN_MODEL_PRECISION=gpt-4o

注意 Base URL 结尾不要带/v1，LangChain 的 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( model=model_id, api_key=api_key, base_url=base_url, temperature=temperature, timeout=60, max_retries=2, )

这里有几个细节值得说。第一，temperature=0是为了让结构化输出稳定，Agent 场景里随机性越小越好。第二，max_retries=2能扛住偶发的网络抖动，但别设太大，否则一个坏请求会拖慢整条图。第三，timeout=60对长文本分析够用，如果你跑的是超长论文，可以调到 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: str

add_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(f"MCP 工具执行错误: {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( command="python", 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( command="python", 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_exceptions=True) 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 found

model not found说明 Model ID 写错了。检查.env里的TAOTOKEN_MODEL_FAST和TAOTOKEN_MODEL_PRECISION是否是有效模型名。不同通道支持的模型列表可能不同，建议先用一个确定可用的模型名测试。

5.5 MCP Server 启动失败

如果报FileNotFoundError或ModuleNotFoundError，检查StdioServerParameters里的args路径是否正确。我建议用绝对路径：

import os PARSER_SERVER = StdioServerParameters( command="python", 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]。如果只写List，LangGraph 会用新值覆盖旧值。这个坑我踩过两次，排查了半天。

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 快得多。