A2A suderinamo atsitiktinio skaičiaus agento kūrimas: žingsnis po žingsnio vadovas, kaip įgyvendinti žemo lygio vykdytojų modelį su „Python“

„Agent-Agent“ (A2A) protokolas yra naujas „Google“ standartas, leidžiantis AI agentams, nepriklausomai nuo jų pagrindinės sistemos ar kūrėjo, kad galėtų sklandžiai bendrauti ir bendradarbiauti. Tai veikia naudojant standartizuotus pranešimus, agento korteles (kurios apibūdina tai, ką agentas gali padaryti) ir vykdymą pagal užduotis, leidžiant agentams sąveikauti per HTTP be pasirinktinės integracijos logikos. A2A leidžia lengviau sukurti keičiamąsias, sąveikaujančias daugialypės terpės sistemas, išsiaiškinant komunikacijos sudėtingumą.

Šiame vadove įdiegsime paprastą demonstracinį agentą, kuris grąžina atsitiktinį numerį, padėdami suprasti pagrindinę A2A protokolo struktūrą ir srautą naudodamiesi praktiniu kodu.

Priklausomybių nustatymas

Pirmiausia nustatysime savo aplinką ir pradėsime įdiegę UV paketo tvarkyklę. „Mac“ ar „Linux“:

curl -LsSf https://astral.sh/uv/install.sh | sh 

„Windows“ („PowerShell“):

powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"

Tada sukursime naują projekto katalogą ir jį inicijuosime naudodami UV

uv init a2a-demo
cd a2a-demo

Dabar galime sukurti ir suaktyvinti virtualią aplinką. „Mac“ ar „Linux“:

uv venv
source .venv/bin/activate

„Windows“:

uv venv
.venv\Scripts\activate

Dabar įdiegsime reikiamas priklausomybes

uv add a2a-sdk python-a2a uvicorn

Pagrindinių elementų įgyvendinimas

Agento vykdytojas (agentas_executor.py)

Šiame žingsnyje mes įgyvendiname pagrindinę savo agento logiką sukurdami agento vykdytoją, kuris yra atsakingas už gaunamų užklausų tvarkymą ir atsakymų grąžinimą A2A formatu. „RandomNumberAgETexecutor“ įvynioja paprastą „RandomNumberAgent“ Tai sukuria atsitiktinį skaičių nuo 1 iki 100. Kai užklausa atsiranda, vykdymo metodas paskambina agento logikai ir perkelia rezultatą į įvykio eilę kaip standartizuotą A2A pranešimą. Ši sąranka sudaro pagrindinę logiką, su kuria A2A klientai gali bendrauti. Peržiūrėkite Pilni kodai „GitHub“

import random
from a2a.server.agent_execution import AgentExecutor
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.utils import new_agent_text_message
from pydantic import BaseModel


class RandomNumberAgent(BaseModel):
    """Generates a random number between 1 and 100"""

    async def invoke(self) -> str:
        number = random.randint(1, 100)
        return f"Random number generated: {number}"


class RandomNumberAgentExecutor(AgentExecutor):

    def __init__(self):
        self.agent = RandomNumberAgent()

    async def execute(self, context: RequestContext, event_queue: EventQueue):
        result = await self.agent.invoke()
        await event_queue.enqueue_event(new_agent_text_message(result))

    async def cancel(self, context: RequestContext, event_queue: EventQueue):
        raise Exception("Cancel not supported")

A2A serverio ir agento kortelės nustatymas (main.py)

Šiame skyriuje mes apibrėžiame metaduomenis, apibūdinančius, ką gali padaryti mūsų agentas – tai vadinama Agento kortelė. Pagalvokite apie tai kaip agento vizitinę kortelę, kurioje yra tokia informacija kaip jos pavadinimas, aprašymas, turimi įgūdžiai, įvesties/išvesties tipai ir versija.

Mes taip pat užregistruojame agento įgūdžius, kurie apibūdina užduotis, kurias ji gali atlikti. Mūsų atveju tai apima įgūdį sugeneruoti atsitiktinį skaičių, tinkamai pažymėti ir su pavyzdžiais.

Kai metaduomenys bus paruošti, mes sukonfigūruojame A2A serverį naudodami A2astarletteapplication. Mes teikiame agento kortelę ir sujungiame ją su savo pasirinktinio agento logika naudodami a Defaultrequesthandlerkuris naudoja „RandomNumberAgETexecutor“ Mes įgyvendinome anksčiau. Galiausiai mes paleidžiame serverį naudodami „UVicorn“, kad agentas galėtų pradėti klausytis gaunamų A2A pranešimų prievade 9999.

Ši sąranka leidžia mūsų agentui gauti standartizuotus A2A pranešimus, juos apdoroti ir reaguoti struktūriškai – laikantis A2A protokolo. Peržiūrėkite Pilni kodai „GitHub“

import uvicorn
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from agent_executor import RandomNumberAgentExecutor


