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

core/engine/src/flows/__init__.py
417 строк14 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
# core/engine/src/flows/__init__.py
"""Flows — готовые мультиагентные рабочие процессы FlowStack.

Этот модуль предоставляет полный набор готовых мультиагентных flows
для решения различных задач: исследования, анализа, дебатов, консенсуса,
генерации идей и других сложных многошаговых процессов.

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

    Flow (базовый класс из primitives.flow)
        ↓
    ┌─────────────────────────────────────────────────────────────────┐
    │ research      — исследование темы (5 агентов, последовательно) │
    │ review        — критический обзор (3 агента, последовательно)  │
    │ debate        — дебаты ЗА/ПРОТИВ (5 агентов, последовательно)  │
    │ code_review   — ревью кода (5 агентов, последовательно)        │
    │ consensus     — консенсус экспертов (8 агентов, параллельно)   │
    │ compare       — сравнение вариантов (6+N, параллельно)         │
    │ brainstorm    — генерация идей (9 агентов, параллельно)        │
    │ analyze       — глубокий анализ (6 агентов, последовательно)   │
    └─────────────────────────────────────────────────────────────────┘

Когда какой flow использовать:

    "Хочу узнать о теме"             → research
    "Хочу критический обзор текста"  → review
    "Решение: делать или нет?"       → debate
    "Проверить код"                  → code_review
    "Согласовать мнения экспертов"   → consensus
    "Выбрать из N вариантов"         → compare
    "Нагенерить идеи"                → brainstorm
    "Понять почему что-то происходит"→ analyze

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

    research:
        Planner → Researcher → Analyst → Writer → Reviewer

    debate:
        Planner → Debater PRO → Debater CON → Analyst → Judge

    consensus:
        Planner → [Expert 1 || Expert 2 || Expert 3] → Moderator → 
        Debater → Judge → Writer

    compare:
        Planner → [Researcher 1 || Researcher 2 || ...] → 
        Analyst → Critic → Judge → Writer

    brainstorm:
        Planner → [Analogies || Patterns || Creative || Reverse] →
        Synthesizer → Evaluator → Judge → Writer

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

    # Самый простой способ — через get_flow()
    from src.flows import get_flow

    flow = get_flow("research")
    async for event in flow.runner({"topic": "Искусственный интеллект"}):
        if event["type"] == "flow_done":
            print(event["output"])

    # Список всех доступных flows
    from src.flows import list_flows, describe_flows

    flows = list_flows()              # [Flow(...), Flow(...), ...]
    descriptions = describe_flows()   # {"research": "...", ...}

    # Создание custom flow через Builder
    from src.flows import CustomFlowBuilder

    flow = (
        CustomFlowBuilder()
        .with_id("my_flow")
        .with_name("My Custom Flow")
        .add_agent("planner")
        .add_agent("researcher")
        .add_agent("writer")
        .build(auto_register=True)
    )

    # Создание из шаблона
    from src.flows import create_from_template

    flow = create_from_template(
        template_name="research_deep",
        flow_id="my_research",
        auto_register=True,
    )

    # Получение flows по категории
    from src.flows import get_flows_by_category

    analysis_flows = get_flows_by_category("analysis")
    # ["analyze", "compare"]
"""
from __future__ import annotations

from typing import Any

import structlog

logger = structlog.get_logger()


# =============================================================================
# Core: базовый класс Flow (из primitives)
# =============================================================================

from src.primitives.flow import Flow


# =============================================================================
# Registry: функции управления реестром (из registry.py)
# =============================================================================

from src.flows.registry import (
    FLOW_REGISTRY,
    get_flow,
    list_flows,
    get_flow_ids,
    flow_exists,
    get_flows_summary,
    register_flow,
    unregister_flow,
)

# Алиас для обратной совместимости
registry = FLOW_REGISTRY


# =============================================================================
# Предопределённые flows (импортируем экземпляры)
# =============================================================================
# При импорте каждого модуля flow автоматически регистрируется в registry

from src.flows.research import research_flow
from src.flows.review_flow import review_flow
from src.flows.debate_flow import debate_flow
from src.flows.code_review_flow import code_review_flow
from src.flows.consensus import consensus_flow
from src.flows.compare import compare_flow
from src.flows.brainstorm import brainstorm_flow
from src.flows.analyze import analyze_flow


# =============================================================================
# Custom flows: Builder и utilities (из custom.py)
# =============================================================================

from src.flows.custom import (
    # Exceptions
    CustomFlowError,
    FlowValidationError,
    FlowAlreadyExistsError,
    
    # Builder
    CustomFlowBuilder,
    AgentConfig,
    
    # Quick creation
    create_custom_flow,
    
    # Loaders
    load_flow_from_yaml,
    load_flow_from_json,
    load_flow_from_dict,
    
    # Exporters
    export_flow_to_yaml,
    export_flow_to_json,
    
    # Management
    register_custom_flow,
    unregister_custom_flow,
    list_custom_flows,
    
    # Runner
    run_custom_flow,
)


# =============================================================================
# Templates: готовые шаблоны (из create.py)
# =============================================================================

from src.flows.create import (
    FlowTemplate,
    TEMPLATES,
    list_templates,
    get_template,
    list_categories,
    search_templates,
    create_from_template,
    create_similar_flow,
    generate_yaml_from_template,
)


