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_KEYsk-你的key TAOTOKEN_BASE_URLhttps://taotoken.net/api AGENT_MODELclaude-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( namename, descriptiondescription, parametersparameters, functionfunc, ) 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( namesearch_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( namecalculate, 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( modelos.getenv(AGENT_MODEL, claude-sonnet-4-20250514), api_keyos.getenv(TAOTOKEN_API_KEY), base_urlos.getenv(TAOTOKEN_BASE_URL, https://taotoken.net/api), temperature0.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(contentSYSTEM_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(contentresult, tool_call_idtool_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(checkpointercheckpointer)这段代码里有几个生产级的关键设计。第一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/apiapi_key 填你的 Keymodel 填模型 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(contenttask)], iteration: 0}, configconfig, ) 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 末尾不要多加/v1OpenAI SDK 会自动拼接写成https://taotoken.net/api/v1会导致路径变成/api/v1/v1/chat/completions。OAuth 相关报错。如果你用的是 Claude Code 的 Anthropic 接入方式可能会遇到 OAuth token 过期的问题。Claude Code 的配置里需要同时设置ANTHROPIC_BASE_URL和ANTHROPIC_API_KEYBase URL 填https://taotoken.net/apiKey 用你在控制台创建的。如果报 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 原型了。接下来就是把它接到真实业务场景里用真实数据打磨工具描述和提示词。这一步没有捷径只能靠迭代。