src/react_agent/tools.py - master - codein/aiaa_llm_app

src/react_agent/tools.py
149 строк5 KB
Sergey
feat: llm app example
11 май 2026, 23:31
11 май 2026, 23:31512464c
from datetime import datetime, timedelta
import random
from typing import Any

from langchain_core.tools import tool

from react_agent.data_samples import (
    SERVICES,
    ALERT_TEMPLATES,
    DIAGNOSTIC_METRICS,
    RECOMMENDATIONS,
    LOG_TEMPLATES,
)


@tool
def get_service_info(entity: str) -> dict[str, Any]:
    """
    Получить детальную информацию о сервисе.

    Аргументы:
        entity (str): Название сервера/сервиса/БД (например, 'payment-api', 'db-master')

    Возвращает:
        dict: Содержит поля:
            - entity: название сервиса
            - type: тип сервиса (API, Database, Cache и т.д.)
            - status: текущий статус (healthy/degraded/down)
            - depends_on: список сервисов, от которых зависит данный
            - depended_by: список сервисов, которые зависят от данного
            - owner: ответственная команда
    """
    if entity not in SERVICES:
        return {"error": f"Service '{entity}' not found"}

    s = SERVICES[entity]

    return {
        "entity": entity,
        "type": s["type"],
        "status": s["status"],
        "depends_on": s["depends_on"],
        "depended_by": s["depended_by"],
        "owner": s["owner"],
    }


@tool
def get_related_alerts(
    entity: str, started_at: datetime, lookback_minutes: int = 30
) -> dict[str, Any]:
    """
    Найти алерты по сервису за указанный период до времени начала инцидента.

    Аргументы:
        entity (str): Название сервера/сервиса/БД
        started_at (datetime): Время начала инцидента
        lookback_minutes (int): Сколько минут до started_at смотреть (по умолчанию 30)

    Возвращает:
        dict: Содержит поля:
            - entity: название сервиса
            - alerts: список алертов, каждый с полями timestamp, level, message
            - count: общее количество алертов
    """
    if entity not in ALERT_TEMPLATES:
        return {"entity": entity, "alerts": [], "count": 0}

    templates = ALERT_TEMPLATES[entity]
    num_alerts = random.randint(2, min(4, len(templates)))
    selected = random.sample(templates, num_alerts)

    alerts = []
    for level, msg in selected:
        offset_minutes = random.randint(0, lookback_minutes)
        offset_seconds = random.randint(0, 59)
        ts = started_at - timedelta(minutes=offset_minutes, seconds=offset_seconds)
        alerts.append({"timestamp": ts.isoformat(), "level": level, "message": msg})

    alerts.sort(key=lambda x: x["timestamp"])

    return {"entity": entity, "alerts": alerts, "count": len(alerts)}


@tool
def run_diagnostic(entity: str) -> dict[str, Any]:
    """
    Запустить диагностику сервиса и получить метрики здоровья с рекомендациями.

    Аргументы:
        entity (str): Название сервера/сервиса/БД

    Возвращает:
        dict: Содержит поля:
            - entity: название сервиса
            - status: текущий статус (healthy/degraded/down)
            - metrics: словарь с метриками (latency_ms, error_rate, cpu, memory и т.д.)
            - recommendations: список строк с рекомендациями по устранению проблем
    """
    if entity not in SERVICES:
        return {"error": f"Service '{entity}' not found"}

    status = SERVICES[entity]["status"]
    metrics = DIAGNOSTIC_METRICS.get(entity, {})

    return {
        "entity": entity,
        "status": status,
        "metrics": metrics,
        "recommendations": RECOMMENDATIONS.get(status, RECOMMENDATIONS["healthy"]),
    }


@tool
def get_logs(
    entity: str, time_range_start: datetime, time_range_end: datetime
) -> dict[str, Any]:
    """
    Получить логи сервиса за указанный временной диапазон.

    Аргументы:
        entity (str): Название сервера/сервиса/БД
        time_range_start (datetime): Начало временного диапазона
        time_range_end (datetime): Конец временного диапазона

    Возвращает:
        dict: Содержит поля:
            - entity: название сервиса
            - logs: список записей логов, каждая с полями timestamp, level, message
            - count: общее количество записей
    """
    if entity not in LOG_TEMPLATES:
        return {"entity": entity, "logs": [], "count": 0}

    templates = LOG_TEMPLATES[entity]
    num_logs = random.randint(2, min(5, len(templates)))
    selected = random.sample(templates, num_logs)

    logs = []
    total_seconds = max(1, int((time_range_end - time_range_start).total_seconds()))

    for level, msg in selected:
        offset_seconds = random.randint(0, total_seconds)
        ts = time_range_start + timedelta(seconds=offset_seconds)
        logs.append({"timestamp": ts.isoformat(), "level": level, "message": msg})

    logs.sort(key=lambda x: x["timestamp"])

    return {"entity": entity, "logs": logs, "count": len(logs)}