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

core/engine/src/flows/extract.py
704 строки24 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:1976704c6
# core/engine/src/flows/extract.py
"""Модуль для извлечения структурированных данных из результатов flows.

Предоставляет функции для парсинга markdown ответов от мультиагентных flows
и преобразования их в структурированные JSON данные для использования в UI,
API и дальнейшей обработки.

Поддерживаемые flows:
- research: извлечение findings, insights, recommendations
- review: извлечение проблем, сильных сторон, оценки качества
- debate: извлечение вердикта, аргументов, анализа
- code_review: извлечение P0/P1/P2 issues, scores, roadmap

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

    from src.flows.extract import extract_flow_result

    # После завершения flow
    result = extract_flow_result(
        flow_id="code_review",
        output=writer_output,
        stages={"research": "...", "analysis": "...", ...},
        metadata={"language": "python", "code_length": 2450}
    )

    # Получаем структурированные данные
    print(result["scores"]["overall"])  # 7.5
    print(result["issues"]["p0"])       # [{"problem": "...", "solution": "..."}]
    print(result["recommendations"])    # ["...", "..."]
"""
from __future__ import annotations

import re
from typing import Any

import structlog

logger = structlog.get_logger()


# ============================================================================
# Regex patterns для парсинга markdown
# ============================================================================

# Паттерны для извлечения scores (1-10)
SCORE_PATTERN = re.compile(
    r"(?:Score|Оценка|Рейтинг)[:\s]*([0-9]+(?:\.[0-9]+)?)\s*/\s*10",
    re.IGNORECASE
)

OVERALL_SCORE_PATTERN = re.compile(
    r"(?:Overall|Общая|Итоговая)[\s\w]*(?:Score|Оценка)[:\s]*([0-9]+(?:\.[0-9]+)?)\s*/?\s*10?",
    re.IGNORECASE
)

# Паттерны для извлечения issues по приоритету
P0_PATTERN = re.compile(
    r"(?:🔴|P0|Critical|Критические?)[^\n]*\n((?:.*?\n)*?)(?=\n(?:🟡|P1|🔵|P2|⚪|P3|$))",
    re.MULTILINE
)

P1_PATTERN = re.compile(
    r"(?:🟡|P1|Major|Важные?)[^\n]*\n((?:.*?\n)*?)(?=\n(?:🔵|P2|⚪|P3|$))",
    re.MULTILINE
)

P2_PATTERN = re.compile(
    r"(?:🔵|P2|Minor|Улучшения?|Рекомендации?)[^\n]*\n((?:.*?\n)*?)(?=\n(?:⚪|P3|$))",
    re.MULTILINE
)

# Паттерн для извлечения effort (в часах)
EFFORT_PATTERN = re.compile(
    r"(?:Effort|Время|Трудозатраты)[:\s]*([0-9]+(?:\.[0-9]+)?)\s*(?:час|hour|h)",
    re.IGNORECASE
)

# Паттерн для извлечения вердикта
VERDICT_PATTERN = re.compile(
    r"(?:Решение|Verdict|Вердикт)[:\s]*(ЗА|ПРОТИВ|КОМПРОМИСС|PRO|CON|COMPROMISE)",
    re.IGNORECASE
)

# Паттерн для извлечения победителя дебатов
WINNER_PATTERN = re.compile(
    r"(?:Победитель|Winner)[:\s]*(ПРО|ЗА|CON|PRO|Ничья|Draw)",
    re.IGNORECASE
)

# Паттерн для извлечения списка рекомендаций
RECOMMENDATION_PATTERN = re.compile(
    r"^\s*[\d]+[\.\)]\s*(.+?)$",
    re.MULTILINE
)


# ============================================================================
# Основные функции извлечения
# ============================================================================

