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

core/engine/src/agents/__init__.py
311 строк12 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
"""Агенты для мультиагентных систем FlowStack.

Этот модуль предоставляет полный набор готовых агентов для построения
мультиагентных систем с использованием LLM.

Архитектура:

    BaseAgent (абстрактный)
        ↓
    LLMAgent (базовый LLM-агент)
        ↓
    ┌─────────────────────────────────────────────────────────┐
    │ PlannerAgent      — планировщик задач (рус.)           │
    │ ExecutorAgent     — исполнитель задач (рус.)           │
    │ ResearchAgent     — простой исследователь (англ.)      │
    │ ResearcherAgent   — расширенный исследователь (рус.)   │
    │ WriterAgent       — писатель (англ.)                   │
    │ AnalystAgent      — аналитик данных (англ.)            │
    │ ReviewerAgent     — ревьюер кода/документов (рус.)     │
    │ CriticAgent       — критик контента/идей (рус.)        │
    │ DebaterAgent      — дебатёр, аргументация (рус.)       │
    │ JudgeAgent        — судья и арбитр (рус.)              │
    └─────────────────────────────────────────────────────────┘

Полный workflow мультиагентной системы:

    1. Planner      → Создаёт детальный план
    2. Debater      → Анализирует плюсы/минусы плана (опционально)
    3. Executor     → Выполняет задачи из плана
       ↓ (делегирует)
    4. Researcher   → Собирает информацию
    5. Analyst      → Анализирует данные
    6. Writer       → Создаёт контент
       ↓
    7. Reviewer     → Проверяет качество
    8. Critic       → Критически оценивает
       ↓
    9. Judge        → Выносит финальный вердикт

Примеры использования:

    # Прямое создание агента
    from src.agents import ResearchAgent, WriterAgent

    agent = ResearchAgent()
    result = await agent.run({"topic": "Искусственный интеллект"})

    # Создание через фабрику
    from src.agents import create_agent

    agent = create_agent(
        "writer",
        name="Technical Writer",
        system_prompt="Ты — технический писатель..."
    )

    # Получение списка всех доступных типов
    from src.agents import get_agent_types, describe_agents

    types = get_agent_types()           # ["planner", "executor", ...]
    descriptions = describe_agents()    # {"planner": "...", ...}
"""
from __future__ import annotations

# =============================================================================
# Базовые классы (из base.py)
# =============================================================================

from .base import (
    BaseAgent,
    LLMAgent,
    Agent,          # алиас для BaseAgent
    AgentState,     # тип для состояния агента (dict[str, Any])
    ResearchAgent,  # простой исследователь (англ.)
    WriterAgent,    # писатель (англ.)
)

# =============================================================================
# Специализированные агенты (из отдельных модулей)
# =============================================================================

from .planner import PlannerAgent       # планировщик задач
from .executor import ExecutorAgent     # исполнитель задач
from .researcher import ResearcherAgent # расширенный исследователь
from .analyst import AnalystAgent       # аналитик данных
from .reviewer import ReviewerAgent     # ревьюер кода/документов
from .critic import CriticAgent         # критик контента/идей
from .debater import DebaterAgent       # дебатёр, аргументация
from .judge import JudgeAgent           # судья и арбитр

# =============================================================================
# Короткие алиасы для удобства
# =============================================================================

Planner = PlannerAgent
Executor = ExecutorAgent
Researcher = ResearcherAgent
Writer = WriterAgent
Analyst = AnalystAgent
Reviewer = ReviewerAgent
Critic = CriticAgent
Debater = DebaterAgent
Judge = JudgeAgent

# =============================================================================
# Публичный API (__all__)
# =============================================================================

__all__ = [
    # Базовые классы
    "BaseAgent",
    "LLMAgent",
    "Agent",
    "AgentState",
    
    # Специализированные агенты
    "PlannerAgent",
    "ExecutorAgent",
    "ResearchAgent",
    "ResearcherAgent",
    "WriterAgent",
    "AnalystAgent",
    "ReviewerAgent",
    "CriticAgent",
    "DebaterAgent",
    "JudgeAgent",
    
    # Короткие алиасы
    "Planner",
    "Executor",
    "Researcher",
    "Writer",
    "Analyst",
    "Reviewer",
    "Critic",
    "Debater",
    "Judge",
    
    # Фабрика и утилиты
    "AGENT_REGISTRY",
    "create_agent",
    "get_agent_types",
    "describe_agents",
]

# =============================================================================
# Реестр агентов (для динамического создания)
# =============================================================================

