Membuat agen Agent2Agent

Agent Runtime memungkinkan Anda mengembangkan dan men-deploy agen menggunakan protokol Agent2Agent (A2A). A2A adalah standar terbuka yang dirancang untuk memungkinkan komunikasi dan kolaborasi yang lancar antara agen AI.

Dokumen ini menjelaskan cara mengembangkan dan menguji agen A2A secara lokal, termasuk menentukan komponen seperti AgentCard dan AgentExecutor.

Untuk mengetahui informasi selengkapnya tentang mengelola agen yang di-deploy, lihat Mengelola agen yang di-deploy.

Alur kerja inti melibatkan langkah-langkah berikut:

  1. Menentukan komponen utama
  2. Membuat agen lokal
  3. Menguji agen lokal

Menentukan komponen agen

Untuk membuat agen A2A, Anda harus menentukan komponen berikut: AgentCard, AgentExecutor, dan LlmAgent ADK.

  • AgentCard berisi dokumen metadata yang menjelaskan kemampuan agen Anda. AgentCard seperti kartu nama yang dapat digunakan agen lain untuk menemukan kemampuan agen Anda. Untuk mengetahui detail selengkapnya, lihat spesifikasi Kartu Agen.
  • AgentExecutor berisi logika inti agen dan menentukan cara agen menangani tugas. Di sinilah Anda menerapkan perilaku agen. Anda dapat membacanya lebih lanjut dalam spesifikasi protokol A2A.
  • Opsional: LlmAgent menentukan agen ADK, termasuk petunjuk sistem, model generatif, dan alatnya.

Menentukan AgentCard

Contoh kode berikut menentukan AgentCard untuk agen nilai tukar mata uang:

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]
)

Menentukan AgentExecutor

Contoh kode berikut menentukan AgentExecutor yang merespons dengan nilai tukar mata uang. Contoh kode ini mengambil instance CurrencyAgent dan menginisialisasi Runner ADK untuk menjalankan permintaan.

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)}")]),
            )

Menentukan LlmAgent

Pertama, tentukan alat nilai tukar mata uang yang akan digunakan LlmAgent:

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)}

Kemudian, tentukan LlmAgent ADK yang menggunakan alat tersebut.

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],
)

Membuat agen lokal

Setelah menentukan komponen agen, buat instance class A2aAgent yang menggunakan AgentCard, AgentExecutor, dan LlmAgent untuk memulai pengujian lokal.

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()

Template Agen A2A membantu Anda membuat layanan yang sesuai dengan A2A. Layanan ini bertindak sebagai wrapper, yang mengabstraksi lapisan konversi dari Anda.

Menguji agen lokal

Agen nilai tukar mata uang mendukung tiga metode berikut:

  • handle_authenticated_agent_card
  • on_message_send
  • on_get_task

Menguji handle_authenticated_agent_card

Kode berikut mengambil kartu terautentikasi agen, yang menjelaskan kemampuan agen.

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

Menguji on_message_send

Kode berikut mensimulasikan klien yang mengirim pesan baru ke agen. A2aAgent membuat tugas baru dan menampilkan ID tugas.

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)

Menguji on_get_task

Kode berikut mengambil status dan hasil tugas. Output menunjukkan bahwa tugas telah selesai dan menyertakan artefak respons "Halo Dunia".

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)

Langkah berikutnya

Panduan

Pelajari lima cara men-deploy agen di Agent Platform Runtime berdasarkan kebutuhan pengembangan Anda.

Panduan

Menggunakan agen Agent2Agent dengan Agent Platform Runtime.

Panduan

Membuat dan men-deploy agen dasar serta menggunakan layanan evaluasi AI Generatif untuk mengevaluasi agen

Pemecahan masalah

Pelajari cara mengatasi error umum saat membuat agen kustom.

Referensi

Temukan referensi dan dukungan untuk Google Agent Platform.

Referensi

Jelajahi contoh Agent2Agent di Python di GitHub.