企业官网建设流程全解析

1. 为什么你的 Agent 总停在 Demo 阶段

AI Agent 这个词在 2026 年已经不算新鲜，但真正把它跑进生产环境的人依然不多。我见过太多项目卡在同一个位置：本地用 LangChain 拼一个 ReAct 循环，调两个工具，跑通一个“查天气+发邮件”的演示，然后就没有然后了。问题不在于模型不够强，而在于从 Demo 到生产之间隔着一整套工程化能力——工具怎么标准化接入、多步任务怎么编排、失败怎么重试、状态怎么持久化、成本怎么控制。

这篇文章要解决的就是这条链路。核心检索词是 AI Agent 生产级实战，我会以 MCP 工具接入和 LangGraph 编排为主线，把 A2A 协作和 PRAR 模式串起来，最终交付一个可复制、可观测、可扩展的 Agent 原型。适合谁？已经写过基础 LLM 调用、想往生产级 Agent 方向走的开发者；或者正在选型 Agent 框架、需要一套能落地的配置参考的团队。

先说清楚一个概念。传统 LLM 是一问一答，模型回答完就结束。AI Agent 不一样，它是“感知-推理-行动-反思”的循环：模型先理解任务，决定调用哪个工具，拿到工具返回结果后继续推理，直到任务完成。这个循环在 2026 年有了标准化的表达，就是 PRAR 模式——Perceive（感知）、Reason（推理）、Act（行动）、Reflect（反思）。MCP 负责把工具调用标准化，LangGraph 负责把循环编排成确定性的状态机，A2A 负责让多个 Agent 之间能互相派活。

我试过用纯手写循环的方式搭 Agent，工具一多就乱，状态管理全靠全局变量，跑长任务必崩。后来换成 LangGraph 的图结构，每个节点职责清晰，状态通过 TypedDict 流转，配合 Checkpointer 还能做断点恢复。这套组合是目前生产环境里最稳的。

下面从环境准备开始，一步步把配置和代码跑通。

2. TaoToken 接入准备与 MCP 工具注册

在写 Agent 之前，得先解决模型调用的问题。生产级 Agent 对模型的推理能力和工具调用准确率要求很高，我实测下来，用统一的 API 网关来管理模型接入会省很多事。TaoToken 提供的就是这样一个入口，Base URL 是https://taotoken.net/api，兼容 OpenAI 的接口格式，所以现有的 OpenAI SDK 代码基本不用改，换个 base_url 和 key 就能用。

你需要先去控制台创建一个 API Key。打开https://taotoken.net/api-keys，登录后新建一个 Key，复制出来保存好。这个 Key 后面会用在环境变量里。注意不要把它硬编码进代码提交到仓库，用.env文件管理。

模型选择上，Agent 场景建议用推理能力强的模型。工具调用准确率直接决定 Agent 能不能跑通，便宜但调不准工具的模型反而更费钱。你可以在模型对话页面先测试一下不同模型对 function calling 的支持情况，地址是https://taotoken.net/models。

环境变量配置如下，创建一个.env文件：

# .env TAOTOKEN_API_KEY=sk-你的key TAOTOKEN_BASE_URL=https://taotoken.net/api AGENT_MODEL=claude-sonnet-4-20250514

然后安装依赖。LangGraph 和 MCP 的 Python SDK 是核心：

pip install langgraph langchain-openai mcp python-dotenv structlog

接下来是 MCP 工具注册。MCP 的核心价值在于把工具定义标准化，任何支持 MCP 的客户端都能复用同一套工具。一个 MCP Server 需要声明三样东西：工具名称、描述、输入参数的 JSON Schema。下面是一个最小可用的 MCP 工具注册示例，包含搜索和代码执行两个工具：

# mcp_tools.py import json from typing import Any from pydantic import BaseModel class ToolDefinition(BaseModel): name: str description: str parameters: dict function: Any class MCPToolRegistry: """MCP 标准工具注册中心""" def __init__(self): self._tools: dict[str, ToolDefinition] = {} def register(self, name: str, description: str, parameters: dict): def decorator(func): self._tools[name] = ToolDefinition( name=name, description=description, parameters=parameters, function=func, ) return func return decorator def get_definitions(self) -> list: """返回 OpenAI function calling 兼容的工具定义""" return [ { "type": "function", "function": { "name": t.name, "description": t.description, "parameters": t.parameters, }, } for t in self._tools.values() ] def call(self, name: str, args: dict) -> str: tool = self._tools.get(name) if not tool: return f"未知工具: {name}" try: return str(tool.function(**args)) except Exception as e: return f"工具执行错误: {e}" registry = MCPToolRegistry() @registry.register( name="search_web", description="搜索互联网获取最新信息", parameters={ "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词"} }, "required": ["query"], }, ) def search_web(query: str) -> str: # 实际项目替换为真实搜索 API return f"关于「{query}」的搜索结果：2026 年 AI Agent 市场规模持续增长..." @registry.register( name="calculate", description="执行数学计算", parameters={ "type": "object", "properties": { "expression": {"type": "string", "description": "数学表达式"} }, "required": ["expression"], }, ) def calculate(expression: str) -> str: try: result = eval(expression, {"__builtins__": {}}, {}) return f"计算结果: {result}" except Exception as e: return f"计算错误: {e}"

