core/engine/src/api/agents.py - main - alexefan136/flowstack

core/engine/src/api/agents.py
451 строка16 KB
Alexander Efanov
upd fix
31 июл 2026, 19:17
31 июл 2026, 19:17d146d86
"""API endpoints для управления AI агентами.

Использует AgentService + AgentRuntime (LLM loop + function calling)
вместо legacy-фабрики классов агентов.

Эндпоинты:
    - GET    /api/v1/agents/types        — список типов агентов (для UI)
    - GET    /api/v1/agents              — список агентов
    - POST   /api/v1/agents              — создать агента
    - GET    /api/v1/agents/stats        — агрегированная статистика
    - GET    /api/v1/agents/{id}         — получить агента
    - PATCH  /api/v1/agents/{id}         — обновить агента
    - DELETE /api/v1/agents/{id}         — удалить агента
    - POST   /api/v1/agents/{id}/run     — запустить (sync/stream)
    - POST   /api/v1/agents/{id}/stop    — остановить (сброс статуса)
    - GET    /api/v1/agents/{id}/stats   — статистика агента
"""

from __future__ import annotations

import json
import uuid
from collections.abc import AsyncIterator
from typing import Any

import structlog
from fastapi import APIRouter, Depends, HTTPException, Query, status
from pydantic import BaseModel
from sse_starlette.sse import EventSourceResponse

from src.api.dependencies import get_agent_repo, get_agent_service, get_user_ctx
from src.db.models import Agent as AgentModel
from src.db.repositories import AgentRepository
from src.middleware.auth import UserContext
from src.primitives import AgentState
from src.schemas.agent import (
    AgentCreate,
    AgentResponse,
    AgentRunRequest,
    AgentRunResponse,
    AgentUpdate,
)
from src.services import AgentService

logger = structlog.get_logger()

router = APIRouter(prefix="/api/v1/agents", tags=["agents"])


# ============================================================================
# AGENT TYPES (для UI — в новой архитектуре типы опциональны)
# ============================================================================

# Агент в новой архитектуре — гибкая конфигурация (system_prompt + model + tools),
# а не экземпляр класса. Типы оставлены как подсказки для UI.
AGENT_TYPES: dict[str, str] = {
    "researcher": "Исследователь — поиск и анализ информации",
    "coder": "Программист — написание и анализ кода",
    "analyst": "Аналитик — анализ данных и подготовка отчётов",
    "writer": "Писатель — создание текстов и контента",
    "assistant": "Ассистент — универсальный помощник",
}


# ============================================================================
# RESPONSE MODELS (специфичные для API)
# ============================================================================


class AgentTypeInfo(BaseModel):
    """Информация о типе агента."""

    type: str
    description: str


class AgentStatsResponse(BaseModel):
    """Агрегированная статистика по агентам."""

    total_agents: int
    total_tasks: int
    total_tokens: int
    total_cost_usd: float
    avg_success_rate: float


# ============================================================================
# HELPER FUNCTIONS
# ============================================================================


def _build_agent_response(agent: AgentModel) -> AgentResponse:
    """Собрать AgentResponse из DB-модели (через from_attributes)."""
    return AgentResponse.model_validate(agent)


def _build_run_response(agent_id: uuid.UUID, state: AgentState) -> AgentRunResponse:
    """Собрать AgentRunResponse из результата выполнения (primitives.AgentState)."""
    return AgentRunResponse(
        agent_id=agent_id,
        status=state.status.value,
        output=state.get_output("output", ""),
        reasoning=state.get_output("reasoning_content"),
        tokens_used=state.tokens_used,
        tool_calls=state.tool_calls,
        duration_ms=state.duration_ms,
    )


# ============================================================================
# ENDPOINTS: типы агентов
# ============================================================================


