Создание финансовых AI-агентов с python-A2A и протоколом Agent-to-Agent от Google

Введение в python-A2A и коммуникацию агентов

Python A2A — это реализация протокола Agent-to-Agent (A2A) от Google на Python. Этот протокол позволяет AI-агентам общаться друг с другом с помощью единого стандартизированного формата, что устраняет необходимость в кастомных интеграциях между сервисами.

Использование декораторов для определения агентов и навыков

Библиотека python-a2a применяет подход с декораторами для упрощения создания агентов. С помощью декораторов @agent и @skill можно легко задать идентичность агента и его функционал, не заботясь о низкоуровневой логике коммуникации.

Установка python-a2a

Для начала установите библиотеку python-a2a командой:

pip install python-a2a

Создание агента для расчёта EMI

Создадим агента EMI Calculator, который рассчитывает ежемесячный платёж (EMI) по заданному кредиту. Мы используем @agent для метаданных и @skill для реализации расчёта.

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import re
 
@agent(
    name="EMI Calculator Agent",
    description="Вычисляет EMI по основному долгу, процентной ставке и сроку кредита",
    version="1.0.0"
)
class EMIAgent(A2AServer):
 
    @skill(
        name="Calculate EMI",
        description="Вычисляет EMI по основному долгу, годовой процентной ставке и сроку в месяцах",
        tags=["emi", "loan", "interest"]
    )
    def calculate_emi(self, principal: float, annual_rate: float, months: int) -> str:
        monthly_rate = annual_rate / (12 * 100)
        emi = (principal * monthly_rate * ((1 + monthly_rate) ** months)) / (((1 + monthly_rate) ** months) - 1)
        return f"EMI по кредиту в ₹{principal:.0f} под {annual_rate:.2f}% на {months} месяцев составляет ₹{emi:.2f}"
 
    def handle_task(self, task):
        input_text = task.message["content"]["text"]
 
        principal_match = re.search(r"₹?(\d{4,10})", input_text)
        rate_match = re.search(r"(\d+(\.\d+)?)\s*%", input_text)
        months_match = re.search(r"(\d+)\s*(months|month)", input_text, re.IGNORECASE)
 
        try:
            principal = float(principal_match.group(1)) if principal_match else 100000
            rate = float(rate_match.group(1)) if rate_match else 10.0
            months = int(months_match.group(1)) if months_match else 12
 
            emi_text = self.calculate_emi(principal, rate, months)
 
        except Exception as e:
            emi_text = f"Извините, не удалось распознать ввод. Ошибка: {e}"
 
        task.artifacts = [{"parts": [{"type": "text", "text": emi_text}]}]
        task.status = TaskStatus(state=TaskState.COMPLETED)
        return task
 
if __name__ == "__main__":
    agent = EMIAgent()
    run_server(agent, port=4737)

Создание агента для корректировки с учётом инфляции

Этот агент вычисляет будущую стоимость суммы с учётом инфляции.

from python_a2a import A2AServer, skill, agent, run_server, TaskStatus, TaskState
import re
 
@agent(
    name="Inflation Adjusted Amount Agent",
    description="Вычисляет будущую стоимость с учётом инфляции",
    version="1.0.0"
)
class InflationAgent(A2AServer):
 
    @skill(
        name="Inflation Adjustment",
        description="Корректирует сумму с учётом инфляции за период",
        tags=["inflation", "adjustment", "future value"]
    )
    def handle_input(self, text: str) -> str:
        try:
            amount_match = re.search(r"₹?(\d{3,10})", text)
            amount = float(amount_match.group(1)) if amount_match else None
 
            rate_match = re.search(r"(\d+(\.\d+)?)\s*(%|percent)", text, re.IGNORECASE)
            rate = float(rate_match.group(1)) if rate_match else None
 
            years_match = re.search(r"(\d+)\s*(years|year)", text, re.IGNORECASE)
            years = int(years_match.group(1)) if years_match else None
 
            if amount is not None and rate is not None and years is not None:
                adjusted = amount * ((1 + rate / 100) ** years)
                return f"Сумма ₹{amount:.2f} с учётом инфляции {rate:.2f}% за {years} лет составит ₹{adjusted:.2f}"
 
            return (
                "Пожалуйста, укажите сумму, уровень инфляции (например, 6%) и срок (например, 5 лет).\n"
                "Пример: 'Сколько будет ₹10000 через 5 лет при 6% инфляции?'"
            )
        except Exception as e:
            return f"Извините, не удалось вычислить. Ошибка: {e}"
 
    def handle_task(self, task):
        text = task.message["content"]["text"]
        result = self.handle_input(text)
 
        task.artifacts = [{"parts": [{"type": "text", "text": result}]}]
        task.status = TaskStatus(state=TaskState.COMPLETED)
        return task
 
if __name__ == "__main__":
    agent = InflationAgent()
    run_server(agent, port=4747)

Создание сети агентов

Запустите оба агента в отдельных терминалах:

python emi_agent.py
python inflation_agent.py

Агенты доступны по REST API на портах 4737 и 4747.

Добавим их в сеть:

from python_a2a import AgentNetwork, A2AClient, AIAgentRouter
 
network = AgentNetwork(name="Economics Calculator")
network.add("EMI", "http://localhost:4737")
network.add("Inflation", "http://localhost:4747")

Использование роутера для маршрутизации запросов

Создаём роутер, который с помощью LLM клиента направляет запросы к нужному агенту:

router = AIAgentRouter(
    llm_client=A2AClient("http://localhost:5000/openai"),
    agent_network=network
)
 
query = "Рассчитай EMI по кредиту ₹200000 под 5% на 18 месяцев."
agent_name, confidence = router.route_query(query)
print(f"Запрос направлен агенту {agent_name} с уверенностью {confidence:.2f}")
 
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Ответ: {response}")
 
query = "Сколько будет стоить ₹1500000 при 9% инфляции за 10 лет?"
agent_name, confidence = router.route_query(query)
print(f"Запрос направлен агенту {agent_name} с уверенностью {confidence:.2f}")
 
agent = network.get_agent(agent_name)
response = agent.ask(query)
print(f"Ответ: {response}")

Данная система демонстрирует, как можно создавать и связывать финансовые агенты для обработки запросов с помощью python-a2a и протокола A2A от Google.