Построение агента случайных чисел, соответствующего стандарту A2A: пошаговое руководство по реализации шаблона низкоуровневого исполнителя на Python
В современном мире, где автоматизация и искусственный интеллект становятся неотъемлемой частью бизнеса, создание эффективных многоагентных систем имеет особое значение. В этой статье мы рассмотрим, как разработать агента случайных чисел, который будет соответствовать стандарту A2A, используя язык программирования Python. Мы подробно разберем каждый шаг процесса и оценим его практическое применение, а также затраты, связанные с реализацией.
Что такое A2A и зачем он нужен?
A2A (Agent-to-Agent) — это новый стандарт, предложенный Google, который позволяет агентам ИИ, независимо от их платформы, взаимодействовать друг с другом. Это упрощает интеграцию и взаимодействие между системами, что критически важно для построения масштабируемых и совместимых многоагентных систем. A2A использует стандартизированные сообщения и карточки агентов, что делает обмен данными более простым и понятным.
Подготовка окружения
Первым шагом в нашем проекте будет установка необходимых инструментов и создание среды разработки. Для этого выполните следующие команды:
- Для Mac или Linux:
curl -LsSf https://astral.sh/uv/install.sh | sh - Для Windows:
powershell -ExecutionPolicy ByPass -c "irm https://astral.sh/uv/install.ps1 | iex"
После установки создайте новый проект и инициализируйте его с помощью:
uv init a2a-demo
Далее создайте виртуальную среду и установите необходимые зависимости:
uv add a2a-sdk python-a2a uvicorn
Реализация основных компонентов
Создание исполнителя агента
В этом разделе мы создадим основной исполнитель нашего агента, который будет обрабатывать входящие запросы. Мы реализуем класс RandomNumberAgentExecutor, который генерирует случайное число от 1 до 100:
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):
async def invoke(self) -> str:
number = random.randint(1, 100)
return f"Сгенерированное случайное число: {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("Отмена не поддерживается")Настройка A2A сервера и карточки агента
Теперь мы определим метаданные, которые описывают, что может делать наш агент. Эти данные будут представлены в виде карточки агента. Затем настроим A2A сервер:
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():
skill = AgentSkill(
id="random_number",
name="Генератор случайных чисел",
description="Генерирует случайное число от 1 до 100",
tags=["случайный", "число", "утилита"],
examples=["Дай мне случайное число", "Брось число", "Случайное"],
)
agent_card = AgentCard(
name="Агент случайных чисел",
description="Агент, который возвращает случайное число от 1 до 100",
url="http://localhost:9999/",
defaultInputModes=["text"],
defaultOutputModes=["text"],
skills=[skill],
version="1.0.0",
capabilities=AgentCapabilities(),
)
request_handler = DefaultRequestHandler(
agent_executor=RandomNumberAgentExecutor(),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=agent_card,
)
uvicorn.run(server.build(), host="0.0.0.0", port=9999)
if __name__ == "__main__":
main()Взаимодействие с агентом через A2AClient
Теперь мы создадим клиент, который будет взаимодействовать с нашим A2A агентом. Клиент будет выполнять три основные задачи:
- Получить карточку агента.
- Инициализировать A2A клиент.
- Отправить сообщение и получить ответ.
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:
resolver = A2ACardResolver(httpx_client=httpx_client, base_url=BASE_URL)
agent_card: AgentCard = await resolver.get_agent_card()
client = A2AClient(httpx_client=httpx_client, agent_card=agent_card)
message_payload = Message(
role=Role.user,
messageId=str(uuid.uuid4()),
parts=[Part(root=TextPart(text="Дай мне случайное число"))],
)
request = SendMessageRequest(
id=str(uuid.uuid4()),
params=MessageSendParams(message=message_payload),
)
response = await client.send_message(request)
print("Ответ:")
print(response.model_dump_json(indent=2))
if __name__ == "__main__":
import asyncio
asyncio.run(main())Запуск агента и тестирование
Чтобы протестировать нашу настройку A2A, сначала запустите сервер агента:
uv run main.py
Затем выполните клиентский скрипт:
uv run client.py
Таким образом, агент обработает запрос и вернет случайное число от 1 до 100. Это простой, но эффективный пример того, как можно реализовать взаимодействие между агентами с использованием A2A.
Заключение
Создание агента, соответствующего стандарту A2A, открывает новые горизонты для автоматизации и интеграции ИИ в бизнес-процессы. Мы рассмотрели ключевые этапы разработки, начиная с настройки окружения и заканчивая реализацией взаимодействия с клиентом. Используя A2A, вы можете значительно упростить интеграцию различных ИИ-решений, что в конечном итоге приведет к снижению затрат и увеличению эффективности.
