Kommunikation mit Multi-Agent mit dem A2A Python SDK

Wenn Sie unter einem Felsen und mit AI zusammenarbeiten, haben Sie wahrscheinlich von Agent2Agent (A2A) -Protokoll gehört, „ein offener Commonplace, der die Kommunikation und Zusammenarbeit zwischen der Zusammenarbeit ermöglicht KI -Agenten”. Es ist immer noch ziemlich neu, aber es wird bereits viel Aufsehen. Da es so intestine mit MCP spielt (was so aussieht, als würde es der Commonplace der Branche werden), wird A2A zu sein Die Commonplace für die Kommunikation mit mehreren Agenten in der Branche.

Als Google das zum ersten Mal fallen ließ ProtokollspezifikationMeine erste Reaktion battle im Grunde: „Okay, cool … aber was soll ich damit machen?“ Zum Glück haben sie diese Woche den Beamten veröffentlicht Python SDK für das Protokoll, additionally spricht es jetzt endlich eine Sprache, die ich verstehe.

In diesem Artikel werden wir uns mit der tatsächlich die Kommunikation zwischen Agenten und Kunden einsetzt. Spoiler: Es ist alles auf aufgabenorientierte Weise. Um die Dinge weniger abstrakt zu machen, bauen wir gemeinsam ein kleines Spielzeugbeispiel auf.

Kommunikation zwischen dem Occasion -Detektorvertreter und einem A2A -Kunden

In unseren Systemen haben wir eine Occasion Detektor AI Agent (verantwortlich für die Erkennung von Ereignissen) und eine Alert AI Agent (Verantwortlich für die Alarmierung des Benutzers der Ereignisse). Da ich mich hier auf das A2A -Protokoll konzentriere, werden beide Mittel als einfache Python -Methoden verspottet, die Saiten zurückgeben. Aber im wirklichen Leben können Sie Ihre Agenten mit jedem Rahmen aufbauen, den Sie mögen (Langgraph, Google ADK, Crewai usw.).

Wir haben drei Zeichen in unserem System, die BenutzerDie Occasion Agent und die Alarmagent. Sie alle kommunizieren mit Verwendung Messages. A Message repräsentiert eine einzige Kommunikationsrunde im A2A -Protokoll. Wir wickeln die Agenten auf A2A -Server. Die Server enthüllen einen HTTP -Endpunkt, der das Protokoll implementiert. Jeder A2A -Server hat Occasion queues Das wirkt als Puffer zwischen der asynchronen Ausführung des Agenten und der Antwort des Servers.

Der A2A -Kunde initiiert die Kommunikation, und wenn zwei Agenten eine kommunizieren müssen A2A -Server kann auch die Rolle von a spielen A2A -Kunde. Das folgende Diagramm zeigt, wie ein Consumer und ein Server innerhalb des Protokolls kommunizieren.

Der EventQueue Geschäfte MessagesAnwesend DutiesAnwesend TaskStatusUpdateEventAnwesend TaskArtifactUpdateEventAnwesend A2AError Und JSONRPCError Objekte. Der Activity Könnte das wichtigste Objekt sein, um zu verstehen, wie man Multi-Agent-Systeme mit A2A erstellt. Nach dem A2A -Dokumentation:

• Wenn ein Kunde eine Nachricht an einen Agenten sendet, kann der Agent feststellen, dass die Erfüllung der Anfrage eine staatliche Aufgabe erfordern muss (z. B. „einen Bericht generieren“, „einen Flug buchen“, „eine Frage beantworten“).
• Jede Aufgabe hat eine eindeutige ID, die vom Agenten definiert wird und durch einen definierten Lebenszyklus (z. submittedAnwesend workingAnwesend input-requiredAnwesend accomplishedAnwesend failed).
• Aufgaben sind staatlich und können mehrere Börsen (Nachrichten) zwischen dem Consumer und dem Server beinhalten.

Denken Sie an a Activity als etwas in Ihrem Multi-Agent-System, das a hat klar Und einzigartig Ziel. Wir haben zwei Aufgaben in unserem System:

1. Ein Ereignis erkennen
2. Alarmieren Sie den Benutzer

Jeder Agent macht sein eigenes Ding (Aufgabe). Erstellen wir den A2A -Server für den Occasion Agent, damit die Dinge greifbarer werden.

Erstellen des A2A -Servers für den Occasion Agent

Erstens: die Agentenkarte. Die Agentenkarte ist JSON -Dokument, mit der andere verfügbare Agenten bekannt sind. :