def extract_scores(markdown: str) -> dict[str, float | None]:
    """
    Извлекает все числовые оценки (scores) из markdown текста.

    Args:
        markdown: Markdown текст с оценками в формате "Score: X/10"

    Returns:
        Словарь с оценками:
        {
            "code_quality": 8.0,
            "architecture": 7.5,
            "maintainability": 6.0,
            "security": 9.0,
            "performance": 7.0,
            "overall": 7.5
        }
    """
    scores: dict[str, float | None] = {
        "code_quality": None,
        "architecture": None,
        "maintainability": None,
        "security": None,
        "performance": None,
        "overall": None,
    }

    # Извлекаем specific scores по ключевым словам
    score_mapping = {
        "code quality": "code_quality",
        "качество кода": "code_quality",
        "architecture": "architecture",
        "архитектура": "architecture",
        "maintainability": "maintainability",
        "поддерживаемость": "maintainability",
        "security": "security",
        "безопасность": "security",
        "performance": "performance",
        "производительность": "performance",
    }

    lines = markdown.split("\n")
    for line in lines:
        line_lower = line.lower()
        for keyword, score_key in score_mapping.items():
            if keyword in line_lower:
                match = SCORE_PATTERN.search(line)
                if match:
                    try:
                        scores[score_key] = float(match.group(1))
                    except (ValueError, IndexError):
                        pass
                break

    # Извлекаем overall score
    overall_match = OVERALL_SCORE_PATTERN.search(markdown)
    if overall_match:
        try:
            scores["overall"] = float(overall_match.group(1))
        except (ValueError, IndexError):
            pass

    # Если overall не найден, вычисляем как среднее
    if scores["overall"] is None:
        valid_scores = [s for s in scores.values() if s is not None]
        if valid_scores:
            scores["overall"] = round(sum(valid_scores) / len(valid_scores), 1)

    return scores


def extract_issues(markdown: str, priority: str = "all") -> list[dict[str, str]]:
    """
    Извлекает проблемы (issues) из markdown текста по приоритету.

    Args:
        markdown: Markdown текст с секциями P0/P1/P2
        priority: Приоритет для извлечения:
            - "p0" — только критические проблемы
            - "p1" — только важные проблемы
            - "p2" — только улучшения
            - "all" — все проблемы (по умолчанию)

    Returns:
        Список словарей с проблемами:
        [
            {
                "problem": "Описание проблемы",
                "impact": "Влияние на систему",
                "solution": "Решение",
                "effort": "4 часа"
            }
        ]
    """
    issues: list[dict[str, str]] = []

    # Выбираем паттерн по приоритету
    patterns = []
    if priority in ("p0", "all"):
        patterns.append(("P0", P0_PATTERN))
    if priority in ("p1", "all"):
        patterns.append(("P1", P1_PATTERN))
    if priority in ("p2", "all"):
        patterns.append(("P2", P2_PATTERN))

    for priority_label, pattern in patterns:
        matches = pattern.findall(markdown)
        for match in matches:
            # Парсим каждый issue в секции
            issue_blocks = match.strip().split("\n\n")
            for block in issue_blocks:
                if not block.strip():
                    continue

                issue: dict[str, str] = {"priority": priority_label}

                # Извлекаем problem
                problem_match = re.search(
                    r"(?:Проблема|Problem)[:\s]*(.+?)(?=\n|$)",
                    block,
                    re.IGNORECASE
                )
                if problem_match:
                    issue["problem"] = problem_match.group(1).strip()

                # Извлекаем impact
                impact_match = re.search(
                    r"(?:Влияние|Impact)[:\s]*(.+?)(?=\n|$)",
                    block,
                    re.IGNORECASE
                )
                if impact_match:
                    issue["impact"] = impact_match.group(1).strip()

                # Извлекаем solution
                solution_match = re.search(
                    r"(?:Решение|Solution)[:\s]*(.+?)(?=\n|$)",
                    block,
                    re.IGNORECASE
                )
                if solution_match:
                    issue["solution"] = solution_match.group(1).strip()

                # Извлекаем effort
                effort_match = EFFORT_PATTERN.search(block)
                if effort_match:
                    issue["effort"] = f"{effort_match.group(1)} часа"

                # Если нашли хотя бы problem, добавляем issue
                if "problem" in issue:
                    issues.append(issue)

    return issues


