Agent2Agent-Agent entwickeln

Vorschau

Diese Funktion unterliegt den „Pre-GA-Angebotsbedingungen“ im Abschnitt „Allgemeine Dienstbedingungen“ der dienstspezifischen Nutzungsbedingungen. Pre-GA-Funktionen sind „wie gesehen“ verfügbar und bieten möglicherweise nur eingeschränkten Support. Weitere Informationen finden Sie in den Beschreibungen der Produkteinführungsphasen.

Auf dieser Seite erfahren Sie, wie Sie einen Agent2Agent-Agenten (A2A) entwickeln und testen. Das A2A-Protokoll ist ein offener Standard, der eine nahtlose Kommunikation und Zusammenarbeit zwischen KI-Agents ermöglichen soll. In diesem Leitfaden wird der lokale Workflow beschrieben, mit dem Sie die Funktionen Ihres Agents vor der Bereitstellung definieren und überprüfen können.

Wenn Sie ein Beispiel für die ersten Schritte mit A2A in Agent Engine sehen möchten, führen Sie das Notebook „Erste Schritte mit A2A in Agent Engine“ in einer der folgenden Umgebungen aus:

In Colab öffnen | In Colab Enterprise öffnen | In Vertex AI Workbench öffnen | Auf GitHub ansehen

Der Kern-Workflow umfasst die folgenden Schritte:

1. Wichtige Komponenten definieren
2. Lokalen Agenten erstellen
3. Lokalen Agenten testen

Agent-Komponenten definieren

Zum Erstellen eines A2A-Agents müssen Sie die folgenden Komponenten definieren: eine AgentCard, eine AgentExecutor und ein ADK LlmAgent.

• AgentCard enthält ein Metadatendokument, in dem die Funktionen Ihres Agents beschrieben werden. AgentCard ist wie eine Visitenkarte, mit der andere Agenten herausfinden können, was Ihr Agent kann. Weitere Informationen finden Sie in der Spezifikation für Agentenkarten.
• AgentExecutor enthält die Kernlogik des Agents und definiert, wie er Aufgaben verarbeitet. Hier implementieren Sie das Verhalten des Agents. Weitere Informationen finden Sie in der A2A-Protokollspezifikation.
• Optional: Mit LlmAgent wird der ADK-Agent definiert, einschließlich seiner Systemanleitung, des generativen Modells und der Tools.

AgentCard definieren

Im folgenden Codebeispiel wird ein AgentCard für einen Agenten für Währungsumrechnungskurse definiert:

from a2a.types import AgentCard, AgentSkill
from vertexai.preview.reasoning_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 eine AgentExecutor definiert, die mit dem Währungskurs antwortet. Es wird eine CurrencyAgent-Instanz verwendet und der ADK Runner wird initialisiert, um Anfragen auszuführen.

import requests
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, UnsupportedOperationError, Part
from a2a.utils import new_agent_text_message
from a2a.utils.errors import ServerError
from google.adk import Runner
from google.adk.agents import LlmAgent
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.sessions import InMemorySessionService
from google.genai import 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):
        raise ServerError(error=UnsupportedOperationError())

    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)
        if not context.current_task:
            await updater.submit()
        await updater.start_work()

        query = context.get_user_input()
        content = types.Content(role='user', parts=[types.Part(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(
                        [TextPart(text=response_text)],
                        name='result',
                    )
                    await updater.complete()
                    return

            await updater.update_status(
                TaskState.failed,
                message=new_agent_text_message('Failed to generate a final response with text content.'),
                final=True
            )

        except Exception as e:
            await updater.update_status(
                TaskState.failed,
                message=new_agent_text_message(f"An error occurred: {str(e)}"),
                final=True,
            )

LlmAgent definieren

Definieren Sie zuerst ein Tool für den Währungsumtausch, das von LlmAgent verwendet werden 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 eine ADK-LlmAgent, die 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 Agents definiert haben, erstellen Sie eine Instanz der Klasse A2aAgent, die AgentCard, AgentExecutor und LlmAgent verwendet, um mit dem lokalen Testen zu beginnen.

from vertexai.preview.reasoning_engines 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-kompatiblen Dienst erstellen. Der Dienst fungiert als Wrapper und abstrahiert die Konvertierungsebene für Sie.

Lokalen Agent testen

Der Agent für Währungsumrechnung unterstützt die folgenden drei Methoden:

• handle_authenticated_agent_card
• on_message_send
• on_get_task

Test handle_authenticated_agent_card

Mit dem folgenden Code wird die authentifizierte Karte des Agents abgerufen, in der die Funktionen des Agents beschrieben werden.

# Test the `authenticated_agent_card` endpoint.
response_get_card = await a2a_agent.handle_authenticated_agent_card(request=None, context=None)
print(response_get_card)

Test on_message_send

Der folgende Code simuliert einen Client, der eine neue Nachricht an den Agent sendet. Mit A2aAgent wird eine neue Aufgabe erstellt und die ID der Aufgabe zurückgegeben.

import json
from starlette.requests import Request
import asyncio

# 1. Define the message payload you want to send.
message_data = {
    "message": {
        "messageId": "local-test-message-id",
        "content":[
              {
                  "text": "What is the exchange rate from USD to EUR today?"
              }
          ],
        "role": "ROLE_USER",
    },
}

# 2. Construct the request
scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "POST",
    "headers": [(b"content-type", b"application/json")],
}

async def receive():
    byte_data = json.dumps(message_data).encode("utf-8")
    return {"type": "http.request", "body": byte_data, "more_body": False}

post_request = Request(scope, receive=receive)

# 3. Call the agent
send_message_response = await a2a_agent.on_message_send(request=post_request, context=None)

print(send_message_response)

Test on_get_task

Mit dem folgenden Code werden der Status und das Ergebnis einer Aufgabe abgerufen. Die Ausgabe zeigt, dass die Aufgabe abgeschlossen ist und das Antwortartefakt „Hello World“ enthält.

from starlette.requests import Request
import asyncio

# 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['task']['id']

# 2. Define the path parameters for the request.
task_data = {"id": task_id_to_get}

# 3. Construct the starlette.requests.Request object directly.
scope = {
    "type": "http",
    "http_version": "1.1",
    "method": "GET",
    "headers": [],
    "query_string": b'',
    "path_params": task_data,
}

async def empty_receive():
    return {"type": "http.disconnect"}

get_request = Request(scope, empty_receive)

# 4. Call the agent's handler to get the task status.
task_status_response = await a2a_agent.on_get_task(request=get_request, context=None)

print(f"Successfully retrieved status for Task ID: {task_id_to_get}")
print("\nFull task status response:")
print(task_status_response)

Nächste Schritte

• Agent2Agent-Beispielrepository
• Agent bewerten
• Agent bereitstellen
• Fehlerbehebung bei der Entwicklung eines Agenten
• Support anfordern