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

core/engine/src/flows/debug.py
705 строк25 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
# core/engine/src/flows/debug.py
"""Debug utilities для FlowStack Flows.

Набор CLI команд для отладки, тестирования и диагностики мультиагентных flows.

Запуск из корня core/engine:
    python -m src.flows.debug <command> [args]

Доступные команды:
    list                     — список всех зарегистрированных flows
    run <id> <input>         — запустить flow с тестовым input
    inspect <id>             — детальная информация о flow
    test <id>                — быстрый тест flow с минимальным input
    extract <id> <file>      — извлечь structured data из markdown файла
    stats                    — статистика по всем flows
    validate                 — валидация всех flows в registry

Примеры:
    python -m src.flows.debug list
    python -m src.flows.debug run research "Искусственный интеллект в медицине"
    python -m src.flows.debug run code_review < code.py
    python -m src.flows.debug inspect debate
    python -m src.flows.debug test research
    python -m src.flows.debug extract code_review report.md -o structured.json
    python -m src.flows.debug stats
    python -m src.flows.debug validate
"""
from __future__ import annotations

import argparse
import asyncio
import json
import sys
import time
from pathlib import Path
from typing import Any, Callable

import structlog

logger = structlog.get_logger()


# ============================================================================
# ANSI цвета для красивого вывода в терминале
# ============================================================================

class Colors:
    """ANSI escape коды для цветного вывода."""
    RESET = "\033[0m"
    BOLD = "\033[1m"
    DIM = "\033[2m"
    
    RED = "\033[31m"
    GREEN = "\033[32m"
    YELLOW = "\033[33m"
    BLUE = "\033[34m"
    MAGENTA = "\033[35m"
    CYAN = "\033[36m"
    WHITE = "\033[37m"
    
    BRIGHT_RED = "\033[91m"
    BRIGHT_GREEN = "\033[92m"
    BRIGHT_YELLOW = "\033[93m"
    BRIGHT_BLUE = "\033[94m"
    BRIGHT_CYAN = "\033[96m"
    BRIGHT_MAGENTA = "\033[95m"


def color(text: str, color_code: str) -> str:
    """Применить цвет к тексту."""
    if not sys.stdout.isatty():
        return text
    return f"{color_code}{text}{Colors.RESET}"


def success(text: str) -> str:
    return color(f"✅ {text}", Colors.BRIGHT_GREEN)


def error(text: str) -> str:
    return color(f"❌ {text}", Colors.BRIGHT_RED)


def warning(text: str) -> str:
    return color(f"⚠️  {text}", Colors.BRIGHT_YELLOW)


def info(text: str) -> str:
    return color(f"ℹ️  {text}", Colors.BRIGHT_CYAN)


def header(text: str) -> str:
    return color(f"\n{'=' * 70}\n{text}\n{'=' * 70}", Colors.BOLD + Colors.BRIGHT_BLUE)


def subheader(text: str) -> str:
    return color(f"\n▶ {text}", Colors.BOLD + Colors.CYAN)


# ============================================================================
# Helpers
# ============================================================================

def _read_stdin() -> str:
    """Прочитать данные из stdin (если они есть)."""
    if sys.stdin.isatty():
        return ""
    return sys.stdin.read()