def extract_recommendations(markdown: str) -> list[str]:
    """
    Извлекает список рекомендаций из markdown текста.

    Ищет секции "Рекомендации", "План действий", "Roadmap" и извлекает
    нумерованные списки.

    Args:
        markdown: Markdown текст с рекомендациями

    Returns:
        Список строк с рекомендациями:
        [
            "Исправить P0 проблемы в первые 2 дня",
            "Провести рефакторинг модуля X",
            "Добавить unit-тесты для критических функций"
        ]
    """
    recommendations: list[str] = []

    # Ищем секции с рекомендациями
    rec_sections = [
        "Рекомендации",
        "План действий",
        "Roadmap",
        "Действия",
        "Recommendations",
        "Action Plan",
    ]

    lines = markdown.split("\n")
    in_rec_section = False

    for line in lines:
        # Проверяем, началась ли секция рекомендаций
        if any(section in line for section in rec_sections):
            in_rec_section = True
            continue

        # Если мы в секции рекомендаций, извлекаем нумерованные пункты
        if in_rec_section:
            # Проверяем, не началась ли новая секция
            if line.startswith("#") or (line.strip() and not line[0].isdigit()):
                in_rec_section = False
                continue

            # Извлекаем нумерованный пункт
            match = RECOMMENDATION_PATTERN.match(line)
            if match:
                recommendations.append(match.group(1).strip())

    return recommendations


def extract_verdict(markdown: str) -> dict[str, str | None]:
    """
    Извлекает вердикт из markdown текста (для debate flow).

    Args:
        markdown: Markdown текст с вердиктом

    Returns:
        Словарь с вердиктом:
        {
            "decision": "ЗА" | "ПРОТИВ" | "КОМПРОМИСС",
            "winner": "ПРО" | "КОН" | "Ничья",
            "reasoning": "Обоснование решения"
        }
    """
    verdict: dict[str, str | None] = {
        "decision": None,
        "winner": None,
        "reasoning": None,
    }

    # Извлекаем decision
    decision_match = VERDICT_PATTERN.search(markdown)
    if decision_match:
        verdict["decision"] = decision_match.group(1).upper()

    # Извлекаем winner
    winner_match = WINNER_PATTERN.search(markdown)
    if winner_match:
        winner = winner_match.group(1)
        # Нормализуем
        if winner in ("ПРО", "CON"):
            verdict["winner"] = "ПРОТИВ"
        elif winner in ("ЗА", "PRO"):
            verdict["winner"] = "ЗА"
        else:
            verdict["winner"] = winner

    # Извлекаем reasoning (текст после "Обоснование:")
    reasoning_match = re.search(
        r"(?:Обоснование|Reasoning)[:\s]*\n*([\s\S]+?)(?=\n#|\n\*\*|$)",
        markdown,
        re.IGNORECASE
    )
    if reasoning_match:
        verdict["reasoning"] = reasoning_match.group(1).strip()

    return verdict


def extract_findings(markdown: str) -> list[str]:
    """
    Извлекает ключевые находки (findings) из markdown текста (для research flow).

    Ищет секцию "Ключевые находки" или "Key Findings" и извлекает пункты.

    Args:
        markdown: Markdown текст с findings

    Returns:
        Список строк с находками
    """
    findings: list[str] = []

    # Ищем секцию findings
    findings_sections = [
        "Ключевые находки",
        "Key Findings",
        "Основные открытия",
        "Главные выводы",
    ]

    lines = markdown.split("\n")
    in_findings_section = False

    for line in lines:
        # Проверяем, началась ли секция findings
        if any(section in line for section in findings_sections):
            in_findings_section = True
            continue

        # Если мы в секции findings, извлекаем пункты
        if in_findings_section:
            # Проверяем, не началась ли новая секция
            if line.startswith("#") or (line.strip() and not line.startswith("-") and not line[0].isdigit()):
                in_findings_section = False
                continue

            # Извлекаем пункт списка
            if line.strip().startswith("-") or line.strip().startswith("*"):
                findings.append(line.strip().lstrip("-* ").strip())
            else:
                match = RECOMMENDATION_PATTERN.match(line)
                if match:
                    findings.append(match.group(1).strip())

    return findings