这里有个关键点：工具数量要控制。单个 Agent 的工具数建议不超过 15 个，超过之后模型选错工具的概率会明显上升。如果业务确实需要很多工具，用懒加载或者按场景分组，不要一次性全塞给模型。

3. LangGraph 编排配置与 PRAR 循环实现

工具注册好了，接下来是编排。LangGraph 的核心思想是把 Agent 的执行过程建模成一张状态图，每个节点是一个处理步骤，边定义流转方向。相比手写 while 循环，图结构的好处是状态显式声明、流程可追溯、支持条件分支和断点恢复。

先定义状态结构。LangGraph 用 TypedDict 描述状态，add_messages是内置的 reducer，负责把新消息追加到消息列表而不是覆盖：

# agent_graph.py import json import os from typing import Annotated, TypedDict from dotenv import load_dotenv from langchain_openai import ChatOpenAI from langchain_core.messages import AnyMessage, add_messages, HumanMessage, SystemMessage, ToolMessage from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from mcp_tools import registry load_dotenv() class AgentState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] iteration: int task_done: bool SYSTEM_PROMPT = """你是一个生产级 AI Agent，遵循 PRAR 模式工作： 1. Perceive：理解用户需求和当前上下文 2. Reason：思考需要哪些工具，制定行动计划 3. Act：调用工具获取信息或执行操作 4. Reflect：评估结果，决定继续还是给出最终答案 规则： - 优先使用工具获取实时信息，不要猜测 - 每次只调用必要的工具 - 工具返回错误时尝试其他方式 - 任务完成后直接给出答案，不要再调用工具""" llm = ChatOpenAI( model=os.getenv("AGENT_MODEL", "claude-sonnet-4-20250514"), api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL", "https://taotoken.net/api"), temperature=0.1, ) def perceive_reason_node(state: AgentState) -> dict: """Perceive + Reason：感知上下文并决策""" messages = state["messages"] if not any(isinstance(m, SystemMessage) for m in messages): messages = [SystemMessage(content=SYSTEM_PROMPT)] + messages response = llm.bind_tools(registry.get_definitions()).invoke(messages) return { "messages": [response], "iteration": state.get("iteration", 0) + 1, } def act_node(state: AgentState) -> dict: """Act：执行工具调用""" last_msg = state["messages"][-1] tool_messages = [] for tool_call in last_msg.tool_calls: name = tool_call["name"] args = tool_call["args"] result = registry.call(name, args) tool_messages.append( ToolMessage(content=result, tool_call_id=tool_call["id"]) ) return {"messages": tool_messages} def should_continue(state: AgentState) -> str: """Reflect：判断是否继续循环""" last_msg = state["messages"][-1] if state.get("iteration", 0) >= 10: return "end" if hasattr(last_msg, "tool_calls") and last_msg.tool_calls: return "act" return "end" # 构建状态图 workflow = StateGraph(AgentState) workflow.add_node("perceive_reason", perceive_reason_node) workflow.add_node("act", act_node) workflow.set_entry_point("perceive_reason") workflow.add_conditional_edges( "perceive_reason", should_continue, {"act": "act", "end": END}, ) workflow.add_edge("act", "perceive_reason") checkpointer = MemorySaver() agent_app = workflow.compile(checkpointer=checkpointer)

这段代码里有几个生产级的关键设计。第一，iteration计数器防止无限循环，超过 10 轮强制结束。第二，MemorySaver作为 Checkpointer，每个节点的状态都会保存，支持断点恢复。第三，should_continue函数实现了 Reflect 逻辑，根据最后一条消息是否包含 tool_calls 来决定下一步。

如果你用的是 Claude Code 或者 Cline 这类工具做开发，配置方式略有不同。以 Cline 的 MCP 配置为例，需要在 settings 里写全三件套——Base URL、Key、Model ID：

