5.1 多智能体协作¶

学习目标¶

了解直接调用其他智能体的方法和缺点
掌握使用A2A协议调用其他智能体的方法

一、直接调用其他智能体¶

1 实施方法¶

将独立智能体做成http服务器，用一个整合智能体去调用所有独立智能体，调用方式、数据格式等由开发团队内部决定。

2 遗留问题¶

多个智能体之间的交互要依赖人工或者单独开发处理系统，工作量很大。
多个智能体如果要交互数据，用什么方式进行交互？交互数据的格式怎样确定？
每次开发多智能体系统都要从头立项开发，成本高，效果也无法保证。
随着智能体的数量增多，他们的组合方式成几何级数增长，无法进行工业化生产。

二、基于A2A协议进行交互¶

1 原理-A2A协议简介¶

A2A

模块 / 组件	功能 / 目的
Agent Card（Agent 卡）	每个 Agent 发布一个结构化的元数据卡，包含自己能做什么 (capabilities)、支持哪些输入／输出类型、信任与安全要求、URI／端点地址等。这是 Agent 被发现和协作的基础。
发现机制（Discovery）	Agent Card 用来被其他 Agent 或者 Agent 管理系统查到。可能有目录服务 (registry)，Agent 查 capabilities，查看哪些 Agent 能处理某类任务。
通信协议	用 HTTP 或 HTTPS 作为传输层。消息格式通常是 JSON 或 JSON-RPC。支持同步请求‐响应，也支持异步／长任务的推送、事件流／SSE 等。
任务委派与状态跟踪	一个 Agent 可以发起任务给另一个 Agent，传输任务描述、上下文、期望的结果。被委派的 Agent 执行任务后，需要反馈中间状态／最终状态／可能的错误／进度更新等。还有唯一的任务 ID 用来追踪。
权限与身份验证	Agent 之间的通信必须确认对方是谁，并且是否有权限执行某些操作。可能用 JWT、OIDC等机制，也可能有权限控制策略。
错误处理与中断	如果某个 Agent 出错，比如任务无法完成，或者响应超时，要有标准方式告诉委托方，也可能要有重试机制或者降级逻辑。
日志与审计	企业环境里尤其重要。谁调用了谁，任务什么时间开始／结束／失败／中断，中间内容怎么变动等，都要有可记录、可追溯的日志。

2 实例演示¶

服务器

from fastapi import FastAPI
from pydantic import BaseModel
import uvicorn


# Define the A2A Task structure
class Task(BaseModel):
    task_id: str
    message: dict


# Initialize FastAPI app
app = FastAPI()


# Define the Agent Card (metadata about the agent)
echo_skill = {
    "name":"echo_query",
    "description":"客户端发出的任何信息，服务器会在前面添加上 'Hello! You said:' 之后返回",
    "examples":"客户端发来任意一句话",
    "input_modes":["text"],
    "output_modes":["text"]
}

agent_card = {
    "name": "EchoAgent",
    "description": "Echoes back your message with a greeting",
    "endpoint": "http://localhost:8000/tasks/send",
    "capabilities": ["text"],
    "skills":[echo_skill]
}


# Expose the Agent Card at /.well-known/agent.json
@app.get("/.well-known/agent.json")
async def get_agent_card():
    return agent_card


# Handle incoming tasks
@app.post("/tasks/send")
async def handle_task(task: Task):
    user_message = task.message.get("parts", [{}])[0].get("text", "")
    response = {
        "task_id": task.task_id,
        "status": "completed",
        "message": {
            "role": "agent",
            "parts": [{"text": f"Hello! You said: {user_message}"}]
        }
    }
    return response


# Run the server
if __name__ == "__main__":
    uvicorn.run(app, host="0.0.0.0", port=8000)

客户端

import requests
import uuid

def getAgentCard():
    # A2A server endpoint
    SERVER_URL = "http://localhost:8000/.well-known/agent.json"

    # Send the task to the server
    response = requests.get(SERVER_URL)
    # Check the response
    if response.status_code == 200:
        result = response.json()
        agent_message = result
        print(f"Agent card: {agent_message}")
    else:
        print(f"Error: {response.status_code}")

def excuteTask():
    # A2A server endpoint
    SERVER_URL = "http://localhost:8000/tasks/send"


    # Create a simple task
    task_id = str(uuid.uuid4())  # Unique task ID
    task = {
        "task_id": task_id,
        "message": {
            "role": "user",
            "parts": [{"text": "Hi, how are you?"}]
        }
    }


    # Send the task to the server
    response = requests.post(SERVER_URL, json=task)
    # Check the response
    if response.status_code == 200:
        result = response.json()
        #agent_message = result["message"]["parts"][0]["text"]
        agent_message = result
        print(f"Agent replied: {agent_message}")
    else:
        print(f"Error: {response.status_code}")

if __name__ == "__main__":
    getAgentCard()
    #excuteTask()