Mit der Agent Runtime können Sie KI-Agenten mit dem Agent2Agent-Protokoll (A2A) entwickeln und bereitstellen. A2A ist ein offener Standard, der eine nahtlose Kommunikation und Zusammenarbeit zwischen KI-Agenten ermöglichen soll.
In diesem Dokument wird beschrieben, wie Sie einen A2A-Agenten lokal entwickeln und testen, einschließlich der Definition von Komponenten wie AgentCard und AgentExecutor.
Weitere Informationen zum Verwalten Ihrer bereitgestellten Agenten finden Sie unter Bereitgestellte Agenten verwalten.
Der Hauptworkflow umfasst die folgenden Schritte:
Agentenkomponenten definieren
Wenn Sie einen A2A-Agenten erstellen möchten, müssen Sie die folgenden Komponenten definieren: einen AgentCard, einen AgentExecutor und einen ADK-LlmAgent.
AgentCardenthält ein Metadatendokument, in dem die Funktionen Ihres Agents beschrieben werden.AgentCardist wie eine Visitenkarte, mit der andere Agenten herausfinden können, was Ihr Agent kann. Weitere Informationen finden Sie in der Spezifikation für Agentenkarten.AgentExecutorenthält die Kernlogik des KI-Agenten und definiert, wie er Aufgaben verarbeitet. Hier implementieren Sie das Verhalten des KI-Agenten. Weitere Informationen findest du in der Spezifikation des A2A Protocol.- Optional:
LlmAgentdefiniert den ADK-Agenten, einschließlich seiner Systemanweisungen, des generativen Modells und der Tools.
AgentCard definieren
Im folgenden Codebeispiel wird ein AgentCard für einen KI-Agenten für Wechselkurse definiert:
from a2a.types import AgentCard, AgentSkill
from vertexai.agent_engines.templates.a2a import create_agent_card
# Define the skill for the CurrencyAgent
currency_skill = AgentSkill(
id='get_exchange_rate',
name='Get Currency Exchange Rate',
description='Retrieves the exchange rate between two currencies on a specified date.',
tags=['Finance', 'Currency', 'Exchange Rate'],
examples=[
'What is the exchange rate from USD to EUR?',
'How many Japanese Yen is 1 US dollar worth today?',
],
)
# Create the agent card using the utility function
agent_card = create_agent_card(
agent_name='Currency Exchange Agent',
description='An agent that can provide currency exchange rates',
skills=[currency_skill]
)
AgentExecutor definieren
Im folgenden Codebeispiel wird ein AgentExecutor definiert, der mit dem Wechselkurs antwortet. Sie verwendet eine CurrencyAgent-Instanz und initialisiert den ADK-Runner, um Anfragen auszuführen.
import requests
from a2a.server.agent_execution.agent_executor import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a import types as a2a_types
from a2a.types import Part
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions.in_memory_session_service import InMemorySessionService
from google.genai import types as genai_types
class CurrencyAgentExecutorWithRunner(AgentExecutor):
"""Executor that takes an LlmAgent instance and initializes the ADK Runner internally."""
def __init__(self, agent: LlmAgent):
self.agent = agent
self.runner = None
def _init_adk(self):
if not self.runner:
self.runner = Runner(
app_name=self.agent.name,
agent=self.agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
task_id = context.task_id
updater = TaskUpdater(
event_queue=event_queue,
task_id=task_id or "",
context_id=context.context_id or "",
)
await updater.cancel()
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
self._init_adk() # Initialize on first execute call
if not context.message:
return
user_id = context.message.metadata.get('user_id') if context.message and context.message.metadata else 'a2a_user'
updater = TaskUpdater(event_queue, context.task_id, context.context_id)
task = a2a_types.Task(
id=context.task_id,
context_id=context.context_id,
status=a2a_types.TaskStatus(state=a2a_types.TaskState.TASK_STATE_SUBMITTED),
history=[context.message] if context.message else [],
)
await event_queue.enqueue_event(task)
await updater.start_work()
query = context.get_user_input()
content = genai_types.Content(role='user', parts=[genai_types.Part.from_text(text=query)])
try:
session = await self.runner.session_service.get_session(
app_name=self.runner.app_name,
user_id=user_id,
session_id=context.context_id,
) or await self.runner.session_service.create_session(
app_name=self.runner.app_name,
user_id=user_id,
session_id=context.context_id,
)
final_event = None
async for event in self.runner.run_async(
session_id=session.id,
user_id=user_id,
new_message=content
):
if event.is_final_response():
final_event = event
if final_event and final_event.content and final_event.content.parts:
response_text = "".join(
part.text for part in final_event.content.parts if hasattr(part, 'text') and part.text
)
if response_text:
await updater.add_artifact(
[Part(text=response_text)],
name='result',
last_chunk=True,
)
await updater.complete()
return
await updater.update_status(
a2a_types.TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text='Failed to generate a final response with text content.')]),
)
except Exception as e:
await updater.update_status(
a2a_types.TaskState.TASK_STATE_FAILED,
message=updater.new_agent_message([Part(text=f"An error occurred: {str(e)}")]),
)
LlmAgent definieren
Definieren Sie zuerst ein Währungsumrechnungstool, das der LlmAgent verwenden soll:
def get_exchange_rate(
currency_from: str = "USD",
currency_to: str = "EUR",
currency_date: str = "latest",
):
"""Retrieves the exchange rate between two currencies on a specified date.
Uses the Frankfurter API (https://api.frankfurter.app/) to obtain
exchange rate data.
"""
try:
response = requests.get(
f"https://api.frankfurter.app/{currency_date}",
params={"from": currency_from, "to": currency_to},
)
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
return {"error": str(e)}
Definieren Sie dann einen ADK-LlmAgent, der das Tool verwendet.
my_llm_agent = LlmAgent(
model='gemini-2.0-flash',
name='currency_exchange_agent',
description='An agent that can provide currency exchange rates.',
instruction="""You are a helpful currency exchange assistant.
Use the get_exchange_rate tool to answer user questions.
If the tool returns an error, inform the user about the error.""",
tools=[get_exchange_rate],
)
Lokalen Agenten erstellen
Nachdem Sie die Komponenten Ihres KI-Agenten definiert haben, erstellen Sie eine Instanz der Klasse A2aAgent, die AgentCard, AgentExecutor und LlmAgent verwendet, um mit dem lokalen Testen zu beginnen.
from vertexai.agent_engines.templates.a2a import A2aAgent
a2a_agent = A2aAgent(
agent_card=agent_card, # Assuming agent_card is defined
agent_executor_builder=lambda: CurrencyAgentExecutorWithRunner(
agent=my_llm_agent,
)
)
a2a_agent.set_up()
Mit der A2A-Agent-Vorlage können Sie einen A2A-konformen Dienst erstellen. Der Dienst fungiert als Wrapper und abstrahiert die Konvertierungsebene für Sie.
Lokalen Agenten testen
Der Agent für die Währungsumrechnung unterstützt die folgenden drei Methoden:
handle_authenticated_agent_cardon_message_sendon_get_task
handle_authenticated_agent_card testen
Mit dem folgenden Code wird die authentifizierte Karte des Agenten abgerufen, die die Funktionen des Agenten beschreibt.
# Test the `authenticated_agent_card` endpoint.
response_get_card = await a2a_agent.handle_authenticated_agent_card(request=None, context=None)
print(response_get_card)
on_message_send testen
Der folgende Code simuliert einen Client, der eine neue Nachricht an den Agenten sendet. Mit A2aAgent wird eine neue Aufgabe erstellt und die ID der Aufgabe zurückgegeben.
from a2a.types import SendMessageRequest, Message, Part
from a2a.server.context import ServerCallContext
# 1. Define the message
message = Message(
role="ROLE_USER",
message_id="local-test-message-id",
parts=[Part(text="What is the exchange rate from USD to EUR today?")]
)
# 2. Construct the request
request = SendMessageRequest(message=message)
# 3. Construct context
context = ServerCallContext()
# 4. Call the agent
send_message_response = await a2a_agent.on_message_send(request=request, context=context)
print(send_message_response)
on_get_task testen
Mit dem folgenden Code werden der Status und das Ergebnis einer Aufgabe abgerufen. Die Ausgabe zeigt, dass die Aufgabe abgeschlossen ist, und enthält das Antwortartefakt „Hello World“.
from a2a.types import GetTaskRequest
# 1. Provide the task_id from the previous step.
# In a real application, you would store and retrieve this ID.
task_id_to_get = send_message_response.id
# 2. Construct the request
request = GetTaskRequest(id=task_id_to_get)
# 3. Call the agent's handler to get the task status.
# Reusing the context constructed in the previous step
task_status_response = await a2a_agent.on_get_task(request=request, context=context)
print(f"Successfully retrieved status for Task ID: {task_id_to_get}")
print("\nFull task status response:")
print(task_status_response)
Nächste Schritte
Agenten bereitstellen
Hier erfahren Sie, wie Sie einen Agenten auf der Agent Platform Runtime auf fünf verschiedene Arten bereitstellen können, je nach Ihren Entwicklungsanforderungen.
KI-Agenten bewerten
Einfachen Agenten erstellen und bereitstellen und mit dem Gen AI Evaluation Service bewerten
Fehlerbehebung bei der Agentenerstellung
Hier erfahren Sie, wie Sie häufige Fehler beim Erstellen benutzerdefinierter KI-Agenten beheben können.