• Identität des Servers
• Fähigkeiten
• Fähigkeiten
• Serviceendpunkt
• URL
• Wie Purchasers authentifizieren und mit dem Agenten interagieren sollten

Lassen Sie uns zunächst die Agentenkarte für den Occasion -Detektor -AI -Agenten definieren (ich habe die Fähigkeiten basierend auf definiert Dieses Beispiel von Google):

agent_card = AgentCard(  
    title='Occasion Detection Agent',  
    description='Detects related occasions and alerts the consumer',  
    url='http://localhost:10008/',  
    model='1.0.0',  
    defaultInputModes=('textual content'),  
    defaultOutputModes=('textual content'),  
    capabilities=AgentCapabilities(streaming=False),  
    authentication={ "schemes": ("primary") },  
    expertise=(  
        AgentSkill(  
            id='detect_events',  
            title='Detect Occasions',  
            description='Detects occasions and alert the consumer',  
            tags=('occasion'),  
        ),  
    ),  
)

Weitere Informationen zur Struktur der Agent -Kartenobjekte finden Sie hier: https://google.github.io/a2a/specification/#55-agentcard-object-structure

Der Agent selbst wird tatsächlich ein Uvicorn -Server sein, additionally erstellen wir die predominant() Methode, um es zum Laufen zu bringen. Die gesamte Anfrage wird von der behandelt DefaultRequestHandler des A2A-Python SDK. Der Handler braucht a TaskStore die Aufgaben aufbewahren und eine AgentExecutor Dies hat die Implementierung der Kernlogik des Agenten (wir werden die erstellen EventAgentExecutor in einer Minute).

Die letzte Komponente der predominant() Methode ist die A2AStarletteApplicationdas ist das Starlette Anwendung, die die Endpunkte des A2A -Protokollservers implementiert. Wir müssen die zur Verfügung stellen Agent Card und die DefaultRequestHandler um es zu initialisieren. Der letzte Schritt besteht nun darin, die App mit Uvicorn auszuführen. Hier ist der vollständige Code der predominant() Verfahren:

import click on  
import uvicorn  
from a2a.sorts import (  
    AgentCard, AgentCapabilities, AgentSkill
) 
from a2a.server.request_handlers import DefaultRequestHandler  
from a2a.server.duties import InMemoryTaskStore  
from a2a.server.apps import A2AStarletteApplication 
 
@click on.command()  
@click on.choice('--host', default='localhost')  
@click on.choice('--port', default=10008)  
def predominant(host: str, port: int):  
    agent_executor = EventAgentExecutor()

    agent_card = AgentCard(  
        title='Occasion Detection Agent',  
        description='Detects related occasions and alerts the consumer',  
        url='http://localhost:10008/',  
        model='1.0.0',  
        defaultInputModes=('textual content'),  
        defaultOutputModes=('textual content'),  
        capabilities=AgentCapabilities(streaming=False),  
        authentication={ "schemes": ("primary") },  
        expertise=(              AgentSkill(                  id='detect_events',                  title='Detect Occasions',                  description='Detects occasions and alert the consumer',                  tags=('occasion'),  
            ),  
        ),  
    )
      
    request_handler = DefaultRequestHandler(  
        agent_executor=agent_executor,  
        task_store=InMemoryTaskStore()  
    ) 
 
    a2a_app = A2AStarletteApplication(  
        agent_card=agent_card,  
        http_handler=request_handler  
    )  

    uvicorn.run(a2a_app.construct(), host=host, port=port)

Erstellen des Eventagentexecutors

Jetzt ist es Zeit, den Kern unseres Agenten aufzubauen und schließlich zu sehen, wie die Aufgaben verwendet werden, damit die Agenten miteinander interagieren. Der EventAgentExecutor Klasse erbt AgentExecutor Schnittstelle und somit müssen wir die implementieren execute() und die cancel() Methoden. Beide nehmen a RequestContext und an EventQueue Objekt als Parameter. Der RequestContext Hält Informationen über die aktuelle Anfrage, die vom Server und die verarbeitet wird EventQueue fungiert als Puffer zwischen der asynchronen Ausführung des Agenten und der Antwort des Servers.

Unser Agent wird nur prüfen, ob die Zeichenfolge “occasion”Ist in der Nachricht, dass der Benutzer gesendet hat (KUSS ✨). Wenn das „occasion”Gibt es da, dann sollten wir den Alert -Agenten anrufen. Wir werden das tun, indem wir a senden Message zu diesem anderen Alarmagenten. Dies ist das Direkte Konfiguration Strategie, was bedeutet, dass wir den Agenten mit einer URL konfigurieren werden, um die Agentenkarte des Alert -Agenten abzurufen. Dazu wird unser Occasion -Agent wie ein A2A -Kunde verhalten.