def main():
    # Define the skill metadata
    skill = AgentSkill(
        id="random_number",
        name="Random Number Generator",
        description="Generates a random number between 1 and 100",
        tags=("random", "number", "utility"),
        examples=("Give me a random number", "Roll a number", "Random"),
    )

    # Define the agent metadata
    agent_card = AgentCard(
        name="Random Number Agent",
        description="An agent that returns a random number between 1 and 100",
        url="http://localhost:9999/",
        defaultInputModes=("text"),
        defaultOutputModes=("text"),
        skills=(skill),
        version="1.0.0",
        capabilities=AgentCapabilities(),
    )

    # Configure the request handler with our custom agent executor
    request_handler = DefaultRequestHandler(
        agent_executor=RandomNumberAgentExecutor(),
        task_store=InMemoryTaskStore(),
    )

    # Create the A2A app server
    server = A2AStarletteApplication(
        http_handler=request_handler,
        agent_card=agent_card,
    )

    # Run the server
    uvicorn.run(server.build(), host="0.0.0.0", port=9999)


if __name__ == "__main__":
    main()

Bendravimas su agentu, naudodamas „A2Aclient“ (client.py)

Tada mes sukuriame klientą, kuris sąveikaus su mūsų A2A agentu. Šis kliento scenarijus atlieka tris pagrindines užduotis:

• Gaukite agento kortelę: Mes pradedame išspręsti agento viešus metaduomenis naudodamiesi „A2AcardResolver“. Tai iškelia „Agent.json“ failą iš .well žinomo galinio taško, kuriame yra esminių detalių, tokių kaip agento vardas, aprašymas, įgūdžiai ir komunikacijos galimybės.

• Inicijuokite A2A klientą: Naudodamiesi „Agent Card“ kortele, mes nustatėme „A2Aclient“, kuris tvarko ryšio protokolą. Šis klientas bus atsakingas už struktūrizuotų pranešimų siuntimą agentui ir atsakymų gavimą.

Siųskite pranešimą ir gaukite atsakymą: Mes sukonstruojame pranešimą su tekstu „Duok man atsitiktinį numerį“, naudodami A2A pranešimo struktūrą (pranešimas, dalis, „Textpart“). Pranešimas siunčiamas kaip „SendMessageQuest“ dalis, kuri įvynioja jį su unikaliu užklausos ID. Kai pranešimas išsiunčiamas, agentas jį apdoroja ir atsako sugeneruotu atsitiktiniu numeriu, kuris vėliau atspausdinamas JSON formatu. Peržiūrėkite Pilni kodai „GitHub“

import uuid
import httpx
from a2a.client import A2ACardResolver, A2AClient
from a2a.types import (
    AgentCard,
    Message,
    MessageSendParams,
    Part,
    Role,
    SendMessageRequest,
    TextPart,
)

PUBLIC_AGENT_CARD_PATH = "/.well-known/agent.json"
BASE_URL = "http://localhost:9999"


async def main() -> None:
    async with httpx.AsyncClient() as httpx_client:
        # Fetch the agent card
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
        try:
            print(f"Fetching public agent card from: {BASE_URL}{PUBLIC_AGENT_CARD_PATH}")
            agent_card: AgentCard = await resolver.get_agent_card()
            print("Agent card fetched successfully:")
            print(agent_card.model_dump_json(indent=2))
        except Exception as e:
            print(f"Error fetching public agent card: {e}")
            return

        # Initialize A2A client with the agent card
        client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)

        # Build message
        message_payload = Message(
            role=Role.user,
            messageId=str(uuid.uuid4()),
            parts=(Part(root=TextPart(text="Give me a random number"))),
        )
        request = SendMessageRequest(
            id=str(uuid.uuid4()),
            params=MessageSendParams(message=message_payload),
        )

        # Send message
        print("Sending message...")
        response = await client.send_message(request)

        # Print response
        print("Response:")
        print(response.model_dump_json(indent=2))


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

Agento vykdymas ir užklausa tą patį

Norėdami išbandyti savo A2A sąranką, pradėsime paleisti agento serverį. Tai atliekama vykdant pagrindinį.py failą, kuris inicijuoja agentą, atskleidžia savo agento kortelę ir pradeda klausytis gaunamų užklausų 9999. Peržiūrėkite Pilni kodai „GitHub“

Kai agentas veikia ir veikia, mes pereisime prie kliento scenarijaus. Klientas pateiks agento metaduomenis, atsiųs struktūrizuotą užklausą naudodamas A2A protokolą ir gaus atsakymą. Mūsų atveju užklausa yra paprastas pranešimas, pavyzdžiui, „Duok man atsitiktinį numerį“, o agentas grąžins skaičių nuo 1 iki 100.

Peržiūrėkite Pilni kodai „GitHub“. Visas šio tyrimo kreditas skirtas šio projekto tyrėjams. Taip pat nedvejodami sekite mus „Twitter“ Ir nepamirškite prisijungti prie mūsų 100K+ ml subreddit ir užsiprenumeruokite Mūsų informacinis biuletenis.

Aš esu civilinės inžinerijos absolventas (2022 m.) Iš Jamia Millia Islamia, Naujajame Delyje, ir labai domisiu duomenų mokslu, ypač neuroniniu tinklais ir jų taikymu įvairiose srityse.