两个AI智能体第一次对话-A2A双Agent协作实战

两个AI智能体第一次对话A2A双Agent协作实战摘要上篇搭了一个Echo Agent这次搭第二个。让翻译Agent和摘要Agent通过A2A协议真正对话跑通双Agent委托任务的全流程。一、上一篇的Agent只会自言自语上篇文章你搭了一个Echo Agent它能收消息、能回消息但本质上是个单人相声——只有你自己跟它对话。真正有意思的场景是两个Agent之间互相发消息、互相委托任务。比如你让翻译Agent把一段中文翻成英文翻译Agent干完了自动把结果丢给摘要Agent摘要Agent提炼出要点再返回给你。整个过程你只需要发一条消息剩下的Agent自己协作完成。今天我就带你搭这个场景。二、先搞清楚双Agent怎么通信回顾单Agent架构上篇的架构很简单你Client→ 发消息 → Echo AgentServer一个Client一个Server。你发一条消息Agent回一条。完事。双Agent架构加一个Agent之后架构变成这样你Client→ 发消息 → 翻译AgentServer ↓ A2A请求 摘要AgentServer ↓ A2A响应 翻译Agent汇总结果 ↓ 最终响应 你Client关键变化翻译Agent既是Server接收你的请求又是Client向摘要Agent发请求。一个Agent可以同时扮演两个角色。这就是A2A的核心设计——Agent之间是平等的谁都可以当Client谁都可以当Server。通信流程拆解整个流程分5步你给翻译Agent发消息“把这段话翻成英文再帮我写个摘要”翻译Agent收到消息先做翻译翻译Agent通过A2A协议把翻译结果发给摘要Agent摘要Agent处理完把摘要返回给翻译Agent翻译Agent把翻译结果摘要一起返回给你三、项目结构这次我们搭两个独立的Agent各自有独立的端口和Agent Card。a2a-dual-agent/ ├── translator/# 翻译Agent│ ├── agent_card.json │ ├── server.py │ └── task_manager.py ├── summarizer/# 摘要Agent│ ├── agent_card.json │ ├── server.py │ └── task_manager.py ├── client.py# 测试客户端├── pyproject.toml └── .env# 存放Ollama配置两个Agent跑在不同的端口翻译Agent在10002摘要Agent在10003。四、搭建摘要Agent先搭被调用的那个为什么先搭摘要Agent因为翻译Agent要调用它得先让它跑起来。4.1 创建项目mkdira2a-dual-agentcda2a-dual-agent uv init--packagedual-agent uv venv .venvsource.venv/bin/activate uvaddgithttps:// github.com/google/A2A#subdirectorysamples/pythonuvaddclick httpx4.2 摘要Agent的Task Manager创建summarizer/ [task_manager.py](http://task_manager.py)from google_a2a.common.server.task_managerimportInMemoryTaskManager from google_a2a.common.typesimport(Artifact, Message, SendTaskRequest, SendTaskResponse, Task, TaskState, TaskStatus,)classSummarizerTaskManager(InMemoryTaskManager):摘要Agent接收文本返回摘要。 asyncdefon_send_task(self, request: SendTaskRequest)-SendTaskResponse: awaitself.upsert_task(request.params)task_idrequest.params.idinput_textrequest.params.message.parts[0].text# 截取前200字作为摘要demo用Ollama生成真实摘要# 这里调用Ollamatry:importhttpx asyncwith httpx.AsyncClient(timeout30.0)as c: respawait c.post(http://localhost:11434/api/generate,json{model: qwen2.5:7b,prompt:(f请用中文对以下内容写一段50字以内的摘要\nf{input_text}),stream:False,},)resultresp.json()summaryresult.get(response,摘要生成失败)except Exception: summaryf摘要原文截取{input_text[:100]}...taskawaitself._update_task(task_idtask_id,task_stateTaskState.COMPLETED,response_textsummary,)returnSendTaskResponse(idrequest.id,resulttask)asyncdef_update_task(self, task_id: str, task_state: TaskState, response_text: str)-Task: taskself.tasks[task_id]parts[{type:text,text:response_text}]task.statusTaskStatus(statetask_state,messageMessage(roleagent,partsparts),)iftask_stateTaskState.COMPLETED: task.artifacts[Artifact(partsparts)]returntask4.3 摘要Agent的Server创建summarizer/ [server.py](http://server.py)importloggingimportclick from google_a2a.common.typesimport(AgentSkill, AgentCapabilities, AgentCard)from google_a2a.common.serverimportA2AServer from task_managerimportSummarizerTaskManager logging.basicConfig(levellogging.INFO)click.command()click.option(--host,defaultlocalhost)click.option(--port,default10003,typeint)defmain(host, port): skillAgentSkill(idsummarize-skill,nameText Summarizer,descriptionSummarizes any text into a concise summary,tags[summary,nlp,text],examples[请帮我总结这篇文章的要点],inputModes[text],outputModes[text],)agent_cardAgentCard(nameSummarizer Agent,descriptionReceives text and returns a concise summary.,urlfhttp://{host}:{port}/,version0.1.0,defaultInputModes[text],defaultOutputModes[text],capabilitiesAgentCapabilities(streamingFalse),skills[skill],)task_managerSummarizerTaskManager()serverA2AServer(agent_cardagent_card,task_managertask_manager,hosthost,portport,)logging.info(fSummarizer Agent running at http://{host}:{port})server.start()if__name____main__:main()4.4 启动摘要Agentcdsummarizer python server.py看到Uvicorn running on http://localhost:10003就OK了。五、搭建翻译Agent它会调用摘要Agent翻译Agent是核心角色它既接收你的请求做翻译又作为Client去调用摘要Agent。5.1 翻译Agent的Task Manager创建translator/ [task_manager.py](http://task_manager.py)importhttpx from google_a2a.common.server.task_managerimportInMemoryTaskManager from google_a2a.common.typesimport(Artifact, Message, SendTaskRequest, SendTaskResponse, Task, TaskState, TaskStatus,)SUMMARIZER_URLhttp://localhost:10003OLLAMA_URLhttp://localhost:11434/api/generateOLLAMA_MODEL qwen2.5:7bclassTranslatorTaskManager(InMemoryTaskManager):翻译Agent翻译文本后调用摘要Agent生成摘要。 asyncdefon_send_task(self, request: SendTaskRequest)-SendTaskResponse: awaitself.upsert_task(request.params)task_idrequest.params.idinput_textrequest.params.message.parts[0].text# 第一步翻译translationawaitself._translate(input_text)# 更新状态为工作中已完成翻译开始摘要awaitself._update_task(task_idtask_id,task_stateTaskState.WORKING,response_textf翻译完成正在生成摘要...\n\n翻译结果\n{translation},)# 第二步调用摘要Agentsummaryawaitself._call_summarizer(translation)# 第三步汇总返回final_text(f【翻译结果】\n{translation}\n\nf【摘要】\n{summary})taskawaitself._update_task(task_idtask_id,task_stateTaskState.COMPLETED,response_textfinal_text,)returnSendTaskResponse(idrequest.id,resulttask)asyncdef_translate(self, text: str)-str:调用Ollama进行翻译。 try: asyncwith httpx.AsyncClient(timeout60.0)as c: respawait c.post(OLLAMA_URL,json{model:OLLAMA_MODEL,prompt:(f请将以下中文翻译成英文只输出翻译结果\nf{text}),stream:False,},)resultresp.json()returnresult.get(response,翻译失败)except Exception as e:returnf翻译出错: {str(e)}asyncdef_call_summarizer(self, text: str)-str:通过A2A协议调用摘要Agent。 try:# 1. 发现摘要Agent获取Agent Cardasyncwith httpx.AsyncClient()as c: card_respawait c.get(f{SUMMARIZER_URL}/.well-known/ agent.json)cardcard_resp.json()endpointcard[url]# 2. 发送A2A请求payload{jsonrpc:2.0,id:summary-request-001,method:message/send,params:{message:{role:user,parts:[{type:text,text:text,}],}},}respawait c.post(endpoint,jsonpayload,headers{Content-Type:application/json},timeout60.0,)resultresp.json()# 3. 解析响应ifresultinresult: task_dataresult[result]artifactstask_data.get(artifacts,[])ifartifacts: partsartifacts[0].get(parts,[])ifparts:returnparts[0].get(text,无摘要)messagetask_data.get(status,{}).get(message,{})partsmessage.get(parts,[])ifparts:returnparts[0].get(text,无摘要)return摘要Agent返回格式异常except Exception as e:returnf调用摘要Agent失败: {str(e)}asyncdef_update_task(self, task_id: str, task_state: TaskState, response_text: str)-Task: taskself.tasks[task_id]parts[{type:text,text:response_text}]task.statusTaskStatus(statetask_state,messageMessage(roleagent,partsparts),)iftask_stateTaskState.COMPLETED: task.artifacts[Artifact(partsparts)]returntask5.2 翻译Agent的Server创建translator/ [server.py](http://server.py)importloggingimportclick from google_a2a.common.typesimport(AgentSkill, AgentCapabilities, AgentCard)from google_a2a.common.serverimportA2AServer from task_managerimportTranslatorTaskManager logging.basicConfig(levellogging.INFO)click.command()click.option(--host,defaultlocalhost)click.option(--port,default10002,typeint)defmain(host, port): skillAgentSkill(idtranslate-skill,nameTranslator,description(Translates Chinese text to English, then delegates summarization to Summarizer Agent),tags[translation,nlp,multi-agent],examples[帮我把这段话翻成英文并写个摘要],inputModes[text],outputModes[text],)agent_cardAgentCard(nameTranslator Agent,description(Translates text and coordinates with Summarizer Agent via A2A protocol.),urlfhttp://{host}:{port}/,version0.1.0,defaultInputModes[text],defaultOutputModes[text],capabilitiesAgentCapabilities(streamingFalse),skills[skill],)task_managerTranslatorTaskManager()serverA2AServer(agent_cardagent_card,task_managertask_manager,hosthost,portport,)logging.info(fTranslator Agent running at http://{host}:{port})server.start()if__name____main__:main()六、启动测试6.1 启动两个Agent开两个终端# 终端1启动摘要Agentcdsummarizerpython server.py# 终端2启动翻译Agentcdtranslatorpython server.py两个终端分别看到Uvicorn启动信息就OK。6.2 用客户端测试创建[client.py](http://client.py)importasyncioimporthttpx from a2a.clientimportA2ACardResolver, ClientConfig, create_client from a2a.helpersimportnew_text_message from a2a.types.a2a_pb2importRole, SendMessageRequest asyncdefmain(): print( 双Agent协作测试 \n)asyncwith httpx.AsyncClient()as httpx_client:# 发现翻译AgentresolverA2ACardResolver(httpx_clienthttpx_client,base_urlhttp://localhost:10002,)agent_cardawait resolver.get_agent_card()print(f连接到: { agent_card.name})print(f技能: {[ s.name for s in agent_card.skills]}\n)# 创建客户端clientawaitcreate_client(agentagent_card,client_configClientConfig(streamingFalse),)# 发送消息text(人工智能正在改变我们工作的方式。从自动写代码到智能客服AI工具已经渗透到各行各业。但不同AI之间的协作仍然是一个巨大的挑战A2A协议正是为了解决这个问题而诞生的。)messagenew_text_message(text,roleRole.ROLE_USER)requestSendMessageRequest(messagemessage)print(f发送: {text}\n)print(等待翻译Agent和摘要Agent协作处理...\n)print(*50)asyncfor chunkinclient.send_message(request):# 提取文本内容ifhasattr(chunk,artifacts)and chunk.artifacts:forpartinchunk.artifacts[0].parts: print(part.text)elifhasattr(chunk,status)and chunk.status.message:forpartinchunk.status.message.parts: print(part.text)await client.close()if__name____main__:asyncio.run(main())运行python client.py6.3 预期输出双Agent协作测试连接到: Translator Agent 技能:[Translator]发送: 人工智能正在改变我们工作的方式... 等待翻译Agent和摘要Agent协作处理...【翻译结果】 Artificial intelligence is changing the way we work... 【摘要】 本文讨论了AI对各行各业的渗透以及A2A协议解决AI协作挑战的意义。看到这个输出说明两个Agent成功通过A2A协议完成了一次完整的协作。七、整个流程发生了什么拆解一下你刚才看到的结果背后发生了什么第1步你的 client.py通过A2A协议把中文文本发给翻译Agent端口10002。第2步翻译Agent收到消息调用本地Ollama大模型进行翻译。第3步翻译Agent通过A2A协议直接用httpx发JSON-RPC请求把英文翻译结果发给摘要Agent端口10003。第4步摘要Agent收到英文文本调用Ollama生成摘要返回给翻译Agent。第5步翻译Agent把翻译结果和摘要拼接在一起作为最终结果返回给你。整个过程你只发了一条消息。两个Agent之间的通信、任务委托、结果汇总全部通过A2A协议自动完成。八、几个值得注意的细节翻译Agent怎么调用摘要Agent注意_call_summarizer这个方法。它做了一件很关键的事——手动构造A2A请求。payload{jsonrpc:2.0,id:summary-request-001,method:message/send,params:{message:{role:user,parts:[{type:text,text:text}],}},}这就是A2A协议的标准格式。翻译Agent作为Client直接用httpx向摘要Agent的端点发送JSON-RPC请求。没有SDK依赖没有魔法就是HTTP JSON。这也是A2A的魅力所在协议就是协议任何能发HTTP请求的代码都能参与。为什么要先获取Agent Card在发请求之前翻译Agent先访问了摘要Agent的/.well-known/ [agent.json](http://agent.json)card_respawait c.get(f{SUMMARIZER_URL}/.well-known/ agent.json)endpointcard[url]这叫能力发现。翻译Agent先搞清楚摘要Agent的地址和能力再决定怎么调用。在真实场景中你可能根据Agent Card里的skill信息来决定要不要调用、调用哪个skill。两个Agent可以独立部署这个demo里两个Agent跑在同一台电脑上。但在生产环境中它们可以跑在不同的服务器、不同的云、甚至不同的厂商平台上。只要网络互通A2A协议就能让它们协作。九、我踩过的3个新坑坑1摘要Agent没启动就测试。翻译Agent调用摘要Agent时连接被拒绝报ConnectionRefusedError。解决方法先启动摘要Agent10003端口再启动翻译Agent10002端口。坑2Ollama没装或没启动。两个Agent都依赖Ollama。如果你没装Ollama两个Agent都会返回翻译失败或摘要生成失败。解决方法ollama serve先启动Ollama服务。坑3JSON-RPC响应格式解析错误。A2A的响应格式跟普通HTTP响应不一样是JSON-RPC格式的。第一次解析时我直接取[response.text](http://response.text)拿到的是原始JSON字符串而不是解析后的对象。解决方法先.json()解析再按result → artifacts → parts → text的路径提取。十、下一步这篇你实现了两个Agent之间的单向委托翻译Agent调摘要Agent。下一篇我们要解决一个更关键的问题安全。当你把Agent暴露到网络上怎么确保只有授权的Agent才能调用A2A协议支持API Key、JWT Bearer、OAuth 2.0三种认证方式下一篇我会带你逐一实现给你的Agent加把锁。下篇预告《给你的A2A Agent加把锁认证鉴权实战指南》—— API Key、JWT Bearer、OAuth 2.0三种认证方式从开发环境到生产环境的完整安全方案。