Erstellen wir den Testamentsvollstrecker Schritt für Schritt. Lassen Sie uns zunächst die Hauptaufgabe erstellen (die Aufgabe zum Erkennen der Ereignisse). Wir müssen a instanziieren TaskUpdater Objekt (eine Helferklasse für Agenten, um Updates für die Occasion -Warteschlange einer Aufgabe zu veröffentlichen), dann die Aufgabe einreichen und ankündigen, dass wir mit dem arbeiten start_work() Verfahren:

from a2a.server.agent_execution import AgentExecutor

class EventAgentExecutor(AgentExecutor):  
    async def execute(self, context: RequestContext, event_queue: EventQueue):  
        task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)  
        task_updater.submit()  
        task_updater.start_work()

Die Nachricht, die der Benutzer an den Agenten sendet, sieht folgendermaßen aus:

send_message_payload = {  
        'message': {  
            'position': 'consumer',  
            'components': ({'sort': 'textual content', 'textual content': f'it has an occasion!'}),  
            'messageId': uuid4().hex,  
        }  
    }

A Half repräsentiert ein eigenes Inhaltsstück innerhalb eines Messageum exportierbare Inhalte als beide darzustellen TextPartAnwesend FilePartoder DataPart. Wir werden a verwenden TextPart Additionally müssen wir es im Testamentsvollstrecker auspacken:

from a2a.server.agent_execution import AgentExecutor

class EventAgentExecutor(AgentExecutor):  
    async def execute(self, context: RequestContext, event_queue: EventQueue):  
        task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)  
        task_updater.submit()  
        task_updater.start_work()

        await asyncio.sleep(1) #let's fake we're really doing one thing

        user_message = context.message.components(0).root.textual content # unwraping the TextPart

Zeit, um die tremendous erweiterte Logik unseres Agenten zu erstellen. Wenn die Nachricht den String nicht hat “occasion”Wir müssen den Alert -Agenten nicht anrufen und die Aufgabe ist erledigt:

from a2a.server.agent_execution import AgentExecutor

class EventAgentExecutor(AgentExecutor):  
    async def execute(self, context: RequestContext, event_queue: EventQueue):  
        task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)  
        task_updater.submit()  
        task_updater.start_work()

        await asyncio.sleep(1) #let's fake we're really doing one thing

        user_message = context.message.components(0).root.textual content # unwraping the TextPart

        if "occasion" not in user_message:  
            task_updater.update_status(  
                TaskState.accomplished,  
                message=task_updater.new_agent_message(components=(TextPart(textual content=f"No occasion detected"))),
            )

Erstellen eines A2A -Purchasers für den Benutzer

Lassen Sie uns einen A2A -Consumer erstellen, damit wir den Agenten so testen können, wie er ist. Der Consumer verwendet die get_client_from_agent_card_url() Methode von A2AClient Klasse, um (zu erraten, was) die Agentenkarte erhalten. Dann wickeln wir die Nachricht in a SendMessageRequest Objekt und senden Sie es mit dem Agenten an den Agenten send_message() Methode des Kunden. Hier ist der vollständige Code:

import httpx  
import asyncio  
from a2a.shopper import A2AClient  
from a2a.sorts import SendMessageRequest, MessageSendParams  
from uuid import uuid4  
from pprint import pprint
  
async def predominant():    
    send_message_payload = {  
        'message': {  
            'position': 'consumer',  
            'components': ({'sort': 'textual content', 'textual content': f'nothing occurring right here'}),  
            'messageId': uuid4().hex,  
        }  
    }  

    async with httpx.AsyncClient() as httpx_client:  
        shopper = await A2AClient.get_client_from_agent_card_url(  
            httpx_client, 'http://localhost:10008'  
        )  
        request = SendMessageRequest(  
            params=MessageSendParams(**send_message_payload)  
        )  
        response = await shopper.send_message(request)  
        pprint(response.model_dump(mode='json', exclude_none=True))  
  
if __name__ == "__main__":  
    asyncio.run(predominant())

Dies geschieht im Terminal, in dem der EventAgent -Server ausgeführt wird:

Und dies ist die Nachricht, die der Kunde sieht:

Die Aufgabe, die Veranstaltung zu erkennen, wurde erstellt und es wurde kein Ereignis festgestellt, schön! Der springende Punkt von A2A besteht jedoch darin, Agenten miteinander zu kommunizieren. Lassen Sie uns additionally den Occasion -Agenten mit dem Alarmagenten sprechen lassen.