@router.get("/types", response_model=list[AgentTypeInfo])
async def list_agent_types() -> list[AgentTypeInfo]:
    """
    Список типов агентов (подсказки для UI).

    Note: в новой архитектуре тип — опциональная метка; поведение агента
    определяется его конфигурацией (system_prompt, model, tools).
    """
    return [
        AgentTypeInfo(type=agent_type, description=description)
        for agent_type, description in AGENT_TYPES.items()
    ]


# ============================================================================
# CRUD ENDPOINTS
# ============================================================================


@router.get("", response_model=list[AgentResponse])
async def list_agents(
    agent_type: str | None = Query(None, description="Фильтр по типу"),
    status_filter: str | None = Query(None, alias="status", description="Фильтр по статусу"),
    limit: int = Query(100, ge=1, le=500),
    offset: int = Query(0, ge=0),
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> list[AgentResponse]:
    """Список агентов в текущем workspace с фильтрацией и пагинацией."""
    agents = await agent_service.list(
        agent_type=agent_type,
        status=status_filter,
        limit=limit,
        offset=offset,
    )

    logger.info(
        "agents.listed",
        count=len(agents),
        agent_type=agent_type,
        status=status_filter,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return [_build_agent_response(agent) for agent in agents]


@router.post("", response_model=AgentResponse, status_code=status.HTTP_201_CREATED)
async def create_agent(
    data: AgentCreate,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> AgentResponse:
    """Создать нового агента."""
    agent = await agent_service.create(data)

    logger.info(
        "agent.created",
        agent_id=str(agent.id),
        name=agent.name,
        agent_type=agent.agent_type,
        tools_count=len(agent.tools or []),
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return _build_agent_response(agent)


@router.get("/stats", response_model=AgentStatsResponse)
async def get_agents_stats(
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_repo: AgentRepository = Depends(get_agent_repo),
) -> AgentStatsResponse:
    """Агрегированная статистика по всем агентам в workspace."""
    stats = await agent_repo.get_agents_stats()

    return AgentStatsResponse(
        total_agents=stats["total_agents"],
        total_tasks=stats["total_tasks"],
        total_tokens=stats["total_tokens"],
        total_cost_usd=stats["total_cost_usd"],
        avg_success_rate=stats["avg_success_rate"],
    )


@router.get("/{agent_id}", response_model=AgentResponse)
async def get_agent(
    agent_id: uuid.UUID,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> AgentResponse:
    """Получить агента по ID."""
    agent = await agent_service.get(agent_id)
    if agent is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    return _build_agent_response(agent)


@router.patch("/{agent_id}", response_model=AgentResponse)
async def update_agent(
    agent_id: uuid.UUID,
    data: AgentUpdate,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> AgentResponse:
    """Обновить агента (частичное обновление)."""
    agent = await agent_service.update(agent_id, data)
    if agent is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    logger.info(
        "agent.updated",
        agent_id=str(agent_id),
        updated_fields=list(data.model_dump(exclude_unset=True).keys()),
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return _build_agent_response(agent)


@router.delete("/{agent_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_agent(
    agent_id: uuid.UUID,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> None:
    """Удалить агента (нельзя удалить запущенного)."""
    agent = await agent_service.get(agent_id)
    if agent is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    if agent.status == "running":
        raise HTTPException(
            status_code=status.HTTP_409_CONFLICT,
            detail=f"Cannot delete agent '{agent.name}' while it is running",
        )

    deleted = await agent_service.delete(agent_id)
    if not deleted:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    logger.info(
        "agent.deleted",
        agent_id=str(agent_id),
        name=agent.name,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )


# ============================================================================
# RUN / STOP ENDPOINTS
# ============================================================================


# ✅ response_model=None — функция возвращает Union (AgentRunResponse | EventSourceResponse),
# FastAPI не может вывести единую Pydantic-схему из EventSourceResponse
@router.post("/{agent_id}/run", response_model=None)
async def run_agent(
    agent_id: uuid.UUID,
    data: AgentRunRequest,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> AgentRunResponse | EventSourceResponse:
    """
    Запустить агента с входными данными.

    Два режима:
    - **stream=true** (default): SSE поток событий (agent_start, reasoning,
      agent_message, tool_call, tool_result, agent_done, error)
    - **stream=false**: JSON с полным ответом
    """
    agent = await agent_service.get(agent_id)
    if agent is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    logger.info(
        "agent.run.requested",
        agent_id=str(agent_id),
        stream=data.stream,
        input_length=len(data.input),
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    if data.stream:
        return EventSourceResponse(_stream_agent_run(agent_id, data.input, agent_service))

    # Non-streaming mode
    state = await agent_service.run(agent_id, data.input)
    return _build_run_response(agent_id, state)


async def _stream_agent_run(
    agent_id: uuid.UUID,
    input_text: str,
    agent_service: AgentService,
) -> AsyncIterator[dict[str, str]]:
    """Генератор SSE событий для streaming выполнения агента."""
    try:
        async for event in agent_service.run_stream(agent_id, input_text):
            event_type = event.get("type", "message")
            yield {
                "event": event_type,
                "data": json.dumps(event, ensure_ascii=False, default=str),
            }
    except HTTPException as e:
        yield {
            "event": "error",
            "data": json.dumps({"type": "error", "error": str(e.detail)}, ensure_ascii=False),
        }
    except Exception as e:
        logger.exception(
            "agent.run.stream.error",
            agent_id=str(agent_id),
            error=str(e),
            error_type=type(e).__name__,
        )
        yield {
            "event": "error",
            "data": json.dumps(
                {
                    "type": "error",
                    "error": str(e),
                    "error_type": type(e).__name__,
                },
                ensure_ascii=False,
            ),
        }


@router.post("/{agent_id}/stop")
async def stop_agent(
    agent_id: uuid.UUID,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_repo: AgentRepository = Depends(get_agent_repo),
) -> dict[str, Any]:
    """
    Остановить агента (сброс статуса).

    Note: агенты выполняются в рамках HTTP-запроса, поэтому stop
    сбрасывает статус в idle (реальная отмена — через CancellationToken
    в будущих версиях).
    """
    agent = await agent_repo.get(agent_id)
    if agent is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    if agent.status != "running":
        return {
            "status": "not_running",
            "agent_id": str(agent_id),
            "message": f"Agent '{agent.name}' is not currently running",
            "current_status": agent.status,
        }

    await agent_repo.update_status(agent_id=agent_id, status="idle", current_task_id=None)

    logger.info(
        "agent.stopped",
        agent_id=str(agent_id),
        name=agent.name,
        user_id=str(user_ctx.user_id),
        workspace_id=user_ctx.workspace_id,
    )

    return {
        "status": "stopped",
        "agent_id": str(agent_id),
        "name": agent.name,
        "message": "Agent status reset to idle",
    }


# ============================================================================
# PER-AGENT STATS
# ============================================================================


@router.get("/{agent_id}/stats")
async def get_agent_stats(
    agent_id: uuid.UUID,
    user_ctx: UserContext = Depends(get_user_ctx),
    agent_service: AgentService = Depends(get_agent_service),
) -> dict[str, Any]:
    """Детальная статистика конкретного агента."""
    agent = await agent_service.get(agent_id)
    if agent is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Agent {agent_id} not found",
        )

    return {
        "agent_id": str(agent.id),
        "name": agent.name,
        "agent_type": agent.agent_type,
        "status": agent.status,
        "statistics": {
            "tasks_count": agent.tasks_count,
            "total_tokens_used": agent.total_tokens_used,
            "total_cost_usd": agent.total_cost_usd,
            "success_rate": agent.success_rate,
        },
        "tools": {
            "builtin": list(agent.tools or []),
            "mcp_servers": list(agent.mcp_server_ids or []),
        },
        "activity": {
            "created_at": agent.created_at.isoformat(),
            "updated_at": agent.updated_at.isoformat(),
            "last_run_at": agent.last_run_at.isoformat() if agent.last_run_at else None,
            "current_task_id": str(agent.current_task_id) if agent.current_task_id else None,
        },
    }