def extract_metadata(
    stages: dict[str, str],
    metadata: dict[str, Any]
) -> dict[str, Any]:
    """
    Извлекает и структурирует метаданные из flow результата.

    Args:
        stages: Словарь с outputs каждого этапа
        metadata: Словарь с метаданными от flow

    Returns:
        Структурированные метаданные:
        {
            "agents_count": 5,
            "total_stages": 5,
            "stage_lengths": {
                "research": 1250,
                "analysis": 1800,
                ...
            },
            "total_output_length": 5600,
            "language": "python",
            "code_length": 2450
        }
    """
    result: dict[str, Any] = {
        "agents_count": metadata.get("agents_count", 0),
        "total_stages": metadata.get("total_stages", 0),
        "stage_lengths": {},
        "total_output_length": 0,
    }

    # Вычисляем длины outputs каждого этапа
    for stage_name, stage_output in stages.items():
        if isinstance(stage_output, str):
            length = len(stage_output)
            result["stage_lengths"][stage_name] = length
            result["total_output_length"] += length

    # Копируем дополнительные метаданные
    for key in ("language", "code_length", "lines_count", "focus", "project_type", "topic"):
        if key in metadata:
            result[key] = metadata[key]

    return result


# ============================================================================
# Главная функция извлечения для каждого flow
# ============================================================================

def extract_flow_result(
    flow_id: str,
    output: str,
    stages: dict[str, str],
    metadata: dict[str, Any],
    tokens: int = 0,
) -> dict[str, Any]:
    """
    Главная функция для извлечения структурированных данных из результата flow.

    Автоматически определяет тип flow и применяет соответствующие парсеры.

    Args:
        flow_id: ID flow ("research", "review", "debate", "code_review")
        output: Финальный markdown output от flow
        stages: Словарь с outputs каждого этапа
        metadata: Словарь с метаданными от flow
        tokens: Общее количество использованных токенов

    Returns:
        Структурированный результат:
        {
            "flow_id": "code_review",
            "scores": {"overall": 7.5, ...},
            "issues": {"p0": [...], "p1": [...], "p2": [...]},
            "recommendations": [...],
            "metadata": {...},
            "tokens": 12500,
            "raw_output": "..."
        }

    Raises:
        ValueError: Если flow_id не поддерживается
    """
    logger.info(
        "extract.flow_result.started",
        flow_id=flow_id,
        output_length=len(output),
        stages_count=len(stages),
    )

    result: dict[str, Any] = {
        "flow_id": flow_id,
        "tokens": tokens,
        "raw_output": output,
        "metadata": extract_metadata(stages, metadata),
    }

    # Применяем специфичные парсеры в зависимости от flow_id
    if flow_id == "code_review":
        result["scores"] = extract_scores(output)
        result["issues"] = {
            "p0": extract_issues(output, "p0"),
            "p1": extract_issues(output, "p1"),
            "p2": extract_issues(output, "p2"),
        }
        result["recommendations"] = extract_recommendations(output)

    elif flow_id == "review":
        result["scores"] = extract_scores(output)
        result["issues"] = {
            "p0": extract_issues(output, "p0"),
            "p1": extract_issues(output, "p1"),
            "p2": extract_issues(output, "p2"),
        }
        result["recommendations"] = extract_recommendations(output)

    elif flow_id == "research":
        result["findings"] = extract_findings(output)
        result["recommendations"] = extract_recommendations(output)

    elif flow_id == "debate":
        result["verdict"] = extract_verdict(output)
        result["recommendations"] = extract_recommendations(output)

    else:
        logger.warning("extract.flow_result.unknown_flow_id", flow_id=flow_id)
        # Для неизвестных flows извлекаем только базовые данные
        result["recommendations"] = extract_recommendations(output)

    logger.info(
        "extract.flow_result.completed",
        flow_id=flow_id,
        extracted_keys=list(result.keys()),
    )

    return result


