src/workflow.py - master - v.chatkin/BKP

src/workflow.py
460 строк17 KB
v.chatkin
Create: Глава_3_Исторические_предпосылки.txt, app.py, main.py, README.md, requirements.txt, .gitkeep, input.md, final_report.md, leader_brief.md, critic_raw.md, expert_3.md, expert_2.md, expert_1.md, critic_parsed.json, expert_5.md, execution.log, expert_4.md, workflow_result.json, config_snapshot.json, execution.log, critic_raw.md, expert_3.md, expert_2.md, expert_1.md, critic_parsed.json, expert_5.md, execution.log, expert_4.md, input.md, final_report.md, leader_brief.md, critic_raw.md, expert_3.md, expert_2.md, expert_1.md, critic_parsed.json, expert_5.md, execution.log, expert_4.md, workflow_result.json, config_snapshot.json, execution.log, critic_raw.md, expert_3.md, expert_2.md, expert_1.md, critic_parsed.json, expert_5.md, execution.log, expert_4.md, config.py, sber_llm.py, models.py, llm_router.py, __init__.py, local_llm.py, utils.py, deepseek_llm.py, autogen_review.py, workflow.py, placeholders.py
10 июн 2026, 12:35
Верифицирован
10 июн 2026, 12:355cfbb91
from __future__ import annotations

from pathlib import Path
from typing import Dict, List, Optional

from .autogen_review import CriticReviewAdapter
from .config import AppConfig
from .llm_router import LLMRouter
from .models import (
    CriticResult,
    ExpertAnswer,
    IterationResult,
    LLMRequest,
    UserInput,
    WorkflowResult,
)
from .placeholders import (
    CRITIC_PLACEHOLDER,
    EDITOR_PLACEHOLDER,
    LEADER_PLACEHOLDER,
    get_expert_placeholder,
)
from .utils import (
    append_log,
    build_input_markdown,
    extract_per_expert_feedback,
    extract_stop_signal,
    make_run_dir,
    save_json,
    save_text,
    validate_critic_technical_format,
)


