שימוש בסוכן Agent2Agent ‏ (A2A)

לפני שמתחילים

במדריך הזה אנחנו יוצאים מנקודת הנחה שקראתם את ההוראות במאמרים הבאים ופעלתם לפיהן:

אחזור מופע של סוכן

כדי לשלוח שאילתה ל-A2aAgent, צריך קודם ליצור מכונה חדשה או לקבל מכונה קיימת.

כדי לקבל את A2aAgent שמתאים למזהה משאב ספציפי:

Agent Platform SDK 

import vertexai
from google.genai import types

PROJECT_ID = "PROJECT_ID"
LOCATION = "LOCATION"
RESOURCE_ID = "RESOURCE_ID"
RESOURCE_NAME = f"projects/{PROJECT_ID}/locations/{LOCATION}/reasoningEngines/{RESOURCE_ID}"

client = vertexai.Client(
    project=PROJECT_ID,
    location=LOCATION,
    http_options=types.HttpOptions(
        api_version="v1beta1",
    )
)

remote_agent = client.agent_engines.get(name=RESOURCE_NAME)

print(remote_agent)

איפה

A2A Python SDK

השיטה הזו משתמשת ב-A2A Python SDK הרשמי, שמספק ספריית לקוח לאינטראקציה עם סוכנים שתואמים ל-A2A. מידע נוסף מופיע במסמכי התיעוד של A2A Python SDK.

קודם כול, מתקינים את ה-SDK:

pip install a2a-sdk>=0.3.4

לאחר מכן, מקבלים את כרטיס הסוכן כדי ליצור מופע של לקוח. ‫A2AClient מטפל בגילוי ובשיחה בשבילכם.

from google.auth import default
from google.auth.transport.requests import Request
from a2a.client import ClientConfig, ClientFactory
from a2a.types import TransportProtocol
import httpx

# We assume 'agent_card' is an existing AgentCard object.

# Fetch credentials for authentication for demo purpose. Use your own auth
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

# Create the client by chaining the factory and config initialization.
factory = ClientFactory(
    ClientConfig(
        supported_transports=[TransportProtocol.http_json], # only support http_json
        use_client_preference=True,
        httpx_client=httpx.AsyncClient(
            headers={
                "Authorization": f"Bearer {credentials.token}",
                "Content-Type": "application/json",
            }
        ),
    )
)
a2a_client = factory.create(agent_card)

ספריית הבקשות של Python

פרוטוקול A2A מבוסס על נקודות קצה רגילות של HTTP. אפשר לקיים אינטראקציה עם נקודות הקצה האלה באמצעות כל לקוח HTTP.

מאחזרים את כתובת ה-URL של A2A מכרטיס הסוכן ומגדירים את כותרות הבקשה.

from google.auth import default
from google.auth.transport.requests import Request

# We assume 'agent_card' is an existing object
a2a_url = agent_card.url

# Get an authentication token for demonstration purposes. Use your own authentication mechanism.
credentials, _ = default(scopes=['https://www.googleapis.com/auth/cloud-platform'])
credentials.refresh(Request())

headers = {
    "Authorization": f"Bearer {credentials.token}",
    "Content-Type": "application/json",
}

כשמשתמשים ב-Agent Platform SDK, האובייקט remote_agent תואם למחלקה AgentEngine שכוללת את המאפיינים הבאים:

  • ‫remote_agent.api_resource עם מידע על הסוכן שנפרס. אפשר גם לקרוא ל-remote_agent.operation_schemas() כדי להחזיר את רשימת הפעולות ש-remote_agent תומך בהן. פרטים נוספים זמינים במאמר בנושא פעולות נתמכות.
  • ‫remote_agent.api_client שמאפשרת אינטראקציות סינכרוניות עם שירותים
  • ‫remote_agent.async_api_client שמאפשרת אינטראקציות אסינכרוניות בין שירותים

בהמשך הקטע הזה נניח שיש לכם מכונת AgentEngine שנקראת remote_agent.

פעולות נתמכות

סוכן A2A שמתארח ב-Agent Runtime חושף קבוצה של פעולות שתואמות ישירות לנקודות הקצה של ה-API של פרוטוקול A2A.

  • ‫on_message_send: שליחת הודעה חדשה לסוכן כדי להתחיל משימה.
  • ‫on_get_task: אחזור הסטטוס ופריטי המידע שנוצרו בתהליך פיתוח (Artifact) של משימה קיימת.
  • ‫on_cancel_task: ביטול משימה שפועלת.
  • ‫handle_authenticated_agent_card: אחזור של היכולות והכישורים המלאים של הסוכן.

אחזור כרטיס הסוכן

שימו לב: כרטיס הנציג הציבורי לא מוצג ב-Agent Runtime. כדי לאחזר את כרטיס הנציג המאומת:

Agent Platform SDK 

response = await remote_agent.handle_authenticated_agent_card()

A2A Python SDK

response = await a2a_client.get_card()

ספריית הבקשות של Python

card_endpoint = f"{a2a_url}/v1/card"
response = httpx.get(card_endpoint, headers=headers)
print(json.dumps(response.json(), indent=4))

שליחת הודעה

כדי לשלוח הודעה:

Agent Platform SDK 

message_data = {
  "messageId": "remote-agent-message-id",
  "role": "user",
  "parts": [{"kind": "text", "text": "What is the exchange rate from USD to EUR today?"}],
}

response = await remote_agent.on_message_send(**message_data)

כדי לשנות את הזמן הקצוב לתפוגה שמוגדר ל-on_message_send, צריך להגדיר את השדה timeout של HttpOptions כשיוצרים את vertexai.Client.

A2A Python SDK

from a2a.types import Message, Part, TextPart
import pprint

message = Message(
    message_id="remote-agent-message-id",
    role="user",
    parts=[Part(root=TextPart(text="What's the currency rate of USD and EUR"))],
)

response_iterator = a2a_client.send_message(message)

async for chunk in response_iterator:
    pprint.pp(chunk)

ספריית הבקשות של Python

import httpx
import json

endpoint = f"{a2a_url}/v1/message:send"

payload = {
    "message": {
        "messageId": "remote-agent-message-id",
        "role": "1",
        "content": [{"text": "What is the exchange rate from USD to EUR today?"}],
    },
    "metadata": {"source": "python_script"},
}

response = httpx.post(endpoint, json=payload, headers=headers)
print(json.dumps(response.json(), indent=4))

קבלת משימה

כדי לקבל משימה ואת הסטטוס שלה

Agent Platform SDK 

task_data = {
    "id": task_id,
}

response = await remote_agent.on_get_task(**task_data)

A2A Python SDK

from a2a.types import TaskQueryParams

task_data ={
    "id":task_id,
}
response = await a2a_client.get_task(TaskQueryParams(**task_data))

ספריית הבקשות של Python

task_end_point = f"{a2a_url}/v1/tasks/{task_id}"
response = httpx.get(task_end_point, headers=headers)
print(json.dumps(response.json(), indent=4))

ביטול משימה

כדי לבטל משימה:

Agent Platform SDK 

task_data = {
    "id": task_id,
}
response = await remote_agent.on_cancel_task(**task_data)

A2A Python SDK

from a2a.types import TaskQueryParams

task_data = {
    "id":task_id,
}
response = await a2a_client.cancel_task(TaskQueryParams(**task_data))

ספריית הבקשות של Python

task_end_point = f"{a2a_url}/v1/tasks/{task_id}:cancel"
response = httpx.post(task_end_point, headers=headers)
print(json.dumps(response.json(), indent=4))

המאמרים הבאים