1. 从五个名词到一条能跑的链路LLM、MCP、A2A、RAG、AI Agent 到底怎么串很多人第一次接触这几个词是在同一篇文章里被同时砸中LLM、MCP 协议、A2A 协议、RAG、AI Agent。每个词单独看都能懂个大概但合在一起就懵了——它们到底是并列关系还是层层嵌套我该先学哪个先把它们放回一条真实的协作链路里看就清楚了。假设你要做一个「自动排查线上告警并给出修复建议」的系统LLM 是这条链路的「大脑」负责理解和生成。它本身只会根据输入预测下一个 token不会主动查数据库也不会调用你的接口。RAG 是给大脑配的「外挂记忆」。LLM 的训练数据有截止时间也不知道你公司内部的文档。RAG 的做法是先把你的知识库切片、向量化存进向量库用户提问时先检索出最相关的几段拼进 prompt 再交给 LLM。这样模型回答时就有据可依而不是凭空编。MCP 协议是给大脑配的「标准手」。LLM 想读文件、查数据库、调 API过去每家都要自己写一套 function calling 的适配代码。MCPModel Context Protocol把这些能力抽象成统一的 Server任何支持 MCP 的客户端都能即插即用。你可以把它理解成「AI 世界的 USB-C 接口」。A2A 协议是「大脑之间的对话规则」。当你有多个 Agent——一个负责查日志、一个负责查代码、一个负责写报告——它们之间怎么传任务、怎么回结果A2AAgent2Agent就是干这个的让不同框架、不同厂商的 Agent 能互相通信而不是各说各话。AI Agent 则是把上面四样东西组装起来的「完整的人」。它有自己的目标、会用 LLM 做决策、会通过 MCP 调工具、会通过 RAG 查资料、会通过 A2A 和其他 Agent 协作。所以它们不是五个平行的概念而是一条链路上的五个角色。这篇就按「从零搭一条多智能体协作链路」的顺序把每个环节的可运行配置都给你模型调用统一走 TaoToken 的 Key省得你为了试不同模型注册一堆账号。适合谁看已经会写 Python、调过至少一次大模型 API、想搞清楚 Agent 工程化落地长什么样的开发者。不需要你精通任何一个协议但需要你愿意动手把配置跑通。2. TaoToken 前置准备一个 Key 打通多模型调用与 MCP 服务端配置在搭链路之前先把「模型调用」这一层统一掉。多智能体场景里不同 Agent 往往需要不同模型规划用推理强的检索改写用便宜的代码生成用专门的。如果每个模型都单独申请 Key、单独记 Base URL配置会迅速失控。TaoToken 在这里的作用是提供一个统一的 OpenAI 兼容入口。你只需要一个 Key、一个 Base URL就能在多个模型之间切换代码里改个 model 字段就行。2.1 获取 Key 与确认接入信息先到控制台创建 API Keyhttps://taotoken.net/console/api-keys?utm_sourcetaotoken_aicg_blog_endutm_contentapi_keysutm_campaignrewrite创建后你会拿到一串以sk-开头的 Key。接入信息固定为配置项值Base URLhttps://taotoken.net/apiAPI Key你创建的sk-...兼容协议OpenAI Chat Completions模型 ID以控制台模型列表为准如gpt-4o-mini、claude-3-5-sonnet等注意 Base URL 结尾不要多加/v1OpenAI SDK 会自己拼/chat/completions。如果你用的是某些要求带/v1的客户端写成https://taotoken.net/api/v1也可以两种都能通。2.2 用环境变量管理 Key不要把 Key 硬编码进代码。统一用环境变量export TAOTOKEN_API_KEYsk-你的key export TAOTOKEN_BASE_URLhttps://taotoken.net/apiWindows PowerShell 用$env:TAOTOKEN_API_KEYsk-你的key $env:TAOTOKEN_BASE_URLhttps://taotoken.net/api2.3 先做一次最小连通性验证在写任何 Agent 逻辑之前先确认 Key 能用。新建test_conn.pyimport os from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) resp client.chat.completions.create( modelgpt-4o-mini, messages[{role: user, content: 只回复两个字通了}], ) print(resp.choices[0].message.content)跑python test_conn.py输出「通了」就说明模型调用层已经打通。这一步别跳过后面 MCP、A2A 出问题时你能快速判断是协议层的问题还是 Key 的问题。2.4 为什么多智能体场景更需要统一入口多 Agent 链路里一次任务可能触发十几次模型调用规划 Agent 调一次、检索改写调一次、每个子 Agent 各调几次、汇总再调一次。如果这些调用分散在四五个平台出问题时你要挨个查余额、查限流、查日志。统一到一个入口后你只需要在一个地方看调用记录排查成本大幅下降。另外MCP 服务端本身也可能需要调用 LLM比如做意图识别A2A 的消息路由有时也要模型判断。这些「基础设施级」的调用同样走这个 Key配置就收敛成一份。3. 可复制配置MCP 服务端、A2A 通信与 RAG 向量库参数这一节是全文的核心三个配置文件你直接抄改就能用。建议按 MCP → RAG → A2A 的顺序搭因为后一个依赖前一个的产物。3.1 MCP 服务端配置stdio 模式MCP Server 的作用是把「工具」暴露给 Agent。我们写一个最小可用的 Server提供两个工具读本地文件、查 RAG 向量库。先装依赖pip install mcp openai chromadb新建mcp_server.pyimport os import json from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent import chromadb app Server(taotoken-demo-server) # 初始化向量库RAG 部分见 3.2 chroma_client chromadb.PersistentClient(path./chroma_db) collection chroma_client.get_or_create_collection(nameknowledge) app.list_tools() async def list_tools(): return [ Tool( nameread_file, description读取指定路径的文本文件内容, inputSchema{ type: object, properties: {path: {type: string}}, required: [path], }, ), Tool( namesearch_knowledge, description在知识库中检索与查询最相关的片段, inputSchema{ type: object, properties: { query: {type: string}, top_k: {type: integer, default: 3}, }, required: [query], }, ), ] app.call_tool() async def call_tool(name: str, arguments: dict): if name read_file: path arguments[path] with open(path, r, encodingutf-8) as f: return [TextContent(typetext, textf.read())] if name search_knowledge: query arguments[query] top_k arguments.get(top_k, 3) results collection.query(query_texts[query], n_resultstop_k) docs results[documents][0] return [TextContent(typetext, text\n---\n.join(docs))] raise ValueError(f未知工具: {name}) async def main(): async with stdio_server() as (read, write): await app.run(read, write, app.create_initialization_options()) if __name__ __main__: import asyncio asyncio.run(main())对应的客户端配置以 Claude Desktop 风格的claude_desktop_config.json为例路径按你实际安装位置改{ mcpServers: { taotoken-demo: { command: python, args: [/absolute/path/to/mcp_server.py], env: { TAOTOKEN_API_KEY: sk-你的key, TAOTOKEN_BASE_URL: https://taotoken.net/api } } } }三件套在这里的体现command是启动方式env里带上了 Base URL 和 Key工具内部如果要调模型Model ID 从环境变量读。这样 Server 本身不绑定任何具体模型。3.2 RAG 向量库参数配置RAG 的效果七成取决于切片和检索参数。下面这份配置是我实测下来比较稳的起点import chromadb from chromadb.utils import embedding_functions # 用 TaoToken 兼容的 embedding 接口若你的模型列表含 embedding 模型 # 也可先用 chromadb 默认的 all-MiniLM-L6-v2 本地模型 embed_fn embedding_functions.DefaultEmbeddingFunction() client chromadb.PersistentClient(path./chroma_db) collection client.get_or_create_collection( nameknowledge, embedding_functionembed_fn, metadata{hnsw:space: cosine}, ) # 切片参数 CHUNK_SIZE 500 # 每片字符数 CHUNK_OVERLAP 80 # 相邻片重叠防止语义被切断 TOP_K 3 # 检索返回条数 SCORE_THRESHOLD 0.35 # 低于此相似度的结果丢弃 def split_text(text: str): chunks [] start 0 while start len(text): end start CHUNK_SIZE chunks.append(text[start:end]) start end - CHUNK_OVERLAP return chunks def ingest(doc_id: str, text: str): chunks split_text(text) collection.add( ids[f{doc_id}-{i} for i in range(len(chunks))], documentschunks, )参数说明hnsw:space选 cosine 是因为文本相似度用余弦更稳CHUNK_SIZE500 是中文场景的经验值英文可以放到 800SCORE_THRESHOLD很关键没有它检索永远返回 top_k 条哪怕全是无关内容反而污染 prompt。3.3 A2A 通信配置Agent Card 消息格式A2A 的核心是 Agent Card——每个 Agent 用一张 JSON 卡片声明自己是谁、能干什么、怎么调用。下面是一个「日志分析 Agent」的卡片{ name: log-analyzer, description: 接收日志文本定位异常模式并返回结构化结论, url: http://localhost:8001/a2a, version: 1.0.0, capabilities: { streaming: false, pushNotifications: false }, skills: [ { id: analyze_log, name: 日志异常分析, description: 输入原始日志输出异常时间段与可能原因, inputModes: [text/plain], outputModes: [application/json] } ], defaultInputModes: [text/plain], defaultOutputModes: [application/json] }调用方按 A2A 的消息格式发任务{ jsonrpc: 2.0, id: task-001, method: tasks/send, params: { id: task-001, message: { role: user, parts: [ {type: text, text: 分析这段日志ERROR 2024-xx-xx timeout ...} ] } } }服务端返回{ jsonrpc: 2.0, id: task-001, result: { id: task-001, status: {state: completed}, artifacts: [ { name: analysis, parts: [ {type: text, text: {\window\:\10:00-10:05\,\cause\:\下游超时\}} ] } ] } }关键点A2A 用 JSON-RPC 2.0 做传输用parts数组支持多模态用artifacts承载结果。你不需要自己发明协议照这个结构填就行。多个 Agent 各自暴露一个/a2a端点编排器按 Agent Card 决定把任务发给谁。4. 验证请求从单 Agent 到多智能体协作链路的完整跑通配置写完现在验证整条链路。分三步先验证 MCP 工具能被调用再验证 RAG 检索有结果最后把两个 Agent 串起来跑一次协作。4.1 验证 MCP 工具调用写一个测试客户端直接调用mcp_server.py里的工具import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): params StdioServerParameters( commandpython, args[mcp_server.py], ) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(可用工具:, [t.name for t in tools.tools]) result await session.call_tool( search_knowledge, {query: 超时告警怎么处理, top_k: 2}, ) print(检索结果:, result.content[0].text) asyncio.run(main())预期输出会列出read_file和search_knowledge两个工具并返回检索到的片段。如果检索为空说明 3.2 的ingest还没执行先灌点数据进去。4.2 验证 RAG 检索质量灌入测试数据后跑一次检索重点看相似度分数res collection.query( query_texts[数据库连接超时], n_results3, include[documents, distances], ) for doc, dist in zip(res[documents][0], res[distances][0]): print(f距离{dist:.4f} | {doc[:60]})距离越小越相关。如果所有距离都接近 1说明 embedding 模型和你的语料不匹配考虑换模型或调整切片大小。4.3 多智能体协作链路跑通现在把「日志分析 Agent」和「报告生成 Agent」串起来。编排器逻辑import os, json, httpx from openai import OpenAI client OpenAI( api_keyos.environ[TAOTOKEN_API_KEY], base_urlos.environ[TAOTOKEN_BASE_URL], ) def call_agent(url: str, text: str): payload { jsonrpc: 2.0, id: task-001, method: tasks/send, params: { id: task-001, message: {role: user, parts: [{type: text, text: text}]}, }, } r httpx.post(url, jsonpayload, timeout60) return r.json()[result][artifacts][0][parts][0][text] # 第一步日志分析 Agent analysis call_agent(http://localhost:8001/a2a, ERROR timeout at 10:03 ...) # 第二步报告生成 Agent内部用 LLM 润色 report client.chat.completions.create( modelgpt-4o-mini, messages[ {role: system, content: 你是运维报告助手把分析结论写成三段式报告。}, {role: user, content: analysis}, ], ) print(report.choices[0].message.content)跑通后你会看到日志文本 → 分析 Agent 返回结构化结论 → 报告 Agent 用 LLM 生成可读报告。整条链路里MCP 负责工具调用RAG 负责知识检索A2A 负责 Agent 间通信LLM 负责生成各司其职。4.4 成功结果的判断标准一次成功的链路跑通应该满足MCP 工具列表非空且能返回内容RAG 检索距离小于 0.5A2A 调用返回status.state completed最终报告包含具体时间窗口和原因而不是「可能是网络问题」这种空话。任何一环不满足回到对应小节排查。5. 本篇常见错排查401、local proxy failed、reading choices、OAuth 报错对照搭链路时最容易卡在几个固定报错上这里按真实错误信息对照排查。5.1 401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是 Key 没读到或写错。检查echo $TAOTOKEN_API_KEY是否有值代码里是否用了os.environ[TAOTOKEN_API_KEY]而不是硬编码。如果 Key 是从控制台复制的注意别把前后空格带进去。还有一种情况Base URL 写成了https://taotoken.net/api/带尾斜杠某些 SDK 会拼出双斜杠导致鉴权失败去掉尾斜杠即可。5.2 local proxy failedhttpx.ConnectError: [Errno 111] Connection refused或客户端提示local proxy failed。这通常是环境里残留了代理配置而代理服务没启动。检查env | grep -i proxy如果有HTTP_PROXY、HTTPS_PROXY指向本地端口先 unset 掉再跑。注意这里说的是清理无效的本地代理残留不是让你去配代理。5.3 reading choices 报错TypeError: NoneType object is not subscriptable出现在resp.choices[0]这一行。说明返回体里没有choices字段。常见原因模型 ID 写错服务端返回了错误 JSON或者流式请求没处理完就取 choices。先打印完整resp看结构确认model字段拼写和控制台一致。5.4 OAuth 相关报错Error: OAuth token expired / invalid_grant如果你用的是 Claude Code 这类带 OAuth 的客户端报这个说明登录态过期。重新走一次登录流程即可。注意 OAuth 和 API Key 是两套体系OAuth 用于客户端登录API Key 用于程序调用。在 MCP Server 的env里要填的是 API Key不是 OAuth token。5.5 MCP 工具列表为空客户端连上了但list_tools返回空。检查mcp_server.py里app.list_tools()装饰器是否写对以及stdio_server是否正确启动。用python mcp_server.py直接跑如果没有报错且进程挂起等待输入说明 Server 本身正常问题在客户端配置的args路径。5.6 A2A 调用超时httpx.ReadTimeout。先确认目标 Agent 的端口在监听curl http://localhost:8001/a2a。如果 Agent 内部要调 LLM超时时间设长一点60 秒起步。另外 A2A 的tasks/send是同步语义长任务建议改用流式或轮询tasks/get。6. 把链路跑稳之后模型切换、成本控制与下一步链路能跑通只是起点真正上线前还有几件事值得做。模型切换要无痛。因为所有调用都走统一的 Base URL 和 Key你只需要改model字段就能在推理模型和便宜模型之间切换。建议在配置里把「规划用模型」「执行用模型」「汇总用模型」分开定义方便按任务类型路由。成本控制靠分层。多智能体链路里模型调用次数是单 Agent 的好几倍。把简单任务意图分类、格式转换交给小模型复杂任务规划、代码生成才用大模型。RAG 的TOP_K也别设太大3 到 5 条通常够用多了既费 token 又稀释相关性。MCP 工具要收敛。不要把所有接口都暴露成工具Agent 面对几十个工具时选择准确率会下降。按场景分组每个 Agent 只挂它真正需要的三五个工具。A2A 的 Agent Card 要写清楚。description和skills是编排器做路由的依据写得越具体任务分发越准。别写「处理各种任务」这种废话。下一步可以尝试的方向给 RAG 加 rerank 模型提升检索精度把 A2A 的同步调用改成流式让长任务有中间反馈用 MCP 的 resources 能力把静态知识也纳入统一管理。这些都可以在现有链路上增量加不用推倒重来。如果你还没开始建议就按这篇的顺序先拿 Key 跑通 2.3 的连通性测试再抄 3.1 的 MCP Server灌点数据验证 RAG最后用 4.3 的编排器把两个 Agent 串起来。跑通一次这五个概念就不再是名词而是你手里能改的配置。