{ "mcpServers": { "taotoken-agent": { "command": "python", "args": ["-m", "mcp_server"], "env": { "TAOTOKEN_BASE_URL": "https://taotoken.net/api", "TAOTOKEN_API_KEY": "sk-你的key", "AGENT_MODEL": "claude-sonnet-4-20250514" } } } }

Codex 的 auth.json 配置类似，核心是三个字段不能少：base_url 指向https://taotoken.net/api，api_key 填你的 Key，model 填模型 ID。这三个缺一个都会报 401 或者 model not found。

4. 本地验证请求与成功结果确认

配置写完了，得跑起来验证。先写一个测试脚本，发一个需要多步工具调用的任务：

# test_agent.py from agent_graph import agent_app from langchain_core.messages import HumanMessage config = {"configurable": {"thread_id": "test-001"}} task = "帮我搜索一下 2026 年 AI Agent 的市场规模，然后计算如果每年增长 30%，3 年后是多少" result = agent_app.invoke( {"messages": [HumanMessage(content=task)], "iteration": 0}, config=config, ) print("=== 执行轨迹 ===") for msg in result["messages"]: msg_type = type(msg).__name__ if hasattr(msg, "tool_calls") and msg.tool_calls: for tc in msg.tool_calls: print(f"[工具调用] {tc['name']}({tc['args']})") elif msg_type == "ToolMessage": print(f"[工具返回] {msg.content[:100]}") elif msg_type == "AIMessage" and msg.content: print(f"[最终回答] {msg.content}")

运行python test_agent.py，正常的话你会看到类似这样的输出：

=== 执行轨迹 === [工具调用] search_web({'query': '2026年 AI Agent 市场规模'}) [工具返回] 关于「2026年 AI Agent 市场规模」的搜索结果：... [工具调用] calculate({'expression': '100 * 1.3 ** 3'}) [工具返回] 计算结果: 219.7 [最终回答] 根据搜索结果，2026 年 AI Agent 市场规模约为...

看到这个轨迹说明 PRAR 循环跑通了：模型先感知任务，推理出需要搜索，调用 search_web，拿到结果后继续推理需要计算，调用 calculate，最后反思任务完成，给出答案。

验证的时候重点看三个地方。第一，工具调用参数是否正确，如果模型传了错误的参数格式，说明工具的 JSON Schema 描述不够清晰。第二，循环次数是否合理，一个简单任务如果跑了 8 轮以上，说明提示词或者工具描述有问题。第三，最终回答是否基于工具返回的真实数据，而不是模型自己编的。

如果你想单独验证模型接入是否正常，可以用模型对话页面直接测试，地址是https://taotoken.net/models，选好模型发一条消息，确认能正常返回。这一步能排除掉大部分接入层的问题。

5. 常见报错排查与修复

跑 Agent 的过程中会遇到几类典型报错，我按出现频率排一下。

401 Unauthorized。这个最常见，基本是 Key 的问题。检查.env里的TAOTOKEN_API_KEY是否复制完整，有没有多余空格。如果用的是 Cline 或者 Codex，检查配置文件里的 key 字段名是否正确。还有一种情况是 base_url 写错了，比如漏了/api后缀或者写成了https://taotoken.net，正确的应该是https://taotoken.net/api。

local proxy failed / connection refused。这个报错通常出现在本地网络环境有特殊配置的时候。先确认能不能直接访问https://taotoken.net/api，用 curl 测一下：

curl -X POST https://taotoken.net/api/v1/chat/completions \ -H "Authorization: Bearer sk-你的key" \ -H "Content-Type: application/json" \ -d '{"model":"claude-sonnet-4-20250514","messages":[{"role":"user","content":"hi"}]}'

如果 curl 能通但代码不通，检查代码里有没有设置http_proxy或https_proxy环境变量，有的话清掉。Python 的 requests 库会自动读取这些变量。

reading choices 报错 / KeyError: 'choices'。这个说明 API 返回的结构和预期不符。大概率是模型 ID 写错了，返回了一个错误信息而不是正常的 completion 结构。检查AGENT_MODEL的值是否在 TaoToken 支持的模型列表里。另外确认 base_url 末尾不要多加/v1，OpenAI SDK 会自动拼接，写成https://taotoken.net/api/v1会导致路径变成/api/v1/v1/chat/completions。

OAuth 相关报错。如果你用的是 Claude Code 的 Anthropic 接入方式，可能会遇到 OAuth token 过期的问题。Claude Code 的配置里需要同时设置ANTHROPIC_BASE_URL和ANTHROPIC_API_KEY，Base URL 填https://taotoken.net/api，Key 用你在控制台创建的。如果报 OAuth 错误，检查是不是混用了两种认证方式。