Den Occasion Agent mit dem Alert Agent sprechen lassen

Um den Occasion Agent mit dem Alert -Agenten zu sprechen, wird der Occasion Agent auch als Kunde fungieren:

from a2a.server.agent_execution import AgentExecutor

ALERT_AGENT_URL = "http://localhost:10009/" 

class EventAgentExecutor(AgentExecutor):  
    async def execute(self, context: RequestContext, event_queue: EventQueue):  
        task_updater = TaskUpdater(event_queue, context.task_id, context.context_id)  
        task_updater.submit()  
        task_updater.start_work()

        await asyncio.sleep(1) #let's fake we're really doing one thing

        user_message = context.message.components(0).root.textual content # unwraping the TextPart

        if "occasion" not in user_message:  
            task_updater.update_status(  
                TaskState.accomplished,  
                message=task_updater.new_agent_message(components=(TextPart(textual content=f"No occasion detected"))),
            )
        else:
            alert_message = task_updater.new_agent_message(components=(TextPart(textual content="Occasion detected!")))

            send_alert_payload = SendMessageRequest(  
                params=MessageSendParams(  
                    message=alert_message  
                )  
            )  

            async with httpx.AsyncClient() as shopper:  
                alert_agent = A2AClient(httpx_client=shopper, url=ALERT_AGENT_URL)  
                response = await alert_agent.send_message(send_alert_payload)  

                if hasattr(response.root, "end result"):  
                    alert_task = response.root.end result  
                    # Polling till the duty is completed
                    whereas alert_task.standing.state not in (  
                        TaskState.accomplished, TaskState.failed, TaskState.canceled, TaskState.rejected  
                    ):  
                        await asyncio.sleep(0.5)  
                        get_resp = await alert_agent.get_task(  
                            GetTaskRequest(params=TaskQueryParams(id=alert_task.id))  
                        )  
                        if isinstance(get_resp.root, GetTaskSuccessResponse):  
                            alert_task = get_resp.root.end result  
                        else:  
                            break  
  
                    # Full the unique job  
                    if alert_task.standing.state == TaskState.accomplished:  
                        task_updater.update_status(  
                            TaskState.accomplished,  
                            message=task_updater.new_agent_message(components=(TextPart(textual content="Occasion detected and alert despatched!"))),  
                        )  
                    else:  
                        task_updater.update_status(  
                            TaskState.failed,  
                            message=task_updater.new_agent_message(components=(TextPart(textual content=f"Did not ship alert: {alert_task.standing.state}"))),  
                        )  
                else:  
                    task_updater.update_status(  
                        TaskState.failed,  
                        message=task_updater.new_agent_message(components=(TextPart(textual content=f"Did not create alert job"))),  
                    )

Wir rufen den Alert -Agenten so an, wie wir den Occasion Agent als Benutzer bezeichnet haben. Wenn die Aufgabe von Alert Agent erledigt ist, erledigen wir die ursprüngliche Occasion -Agent -Aufgabe. Rufen wir den Occasion Agent erneut an, diesmal jedoch mit einem Ereignis:

Die Schönheit hier ist, dass wir einfach den Alarmagenten bezeichnet haben und nichts darüber wissen müssen, wie es den Benutzer alarmiert. Wir senden einfach eine Nachricht an sie und warten darauf, dass es fertig ist.

Der Alert -Agent ist dem Occasion Agent sehr ähnlich. Sie können den gesamten Code hier überprüfen: https://github.com/dmesquita/multi-agent-communication-a2a-python

Letzte Gedanken

Verstehen, wie man Multi-Agent-Systeme mit A2A aufbaut Senden Sie Nachrichten, um die Agenten ihr Ding zu machen. Alles, was Sie tun müssen, um Ihre Agenten in A2A zu integrieren AgentExecutor und führen Sie den Agenten als Server aus.

Ich hoffe, dieser Artikel hat Ihnen auf Ihrer A2A -Reise geholfen, danke fürs Lesen!

Referenzen

(1) Padgham, Lin und Michael Winikoff. Entwicklung intelligenter Agentensysteme: Ein praktischer Leitfaden. John Wiley & Sons, 2005.

(2) https://github.com/google/a2a-python

(3) https://github.com/google/a2a-python/tree/predominant/examples/google_adk

(4) https://builders.googleblog.com/en/agents-adk-agent-engine-a2-enhancements-google-io/