class SituationAnalysisWorkflow:
    """Последовательный workflow ситуационного анализа."""

    def __init__(self, config: AppConfig) -> None:
        self.config = config
        self.router = LLMRouter(config)
        self.critic_adapter = CriticReviewAdapter(self.router)

    def run(self, user_input: UserInput) -> WorkflowResult:
        self._validate_input(user_input)

        run_dir = make_run_dir(self.config.runs_dir, user_input.input_type)
        warnings: List[str] = []
        append_log(run_dir, "Workflow started")

        save_text(
            run_dir / "input.md",
            build_input_markdown(
                user_input.topic, user_input.input_type, user_input.situation
            ),
        )
        save_json(run_dir / "config_snapshot.json", self.config)

        leader_brief = self._run_leader(user_input=user_input, run_dir=run_dir)

        latest_expert_answers: Dict[str, str] = {}
        previous_feedback: Dict[str, str] = {}
        iterations: List[IterationResult] = []

        for iteration in range(1, self.config.max_iterations + 1):
            append_log(run_dir, f"Iteration {iteration} started")
            iteration_dir = run_dir / f"iteration_{iteration:02d}"
            iteration_dir.mkdir(parents=True, exist_ok=True)

            expert_answers = self._run_experts(
                user_input=user_input,
                leader_brief=leader_brief,
                iteration=iteration,
                run_dir=iteration_dir,
                latest_expert_answers=latest_expert_answers,
                previous_feedback=previous_feedback,
            )

            for expert_id, answer in expert_answers.items():
                latest_expert_answers[expert_id] = answer.text

            critic_result = self._run_critic(
                user_input=user_input,
                leader_brief=leader_brief,
                iteration=iteration,
                current_answers=latest_expert_answers,
                previous_feedback=previous_feedback,
                run_dir=iteration_dir,
                warnings=warnings,
            )

            iterations.append(
                IterationResult(
                    iteration=iteration,
                    expert_answers=expert_answers,
                    critic_result=critic_result,
                )
            )

            if critic_result.stop_signal is True:
                append_log(
                    run_dir, f"Iteration stopped by critic at iteration {iteration}"
                )
                break

            if critic_result.stop_signal is None:
                warning = (
                    f"Критика на итерации {iteration} не удалось технически разобрать. "
                    "Итерационный цикл завершен, редактор получит последние доступные материалы."
                )
                warnings.append(warning)
                append_log(run_dir, warning)
                break

            previous_feedback = critic_result.per_expert_feedback
            append_log(
                run_dir, f"Iteration {iteration} finished; next iteration required"
            )

        final_report = self._run_editor(
            user_input=user_input,
            leader_brief=leader_brief,
            latest_expert_answers=latest_expert_answers,
            iterations=iterations,
            run_dir=run_dir,
        )

        result = WorkflowResult(
            topic=user_input.topic,
            input_type=user_input.input_type,
            run_dir=str(run_dir),
            leader_brief=leader_brief,
            final_report=final_report,
            iterations=iterations,
            warnings=warnings,
        )
        save_json(run_dir / "workflow_result.json", result)
        append_log(run_dir, "Workflow finished")
        return result

    @staticmethod
    def _validate_input(user_input: UserInput) -> None:
        if not user_input.topic or not user_input.topic.strip():
            raise ValueError("Параметр topic обязателен и не может быть пустым.")

    def _run_leader(self, user_input: UserInput, run_dir: Path) -> str:
        append_log(run_dir, "Leader stage started")
        request = LLMRequest(
            role="leader",
            role_id="leader",
            system_prompt=LEADER_PLACEHOLDER,
            user_prompt=self._build_leader_user_prompt(user_input),
        )
        response = self.router.generate(request)
        save_text(run_dir / "leader_brief.md", response.text)
        append_log(run_dir, f"Leader stage finished via backend={response.backend}")
        return response.text

    def _run_experts(
        self,
        user_input: UserInput,
        leader_brief: str,
        iteration: int,
        run_dir: Path,
        latest_expert_answers: Dict[str, str],
        previous_feedback: Dict[str, str],
    ) -> Dict[str, ExpertAnswer]:
        answers: Dict[str, ExpertAnswer] = {}

        for expert_id in self.config.expert_ids:
            append_log(
                run_dir, f"Expert stage started: {expert_id}, iteration={iteration}"
            )
            request = LLMRequest(
                role="expert",
                role_id=expert_id,
                system_prompt=get_expert_placeholder(expert_id),
                user_prompt=self._build_expert_user_prompt(
                    user_input=user_input,
                    leader_brief=leader_brief,
                    expert_id=expert_id,
                    iteration=iteration,
                    previous_answer=latest_expert_answers.get(expert_id),
                    critic_feedback=previous_feedback.get(expert_id),
                ),
            )

            try:
                response = self.router.generate(request)
                text = response.text
                error = None
                append_log(
                    run_dir,
                    f"Expert stage finished: {expert_id}, backend={response.backend}",
                )
            except (
                Exception
            ) as exc:  # noqa: BLE001 - здесь важно сохранить ошибку и продолжить workflow.
                error = str(exc)
                text = f"[ERROR] {expert_id} did not return an answer. Error: {error}"
                append_log(run_dir, f"Expert stage error: {expert_id}: {error}")

            answer = ExpertAnswer(
                expert_id=expert_id,
                iteration=iteration,
                text=text,
                error=error,
            )
            answers[expert_id] = answer
            save_text(run_dir / f"{expert_id}.md", text)

        return answers

    def _run_critic(
        self,
        user_input: UserInput,
        leader_brief: str,
        iteration: int,
        current_answers: Dict[str, str],
        previous_feedback: Dict[str, str],
        run_dir: Path,
        warnings: List[str],
    ) -> CriticResult:
        append_log(run_dir, f"Critic stage started, iteration={iteration}")
        request = LLMRequest(
            role="critic",
            role_id="critic",
            system_prompt=CRITIC_PLACEHOLDER,
            user_prompt=self._build_critic_user_prompt(
                user_input=user_input,
                leader_brief=leader_brief,
                current_answers=current_answers,
                previous_feedback=previous_feedback,
            ),
        )

        response = self.critic_adapter.review(request)
        raw_text = response.text
        save_text(run_dir / "critic_raw.md", raw_text)

        parse_error = validate_critic_technical_format(raw_text, self.config.expert_ids)
        if parse_error:
            append_log(run_dir, f"Critic technical parse error: {parse_error}")
            raw_text = self._retry_critic_format(
                original_text=raw_text,
                original_request=request,
                run_dir=run_dir,
                warnings=warnings,
            )
            parse_error = validate_critic_technical_format(
                raw_text, self.config.expert_ids
            )

        stop_signal = extract_stop_signal(raw_text)
        feedback = extract_per_expert_feedback(raw_text, self.config.expert_ids)

        result = CriticResult(
            iteration=iteration,
            raw_text=raw_text,
            stop_signal=stop_signal,
            per_expert_feedback=feedback,
            parse_error=parse_error,
        )
        save_json(run_dir / "critic_parsed.json", result)
        append_log(
            run_dir,
            f"Critic stage finished, stop_signal={stop_signal}, backend={response.backend}",
        )
        return result

    def _retry_critic_format(
        self,
        original_text: str,
        original_request: LLMRequest,
        run_dir: Path,
        warnings: List[str],
    ) -> str:
        append_log(run_dir, "Critic format retry started")
        retry_prompt = (
            "Предыдущий ответ критика невозможно технически разобрать для workflow.\n"
            "Нужно вернуть тот же смысл, но строго в минимальном служебном формате:\n\n"
            "# SUMMARY\n...\n\n"
            "# PER_EXPERT_FEEDBACK\n\n"
            + "\n\n".join(
                f"## {expert_id}\n..." for expert_id in self.config.expert_ids
            )
            + "\n\n# STOP_SIGNAL\ntrue или false\n\n"
            "Предыдущий ответ критика:\n\n"
            f"{original_text}"
        )
        retry_request = LLMRequest(
            role="critic",
            role_id="critic",
            system_prompt=original_request.system_prompt,
            user_prompt=retry_prompt,
        )

        try:
            retry_response = self.critic_adapter.review(retry_request)
            save_text(run_dir / "critic_retry_raw.md", retry_response.text)
            append_log(
                run_dir,
                f"Critic format retry finished via backend={retry_response.backend}",
            )
            return retry_response.text
        except Exception as exc:  # noqa: BLE001
            warning = f"Повторный запрос к критику завершился ошибкой: {exc}"
            warnings.append(warning)
            append_log(run_dir, warning)
            return original_text

    def _run_editor(
        self,
        user_input: UserInput,
        leader_brief: str,
        latest_expert_answers: Dict[str, str],
        iterations: List[IterationResult],
        run_dir: Path,
    ) -> str:
        append_log(run_dir, "Editor stage started")
        request = LLMRequest(
            role="editor",
            role_id="editor",
            system_prompt=EDITOR_PLACEHOLDER,
            user_prompt=self._build_editor_user_prompt(
                user_input=user_input,
                leader_brief=leader_brief,
                latest_expert_answers=latest_expert_answers,
                iterations=iterations,
            ),
        )
        response = self.router.generate(request)
        save_text(run_dir / "final_report.md", response.text)
        append_log(run_dir, f"Editor stage finished via backend={response.backend}")
        return response.text

    @staticmethod
    def _build_leader_user_prompt(user_input: UserInput) -> str:
        situation = user_input.situation or "Не указано."
        return (
            f"Тип входа: {user_input.input_type}\n\n"
            f"Тема анализа:\n{user_input.topic}\n\n"
            f"Дополнительное описание ситуации:\n{situation}\n"
        )

    @staticmethod
    def _build_expert_user_prompt(
        user_input: UserInput,
        leader_brief: str,
        expert_id: str,
        iteration: int,
        previous_answer: Optional[str],
        critic_feedback: Optional[str],
    ) -> str:
        previous = (
            previous_answer or "Отсутствует: это первая итерация для данного эксперта."
        )
        feedback = critic_feedback or "Отсутствуют."
        return (
            f"Эксперт: {expert_id}\n"
            f"Итерация: {iteration}\n\n"
            f"Тема анализа:\n{user_input.topic}\n\n"
            "Бриф руководителя как основная аналитическая рамка:\n"
            f"{leader_brief}\n\n"
            "Предыдущий ответ этого эксперта:\n"
            f"{previous}\n\n"
            "Замечания критика по этому эксперту:\n"
            f"{feedback}\n"
        )

    @staticmethod
    def _build_critic_user_prompt(
        user_input: UserInput,
        leader_brief: str,
        current_answers: Dict[str, str],
        previous_feedback: Dict[str, str],
    ) -> str:
        situation = user_input.situation or "Не указано."
        answers_block = "\n\n".join(
            f"## {expert_id}\n{answer}" for expert_id, answer in current_answers.items()
        )
        previous_feedback_block = (
            "\n\n".join(
                f"## {expert_id}\n{feedback}"
                for expert_id, feedback in previous_feedback.items()
            )
            or "Отсутствует."
        )

        return (
            f"Тип входа: {user_input.input_type}\n\n"
            f"Тема анализа:\n{user_input.topic}\n\n"
            f"Дополнительное описание ситуации:\n{situation}\n\n"
            "Бриф руководителя:\n"
            f"{leader_brief}\n\n"
            "Предыдущие замечания критика, если были:\n"
            f"{previous_feedback_block}\n\n"
            "Текущий пакет экспертных ответов:\n"
            f"{answers_block}\n"
        )

    @staticmethod
    def _build_editor_user_prompt(
        user_input: UserInput,
        leader_brief: str,
        latest_expert_answers: Dict[str, str],
        iterations: List[IterationResult],
    ) -> str:
        situation = user_input.situation or "Не указано."

        # Сборка всех версий ответов экспертов по итерациям
        all_versions_block_parts = []

        # Группируем по экспертам для показа эволюции
        all_experts = set()
        for iteration in iterations:
            all_experts.update(iteration.expert_answers.keys())
        all_experts.update(latest_expert_answers.keys())

        for expert_id in sorted(all_experts):
            expert_section = [f"### {expert_id}"]
            for iteration in iterations:
                if expert_id in iteration.expert_answers:
                    answer = iteration.expert_answers[expert_id]
                    expert_section.append(f"**Итерация {answer.iteration}:**")
                    if answer.error:
                        expert_section.append(f"ОШИБКА: {answer.error}")
                    else:
                        expert_section.append(answer.text)
                    expert_section.append("")
            all_versions_block_parts.append("\n".join(expert_section))

        all_versions_block = (
            "\n\n".join(all_versions_block_parts) or "Нет доступных версий."
        )

        latest_answers_block = "\n\n".join(
            f"## {expert_id}\n{answer}"
            for expert_id, answer in latest_expert_answers.items()
        )
        critic_history_block = (
            "\n\n".join(
                f"## Итерация {item.iteration}\n{item.critic_result.raw_text}"
                for item in iterations
            )
            or "Критик не дал доступных результатов."
        )

        return (
            f"Тип входа: {user_input.input_type}\n\n"
            f"Тема анализа:\n{user_input.topic}\n\n"
            f"Дополнительное описание ситуации:\n{situation}\n\n"
            "Бриф руководителя как контекст и рамка:\n"
            f"{leader_brief}\n\n"
            "Последние версии экспертных ответов. Использовать первоочередно:\n"
            f"{latest_answers_block}\n\n"
            "История всех версий экспертных ответов по итерациям. Учитывать эволюцию мнений:\n"
            f"{all_versions_block}\n\n"
            "История замечаний и выводов критика по итерациям. Использовать первоочередно:\n"
            f"{critic_history_block}\n"
        )