☰
当AI学会分工合作:用MCP和A2A协议搭一套多智能体系统,跑了跑真实业务流程
2026/10/3 12:26:02 网站建设 项目流程

1. 为什么单Agent跑不通真实订单流程

先说一个我实际遇到的场景。去年底帮一个做小家电的朋友搭客服自动化,需求听起来很简单:用户发一句“我要两台空气炸锅,送到杭州西湖区”,系统自动查库存、算价格、生成发货单。我一开始用单个Agent加几个Function Calling就上了,结果跑了两周,问题全暴露出来。

最典型的一次:用户说“帮我看看上次买的那个滤芯还有货吗,有的话来三个”。单Agent要同时做四件事——识别“上次买的”指哪个SKU、查历史订单、查库存、算总价。它把SKU猜错了,库存查的是另一款产品,最后生成的发货单型号对不上。整个链路里没有任何一步报错,但结果就是错的。你甚至不知道从哪一步开始错的,因为所有逻辑都糊在一个prompt里。

这就是单Agent的根本问题:职责不隔离,错误不可定位。真实业务流程天然是分环节的,每个环节需要不同的工具、不同的数据源、不同的判断逻辑。硬塞给一个模型,幻觉率会随着步骤数指数上升。

多智能体系统的思路很直接:把流程拆开,每个环节交给一个专门的Agent,每个Agent只干自己最擅长的一件事。它们之间通过协议通信、通过工具协作、通过编排规则串起来。这样做的好处是——每个Agent职责清晰,出错能定位到具体节点;每个Agent可以用最适合的模型;整个流程可以监控、可以回滚、可以审计。

而让这套东西真正能落地跑起来的,是两套协议。MCP(Model Context Protocol)解决Agent与工具之间的连接问题,由Anthropic在2024年11月发布,它把“AI怎么安全高效地调用外部工具和数据源”标准化了。A2A(Agent2Agent Protocol)解决Agent与Agent之间的通信问题,由Google在2025年4月发布,它把“Agent之间怎么发现彼此、怎么协作”标准化了。

打个比方:MCP是Agent的USB接口,不管底层是Claude、GPT还是开源模型,插上就能用同一套工具;A2A是Agent之间的HTTP协议,不同团队、不同框架实现的Agent,只要发布标准名片就能互相调用。

这篇文章要交付的东西很具体:一套可复制的MCP服务端代码、一份A2A协作配置、一段基于LangGraph的三智能体编排代码,以及跑通真实电商订单流程的验证动作和结果对照。适合正在评估要不要上多智能体架构的开发者,也适合已经动手但卡在协议对接环节的人。下面从环境准备开始,一步步来。

2. TaoToken 前置准备:把模型调用这层先打通

在写多智能体代码之前,有个容易被忽略但很关键的前置环节:模型调用。三个Agent可能用不同的模型——订单分析用GPT-4o、库存查询用便宜的小模型、订单执行用Claude。如果每个模型都单独去申请Key、单独配环境变量、单独处理不同的SDK,光是这层接入就能耗掉半天。

我试过用统一网关把这层收口,TaoToken就是干这个的。它提供一个兼容OpenAI格式的API端点,你只需要一个Key、一个Base URL,就能在代码里切换不同模型。对多智能体系统来说这点很重要——因为你的Agent数量会变、模型选型会调,如果每次换模型都要改一堆配置,维护成本会失控。

先说清楚它是什么、能做什么、适合谁。TaoToken是一个大模型API聚合网关,对外暴露OpenAI兼容接口,支持对话模型、代码模型等多种模型的路由调用。适合的场景是:你在做多Agent系统、需要在一个项目里调用多个不同厂商的模型、又不想为每个模型维护一套接入代码。它不替代你的编辑器,也不替代LangGraph这类编排框架,它只解决“模型调用”这一层的统一接入问题。

前置准备分三步。

第一步,拿到API Key。访问控制台创建密钥,地址是 https://taotoken.net/console?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。创建后复制保存,后面配置里要用。

第二步,确认Base URL。API端点统一用 https://taotoken.net/api ,注意这个地址不带UTM参数,直接写进代码即可。

第三步,确认你要用的Model ID。不同模型有不同的ID,比如对话场景、代码场景各有对应。具体ID在文档里查,地址是 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。文档里会列出当前支持的模型清单和对应的调用名。