工具调用返回空 / 模型不调工具。这个不是报错但很常见。原因是工具的 description 写得太模糊，模型不知道什么时候该用。把 description 写具体，比如“搜索互联网获取最新信息”比“搜索”好得多。另外确认bind_tools传进去的定义格式正确，必须是 OpenAI function calling 的标准格式。

循环不终止。Agent 一直调工具不结束，通常是should_continue的判断逻辑有问题。检查最后一条消息的tool_calls字段是否为空列表而不是 None，空列表在 Python 里是 falsy 的，但hasattr判断会通过。建议用if last_msg.tool_calls:而不是if hasattr(last_msg, "tool_calls")。

排查的时候养成看日志的习惯。在act_node里加一行print(f"调用工具: {name}, 参数: {args}")，能快速定位是模型选错了工具还是参数传错了。

6. 从单 Agent 到 A2A 协作的扩展路径

单 Agent 跑通之后，下一步是扩展。生产环境里很多任务不是单个 Agent 能搞定的，需要多个专业 Agent 协作。2026 年主流的协作模式是层级化指挥链：一个中央协调器加 2 到 4 个专业工作者。

用 LangGraph 实现这个模式很直接，在状态图里加条件分支就行：

# multi_agent.py from typing import TypedDict, Annotated from langgraph.graph import StateGraph, END from langchain_core.messages import AnyMessage, add_messages class MultiAgentState(TypedDict): messages: Annotated[list[AnyMessage], add_messages] next_agent: str results: dict def orchestrator(state: MultiAgentState) -> dict: """中央协调器：分析任务类型，决定派给哪个 Agent""" task = state["messages"][-1].content if "搜索" in task or "调研" in task: return {"next_agent": "researcher"} elif "计算" in task or "分析" in task: return {"next_agent": "analyst"} else: return {"next_agent": "general"} def researcher(state: MultiAgentState) -> dict: """调研 Agent：负责信息收集""" # 这里接入调研专用的工具集和提示词 return {"results": {"research": "调研完成"}, "next_agent": "done"} def analyst(state: MultiAgentState) -> dict: """分析 Agent：负责数据处理""" return {"results": {"analysis": "分析完成"}, "next_agent": "done"} workflow = StateGraph(MultiAgentState) workflow.add_node("orchestrator", orchestrator) workflow.add_node("researcher", researcher) workflow.add_node("analyst", analyst) workflow.set_entry_point("orchestrator") workflow.add_conditional_edges( "orchestrator", lambda s: s["next_agent"], {"researcher": "researcher", "analyst": "analyst", "general": END}, ) workflow.add_edge("researcher", END) workflow.add_edge("analyst", END) multi_agent_app = workflow.compile()

如果要做跨团队的 Agent 协作，就需要 A2A 协议了。A2A 的核心是 Agent Card，每个 Agent 暴露一个描述自己能力的 JSON 端点，其他 Agent 通过这个端点发现和派发任务：

# a2a_agent_card.py agent_card = { "name": "research-agent", "description": "负责市场调研和信息收集", "capabilities": ["web_search", "data_collection"], "endpoint": "https://your-domain.com/a2a", "version": "1.0", }

A2A 和 MCP 不是竞争关系。MCP 解决的是 Agent 到工具的垂直连接，A2A 解决的是 Agent 到 Agent 的水平协作。一个完整的生产系统里两者都会用到。

生产化还需要补几块能力。成本控制上，给每次任务设一个 token 上限和费用上限，超了就中断。安全边界上，对工具调用的参数做危险模式检查，比如 SQL 里的 DROP TABLE、shell 里的 rm -rf。可观测性上，每次工具调用记录 trace_id、耗时、token 消耗，方便事后排查。这些能力不需要一开始就全上，但架构上要留好扩展点。

如果你打算长期做 Agent 方向的开发，建议把 Coding Plan 用起来，地址是https://taotoken.net/coding-plan，里面有更完整的工程化实践和持续更新的配置模板。接入文档在https://taotoken.net/doc，遇到配置问题可以先查文档。API Key 管理在https://taotoken.net/api-keys，建议给不同项目建不同的 Key，方便追踪用量。

整套跑下来，你手里应该有一个能多步推理、能调工具、能断点恢复的 Agent 原型了。接下来就是把它接到真实业务场景里，用真实数据打磨工具描述和提示词。这一步没有捷径，只能靠迭代。