AGENT_REGISTRY: dict[str, type[BaseAgent]] = {
    # Стратегические агенты
    "planner": PlannerAgent,
    "executor": ExecutorAgent,
    
    # Исследовательские агенты
    "research": ResearchAgent,
    "researcher": ResearcherAgent,
    "analyst": AnalystAgent,
    
    # Создательные агенты
    "writer": WriterAgent,
    
    # Агенты проверки качества
    "reviewer": ReviewerAgent,
    "critic": CriticAgent,
    "debater": DebaterAgent,
    
    # Агенты принятия решений
    "judge": JudgeAgent,
}

# =============================================================================
# Фабрика агентов
# =============================================================================

def create_agent(
    agent_type: str,
    name: str | None = None,
    system_prompt: str | None = None,
) -> BaseAgent:
    """Фабрика для создания агентов по типу.
    
    Args:
        agent_type: Тип агента. Доступные значения:
            - "planner" — планировщик задач
            - "executor" — исполнитель задач
            - "research" — простой исследователь (англ.)
            - "researcher" — расширенный исследователь (рус.)
            - "writer" — писатель
            - "analyst" — аналитик данных
            - "reviewer" — ревьюер кода/документов
            - "critic" — критик контента/идей
            - "debater" — дебатёр, аргументация
            - "judge" — судья и арбитр
        name: Кастомное имя агента (опционально).
        system_prompt: Кастомный system prompt (опционально).
        
    Returns:
        Экземпляр агента указанного типа.
        
    Raises:
        ValueError: Если agent_type не найден в реестре.
        
    Example:
        >>> agent = create_agent("researcher", name="Deep Researcher")
        >>> result = await agent.run({"topic": "Machine Learning"})
        
        >>> judge = create_agent(
        ...     "judge",
        ...     name="Quality Judge",
        ...     system_prompt="Ты — строгий судья..."
        ... )
    """
    agent_class = AGENT_REGISTRY.get(agent_type.lower())
    if agent_class is None:
        available = ", ".join(sorted(AGENT_REGISTRY.keys()))
        raise ValueError(
            f"Unknown agent type: '{agent_type}'. "
            f"Available types: {available}"
        )
    
    # Формируем kwargs для кастомных параметров
    kwargs: dict[str, str] = {}
    if name is not None:
        kwargs["name"] = name
    if system_prompt is not None:
        kwargs["system_prompt"] = system_prompt
    
    return agent_class(**kwargs)


def get_agent_types() -> list[str]:
    """Получить список всех доступных типов агентов.
    
    Returns:
        Отсортированный список строк — ключей из AGENT_REGISTRY.
        
    Example:
        >>> types = get_agent_types()
        >>> print(types)
        ['analyst', 'critic', 'debater', 'executor', 'judge', ...]
    """
    return sorted(AGENT_REGISTRY.keys())


def describe_agents() -> dict[str, str]:
    """Получить описания всех доступных агентов.
    
    Извлекает первую строку docstring каждого класса агента.
    
    Returns:
        Словарь {agent_type: description}.
        
    Example:
        >>> descriptions = describe_agents()
        >>> for agent_type, desc in descriptions.items():
        ...     print(f"{agent_type}: {desc}")
    """
    descriptions: dict[str, str] = {}
    
    for agent_type, agent_class in sorted(AGENT_REGISTRY.items()):
        doc = agent_class.__doc__ or ""
        # Берём первую непустую строку docstring
        first_line = ""
        for line in doc.splitlines():
            stripped = line.strip()
            if stripped:
                first_line = stripped
                break
        descriptions[agent_type] = first_line
    
    return descriptions


# =============================================================================
# Группировка агентов по ролям (для UI и оркестраторов)
# =============================================================================

AGENT_ROLES: dict[str, list[str]] = {
    "strategic": ["planner", "executor"],
    "research": ["research", "researcher", "analyst"],
    "creation": ["writer"],
    "quality": ["reviewer", "critic", "debater"],
    "decision": ["judge"],
}


def get_agents_by_role(role: str) -> list[str]:
    """Получить список агентов по их роли в workflow.
    
    Args:
        role: Одна из ролей:
            - "strategic" — планирование и выполнение
            - "research" — исследование и анализ
            - "creation" — создание контента
            - "quality" — проверка качества
            - "decision" — принятие решений
            
    Returns:
        Список типов агентов для данной роли.
        
    Raises:
        ValueError: Если роль не найдена.
    """
    if role not in AGENT_ROLES:
        available = ", ".join(sorted(AGENT_ROLES.keys()))
        raise ValueError(
            f"Unknown role: '{role}'. "
            f"Available roles: {available}"
        )
    return AGENT_ROLES[role]