def _format_duration(seconds: float) -> str:
    """Форматировать длительность в человекочитаемый вид."""
    if seconds < 1:
        return f"{seconds * 1000:.0f}ms"
    if seconds < 60:
        return f"{seconds:.2f}s"
    minutes = int(seconds // 60)
    secs = seconds % 60
    return f"{minutes}m {secs:.1f}s"


def _print_event(event: dict[str, Any]) -> None:
    """Красиво вывести SSE событие."""
    event_type = event.get("type", "unknown")
    
    if event_type == "agent_start":
        agent = event.get("agent", "?")
        print(color(f"🤖 Агент '{agent}' запущен", Colors.BRIGHT_CYAN))
    
    elif event_type == "agent_message":
        content = event.get("content", "")
        print(content, end="", flush=True)
    
    elif event_type == "agent_done":
        agent = event.get("agent", "?")
        tokens = event.get("tokens", 0)
        model = event.get("model", "")
        print()
        print(color(
            f"✓ Агент '{agent}' завершил · {tokens:,} токенов · {model}",
            Colors.BRIGHT_GREEN
        ))
    
    elif event_type == "error":
        err = event.get("error", "Unknown")
        agent = event.get("agent", "")
        prefix = f" в агенте '{agent}'" if agent else ""
        print(error(f"Ошибка{prefix}: {err}"))
    
    elif event_type == "flow_done":
        flow_id = event.get("flow_id", "?")
        tokens = event.get("tokens", 0)
        stages = event.get("stages", {})
        print()
        print(color(f"🏁 Flow '{flow_id}' завершён", Colors.BOLD + Colors.BRIGHT_GREEN))
        print(color(f"   • Токенов: {tokens:,}", Colors.DIM))
        print(color(f"   • Этапов: {len(stages)}", Colors.DIM))
        for stage_name, stage_output in stages.items():
            if isinstance(stage_output, str):
                print(color(f"   • {stage_name}: {len(stage_output):,} символов", Colors.DIM))


# ============================================================================
# Commands
# ============================================================================

async def cmd_list(args: argparse.Namespace) -> int:
    """Список всех зарегистрированных flows."""
    from src.flows.registry import list_flows
    
    flows = list_flows()
    
    if not flows:
        print(warning("Нет зарегистрированных flows"))
        return 1
    
    print(header(f"Зарегистрированные flows ({len(flows)})"))
    
    for flow in flows:
        print(color(f"\n📋 {flow.id}", Colors.BOLD + Colors.BRIGHT_MAGENTA))
        print(f"   {color('Name:', Colors.DIM)} {flow.name}")
        print(f"   {color('Description:', Colors.DIM)} {flow.description}")
        print(f"   {color('Agents:', Colors.DIM)} {', '.join(flow.agents)}")
        print(f"   {color('Runner:', Colors.DIM)} {flow.runner.__name__ if flow.runner else 'None'}")
    
    print()
    print(info(f"Используй: python -m src.flows.debug run <id> \"<input>\""))
    return 0


async def cmd_run(args: argparse.Namespace) -> int:
    """Запуск flow с тестовым input."""
    from src.flows.registry import get_flow
    
    flow = get_flow(args.flow_id)
    if flow is None:
        print(error(f"Flow '{args.flow_id}' не найден"))
        print(info("Доступные flows:"))
        from src.flows.registry import list_flows
        for f in list_flows():
            print(f"   - {f.id}")
        return 1
    
    # Собираем input
    input_text = args.input or _read_stdin()
    if not input_text.strip():
        print(error("Не указан input. Используй: python -m src.flows.debug run <id> \"<input>\""))
        print(info("Или pipe через stdin: cat file.py | python -m src.flows.debug run code_review"))
        return 1
    
    input_data: dict[str, Any] = {"input": input_text}
    if args.context:
        input_data["context"] = args.context
    if args.language:
        input_data["language"] = args.language
    if args.focus:
        input_data["focus"] = args.focus
    
    print(header(f"Запуск flow: {flow.name}"))
    print(color(f"ID: {flow.id}", Colors.DIM))
    print(color(f"Agents: {', '.join(flow.agents)}", Colors.DIM))
    print(color(f"Input length: {len(input_text):,} символов", Colors.DIM))
    print()
    
    start_time = time.time()
    final_output = ""
    final_tokens = 0
    stages_data = {}
    
    if flow.runner is None:
        print(error(f"Flow '{flow.id}' не имеет runner"))
        return 1

    try:
        async for event in flow.runner(input_data):
            _print_event(event)
            if event.get("type") == "flow_done":
                final_output = event.get("output", "")
                final_tokens = event.get("tokens", 0)
                stages_data = event.get("stages", {})
    except KeyboardInterrupt:
        print(warning("\nПрервано пользователем"))
        return 130
    except Exception as e:
        logger.exception("flows.debug.run.failed")
        print(error(f"Ошибка: {e}"))
        return 1
    
    duration = time.time() - start_time
    print()
    print(color(f"Время выполнения: {_format_duration(duration)}", Colors.DIM))
    
    # Сохранение результата
    if args.output:
        output_path = Path(args.output)
        output_path.write_text(final_output, encoding="utf-8")
        print(success(f"Результат сохранён в {output_path}"))
    
    # Извлечение structured data
    if args.extract:
        try:
            from src.flows.extract import extract_flow_result
            structured = extract_flow_result(
                flow_id=flow.id,
                output=final_output,
                stages=stages_data,
                metadata={},
                tokens=final_tokens,
            )
            print(subheader("Structured data:"))
            print(json.dumps(structured, ensure_ascii=False, indent=2))
            
            if args.structured_output:
                structured_path = Path(args.structured_output)
                structured_path.write_text(
                    json.dumps(structured, ensure_ascii=False, indent=2),
                    encoding="utf-8"
                )
                print(success(f"Structured data сохранены в {structured_path}"))
        except Exception as e:
            print(warning(f"Не удалось извлечь structured данные: {e}"))
    
    return 0


async def cmd_inspect(args: argparse.Namespace) -> int:
    """Детальная информация о flow."""
    from src.flows.registry import get_flow
    
    flow = get_flow(args.flow_id)
    if flow is None:
        print(error(f"Flow '{args.flow_id}' не найден"))
        return 1
    
    print(header(f"Flow: {flow.name}"))
    
    print(subheader("Basic Info"))
    print(f"  {color('ID:', Colors.BOLD)} {flow.id}")
    print(f"  {color('Name:', Colors.BOLD)} {flow.name}")
    print(f"  {color('Description:', Colors.BOLD)} {flow.description}")
    
    print(subheader("Agents"))
    for i, agent in enumerate(flow.agents, 1):
        print(f"  {i}. {color(agent, Colors.BRIGHT_CYAN)}")
    
    print(subheader("Runner"))
    if flow.runner:
        print(f"  {color('Function:', Colors.BOLD)} {flow.runner.__name__}")
        print(f"  {color('Module:', Colors.BOLD)} {flow.runner.__module__}")
        
        # Показываем signature
        import inspect
        sig = inspect.signature(flow.runner)
        print(f"  {color('Signature:', Colors.BOLD)} {sig}")
        
        # Показываем docstring
        if flow.runner.__doc__:
            print(f"  {color('Docstring:', Colors.BOLD)}")
            for line in flow.runner.__doc__.split("\n")[:5]:
                print(f"    {line.strip()}")
    else:
        print(warning("  Runner не установлен"))
    
    print(subheader("Validation"))
    issues = []
    
    if not flow.id:
        issues.append("ID не установлен")
    if not flow.name:
        issues.append("Name не установлен")
    if not flow.description:
        issues.append("Description не установлен")
    if not flow.agents:
        issues.append("Список agents пуст")
    if not flow.runner:
        issues.append("Runner не установлен")
    
    if issues:
        for issue in issues:
            print(warning(f"  {issue}"))
    else:
        print(success("  Все проверки пройдены"))
    
    return 0


async def cmd_test(args: argparse.Namespace) -> int:
    """Быстрый тест flow с минимальным input."""
    from src.flows.registry import get_flow
    
    flow = get_flow(args.flow_id)
    if flow is None:
        print(error(f"Flow '{args.flow_id}' не найден"))
        return 1
    
    # Тестовые inputs для разных flows
    test_inputs = {
        "research": "Искусственный интеллект в медицине",
        "review": "Тестовый контент для ревью",
        "debate": "Стоит ли переходить на микросервисы?",
        "code_review": """
def calculate_total(items):
    total = 0
    for item in items:
        total += item['price'] * item['quantity']
    return total
""",
    }
    
    input_text = test_inputs.get(flow.id, "Тестовый input")
    input_data = {"input": input_text}
    
    print(header(f"Быстрый тест: {flow.name}"))
    print(color(f"Test input: {input_text}", Colors.DIM))
    print()
    
    start_time = time.time()
    events_count = 0
    agents_started = []
    
    if flow.runner is None:
        print(error(f"Flow '{flow.id}' не имеет runner"))
        return 1

    try:
        async for event in flow.runner(input_data):
            events_count += 1
            event_type = event.get("type")
            
            if event_type == "agent_start":
                agent = event.get("agent", "?")
                agents_started.append(agent)
                print(color(f"🤖 {agent}", Colors.BRIGHT_CYAN))
            elif event_type == "agent_done":
                agent = event.get("agent", "?")
                tokens = event.get("tokens", 0)
                print(color(f"✓ {agent} ({tokens:,} tokens)", Colors.BRIGHT_GREEN))
            elif event_type == "error":
                print(error(f"Ошибка: {event.get('error')}"))
                return 1
            elif event_type == "flow_done":
                print(success(f"Flow завершён успешно"))
    
    except KeyboardInterrupt:
        print(warning("\nПрервано"))
        return 130
    except Exception as e:
        logger.exception("flows.debug.test.failed")
        print(error(f"Ошибка: {e}"))
        return 1
    
    duration = time.time() - start_time
    
    print()
    print(subheader("Test Results"))
    print(f"  {color('Duration:', Colors.BOLD)} {_format_duration(duration)}")
    print(f"  {color('Events:', Colors.BOLD)} {events_count}")
    print(f"  {color('Agents executed:', Colors.BOLD)} {', '.join(agents_started)}")
    print()
    print(success("Тест пройден успешно!"))
    
    return 0


async def cmd_extract(args: argparse.Namespace) -> int:
    """Извлечь структурированные данные из markdown файла."""
    from src.flows.extract import extract_flow_result
    
    file_path = Path(args.file)
    if not file_path.exists():
        print(error(f"Файл не найден: {file_path}"))
        return 1
    
    content = file_path.read_text(encoding="utf-8")
    
    print(header(f"Extract from {file_path.name}"))
    print(color(f"Flow: {args.flow_id}", Colors.DIM))
    print(color(f"Content length: {len(content):,} символов", Colors.DIM))
    print()
    
    try:
        result = extract_flow_result(
            flow_id=args.flow_id,
            output=content,
            stages={},
            metadata={},
        )
        
        if args.output:
            output_path = Path(args.output)
            output_path.write_text(
                json.dumps(result, ensure_ascii=False, indent=2),
                encoding="utf-8"
            )
            print(success(f"Результат сохранён в {output_path}"))
        else:
            print(json.dumps(result, ensure_ascii=False, indent=2))
        
        return 0
    
    except Exception as e:
        logger.exception("flows.debug.extract.failed")
        print(error(f"Ошибка извлечения: {e}"))
        return 1


async def cmd_stats(args: argparse.Namespace) -> int:
    """Статистика по всем flows."""
    from src.flows.registry import list_flows
    
    flows = list_flows()
    
    if not flows:
        print(warning("Нет зарегистрированных flows"))
        return 1
    
    print(header("Flows Statistics"))
    
    # Общая статистика
    total_flows = len(flows)
    total_agents = sum(len(f.agents) for f in flows)
    unique_agents = set()
    for flow in flows:
        unique_agents.update(flow.agents)
    
    print(subheader("Summary"))
    print(f"  {color('Total flows:', Colors.BOLD)} {total_flows}")
    print(f"  {color('Total agents (with duplicates):', Colors.BOLD)} {total_agents}")
    print(f"  {color('Unique agents:', Colors.BOLD)} {len(unique_agents)}")
    print(f"  {color('Avg agents per flow:', Colors.BOLD)} {total_agents / total_flows:.1f}")
    
    # Статистика по flows
    print(subheader("Flows Breakdown"))
    for flow in sorted(flows, key=lambda f: len(f.agents), reverse=True):
        print(f"  {color(flow.id, Colors.BRIGHT_MAGENTA)}: {len(flow.agents)} agents")
    
    # Статистика по агентам
    print(subheader("Agent Usage"))
    agent_counts: dict[str, int] = {}
    for flow in flows:
        for agent in flow.agents:
            agent_counts[agent] = agent_counts.get(agent, 0) + 1
    
    for agent, count in sorted(agent_counts.items(), key=lambda x: x[1], reverse=True):
        flows_using = [f.id for f in flows if agent in f.agents]
        print(f"  {color(agent, Colors.BRIGHT_CYAN)}: {count} flows ({', '.join(flows_using)})")
    
    return 0


async def cmd_validate(args: argparse.Namespace) -> int:
    """Валидация всех flows в registry."""
    from src.flows.registry import list_flows
    
    flows = list_flows()
    
    if not flows:
        print(warning("Нет flows для валидации"))
        return 1
    
    print(header(f"Validating {len(flows)} flows"))
    
    all_valid = True
    
    for flow in flows:
        print(color(f"\n📋 {flow.id}", Colors.BOLD + Colors.BRIGHT_MAGENTA))
        
        issues = []
        
        # Проверка обязательных полей
        if not flow.id:
            issues.append("ID не установлен")
        if not flow.name:
            issues.append("Name не установлен")
        if not flow.description:
            issues.append("Description не установлен")
        if not flow.agents:
            issues.append("Список agents пуст")
        if not flow.runner:
            issues.append("Runner не установлен")
        
        # Проверка runner
        if flow.runner:
            import inspect
            sig = inspect.signature(flow.runner)
            params = list(sig.parameters.keys())
            if "input_data" not in params:
                issues.append(f"Runner должен принимать параметр 'input_data', но имеет: {params}")
        
        # Проверка agents в registry
        try:
            from src.agents import AGENT_REGISTRY
            for agent in flow.agents:
                # Учитываем что могут быть варианты типа debater_pro, debater_con
                base_agent = agent.split("_")[0] if "_" in agent else agent
                if base_agent not in AGENT_REGISTRY:
                    issues.append(f"Agent '{agent}' не найден в AGENT_REGISTRY")
        except Exception as e:
            issues.append(f"Не удалось проверить agents: {e}")
        
        if issues:
            all_valid = False
            for issue in issues:
                print(warning(f"  {issue}"))
        else:
            print(success("  Все проверки пройдены"))
    
    print()
    if all_valid:
        print(color("🎉 Все flows валидны!", Colors.BOLD + Colors.BRIGHT_GREEN))
        return 0
    else:
        print(color("⚠️  Некоторые flows имеют проблемы", Colors.BOLD + Colors.BRIGHT_YELLOW))
        return 1


# ============================================================================
# Argument Parser
# ============================================================================

def build_parser() -> argparse.ArgumentParser:
    """Создать парсер аргументов."""
    parser = argparse.ArgumentParser(
        prog="python -m src.flows.debug",
        description="FlowStack Flows Debug Utilities",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
Примеры:
  python -m src.flows.debug list
  python -m src.flows.debug run research "AI in medicine"
  python -m src.flows.debug run code_review < code.py
  python -m src.flows.debug run code_review "$(cat code.py)" -o report.md --extract
  python -m src.flows.debug inspect debate
  python -m src.flows.debug test research
  python -m src.flows.debug extract code_review report.md -o structured.json
  python -m src.flows.debug stats
  python -m src.flows.debug validate
        """,
    )
    
    subparsers = parser.add_subparsers(dest="command", help="Команда")
    
    # list
    subparsers.add_parser("list", help="Список всех зарегистрированных flows")
    
    # run <id> <input>
    p_run = subparsers.add_parser("run", help="Запустить flow")
    p_run.add_argument("flow_id", help="ID flow (research, review, debate, code_review)")
    p_run.add_argument("input", nargs="?", help="Входные данные (или stdin)")
    p_run.add_argument("--context", "-c", help="Дополнительный контекст")
    p_run.add_argument("--language", "-l", help="Язык кода (для code_review)")
    p_run.add_argument("--focus", "-f", help="Фокус ревью")
    p_run.add_argument("--output", "-o", help="Сохранить markdown результат в файл")
    p_run.add_argument("--extract", "-e", action="store_true", 
                       help="Извлечь structured data")
    p_run.add_argument("--structured-output", "-s", 
                       help="Сохранить structured data в JSON файл")
    
    # inspect <id>
    p_inspect = subparsers.add_parser("inspect", help="Детальная информация о flow")
    p_inspect.add_argument("flow_id", help="ID flow")
    
    # test <id>
    p_test = subparsers.add_parser("test", help="Быстрый тест flow")
    p_test.add_argument("flow_id", help="ID flow")
    
    # extract <id> <file>
    p_extract = subparsers.add_parser("extract", help="Извлечь structured data")
    p_extract.add_argument("flow_id", help="ID flow")
    p_extract.add_argument("file", help="Markdown файл")
    p_extract.add_argument("--output", "-o", help="Сохранить JSON в файл")
    
    # stats
    subparsers.add_parser("stats", help="Статистика по всем flows")
    
    # validate
    subparsers.add_parser("validate", help="Валидация всех flows")
    
    return parser


# ============================================================================
# Command registry
# ============================================================================

COMMANDS: dict[str, Callable] = {
    "list": cmd_list,
    "run": cmd_run,
    "inspect": cmd_inspect,
    "test": cmd_test,
    "extract": cmd_extract,
    "stats": cmd_stats,
    "validate": cmd_validate,
}


# ============================================================================
# Main
# ============================================================================

async def main() -> int:
    """Точка входа."""
    # Загружаем .env если есть
    try:
        from dotenv import load_dotenv
        load_dotenv()
    except ImportError:
        pass
    
    # Настраиваем минимальный logging
    structlog.configure(
        processors=[
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.add_log_level,
            structlog.dev.ConsoleRenderer(),
        ],
        wrapper_class=structlog.make_filtering_bound_logger(20),  # INFO
    )
    
    parser = build_parser()
    args = parser.parse_args()
    
    if not args.command:
        parser.print_help()
        return 1
    
    handler = COMMANDS.get(args.command)
    if handler is None:
        print(error(f"Unknown command: {args.command}"))
        parser.print_help()
        return 1
    
    try:
        return await handler(args)
    except KeyboardInterrupt:
        print(warning("\nПрервано"))
        return 130
    except Exception as e:
        logger.exception("flows.debug.main.failed")
        print(error(f"Unexpected error: {e}"))
        return 1


if __name__ == "__main__":
    sys.exit(asyncio.run(main()))