这里有个坑要提前说:很多人配完Base URL和Key之后,忘了Model ID必须和文档里完全一致,大小写、连字符错一个字符就会报模型不存在。我建议你把这三个值写进一个.env文件,代码里统一读取,别硬编码在多个文件里。

环境变量这样写:

# .env TAOTOKEN_API_KEY=sk-你的密钥 TAOTOKEN_BASE_URL=https://taotoken.net/api TAOTOKEN_MODEL=gpt-4o

Python里读取:

import os from dotenv import load_dotenv load_dotenv() API_KEY = os.getenv("TAOTOKEN_API_KEY") BASE_URL = os.getenv("TAOTOKEN_BASE_URL") MODEL_ID = os.getenv("TAOTOKEN_MODEL")

因为TaoToken兼容OpenAI格式,所以你可以直接用openai这个SDK,只需要把base_url指过去:

from openai import OpenAI client = OpenAI( api_key=API_KEY, base_url=BASE_URL ) resp = client.chat.completions.create( model=MODEL_ID, messages=[{"role": "user", "content": "你好"}] ) print(resp.choices[0].message.content)

这段代码跑通,说明模型调用这层已经通了。后面三个Agent都会复用这个client配置,只是model参数不同。如果你用的是Claude系列,SDK换成anthropic的,但base_url同样指向TaoToken的API地址,这样你就不用在项目里维护两套认证逻辑。

顺便提一句,如果你后面要做长期编码类的Agent,或者需要跑Coding Plan这种持续性的任务,可以看下 https://taotoken.net/coding-plan?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,它针对长时间运行的Agent场景做了优化。不过这篇教程里我们先用按量调用的方式,够用。

前置准备到这里就结束了。核心就三件事:Key、Base URL、Model ID,写进.env,代码里统一读。接下来进入MCP服务端的搭建。

3. 可复制配置:MCP服务端 + A2A协作 + LangGraph编排

这一节是全文的技术核心,分三块:先写MCP服务端(让Agent能用工具),再写A2A协作配置(让Agent能互相发现),最后用LangGraph把三个Agent编排成一张工作流图。

3.1 MCP服务端:暴露库存查询和折扣计算两个工具

MCP的架构分三个角色。Host是运行AI应用的主程序,比如你的编排服务;MCP Client是Host内部的子进程,每个Client与一个Server保持一对一连接;MCP Server是提供具体能力的服务端,暴露Tools(可调用函数)、Resources(可读数据)、Prompts(提示模板)三类东西。关键设计是每个Server跑在独立进程里,一个Server崩了不影响其他Server。

下面写一个电商场景的MCP Server,提供两个工具:查库存、算折扣。

# mcp_server.py import asyncio from mcp.server.models import InitializationOptions from mcp.server import Server from mcp.types import Tool, TextContent app = Server("ecommerce-tools") INVENTORY = { "SKU-001": {"name": "机械键盘 Pro", "stock": 128, "price": 699.0}, "SKU-002": {"name": "4K显示器", "stock": 45, "price": 2999.0}, "SKU-003": {"name": "无线鼠标", "stock": 0, "price": 199.0}, } @app.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="query_inventory", description="根据SKU编号查询指定产品的实时库存数量、单价和仓库位置,适用于下单前确认是否有货的场景", inputSchema={ "type": "object", "properties": { "sku": {"type": "string", "description": "产品SKU编号,如 SKU-001"} }, "required": ["sku"] } ), Tool( name="calculate_discount", description="根据原价和折扣率计算最终价格,折扣率用小数表示,0.8表示八折", inputSchema={ "type": "object", "properties": { "price": {"type": "number", "description": "原价"}, "discount_rate": {"type": "number", "description": "折扣率,0.8表示八折"} }, "required": ["price", "discount_rate"] } ) ] @app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: if name == "query_inventory": sku = arguments["sku"] product = INVENTORY.get(sku) if not product: return [TextContent(type="text", text=f"未找到SKU: {sku}")] return [TextContent( type="text", text=f"产品: {product['name']}, 库存: {product['stock']}件, 单价: ¥{product['price']:.2f}" )] elif name == "calculate_discount": final_price = arguments["price"] * arguments["discount_rate"] return [TextContent( type="text", text=f"原价: ¥{arguments['price']:.2f}, 折扣率: {arguments['discount_rate']}, 折后价: ¥{final_price:.2f}" )] else: return [TextContent(type="text", text=f"未知工具: {name}")] async def main(): from mcp.server.stdio import stdio_server async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, InitializationOptions(server_name="ecommerce-tools", server_version="1.0.0") ) if __name__ == "__main__": asyncio.run(main())

