1. 为什么单Agent跑不动复杂任务A2A协议多Agent协作系统到底解决什么问题先说结论A2AAgent-to-Agent协议是一套让不同框架、不同语言写出来的 AI Agent 能互相“对话”和“派活”的开放标准。它能做的事是把原本各自为战的 Agent 串成一条流水线适合谁适合已经用 LangChain、LangGraph、CrewAI 甚至自研框架写过单个 Agent但一遇到“查天气 订票 汇总”这种复合任务就靠 if-else 硬编码的开发者。我试过最原始的写法一个 Python 文件里塞三个函数主函数里用关键词判断用户想干嘛然后依次调用。跑 demo 没问题一旦要加第四个能力整个文件就开始失控——函数互相 import、状态传来传去、报错定位不到源头。这就是单 Agent 架构的天花板。A2A 的核心思路其实很朴素把每个 Agent 当成一个独立的微服务。每个 Agent 有自己的地址URL、自己的“名片”AgentCard、自己的技能列表AgentSkill。主控 Agent 不需要知道天气 Agent 内部是用 Function Calling 还是 LangGraph它只需要知道“这个地址能查天气”然后把任务打包成标准格式发过去等结果回来。这套机制解决三个具体痛点。第一是能力单一一个 Agent 只擅长一个领域A2A 让它们组合。第二是接口各异LangChain 的 Agent 和自研框架的 Agent 通信方式完全不同A2A 用统一的 Task/Message/Artifact 结构抹平差异。第三是协作困难以前靠手写胶水代码把 Agent 硬编码在一起现在靠 AgentNetwork 做服务发现加一个新 Agent 只需要 network.add() 一行。用一个生活化的类比你是一个旅行规划师客户说“帮我规划北京到上海的出差顺便看看天气”。如果你单打独斗得自己查天气、自己订票、自己拼结果。但如果你有一个团队——“天气专家”和“票务专家”——你只需要发一封标准化的工作邮件给他们他们干完活把结果填在交接单上送回来。A2A 协议就是这套“标准化工作邮件模板”和“任务交接单”。任务在 A2A 里有明确的生命周期SUBMITTED已提交→ WORKING进行中→ COMPLETED已完成中间可能进入 INPUT_REQUIRED需要补充信息或 FAILED失败。这个状态机很重要它让主控 Agent 能知道每个子任务当前处于什么阶段而不是发出去就干等。理解了这层后面的代码就不是“抄一遍”而是“我知道每一行在解决什么问题”。接下来我会用 TaoToken 作为统一的 LLM 接入通道把 Key 和 Base URL 收敛到一处然后用 LangGraph 编排 Agent 角色从零搭一套能跑通的多 Agent 协作系统。2. TaoToken 统一 Key 接入 LLM多 Agent 共享一条 API 通道的前置准备多 Agent 系统有个容易被忽略的工程问题每个 Agent 都要调 LLM如果每个 Agent 各自配置 Key、各自写 base_url维护成本会随 Agent 数量线性增长。更麻烦的是天气 Agent 用 GPT-4o-mini、票务 Agent 用另一个模型时你得在三个文件里改三处配置。TaoToken 在这里扮演的角色是“统一入口”。它提供 OpenAI 兼容的 API 通道你只需要一个 Key、一个 Base URL所有 Agent 共用同一份配置。模型切换、额度管理、通道稳定性都在这一层收敛Agent 代码里只关心业务逻辑。前置准备分三步。第一步拿到 API Key。访问 https://taotoken.net/api-keys 创建密钥复制出来形如sk-xxxxxxxx的字符串。第二步确认 Base URL 是https://taotoken.net/api注意这个地址不带任何路径后缀OpenAI SDK 会自动拼接/v1/chat/completions。第三步确认你要用的 Model ID比如gpt-4o-mini、claude-3-5-sonnet这类具体以控制台模型列表为准。这里有个坑要提前说很多人在.env里把 Base URL 写成https://taotoken.net/api/v1结果 SDK 又拼了一次/v1变成/api/v1/v1/chat/completions直接 404。正确写法就是https://taotoken.net/api让 SDK 自己补。项目根目录建一个.env文件内容如下MODEL_API_KEYsk-你的TaoToken密钥 MODEL_BASE_URLhttps://taotoken.net/api MODEL_NAMEgpt-4o-mini然后写一个common/llm.py作为所有 Agent 共享的 LLM 工厂。这个文件只做一件事读环境变量返回一个配置好的 ChatOpenAI 实例。import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI env_path os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), .env ) load_dotenv(env_path) MODEL_API_KEY os.getenv(MODEL_API_KEY) MODEL_BASE_URL os.getenv(MODEL_BASE_URL) MODEL_NAME os.getenv(MODEL_NAME) my_llm ChatOpenAI( api_keyMODEL_API_KEY, base_urlMODEL_BASE_URL, modelMODEL_NAME, temperature0, ) if __name__ __main__: print(f模型{MODEL_NAME}) print(f接口{MODEL_BASE_URL}) print(测试输出, end) for chunk in my_llm.stream(你好请用一句话介绍你自己。): print(chunk.content, flushTrue, end) print()temperature0是刻意设的。意图识别和信息抽取这类任务需要稳定输出温度高了会导致同一个输入两次跑出不同结果调试时非常痛苦。跑一下python common/llm.py如果能看到模型正常流式输出说明 TaoToken 通道打通了。这一步是整个系统的地基地基不稳后面全是玄学问题。如果这里报 401先检查 Key 有没有复制完整如果报连接错误检查 Base URL 是不是多写了/v1。依赖安装清单pip install python-a2a langchain-openai langchain-core langgraph pydantic python-dotenvpython-a2a是 Python 社区对 A2A 协议的实现库封装了 Server 启动、Client 通信、消息序列化这些底层细节。装完之后我们就有了一套标准的 Agent 通信骨架。3. 可复制的 Agent 注册与任务分发配置LangGraph 编排三角色这一节给出完整可复制的配置。系统采用星型拓扑一个总控 AgentOrchestrator作为中心协调两个专家 Agent天气、票务。每个专家 Agent 是一个独立的 A2A Server监听不同端口。先看目录结构后面所有路径都以此为准a2a_travel_system/ ├── common/ │ └── llm.py ├── weather_agent/ │ ├── parse_weather.py │ └── weather_agent.py ├── ticket_agent/ │ ├── ticket_workflow.py │ └── ticket_agent.py ├── orchestrator/ │ ├── intent.py │ └── orchestrator.py ├── .env └── requirements.txt3.1 天气专家 Agent 的注册配置天气 Agent 用 LangChain Function Calling 实现LLM 自己判断用户是不是在问天气自己从自然语言里抽出城市名然后调用get_weather工具。weather_agent/parse_weather.pyfrom langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage from common.llm import my_llm tool def get_weather(city: str) - str: 根据城市名称返回该城市的当前天气情况。支持全国主要城市。 weather_db { 北京: 晴28°C微风, 上海: 多云22°C东南风3级, 广州: 小雨19°C湿度85%, 深圳: 阵雨26°C体感温度29°C, 成都: 阴24°C空气质量良, 杭州: 多云25°C适合出行, } return weather_db.get(city, f未找到 {city} 的天气信息。) tools [get_weather] llm_with_tools my_llm.bind_tools(tools) def parse_weather_text(user_input: str) - str: response llm_with_tools.invoke([HumanMessage(contentuser_input)]) if response.tool_calls: tool_results [] for tool_call in response.tool_calls: result get_weather.invoke(tool_call[args]) tool_results.append(result) messages [ HumanMessage(contentuser_input), response, ToolMessage( content\n.join(tool_results), tool_call_idresponse.tool_calls[0][id] ) ] final_response llm_with_tools.invoke(messages) return final_response.content return response.contentweather_agent/weather_agent.py把这个逻辑包成 A2A Serverfrom python_a2a import ( A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState ) from weather_agent.parse_weather import parse_weather_text agent_card AgentCard( nameWeatherExpert, description负责查询天气的专家能够根据城市名称返回实时天气信息。, urlhttp://127.0.0.1:5008, skills[ AgentSkill( nameget_weather, description根据城市查询天气支持全国主要城市。, examples[北京天气怎么样, 查一下上海天气, 广州今天下雨吗] ) ] ) class WeatherExpertServer(A2AServer): def __init__(self): super().__init__(agent_cardagent_card) def handle_task(self, task): print( [天气专家] 收到任务) text (task.message or {}).get(content, {}).get(text, ) result parse_weather_text(text) task.artifacts [{parts: [{type: text, text: result}]}] task.status TaskStatus(stateTaskState.COMPLETED) print(f [天气专家] 已返回{result}) return task if __name__ __main__: server WeatherExpertServer() print(f [天气专家] 启动成功{server.agent_card.url}) run_server(server, host127.0.0.1, port5008, debugTrue)3.2 票务专家 Agent 的 LangGraph 工作流配置票务 Agent 的逻辑更复杂需要两步先从自然语言里抽取出出发城市、目的城市、出行时间再校验信息完整性。用 LangGraph 建一个两节点线性工作流。ticket_agent/ticket_workflow.pyfrom typing import TypedDict from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from langgraph.constants import END, START from langgraph.graph import StateGraph from pydantic import BaseModel, Field from common.llm import my_llm class AgentState(TypedDict): input: str start_city: str end_city: str departure_time: str output: str class TicketBookingExtract(BaseModel): start_city: str Field(default, description出发城市未提及填空字符串) end_city: str Field(default, description目的城市未提及填空字符串) departure_time: str Field(default, description出行时间未提及填空字符串) _extract_parser PydanticOutputParser(pydantic_objectTicketBookingExtract) _extract_prompt ChatPromptTemplate.from_messages([ (system, 你是火车票预订信息抽取助手。从用户自然语言中提取出发城市、目的城市和出行时间。 城市使用标准中文地名不要编造用户未提及的信息。 任一字段抽不出来时输出空字符串不要用未知无等占位词。), (user, 请从以下内容抽取订票信息。\n{format_instructions}\n\n用户输入{user_input}), ]).partial(format_instructions_extract_parser.get_format_instructions()) _extract_chain _extract_prompt | my_llm | _extract_parser _PLACEHOLDER_VALUES frozenset({未知, 无, 未提及, null, none, n/a, 待定}) def _clean_field(value: str) - str: s (value or ).strip() if not s or s.lower() in _PLACEHOLDER_VALUES: return return s def extract_city_node(state: AgentState) - AgentState: user_input state.get(input, ) extracted _extract_chain.invoke({user_input: user_input}) state[start_city] _clean_field(extracted.start_city) state[end_city] _clean_field(extracted.end_city) state[departure_time] _clean_field(extracted.departure_time) return state def order_ticket_node(state: AgentState) - AgentState: start_city state.get(start_city, ) end_city state.get(end_city, ) departure_time state.get(departure_time, ) missing [] if not start_city: missing.append(出发城市) if not end_city: missing.append(目的城市) if not departure_time: missing.append(出行时间) if missing: state[output] ( f订票信息不完整还需要您补充以下信息{、.join(missing)}。\n f请提供完整信息后重新下单。 ) else: state[output] ( f 已为您成功预订火车票\n f 线路{start_city} → {end_city}\n f 时间{departure_time}\n f 座位08车 06F靠窗\n f 票价二等座 ¥553.0 ) return state def build_langgraph(): graph_builder StateGraph(AgentState) graph_builder.add_node(extract_city_node, extract_city_node) graph_builder.add_node(order_ticket_node, order_ticket_node) graph_builder.add_edge(START, extract_city_node) graph_builder.add_edge(extract_city_node, order_ticket_node) graph_builder.add_edge(order_ticket_node, END) return graph_builder.compile() graph build_langgraph() def parse_ticket_text(text: str) - str: result graph.invoke({input: text}) return result.get(output, 处理出错。)ticket_agent/ticket_agent.py同样包成 A2A Server端口 5009结构跟天气 Agent 一致只是handle_task里调用parse_ticket_text。3.3 总控 Agent 的意图识别与任务分发配置总控 Agent 用 LLM 做结构化意图识别判断用户是否需要查天气、是否需要订票然后并发调用对应专家。orchestrator/intent.pyfrom langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from common.llm import my_llm class WeatherTicketIntent(BaseModel): need_weather: bool Field(description用户是否需要查询天气、气温、是否下雨等气象信息) need_ticket: bool Field(description用户是否需要预订或购买火车票、高铁票等出行票务) _intent_parser PydanticOutputParser(pydantic_objectWeatherTicketIntent) _intent_prompt ChatPromptTemplate.from_messages([ (system, 你是旅行助手意图识别模块。根据用户的一句话判断是否需要查天气、是否需要订票。\n 规则\n 1. 只根据用户明确或合理隐含的需求判断不要臆测\n 2. 与天气、票务无关的闲聊两项均为 false\n 3. 同一句话里可能同时需要查天气和订票此时对应字段均为 true\n 4. 出差旅行等隐含出行需求时need_ticket 可为 true\n 5. 提到某个城市但没有明确出行意图时不要设 need_ticket 为 true), (user, 请判断以下用户输入的意图。\n{format_instructions}\n\n用户输入{user_input}), ]).partial(format_instructions_intent_parser.get_format_instructions()) _intent_chain _intent_prompt | my_llm | _intent_parser def detect_weather_ticket_intent(user_input: str) - WeatherTicketIntent: text (user_input or ).strip() if not text: return WeatherTicketIntent(need_weatherFalse, need_ticketFalse) return _intent_chain.invoke({user_input: text})orchestrator/orchestrator.py用 AgentNetwork 注册两个专家用 asyncio.gather 并发派发import asyncio import uuid from python_a2a import AgentNetwork, Task, Message, MessageRole, TextContent from orchestrator.intent import detect_weather_ticket_intent network AgentNetwork(nameTravelTeam) network.add(WeatherExpert, http://127.0.0.1:5008) network.add(TicketExpert, http://127.0.0.1:5009) class Orchestrator: def __init__(self, network): self.network network def make_task(self, text: str) - Task: message Message(roleMessageRole.USER, contentTextContent(texttext)) return Task(idftask-{uuid.uuid4()}, messagemessage.to_dict()) async def call_agent(self, agent_name: str, text: str): print(f [总控Agent] 正在调用「{agent_name}」) client self.network.get_agent(agent_name) task self.make_task(text) result_task await client.send_task_async(task) result result_task.artifacts[0][parts][0][text] print(f [总控Agent] 收到「{agent_name}」回复) return agent_name, result async def handle_user_request(self, text: str) - str: print(\n * 60) print(f [总控Agent] 收到用户请求{text}) intent detect_weather_ticket_intent(text) print(f [总控Agent] 意图识别查天气{intent.need_weather}订票{intent.need_ticket}) jobs [] if intent.need_weather: jobs.append(self.call_agent(WeatherExpert, text)) if intent.need_ticket: jobs.append(self.call_agent(TicketExpert, text)) if not jobs: return 抱歉我目前只会查天气和订火车票。 results await asyncio.gather(*jobs, return_exceptionsTrue) final_parts [] for item in results: if isinstance(item, Exception): final_parts.append(f 调用失败{item}) else: agent_name, result item final_parts.append(f【{agent_name}】\n{result}) return \n\n.join(final_parts) async def main(): orchestrator Orchestrator(network) while True: text input(\n 您).strip() if not text: continue if text.lower() quit: break result await orchestrator.handle_user_request(text) print(f\n 助手\n{result}) if __name__ __main__: asyncio.run(main())三件套配置到这里就齐了Base URL 是https://taotoken.net/apiKey 在.env的MODEL_API_KEYModel ID 是MODEL_NAME。所有 Agent 共享common/llm.py改模型只改一处。4. 验证请求与成功结果一次完整的多 Agent 协作链路演示启动顺序有讲究先起两个专家 Server再起总控 Client。因为总控启动时会通过 AgentNetwork 去连专家地址专家没起来会连接失败。三个终端分别执行# 终端 1 python weather_agent/weather_agent.py # 终端 2 python ticket_agent/ticket_agent.py # 终端 3 python orchestrator/orchestrator.py专家启动成功会打印监听地址总控启动后会打印通讯录。现在输入一个复合请求我明天要去上海出差帮我订张北京到上海的火车票顺便看看上海天气怎么样预期输出关键日志 [总控Agent] 收到用户请求我明天要去上海出差... [总控Agent] 意图识别查天气True订票True [总控Agent] 正在调用「WeatherExpert」 [总控Agent] 正在调用「TicketExpert」 [天气专家] 收到任务 [票务专家] 收到任务 [总控Agent] 收到「WeatherExpert」回复 [总控Agent] 收到「TicketExpert」回复 助手 【WeatherExpert】 上海今天多云22°C东南风3级适合出行。 【TicketExpert】 已为您成功预订火车票 线路北京 → 上海 时间明天 座位08车 06F靠窗 票价二等座 ¥553.0这里有两个关键观察点。第一两个专家任务是并发执行的总耗时取决于最慢的那个而不是两者之和这是asyncio.gather的效果也是 A2A 异步优先设计理念的体现。第二票务专家能识别出用户只说了“去上海”但没说“从哪出发”和“具体时间”时会主动提示补充信息而不是盲目返回错误结果。再测一个信息不完整的场景我想买张去广州的票预期票务专家返回订票信息不完整还需要您补充以下信息出发城市、出行时间。 请提供完整信息后重新下单。这说明 LangGraph 工作流里的校验节点在起作用。如果这里直接返回了订票成功说明_clean_field没生效LLM 把“未知”当成了有效值需要检查_PLACEHOLDER_VALUES集合是否覆盖了模型实际输出的占位词。验证天气 Agent 单独工作杭州天气怎么样预期返回杭州的天气信息。如果返回“未找到杭州的天气信息”说明 Function Calling 没触发LLM 直接把用户输入当成了普通对话。检查llm_with_tools.bind_tools(tools)是否执行以及get_weather的 docstring 是否清晰——工具描述写得越明确模型越容易正确调用。5. 本篇常见错误排查401、local proxy failed、reading choices 逐个击破这一节按真实报错来。多 Agent 系统涉及多个进程、多个网络请求出错时定位链路长先把高频错误列清楚。错误一401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是.env里的MODEL_API_KEY没被正确加载。load_dotenv的路径是按common/llm.py往上推三层找到项目根目录如果你的目录结构和本文不一致路径会错。排查方法在llm.py里加一行print(MODEL_API_KEY[:8])看是否打印出sk-开头的前缀。如果打印出None说明环境变量没读到检查.env文件是否在项目根目录、变量名是否拼写正确。另一个可能是 Key 本身失效或额度耗尽去 https://taotoken.net/api-keys 确认密钥状态。错误二local proxy failed / Connection erroropenai.APIConnectionError: Connection error. httpx.ConnectError: [Errno 111] Connection refused这个报错分两种场景。场景一总控 Agent 连不上专家 Agent说明专家 Server 没启动或端口不对。检查network.add(WeatherExpert, http://127.0.0.1:5008)里的端口和run_server(server, host127.0.0.1, port5008)是否一致。场景二LLM 请求连不上 TaoToken检查MODEL_BASE_URL是否写成了https://taotoken.net/api/v1多写的/v1会导致路径拼接错误。正确值就是https://taotoken.net/api。错误三reading choices / KeyError: choicesKeyError: choices TypeError: Cannot read properties of undefined (reading choices)这个错误说明 API 返回的 JSON 结构里没有choices字段通常是请求根本没到 LLM 层而是被某个中间层拦截返回了错误页。最常见原因是 Base URL 写错请求打到了一个返回 HTML 的地址。用 curl 直接测一下curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-你的密钥 \ -H Content-Type: application/json \ -d {model:gpt-4o-mini,messages:[{role:user,content:hi}]}如果 curl 返回正常 JSON说明通道没问题问题在代码里的 base_url 配置。如果 curl 也报错检查 Key 和 Model ID。错误四OAuth / 认证相关报错Error: OAuth token expired Authentication failed: invalid_grant这类报错一般出现在你用了需要 OAuth 的模型通道但 TaoToken 走的是 API Key 认证不应该出现 OAuth 流程。如果看到这个报错检查是不是误装了某个需要 OAuth 的 SDK 插件或者.env里混入了其他平台的配置变量。清理掉无关变量只保留MODEL_API_KEY、MODEL_BASE_URL、MODEL_NAME三个。错误五AgentCard 注册失败 / 404 on agent cardFailed to fetch agent card from http://127.0.0.1:5008/.well-known/agent.jsonA2A 协议要求 Server 暴露/.well-known/agent.json路径返回 AgentCard。如果这个路径 404说明run_server没正确挂载路由。检查python-a2a版本是否过旧升级到最新版。另外确认agent_card的url字段和实际监听地址一致协议、IP、端口三者都要对上。错误六LangGraph 节点返回类型错误InvalidUpdateError: Expected dict, got NoneLangGraph 的节点函数必须返回 state 字典。如果你在extract_city_node里忘了return state就会报这个。检查每个add_node对应的函数是否都有返回值且返回值类型是AgentState。排查顺序建议先 curl 测 LLM 通道再单独启动每个专家 Agent 用 A2AClient 测最后起总控测全链路。逐层验证比一上来就跑全链路高效得多。6. 语义一致的 CTA把统一 Key 和多 Agent 编排用到真实项目里跑通这套 demo 之后你手里其实已经有了一个可扩展的骨架。加一个新专家 Agent 只需要三步写业务逻辑、包成 A2AServer、在network.add()里注册一行。总控的意图识别模块加一个布尔字段就能把新专家纳入调度。真实项目里天气和票务的数据源要换成真实 API比如天气接和风或高德票务接 12306 或携程开放平台。LLM 通道继续用 TaoToken 统一管理模型可以按 Agent 分别指定——意图识别用便宜快的小模型信息抽取用结构化输出稳定的模型这样成本和效果都能兼顾。如果你想把模型对话能力也接进来做调试可以访问 https://taotoken.net/models 直接测试不同模型的输出效果确认哪个 Model ID 在你的场景下表现最好。接入文档在 https://taotoken.net/doc里面有 OpenAI 兼容接口的完整参数说明。长期跑编码类 Agent 或需要稳定额度的场景可以看看 https://taotoken.net/coding-plan按用量规划比临时充值更可控。最后留一个实用技巧多 Agent 系统调试时给每个 Agent 的日志加上统一前缀比如[天气专家]、[票务专家]三个终端并排看日志时能一眼分清谁在说话。这个习惯在 Agent 数量超过三个之后会救命。
1. 为什么单Agent跑不动复杂任务A2A协议多Agent协作系统到底解决什么问题先说结论A2AAgent-to-Agent协议是一套让不同框架、不同语言写出来的 AI Agent 能互相“对话”和“派活”的开放标准。它能做的事是把原本各自为战的 Agent 串成一条流水线适合谁适合已经用 LangChain、LangGraph、CrewAI 甚至自研框架写过单个 Agent但一遇到“查天气 订票 汇总”这种复合任务就靠 if-else 硬编码的开发者。我试过最原始的写法一个 Python 文件里塞三个函数主函数里用关键词判断用户想干嘛然后依次调用。跑 demo 没问题一旦要加第四个能力整个文件就开始失控——函数互相 import、状态传来传去、报错定位不到源头。这就是单 Agent 架构的天花板。A2A 的核心思路其实很朴素把每个 Agent 当成一个独立的微服务。每个 Agent 有自己的地址URL、自己的“名片”AgentCard、自己的技能列表AgentSkill。主控 Agent 不需要知道天气 Agent 内部是用 Function Calling 还是 LangGraph它只需要知道“这个地址能查天气”然后把任务打包成标准格式发过去等结果回来。这套机制解决三个具体痛点。第一是能力单一一个 Agent 只擅长一个领域A2A 让它们组合。第二是接口各异LangChain 的 Agent 和自研框架的 Agent 通信方式完全不同A2A 用统一的 Task/Message/Artifact 结构抹平差异。第三是协作困难以前靠手写胶水代码把 Agent 硬编码在一起现在靠 AgentNetwork 做服务发现加一个新 Agent 只需要 network.add() 一行。用一个生活化的类比你是一个旅行规划师客户说“帮我规划北京到上海的出差顺便看看天气”。如果你单打独斗得自己查天气、自己订票、自己拼结果。但如果你有一个团队——“天气专家”和“票务专家”——你只需要发一封标准化的工作邮件给他们他们干完活把结果填在交接单上送回来。A2A 协议就是这套“标准化工作邮件模板”和“任务交接单”。任务在 A2A 里有明确的生命周期SUBMITTED已提交→ WORKING进行中→ COMPLETED已完成中间可能进入 INPUT_REQUIRED需要补充信息或 FAILED失败。这个状态机很重要它让主控 Agent 能知道每个子任务当前处于什么阶段而不是发出去就干等。理解了这层后面的代码就不是“抄一遍”而是“我知道每一行在解决什么问题”。接下来我会用 TaoToken 作为统一的 LLM 接入通道把 Key 和 Base URL 收敛到一处然后用 LangGraph 编排 Agent 角色从零搭一套能跑通的多 Agent 协作系统。2. TaoToken 统一 Key 接入 LLM多 Agent 共享一条 API 通道的前置准备多 Agent 系统有个容易被忽略的工程问题每个 Agent 都要调 LLM如果每个 Agent 各自配置 Key、各自写 base_url维护成本会随 Agent 数量线性增长。更麻烦的是天气 Agent 用 GPT-4o-mini、票务 Agent 用另一个模型时你得在三个文件里改三处配置。TaoToken 在这里扮演的角色是“统一入口”。它提供 OpenAI 兼容的 API 通道你只需要一个 Key、一个 Base URL所有 Agent 共用同一份配置。模型切换、额度管理、通道稳定性都在这一层收敛Agent 代码里只关心业务逻辑。前置准备分三步。第一步拿到 API Key。访问 https://taotoken.net/api-keys 创建密钥复制出来形如sk-xxxxxxxx的字符串。第二步确认 Base URL 是https://taotoken.net/api注意这个地址不带任何路径后缀OpenAI SDK 会自动拼接/v1/chat/completions。第三步确认你要用的 Model ID比如gpt-4o-mini、claude-3-5-sonnet这类具体以控制台模型列表为准。这里有个坑要提前说很多人在.env里把 Base URL 写成https://taotoken.net/api/v1结果 SDK 又拼了一次/v1变成/api/v1/v1/chat/completions直接 404。正确写法就是https://taotoken.net/api让 SDK 自己补。项目根目录建一个.env文件内容如下MODEL_API_KEYsk-你的TaoToken密钥 MODEL_BASE_URLhttps://taotoken.net/api MODEL_NAMEgpt-4o-mini然后写一个common/llm.py作为所有 Agent 共享的 LLM 工厂。这个文件只做一件事读环境变量返回一个配置好的 ChatOpenAI 实例。import os from dotenv import load_dotenv from langchain_openai import ChatOpenAI env_path os.path.join( os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))), .env ) load_dotenv(env_path) MODEL_API_KEY os.getenv(MODEL_API_KEY) MODEL_BASE_URL os.getenv(MODEL_BASE_URL) MODEL_NAME os.getenv(MODEL_NAME) my_llm ChatOpenAI( api_keyMODEL_API_KEY, base_urlMODEL_BASE_URL, modelMODEL_NAME, temperature0, ) if __name__ __main__: print(f模型{MODEL_NAME}) print(f接口{MODEL_BASE_URL}) print(测试输出, end) for chunk in my_llm.stream(你好请用一句话介绍你自己。): print(chunk.content, flushTrue, end) print()temperature0是刻意设的。意图识别和信息抽取这类任务需要稳定输出温度高了会导致同一个输入两次跑出不同结果调试时非常痛苦。跑一下python common/llm.py如果能看到模型正常流式输出说明 TaoToken 通道打通了。这一步是整个系统的地基地基不稳后面全是玄学问题。如果这里报 401先检查 Key 有没有复制完整如果报连接错误检查 Base URL 是不是多写了/v1。依赖安装清单pip install python-a2a langchain-openai langchain-core langgraph pydantic python-dotenvpython-a2a是 Python 社区对 A2A 协议的实现库封装了 Server 启动、Client 通信、消息序列化这些底层细节。装完之后我们就有了一套标准的 Agent 通信骨架。3. 可复制的 Agent 注册与任务分发配置LangGraph 编排三角色这一节给出完整可复制的配置。系统采用星型拓扑一个总控 AgentOrchestrator作为中心协调两个专家 Agent天气、票务。每个专家 Agent 是一个独立的 A2A Server监听不同端口。先看目录结构后面所有路径都以此为准a2a_travel_system/ ├── common/ │ └── llm.py ├── weather_agent/ │ ├── parse_weather.py │ └── weather_agent.py ├── ticket_agent/ │ ├── ticket_workflow.py │ └── ticket_agent.py ├── orchestrator/ │ ├── intent.py │ └── orchestrator.py ├── .env └── requirements.txt3.1 天气专家 Agent 的注册配置天气 Agent 用 LangChain Function Calling 实现LLM 自己判断用户是不是在问天气自己从自然语言里抽出城市名然后调用get_weather工具。weather_agent/parse_weather.pyfrom langchain_core.tools import tool from langchain_core.messages import HumanMessage, ToolMessage from common.llm import my_llm tool def get_weather(city: str) - str: 根据城市名称返回该城市的当前天气情况。支持全国主要城市。 weather_db { 北京: 晴28°C微风, 上海: 多云22°C东南风3级, 广州: 小雨19°C湿度85%, 深圳: 阵雨26°C体感温度29°C, 成都: 阴24°C空气质量良, 杭州: 多云25°C适合出行, } return weather_db.get(city, f未找到 {city} 的天气信息。) tools [get_weather] llm_with_tools my_llm.bind_tools(tools) def parse_weather_text(user_input: str) - str: response llm_with_tools.invoke([HumanMessage(contentuser_input)]) if response.tool_calls: tool_results [] for tool_call in response.tool_calls: result get_weather.invoke(tool_call[args]) tool_results.append(result) messages [ HumanMessage(contentuser_input), response, ToolMessage( content\n.join(tool_results), tool_call_idresponse.tool_calls[0][id] ) ] final_response llm_with_tools.invoke(messages) return final_response.content return response.contentweather_agent/weather_agent.py把这个逻辑包成 A2A Serverfrom python_a2a import ( A2AServer, run_server, AgentCard, AgentSkill, TaskStatus, TaskState ) from weather_agent.parse_weather import parse_weather_text agent_card AgentCard( nameWeatherExpert, description负责查询天气的专家能够根据城市名称返回实时天气信息。, urlhttp://127.0.0.1:5008, skills[ AgentSkill( nameget_weather, description根据城市查询天气支持全国主要城市。, examples[北京天气怎么样, 查一下上海天气, 广州今天下雨吗] ) ] ) class WeatherExpertServer(A2AServer): def __init__(self): super().__init__(agent_cardagent_card) def handle_task(self, task): print( [天气专家] 收到任务) text (task.message or {}).get(content, {}).get(text, ) result parse_weather_text(text) task.artifacts [{parts: [{type: text, text: result}]}] task.status TaskStatus(stateTaskState.COMPLETED) print(f [天气专家] 已返回{result}) return task if __name__ __main__: server WeatherExpertServer() print(f [天气专家] 启动成功{server.agent_card.url}) run_server(server, host127.0.0.1, port5008, debugTrue)3.2 票务专家 Agent 的 LangGraph 工作流配置票务 Agent 的逻辑更复杂需要两步先从自然语言里抽取出出发城市、目的城市、出行时间再校验信息完整性。用 LangGraph 建一个两节点线性工作流。ticket_agent/ticket_workflow.pyfrom typing import TypedDict from langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from langgraph.constants import END, START from langgraph.graph import StateGraph from pydantic import BaseModel, Field from common.llm import my_llm class AgentState(TypedDict): input: str start_city: str end_city: str departure_time: str output: str class TicketBookingExtract(BaseModel): start_city: str Field(default, description出发城市未提及填空字符串) end_city: str Field(default, description目的城市未提及填空字符串) departure_time: str Field(default, description出行时间未提及填空字符串) _extract_parser PydanticOutputParser(pydantic_objectTicketBookingExtract) _extract_prompt ChatPromptTemplate.from_messages([ (system, 你是火车票预订信息抽取助手。从用户自然语言中提取出发城市、目的城市和出行时间。 城市使用标准中文地名不要编造用户未提及的信息。 任一字段抽不出来时输出空字符串不要用未知无等占位词。), (user, 请从以下内容抽取订票信息。\n{format_instructions}\n\n用户输入{user_input}), ]).partial(format_instructions_extract_parser.get_format_instructions()) _extract_chain _extract_prompt | my_llm | _extract_parser _PLACEHOLDER_VALUES frozenset({未知, 无, 未提及, null, none, n/a, 待定}) def _clean_field(value: str) - str: s (value or ).strip() if not s or s.lower() in _PLACEHOLDER_VALUES: return return s def extract_city_node(state: AgentState) - AgentState: user_input state.get(input, ) extracted _extract_chain.invoke({user_input: user_input}) state[start_city] _clean_field(extracted.start_city) state[end_city] _clean_field(extracted.end_city) state[departure_time] _clean_field(extracted.departure_time) return state def order_ticket_node(state: AgentState) - AgentState: start_city state.get(start_city, ) end_city state.get(end_city, ) departure_time state.get(departure_time, ) missing [] if not start_city: missing.append(出发城市) if not end_city: missing.append(目的城市) if not departure_time: missing.append(出行时间) if missing: state[output] ( f订票信息不完整还需要您补充以下信息{、.join(missing)}。\n f请提供完整信息后重新下单。 ) else: state[output] ( f 已为您成功预订火车票\n f 线路{start_city} → {end_city}\n f 时间{departure_time}\n f 座位08车 06F靠窗\n f 票价二等座 ¥553.0 ) return state def build_langgraph(): graph_builder StateGraph(AgentState) graph_builder.add_node(extract_city_node, extract_city_node) graph_builder.add_node(order_ticket_node, order_ticket_node) graph_builder.add_edge(START, extract_city_node) graph_builder.add_edge(extract_city_node, order_ticket_node) graph_builder.add_edge(order_ticket_node, END) return graph_builder.compile() graph build_langgraph() def parse_ticket_text(text: str) - str: result graph.invoke({input: text}) return result.get(output, 处理出错。)ticket_agent/ticket_agent.py同样包成 A2A Server端口 5009结构跟天气 Agent 一致只是handle_task里调用parse_ticket_text。3.3 总控 Agent 的意图识别与任务分发配置总控 Agent 用 LLM 做结构化意图识别判断用户是否需要查天气、是否需要订票然后并发调用对应专家。orchestrator/intent.pyfrom langchain_core.output_parsers import PydanticOutputParser from langchain_core.prompts import ChatPromptTemplate from pydantic import BaseModel, Field from common.llm import my_llm class WeatherTicketIntent(BaseModel): need_weather: bool Field(description用户是否需要查询天气、气温、是否下雨等气象信息) need_ticket: bool Field(description用户是否需要预订或购买火车票、高铁票等出行票务) _intent_parser PydanticOutputParser(pydantic_objectWeatherTicketIntent) _intent_prompt ChatPromptTemplate.from_messages([ (system, 你是旅行助手意图识别模块。根据用户的一句话判断是否需要查天气、是否需要订票。\n 规则\n 1. 只根据用户明确或合理隐含的需求判断不要臆测\n 2. 与天气、票务无关的闲聊两项均为 false\n 3. 同一句话里可能同时需要查天气和订票此时对应字段均为 true\n 4. 出差旅行等隐含出行需求时need_ticket 可为 true\n 5. 提到某个城市但没有明确出行意图时不要设 need_ticket 为 true), (user, 请判断以下用户输入的意图。\n{format_instructions}\n\n用户输入{user_input}), ]).partial(format_instructions_intent_parser.get_format_instructions()) _intent_chain _intent_prompt | my_llm | _intent_parser def detect_weather_ticket_intent(user_input: str) - WeatherTicketIntent: text (user_input or ).strip() if not text: return WeatherTicketIntent(need_weatherFalse, need_ticketFalse) return _intent_chain.invoke({user_input: text})orchestrator/orchestrator.py用 AgentNetwork 注册两个专家用 asyncio.gather 并发派发import asyncio import uuid from python_a2a import AgentNetwork, Task, Message, MessageRole, TextContent from orchestrator.intent import detect_weather_ticket_intent network AgentNetwork(nameTravelTeam) network.add(WeatherExpert, http://127.0.0.1:5008) network.add(TicketExpert, http://127.0.0.1:5009) class Orchestrator: def __init__(self, network): self.network network def make_task(self, text: str) - Task: message Message(roleMessageRole.USER, contentTextContent(texttext)) return Task(idftask-{uuid.uuid4()}, messagemessage.to_dict()) async def call_agent(self, agent_name: str, text: str): print(f [总控Agent] 正在调用「{agent_name}」) client self.network.get_agent(agent_name) task self.make_task(text) result_task await client.send_task_async(task) result result_task.artifacts[0][parts][0][text] print(f [总控Agent] 收到「{agent_name}」回复) return agent_name, result async def handle_user_request(self, text: str) - str: print(\n * 60) print(f [总控Agent] 收到用户请求{text}) intent detect_weather_ticket_intent(text) print(f [总控Agent] 意图识别查天气{intent.need_weather}订票{intent.need_ticket}) jobs [] if intent.need_weather: jobs.append(self.call_agent(WeatherExpert, text)) if intent.need_ticket: jobs.append(self.call_agent(TicketExpert, text)) if not jobs: return 抱歉我目前只会查天气和订火车票。 results await asyncio.gather(*jobs, return_exceptionsTrue) final_parts [] for item in results: if isinstance(item, Exception): final_parts.append(f 调用失败{item}) else: agent_name, result item final_parts.append(f【{agent_name}】\n{result}) return \n\n.join(final_parts) async def main(): orchestrator Orchestrator(network) while True: text input(\n 您).strip() if not text: continue if text.lower() quit: break result await orchestrator.handle_user_request(text) print(f\n 助手\n{result}) if __name__ __main__: asyncio.run(main())三件套配置到这里就齐了Base URL 是https://taotoken.net/apiKey 在.env的MODEL_API_KEYModel ID 是MODEL_NAME。所有 Agent 共享common/llm.py改模型只改一处。4. 验证请求与成功结果一次完整的多 Agent 协作链路演示启动顺序有讲究先起两个专家 Server再起总控 Client。因为总控启动时会通过 AgentNetwork 去连专家地址专家没起来会连接失败。三个终端分别执行# 终端 1 python weather_agent/weather_agent.py # 终端 2 python ticket_agent/ticket_agent.py # 终端 3 python orchestrator/orchestrator.py专家启动成功会打印监听地址总控启动后会打印通讯录。现在输入一个复合请求我明天要去上海出差帮我订张北京到上海的火车票顺便看看上海天气怎么样预期输出关键日志 [总控Agent] 收到用户请求我明天要去上海出差... [总控Agent] 意图识别查天气True订票True [总控Agent] 正在调用「WeatherExpert」 [总控Agent] 正在调用「TicketExpert」 [天气专家] 收到任务 [票务专家] 收到任务 [总控Agent] 收到「WeatherExpert」回复 [总控Agent] 收到「TicketExpert」回复 助手 【WeatherExpert】 上海今天多云22°C东南风3级适合出行。 【TicketExpert】 已为您成功预订火车票 线路北京 → 上海 时间明天 座位08车 06F靠窗 票价二等座 ¥553.0这里有两个关键观察点。第一两个专家任务是并发执行的总耗时取决于最慢的那个而不是两者之和这是asyncio.gather的效果也是 A2A 异步优先设计理念的体现。第二票务专家能识别出用户只说了“去上海”但没说“从哪出发”和“具体时间”时会主动提示补充信息而不是盲目返回错误结果。再测一个信息不完整的场景我想买张去广州的票预期票务专家返回订票信息不完整还需要您补充以下信息出发城市、出行时间。 请提供完整信息后重新下单。这说明 LangGraph 工作流里的校验节点在起作用。如果这里直接返回了订票成功说明_clean_field没生效LLM 把“未知”当成了有效值需要检查_PLACEHOLDER_VALUES集合是否覆盖了模型实际输出的占位词。验证天气 Agent 单独工作杭州天气怎么样预期返回杭州的天气信息。如果返回“未找到杭州的天气信息”说明 Function Calling 没触发LLM 直接把用户输入当成了普通对话。检查llm_with_tools.bind_tools(tools)是否执行以及get_weather的 docstring 是否清晰——工具描述写得越明确模型越容易正确调用。5. 本篇常见错误排查401、local proxy failed、reading choices 逐个击破这一节按真实报错来。多 Agent 系统涉及多个进程、多个网络请求出错时定位链路长先把高频错误列清楚。错误一401 Unauthorizedopenai.AuthenticationError: Error code: 401 - {error: {message: Invalid API key}}原因通常是.env里的MODEL_API_KEY没被正确加载。load_dotenv的路径是按common/llm.py往上推三层找到项目根目录如果你的目录结构和本文不一致路径会错。排查方法在llm.py里加一行print(MODEL_API_KEY[:8])看是否打印出sk-开头的前缀。如果打印出None说明环境变量没读到检查.env文件是否在项目根目录、变量名是否拼写正确。另一个可能是 Key 本身失效或额度耗尽去 https://taotoken.net/api-keys 确认密钥状态。错误二local proxy failed / Connection erroropenai.APIConnectionError: Connection error. httpx.ConnectError: [Errno 111] Connection refused这个报错分两种场景。场景一总控 Agent 连不上专家 Agent说明专家 Server 没启动或端口不对。检查network.add(WeatherExpert, http://127.0.0.1:5008)里的端口和run_server(server, host127.0.0.1, port5008)是否一致。场景二LLM 请求连不上 TaoToken检查MODEL_BASE_URL是否写成了https://taotoken.net/api/v1多写的/v1会导致路径拼接错误。正确值就是https://taotoken.net/api。错误三reading choices / KeyError: choicesKeyError: choices TypeError: Cannot read properties of undefined (reading choices)这个错误说明 API 返回的 JSON 结构里没有choices字段通常是请求根本没到 LLM 层而是被某个中间层拦截返回了错误页。最常见原因是 Base URL 写错请求打到了一个返回 HTML 的地址。用 curl 直接测一下curl -X POST https://taotoken.net/api/v1/chat/completions \ -H Authorization: Bearer sk-你的密钥 \ -H Content-Type: application/json \ -d {model:gpt-4o-mini,messages:[{role:user,content:hi}]}如果 curl 返回正常 JSON说明通道没问题问题在代码里的 base_url 配置。如果 curl 也报错检查 Key 和 Model ID。错误四OAuth / 认证相关报错Error: OAuth token expired Authentication failed: invalid_grant这类报错一般出现在你用了需要 OAuth 的模型通道但 TaoToken 走的是 API Key 认证不应该出现 OAuth 流程。如果看到这个报错检查是不是误装了某个需要 OAuth 的 SDK 插件或者.env里混入了其他平台的配置变量。清理掉无关变量只保留MODEL_API_KEY、MODEL_BASE_URL、MODEL_NAME三个。错误五AgentCard 注册失败 / 404 on agent cardFailed to fetch agent card from http://127.0.0.1:5008/.well-known/agent.jsonA2A 协议要求 Server 暴露/.well-known/agent.json路径返回 AgentCard。如果这个路径 404说明run_server没正确挂载路由。检查python-a2a版本是否过旧升级到最新版。另外确认agent_card的url字段和实际监听地址一致协议、IP、端口三者都要对上。错误六LangGraph 节点返回类型错误InvalidUpdateError: Expected dict, got NoneLangGraph 的节点函数必须返回 state 字典。如果你在extract_city_node里忘了return state就会报这个。检查每个add_node对应的函数是否都有返回值且返回值类型是AgentState。排查顺序建议先 curl 测 LLM 通道再单独启动每个专家 Agent 用 A2AClient 测最后起总控测全链路。逐层验证比一上来就跑全链路高效得多。6. 语义一致的 CTA把统一 Key 和多 Agent 编排用到真实项目里跑通这套 demo 之后你手里其实已经有了一个可扩展的骨架。加一个新专家 Agent 只需要三步写业务逻辑、包成 A2AServer、在network.add()里注册一行。总控的意图识别模块加一个布尔字段就能把新专家纳入调度。真实项目里天气和票务的数据源要换成真实 API比如天气接和风或高德票务接 12306 或携程开放平台。LLM 通道继续用 TaoToken 统一管理模型可以按 Agent 分别指定——意图识别用便宜快的小模型信息抽取用结构化输出稳定的模型这样成本和效果都能兼顾。如果你想把模型对话能力也接进来做调试可以访问 https://taotoken.net/models 直接测试不同模型的输出效果确认哪个 Model ID 在你的场景下表现最好。接入文档在 https://taotoken.net/doc里面有 OpenAI 兼容接口的完整参数说明。长期跑编码类 Agent 或需要稳定额度的场景可以看看 https://taotoken.net/coding-plan按用量规划比临时充值更可控。最后留一个实用技巧多 Agent 系统调试时给每个 Agent 的日志加上统一前缀比如[天气专家]、[票务专家]三个终端并排看日志时能一眼分清谁在说话。这个习惯在 Agent 数量超过三个之后会救命。
