‫Google משתמשת בטכנולוגיית AI כדי לתרגם תוכן לשפה המועדפת עליך. בתרגומים כאלו עשויות להיות שגיאות.

יצירת סוכן Agent2Agent

‫Agent Runtime מאפשר לכם לפתח ולפרוס סוכנים באמצעות פרוטוקול Agent2Agent ‏ (A2A). ‫A2A הוא תקן פתוח שנועד לאפשר תקשורת ושיתוף פעולה חלקים בין סוכני AI.

במאמר הזה מוסבר איך לפתח ולבדוק סוכן A2A באופן מקומי, כולל הגדרת רכיבים כמו AgentCard ו-AgentExecutor.

מידע נוסף על ניהול הסוכנים שהופעלו זמין במאמר ניהול סוכנים שהופעלו.

תהליך העבודה המרכזי כולל את השלבים הבאים:

הגדרת רכיבים מרכזיים
יצירת סוכן מקומי
בדיקת הסוכן המקומי

הגדרת רכיבי הסוכן

כדי ליצור סוכן A2A, צריך להגדיר את הרכיבים הבאים: AgentCard, AgentExecutor ו-ADK LlmAgent.

‫AgentCard מכיל מסמך מטא-נתונים שמתאר את היכולות של הסוכן. AgentCard הוא כמו כרטיס ביקור שסוכנים אחרים יכולים להשתמש בו כדי לגלות מה הסוכן שלכם יכול לעשות. פרטים נוספים זמינים במפרט של כרטיס הסוכן.
‫AgentExecutor מכיל את הלוגיקה הבסיסית של הסוכן ומגדיר איך הוא מטפל במשימות. כאן מגדירים את ההתנהגות של הסוכן. מידע נוסף בנושא זמין במפרט של פרוטוקול A2A.
אופציונלי: LlmAgent מגדיר את סוכן ה-ADK, כולל הוראות המערכת, המודל הגנרטיבי והכלים שלו.

הגדרה של `AgentCard`

בדוגמת הקוד הבאה מוגדר AgentCard לסוכן של שער חליפין:

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`

בדוגמה הבאה של קוד מוגדרת פונקציה AgentExecutor שמחזירה את שער החליפין של המטבע. היא מקבלת מופע CurrencyAgent ומאתחלת את ADK Runner כדי לבצע בקשות.

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`

קודם כול, מגדירים כלי להמרת מטבעות שבו 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)}

לאחר מכן, מגדירים ADK‏ LlmAgent שמשתמש בכלי.

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

יצירת סוכן מקומי

אחרי שמגדירים את רכיבי הנציג, יוצרים מופע של המחלקה A2aAgent שמשתמשת ב-AgentCard, ב-AgentExecutor וב-LlmAgent כדי להתחיל בדיקה מקומית.

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

תבנית הסוכן A2A עוזרת לכם ליצור שירות שתואם ל-A2A. השירות פועל כעטיפה, ומסתיר מכם את שכבת ההמרה.

בדיקת הסוכן המקומי

סוכן שער החליפין תומך בשלוש השיטות הבאות:

handle_authenticated_agent_card
on_message_send
on_get_task

בדיקה של `handle_authenticated_agent_card`

הקוד הבא מאחזר את הכרטיס המאומת של הסוכן, שמתאר את היכולות שלו.

# 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`

הקוד הבא מדמה לקוח ששולח הודעה חדשה לסוכן. הפונקציה A2aAgent יוצרת משימה חדשה ומחזירה את המזהה שלה.

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`

הקוד הבא מאחזר את הסטטוס ואת התוצאה של משימה. הפלט מראה שהמשימה הושלמה וכולל את ארטיפקט התגובה 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)

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

מדריך

יצירת סוכן Agent2Agent

הגדרת רכיבי הסוכן

הגדרה של `AgentCard`

הגדרה של `AgentExecutor`

הגדרה של `LlmAgent`

יצירת סוכן מקומי

בדיקת הסוכן המקומי

בדיקה של `handle_authenticated_agent_card`

בדיקה של `on_message_send`

בדיקה של `on_get_task`

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

פריסת סוכנים

שימוש בסוכן Agent2Agent

הערכת הסוכנים

פתרון בעיות ביצירת סוכנים

קבלת תמיכה

מאגר דוגמאות של Agent2Agent

יצירת סוכן Agent2Agent קל לארגן דפים בעזרת אוספים אפשר לשמור ולסווג תוכן על סמך ההעדפות שלך.

הגדרת רכיבי הסוכן

הגדרה של AgentCard

הגדרה של AgentExecutor

הגדרה של LlmAgent

יצירת סוכן מקומי

בדיקת הסוכן המקומי

בדיקה של handle_authenticated_agent_card

בדיקה של on_message_send

בדיקה של on_get_task

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

פריסת סוכנים

שימוש בסוכן Agent2Agent

הערכת הסוכנים

פתרון בעיות ביצירת סוכנים

קבלת תמיכה

מאגר דוגמאות של Agent2Agent

יצירת סוכן Agent2Agent

הגדרה של `AgentCard`

הגדרה של `AgentExecutor`

הגדרה של `LlmAgent`

בדיקה של `handle_authenticated_agent_card`

בדיקה של `on_message_send`

בדיקה של `on_get_task`