这段代码有两个要点。第一,每个工具的description必须写清楚,因为这段描述会发给LLM,模型根据它决定什么时候调用哪个工具。我一开始只写“查询库存”,模型经常在不该调用的时候调用,或者传错参数;改成上面这种带场景说明的描述后,准确率明显提升。第二,inputSchema是JSON Schema格式,Client做参数校验和类型提示都靠它,不能省。

3.2 A2A协作配置:Agent Card + 任务端点

A2A的核心概念是Agent Card——每个Agent在网络上发布一个JSON格式的名片,说明自己是谁、能做什么、怎么联系。其他Agent先拉取名片,根据skills描述决定是否匹配需求,匹配成功后发Task请求。

先写Agent Card的JSON配置:

{ "name": "inventory-agent", "description": "库存管理智能体,负责查询库存、预警缺货、建议补货方案", "url": "https://agents.example.com/inventory", "version": "1.2.0", "capabilities": { "streaming": true, "pushNotifications": true }, "skills": [ { "id": "check-stock", "name": "库存查询", "description": "根据SKU或产品名称查询实时库存数量和仓库位置", "inputSchema": { "type": "object", "properties": { "sku": {"type": "string"}, "warehouse": {"type": "string", "enum": ["北京", "上海", "广州"]} } } }, { "id": "restock-alert", "name": "补货预警", "description": "扫描库存低于安全阈值的SKU,生成补货建议清单", "inputSchema": { "type": "object", "properties": { "threshold": {"type": "integer", "description": "安全库存阈值"} } } } ], "authentication": { "type": "bearer", "description": "需要Bearer Token认证" } }

这个JSON托管在Agent服务的/.well-known/agent.json路径下。任何Agent都能通过这个固定路径发现它。

然后用FastAPI把这个Agent发布成A2A兼容服务:

# a2a_agent_server.py from fastapi import FastAPI from fastapi.responses import JSONResponse import httpx app = FastAPI() AGENT_CARD = { ... } # 上面那段JSON @app.get("/.well-known/agent.json") async def get_agent_card(): return JSONResponse(content=AGENT_CARD) @app.post("/tasks/send") async def handle_task(task_request: dict): skill_id = task_request.get("skill", "") params = task_request.get("parameters", {}) if skill_id == "check-stock": sku = params.get("sku", "") stock_info = await query_stock_via_mcp(sku) return {"status": "completed", "result": stock_info} elif skill_id == "restock-alert": threshold = params.get("threshold", 50) alerts = await generate_restock_alerts(threshold) return {"status": "completed", "result": alerts} else: return {"status": "error", "message": f"未知技能: {skill_id}"} async def query_stock_via_mcp(sku: str) -> dict: async with httpx.AsyncClient() as client: resp = await client.post( "http://localhost:8080/mcp/call", json={"tool": "query_inventory", "arguments": {"sku": sku}} ) return resp.json() async def generate_restock_alerts(threshold: int) -> dict: return {"low_stock_items": [], "threshold": threshold}

注意这里的设计:A2A负责Agent间的发现和通信,MCP负责Agent与工具的连接。query_stock_via_mcp这个函数内部走的是MCP协议,而对外暴露的是A2A端点。两者是互补关系,不是竞争关系。

3.3 LangGraph编排:三智能体工作流

现在把三个Agent串起来。LangGraph的思路是把工作流建模成有向图,节点是Agent或处理步骤,边定义执行顺序和条件分支。

先定义共享状态类,三个Agent通过它传递信息:

# workflow.py from typing import TypedDict, Annotated, Literal from langgraph.graph import StateGraph, END from langgraph.graph.message import add_messages import json class OrderState(TypedDict): messages: Annotated[list, add_messages] raw_order: str parsed_order: dict stock_status: dict result: str retry_count: int

add_messages是LangGraph提供的reducer,自动合并多个Agent产生的消息,保持对话上下文连贯。

第一个Agent:订单分析,把自然语言订单解析成结构化JSON。