# ============================================================================
# Вспомогательные функции
# ============================================================================

def extract_code_examples(markdown: str) -> list[dict[str, str]]:
    """
    Извлекает примеры кода из markdown текста.

    Ищет code blocks в формате ```language\\ncode``` и извлекает их.

    Args:
        markdown: Markdown текст с примерами кода

    Returns:
        Список словарей с примерами:
        [
            {
                "language": "python",
                "code": "def example():\\n    pass",
                "context": "Контекст примера (если есть)"
            }
        ]
    """
    examples: list[dict[str, str]] = []

    # Паттерн для code blocks
    code_block_pattern = re.compile(
        r"```(\w+)?\n([\s\S]+?)```",
        re.MULTILINE
    )

    matches = code_block_pattern.findall(markdown)
    for language, code in matches:
        example = {
            "language": language or "text",
            "code": code.strip(),
        }
        examples.append(example)

    return examples


def extract_sections(markdown: str) -> dict[str, str]:
    """
    Разбивает markdown текст на секции по заголовкам.

    Args:
        markdown: Markdown текст с заголовками

    Returns:
        Словарь секций:
        {
            "Executive Summary": "...",
            "Сильные стороны": "...",
            "Критические проблемы": "..."
        }
    """
    sections: dict[str, str] = {}

    # Паттерн для заголовков
    heading_pattern = re.compile(r"^#{1,3}\s+(.+?)$", re.MULTILINE)

    lines = markdown.split("\n")
    current_section = "Introduction"
    current_content: list[str] = []

    for line in lines:
        heading_match = heading_pattern.match(line)
        if heading_match:
            # Сохраняем предыдущую секцию
            if current_content:
                sections[current_section] = "\n".join(current_content).strip()

            # Начинаем новую секцию
            current_section = heading_match.group(1).strip()
            current_content = []
        else:
            current_content.append(line)

    # Сохраняем последнюю секцию
    if current_content:
        sections[current_section] = "\n".join(current_content).strip()

    return sections


def summarize_result(result: dict[str, Any]) -> dict[str, Any]:
    """
    Создаёт краткую сводку результата flow.

    Args:
        result: Полный результат от extract_flow_result

    Returns:
        Краткая сводка:
        {
            "flow_id": "code_review",
            "overall_score": 7.5,
            "critical_issues": 2,
            "important_issues": 5,
            "recommendations_count": 8,
            "tokens_used": 12500
        }
    """
    summary: dict[str, Any] = {
        "flow_id": result.get("flow_id"),
        "tokens_used": result.get("tokens", 0),
    }

    # Добавляем overall score если есть
    if "scores" in result and "overall" in result["scores"]:
        summary["overall_score"] = result["scores"]["overall"]

    # Считаем количество issues по приоритетам
    if "issues" in result:
        summary["critical_issues"] = len(result["issues"].get("p0", []))
        summary["important_issues"] = len(result["issues"].get("p1", []))
        summary["minor_issues"] = len(result["issues"].get("p2", []))

    # Считаем количество recommendations
    if "recommendations" in result:
        summary["recommendations_count"] = len(result["recommendations"])

    # Для debate добавляем verdict
    if "verdict" in result:
        summary["verdict"] = result["verdict"].get("decision")

    # Для research добавляем findings count
    if "findings" in result:
        summary["findings_count"] = len(result["findings"])

    return summary


# ============================================================================
# Public API
# ============================================================================

__all__ = [
    # Главная функция
    "extract_flow_result",
    
    # Специфичные экстракторы
    "extract_scores",
    "extract_issues",
    "extract_recommendations",
    "extract_verdict",
    "extract_findings",
    "extract_metadata",
    
    # Вспомогательные функции
    "extract_code_examples",
    "extract_sections",
    "summarize_result",
]