# =============================================================================
# Extract: извлечение структурированных данных (из extract.py)
# =============================================================================

from src.flows.extract import (
    extract_flow_result,
    extract_scores,
    extract_issues,
    extract_recommendations,
    extract_verdict,
    extract_findings,
    extract_metadata,
    extract_code_examples,
    extract_sections,
    summarize_result,
)


# =============================================================================
# Авто-регистрация: гарантируем что все flows в реестре
# =============================================================================

_ALL_PREDEFINED_FLOWS = [
    research_flow,
    review_flow,
    debate_flow,
    code_review_flow,
    consensus_flow,
    compare_flow,
    brainstorm_flow,
    analyze_flow,
]

for _flow in _ALL_PREDEFINED_FLOWS:
    if _flow.id not in FLOW_REGISTRY:
        try:
            register_flow(_flow)
        except Exception as e:
            logger.warning(
                "flows.init.auto_register_failed",
                flow_id=_flow.id,
                error=str(e),
            )


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

__all__ = [
    # === Core ===
    "Flow",
    
    # === Registry ===
    "FLOW_REGISTRY",
    "registry",                   # алиас для обратной совместимости
    "get_flow",                   # ⭐ главный способ получения flow
    "list_flows",
    "get_flow_ids",
    "flow_exists",
    "get_flows_summary",
    "register_flow",
    "unregister_flow",
    
    # === Предопределённые flows (экземпляры) ===
    "research_flow",
    "review_flow",
    "debate_flow",
    "code_review_flow",
    "consensus_flow",
    "compare_flow",
    "brainstorm_flow",
    "analyze_flow",
    
    # === Custom flows: exceptions ===
    "CustomFlowError",
    "FlowValidationError",
    "FlowAlreadyExistsError",
    
    # === Custom flows: builder ===
    "CustomFlowBuilder",
    "AgentConfig",
    "create_custom_flow",
    
    # === Custom flows: loaders/exporters ===
    "load_flow_from_yaml",
    "load_flow_from_json",
    "load_flow_from_dict",
    "export_flow_to_yaml",
    "export_flow_to_json",
    
    # === Custom flows: management ===
    "register_custom_flow",
    "unregister_custom_flow",
    "list_custom_flows",
    "run_custom_flow",
    
    # === Templates ===
    "FlowTemplate",
    "TEMPLATES",
    "list_templates",
    "get_template",
    "list_categories",
    "search_templates",
    "create_from_template",
    "create_similar_flow",
    "generate_yaml_from_template",
    
    # === Extract ===
    "extract_flow_result",
    "extract_scores",
    "extract_issues",
    "extract_recommendations",
    "extract_verdict",
    "extract_findings",
    "extract_metadata",
    "extract_code_examples",
    "extract_sections",
    "summarize_result",
    
    # === Утилиты (определены ниже) ===
    "describe_flows",
    "get_flows_by_category",
    "FLOW_CATEGORIES",
]


# =============================================================================
# Утилиты: описания flows
# =============================================================================

def describe_flows() -> dict[str, str]:
    """Получить описания всех зарегистрированных flows.
    
    Извлекает первую строку docstring каждого flow.
    
    Returns:
        Словарь {flow_id: description}.
        
    Example:
        >>> descriptions = describe_flows()
        >>> for flow_id, desc in descriptions.items():
        ...     print(f"{flow_id}: {desc}")
    """
    descriptions: dict[str, str] = {}
    
    for flow_id, flow in sorted(FLOW_REGISTRY.items()):
        # Пытаемся взять описание из атрибута description
        desc = getattr(flow, "description", None)
        
        # Если нет — берём из runner docstring
        if not desc and flow.runner and flow.runner.__doc__:
            first_line = ""
            for line in flow.runner.__doc__.splitlines():
                stripped = line.strip()
                if stripped:
                    first_line = stripped
                    break
            desc = first_line
        
        descriptions[flow_id] = desc or "No description"
    
    return descriptions


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

FLOW_CATEGORIES: dict[str, list[str]] = {
    "research": ["research", "analyze"],
    "review": ["review", "code_review"],
    "decision": ["debate", "consensus", "compare"],
    "ideation": ["brainstorm"],
}


def get_flows_by_category(category: str) -> list[str]:
    """Получить список flows по их категории.
    
    Args:
        category: Одна из категорий:
            - "research" — исследование и анализ (research, analyze)
            - "review" — обзор и ревью (review, code_review)
            - "decision" — принятие решений (debate, consensus, compare)
            - "ideation" — генерация идей (brainstorm)
            
    Returns:
        Список ID flows для данной категории.
        
    Raises:
        ValueError: Если категория не найдена.
        
    Example:
        >>> get_flows_by_category("decision")
        ['debate', 'consensus', 'compare']
    """
    if category not in FLOW_CATEGORIES:
        available = ", ".join(sorted(FLOW_CATEGORIES.keys()))
        raise ValueError(
            f"Unknown category: '{category}'. "
            f"Available categories: {available}"
        )
    return FLOW_CATEGORIES[category]


# =============================================================================
# Логирование инициализации
# =============================================================================

logger.info(
    "flows.initialized",
    total_flows=len(FLOW_REGISTRY),
    flow_ids=sorted(FLOW_REGISTRY.keys()),
    templates_count=len(TEMPLATES),
    categories=list(FLOW_CATEGORIES.keys()),
)