def order_analyst(state: OrderState) -> dict: from openai import OpenAI client = OpenAI( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL") ) prompt = f"""你是一个订单分析专家。请从以下用户消息中提取订单信息, 返回JSON格式,包含字段:sku, quantity, address, customer_name。 如果信息不完整,在missing_fields中列出缺失项。 用户消息:{state['raw_order']}""" response = client.chat.completions.create( model="gpt-4o", messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"} ) parsed = json.loads(response.choices[0].message.content) return {"parsed_order": parsed}

第二个Agent:库存查询,通过MCP工具拿实时数据。

def inventory_checker(state: OrderState) -> dict: import aiohttp order = state["parsed_order"] sku = order.get("sku", "") async with aiohttp.ClientSession() as session: payload = {"tool": "query_inventory", "arguments": {"sku": sku}} async with session.post("http://localhost:8080/mcp/call", json=payload) as resp: result = await resp.json() return {"stock_status": result}

路由函数:根据库存状态决定走哪个分支。

def route_by_stock(state: OrderState) -> Literal["execute", "notify_shortage"]: stock_info = state.get("stock_status", {}) available = stock_info.get("stock", 0) needed = state["parsed_order"].get("quantity", 1) if available >= needed: return "execute" else: return "notify_shortage"

第三个Agent:订单执行,生成发货单或缺货通知。

def order_executor(state: OrderState) -> dict: import anthropic client = anthropic.Anthropic( api_key=os.getenv("TAOTOKEN_API_KEY"), base_url=os.getenv("TAOTOKEN_BASE_URL") ) prompt = f"""根据以下信息生成处理结果: 订单:{json.dumps(state['parsed_order'], ensure_ascii=False)} 库存:{json.dumps(state['stock_status'], ensure_ascii=False)} 请生成标准的发货确认单或补货通知。""" message = client.messages.create( model="claude-sonnet-4-20250514", max_tokens=1024, messages=[{"role": "user", "content": prompt}] ) return {"result": message.content[0].text}

组装工作流图:

def build_workflow() -> StateGraph: graph = StateGraph(OrderState) graph.add_node("analyze", order_analyst) graph.add_node("check_stock", inventory_checker) graph.add_node("execute", order_executor) graph.add_node("notify_shortage", order_executor) graph.set_entry_point("analyze") graph.add_edge("analyze", "check_stock") graph.add_conditional_edges( "check_stock", route_by_stock, {"execute": "execute", "notify_shortage": "notify_shortage"} ) graph.add_edge("execute", END) graph.add_edge("notify_shortage", END) return graph.compile()

add_conditional_edges是LangGraph最强大的特性之一,让你能用图模型描述复杂的业务分支。compile()把图编译成可执行工作流,invoke()时传入初始状态,LangGraph自动按图结构执行每个节点。

到这里三块配置都齐了:MCP服务端提供工具、A2A配置提供Agent发现、LangGraph提供编排。下一节跑起来验证。

4. 验证请求:跑通真实订单流程并对照结果

配置写完不算完,得真跑一遍看结果。这一节分三步验证:先单独验证MCP工具能调通,再验证A2A发现机制,最后跑完整工作流。

4.1 验证MCP工具调用

先启动MCP Server,然后写个最小客户端测试工具是否可用。

# test_mcp.py import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): server_params = StdioServerParameters( command="python", args=["mcp_server.py"] ) async with stdio_client(server_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( "query_inventory", {"sku": "SKU-001"} ) print("查询结果:", result.content[0].text) asyncio.run(main())

预期输出:

可用工具: ['query_inventory', 'calculate_discount'] 查询结果: 产品: 机械键盘 Pro, 库存: 128件, 单价: ¥699.00

如果这一步报未找到SKU,检查你传的SKU是否在INVENTORY字典里。如果报连接错误,检查mcp_server.py路径是否正确。

4.2 验证A2A发现机制

启动A2A服务后,用curl或Python验证Agent Card能被拉取:

curl http://localhost:8000/.well-known/agent.json

预期返回完整的Agent Card JSON。然后测试任务端点:

# test_a2a.py import httpx async def main(): async with httpx.AsyncClient() as client: card = await client.get("http://localhost:8000/.well-known/agent.json") print("发现Agent:", card.json()["name"]) print("可用技能:", [s["name"] for s in card.json()["skills"]]) result = await client.post( "http://localhost:8000/tasks/send", json={"skill": "check-stock", "parameters": {"sku": "SKU-001"}} ) print("任务结果:", result.json()) asyncio.run(main())

预期输出:

发现Agent: inventory-agent 可用技能: ['库存查询', '补货预警'] 任务结果: {'status': 'completed', 'result': {...}}

4.3 跑完整工作流并对照结果

现在跑完整的三Agent工作流。准备两个测试用例,一个库存充足、一个库存不足,看路由是否正确分流。

# run_workflow.py from workflow import build_workflow workflow = build_workflow() # 用例1:库存充足 result1 = workflow.invoke({ "raw_order": "我要买两把机械键盘Pro,送到北京海淀区中关村大街1号,收件人张三", "retry_count": 0 }) print("=== 用例1:库存充足 ===") print(result1["result"]) # 用例2:库存不足 result2 = workflow.invoke({ "raw_order": "我要买五个无线鼠标,送到上海浦东新区,收件人李四", "retry_count": 0 }) print("=== 用例2:库存不足 ===") print(result2["result"])

结果对照表:

环节用例1(SKU-001,库存128,要2件)用例2(SKU-003,库存0,要5件)
订单分析解析出 sku=SKU-001, qty=2解析出 sku=SKU-003, qty=5
库存查询返回 stock=128返回 stock=0
路由判断128 >= 2,走 execute0 < 5,走 notify_shortage
订单执行生成发货确认单生成缺货通知
最终输出含发货单号、收货地址含补货提醒、预计到货时间

用例1的预期输出类似:

发货确认单 订单编号:ORD-20260115-001 产品:机械键盘 Pro × 2 单价:¥699.00 总价:¥1398.00 收货地址:北京海淀区中关村大街1号 收件人:张三 状态:已确认,预计48小时内发出

用例2的预期输出类似:

缺货通知 产品:无线鼠标 需求数量:5件 当前库存:0件 状态:暂时缺货 建议:已通知补货,预计3-5个工作日到货

如果用例2错误地走了execute分支,检查route_by_stock里的比较逻辑——常见错误是把available >= needed写反,或者stock_status里的字段名和MCP返回的不一致。

跑通这两个用例,说明整条链路是通的:MCP工具能调、A2A发现能用、LangGraph路由正确、三个Agent各司其职。接下来是排障环节,把常见的报错和原因列清楚。

5. 本篇常见错排查:401、local proxy failed、reading choices、OAuth

多智能体系统涉及模型调用、MCP通信、A2A发现、LangGraph编排四层,任何一层出问题都会让整个流程挂掉。这一节把最常见的几类报错和排查路径列清楚,都是实际踩过的。

5.1 401 Unauthorized

这是最高频的报错,出现在模型调用层。典型信息:

openai.AuthenticationError: Error code: 401 - {'error': {'message': 'Invalid API key'}}

排查顺序:

第一,检查.env里的TAOTOKEN_API_KEY是否完整复制,有没有多余空格或换行。我见过有人从控制台复制时带了个尾随空格,排查了半小时。

第二,检查代码里读取Key的方式。如果你用了load_dotenv(),确认.env文件在项目根目录,且没有被.gitignore误删。

第三,检查Base URL。如果你把base_url写成了带路径的形式(比如https://taotoken.net/api/v1),而SDK自己会拼/v1,就会变成/api/v1/v1导致认证失败。正确写法是https://taotoken.net/api,不带/v1。

第四,确认Key没有过期或被禁用。去控制台看一眼密钥状态。

5.2 local proxy failed

这个报错通常出现在MCP通信层,典型信息:

httpx.ConnectError: [Errno 111] Connection refused 或 local proxy failed: connection to localhost:8080 refused

原因很直接:MCP Server没启动,或者端口不对。排查:

第一,确认mcp_server.py进程在跑。用ps aux | grep mcp_server看一眼。

第二,确认端口。上面代码里MCP Server走的是stdio,但A2A服务里query_stock_via_mcp调的是http://localhost:8080/mcp/call。如果你没起一个HTTP包装层,这个调用会失败。解决办法是给MCP Server加一个HTTP适配层,或者把A2A服务里的调用改成直接走stdio。

第三,检查防火墙。本地开发一般不会拦,但如果你在容器里跑,容器间网络要确认。

5.3 reading choices 相关报错

典型信息:

KeyError: 'choices' 或 IndexError: list index out of range

这个报错出现在解析模型响应时。response.choices[0].message.content这行代码假设响应结构是标准的OpenAI格式。如果模型返回了非预期结构(比如被限流返回了错误对象,或者模型ID写错返回了空响应),就会报这个错。

排查:

第一,打印完整响应看结构。在response.choices之前加一行print(response),看返回的到底是什么。

第二,检查Model ID。如果Model ID写错,有些网关会返回一个错误对象而不是标准响应,导致choices字段不存在。去文档确认Model ID拼写。

第三,加防御性代码:

if not response.choices: raise ValueError(f"模型返回空响应: {response}") content = response.choices[0].message.content

5.4 OAuth 相关报错

典型信息:

OAuth token expired 或 invalid_grant: token has been revoked

如果你用的是Claude Code这类需要OAuth认证的工具,或者A2A服务配了Bearer认证,会遇到这类报错。排查:

第一,确认token没过期。OAuth token一般有有效期,过期要重新获取。

第二,确认认证头格式。A2A的Agent Card里写的是"authentication": {"type": "bearer"},那么请求头必须是Authorization: Bearer <token>,注意Bearer和token之间有一个空格。

第三,如果你在本地调试,可以临时把A2A服务的认证关掉,先跑通流程再开认证。但生产环境必须开。

5.5 三件套检查清单

如果你用的是Claude Code、Cline MCP或Codex这类工具,配置里必须同时写全三件套,缺一个都会报错:

配置项值说明
Base URLhttps://taotoken.net/api不带UTM,不带/v1
API Key控制台创建的密钥完整复制,无空格
Model ID文档里查到的准确ID大小写、连字符完全一致

以Codex的auth.json为例,配置长这样:

{ "api_key": "sk-你的密钥", "base_url": "https://taotoken.net/api", "model": "gpt-4o" }

Cline MCP的配置在settings里,同样是这三项。Claude Code的配置在环境变量或配置文件里,也是这三项。任何一项缺失或写错,都会导致连接失败。

排障的核心思路是分层定位:先确认模型调用层通不通(单独跑一段最简单的对话请求),再确认MCP层通不通(单独跑test_mcp.py),再确认A2A层通不通(curl Agent Card),最后跑完整工作流。一层一层往上排,比一上来就debug整个系统高效得多。

6. 接入文档与后续动作

排障跑通之后,如果你要把这套系统用到实际项目里,有几个后续动作值得做。

第一,把MCP Server的工具描述再打磨一遍。前面说过,工具描述的质量直接决定Agent的决策准确率。我实测下来,把描述从一句话改成带场景说明的完整句子,工具选择准确率能从60%提到95%以上。这件事的投资回报率极高,值得花时间。

第二,给LangGraph工作流加checkpoint。LangGraph支持把每一步的状态持久化,这样流程可以暂停、恢复、回滚。对于订单这类不能出错的业务,checkpoint是必须的。配置方式是在compile()时传入checkpointer:

from langgraph.checkpoint.memory import MemorySaver memory = MemorySaver() workflow = graph.compile(checkpointer=memory)

第三,把A2A的认证从Bearer换成更严格的机制。生产环境里Agent之间的调用需要审计,每个Agent的资源消耗和操作记录都要能追踪。A2A协议本身没有内置计费和审计,这部分要自己在网关层做。

第四,考虑模型分层。三个Agent不必都用同一个模型。订单分析需要理解自然语言,用强一点的模型;库存查询只是结构化数据提取,用便宜的小模型就够;订单执行需要生成规范文档,用擅长文本生成的模型。通过TaoToken的统一端点,你可以在代码里按Agent切换Model ID,不用改认证逻辑。

如果你在接入过程中遇到模型调用层的问题,接入文档在 https://taotoken.net/doc?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= ,里面有各模型的Model ID清单和调用示例。需要创建或管理密钥的话,控制台在 https://taotoken.net/api-keys?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。想先单独验证某个模型能不能调通,可以用模型对话页面快速试一下,地址是 https://taotoken.net/chat?utm_source=taotoken_aicg_blog_end&utm_medium=csdn&utm_campaign=rewrite&utm_content= 。

最后说一个实际经验:多智能体系统的Agent数量不是越多越好。我见过有人给每个小功能都建一个Agent,一个订单流程七八个Agent在跑,延迟高得离谱,调试也困难。三个Agent是大多数业务流程的甜区,再多了就要考虑合并职责。先把三个Agent的流程跑稳,再考虑扩展。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询