/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/flows/analyze.py
1 236 строк
46 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
# core/engine/src/flows/analyze.py """Analyze flow — мультиагентный глубокий анализ ситуации, данных или проблемы. Последовательность агентов: 1. PlannerAgent — определяет scope анализа и методологию 2. ResearcherAgent — собирает факты, данные и контекст 3. AnalystAgent — проводит root cause analysis и выявляет паттерны 4. CriticAgent — проверяет на когнитивные искажения и слепые пятна 5. AnalystAgent (2-й) — строит прогнозы и сценарии развития 6. WriterAgent — формирует финальный аналитический отчёт Философия flow: В отличие от research_flow (сбор информации о теме), analyze_flow фокусируется на **глубоком анализе конкретной ситуации** с целью понять корневые причины, выявить паттерны и построить прогнозы. Использует классические методики анализа: - **5 Whys** — поиск корневых причин - **Fishbone Diagram** — выявление всех факторов - **Pattern Recognition** — поиск повторяющихся паттернов - **Scenario Planning** — построение сценариев развития - **Data-Driven Insights** — выводы на основе данных Идеально подходит для: - Анализ бизнес-ситуаций и метрик - Root cause analysis инцидентов - Анализ трендов и паттернов - Диагностика проблем в процессах - Подготовка к принятию стратегических решений - Пост-мортем анализ проектов Отличие от других flows: - research_flow: "что известно о X" (сбор информации) - analyze_flow: "почему происходит X, что будет дальше" (глубокий анализ) - compare_flow: "какой вариант лучше" (сравнение) - brainstorm_flow: "как решить X" (генерация идей) Особенности реализации: - Специализированный root cause analysis (5 Whys + Fishbone) - Двойной проход AnalystAgent (анализ + прогнозы) - CriticAgent для выявления когнитивных искажений в анализе - Сценарный анализ (оптимистичный/пессимистичный/базовый) - Data-driven подход с конкретными метриками События SSE (streaming): - agent_start: агент начал работу - agent_message: агент генерирует контент (стриминг) - agent_done: агент завершил работу - error: ошибка в одном из агентов - flow_done: финальное событие с аналитическим отчётом """ from __future__ import annotations from typing import Any, AsyncIterator import structlog from src.agents import ( PlannerAgent, ResearcherAgent, AnalystAgent, CriticAgent, WriterAgent, ) from src.primitives.flow import Flow logger = structlog.get_logger() # ============================================================================ # Specialized system prompts # ============================================================================ PLANNER_ANALYZE_PROMPT = """Ты — эксперт по планированию аналитических исследований. Твоя задача — разработать **методологию глубокого анализа** конкретной ситуации, проблемы или набора данных. ## Твоя роль: - Определить scope и границы анализа - Выбрать подходящие аналитические фреймворки - Определить ключевые вопросы для ответа - Задать метрики успеха анализа ## Принципы работы: 1. **Чёткий scope** — что анализируем и что НЕ анализируем 2. **Right frameworks** — подходящие методики для конкретного типа анализа 3. **Измеримость** — какие данные/метрики нужны 4. **Actionability** — анализ должен вести к действиям ## Формат ответа: ### 🎯 Scope анализа **Объект анализа:** (что именно анализируем) **Границы:** (что включено/исключено) **Цель анализа:** (какие вопросы должны ответить) ### 📋 Ключевые вопросы для анализа 1. **Дескриптивные:** Что происходит? (факты) 2. **Диагностические:** Почему это происходит? (причины) 3. **Предиктивные:** Что будет дальше? (прогнозы) 4. **Прескриптивные:** Что делать? (рекомендации) ### 🛠️ Аналитические фреймворки Для каждого фреймворка: - **Название:** (5 Whys, Fishbone, SWOT, PESTEL, etc.) - **Применение:** (для какого аспекта анализа) - **Ожидаемый результат:** (что получим) ### 📊 Необходимые данные и метрики - **Количественные:** (числа, статистика) - **Качественные:** (интервью, наблюдения) - **Источники:** (где брать) ### ✅ Критерии успешного анализа (Когда считать анализ завершённым и качественным) ### ⚠️ Типичные ловушки (Чего избегать при анализе) Отвечай на русском языке. Создай практичную методологию.""" RESEARCHER_FACTS_PROMPT = """Ты — аналитик-исследователь, специализирующийся на **сборе фактов и данных** для последующего глубокого анализа. Твоя задача — собрать **объективную фактологическую базу** без интерпретаций и выводов. Только факты, данные, контекст. ## Твоя роль: - Собирать релевантные факты и данные - Описывать текущую ситуацию объективно - Фиксировать контекст и ограничения - Выявлять информационные пробелы ## Принципы работы: 1. **Объективность** — только факты, без интерпретаций 2. **Полнота** — охватить все релевантные аспекты 3. **Верифицируемость** — указывать источники 4. **Структурированность** — чёткая организация данных ## Формат ответа: ### 📊 Фактологическая база #### Текущая ситуация (Объективное описание того, что происходит) #### Ключевые метрики и данные | Метрика | Значение | Тренд | Источник | |---------|----------|-------|----------| #### Хронология событий (Если применимо — timeline ключевых событий) #### Контекст и ограничения - **Внутренние факторы:** ... - **Внешние факторы:** ... - **Ограничения:** ... #### Stakeholders (Кто вовлечён и как) ### 🔍 Информационные пробелы (Каких данных не хватает для полного анализа) ### 📚 Источники информации (Откуда получены данные) ### ❓ Вопросы для уточнения (Что нужно прояснить перед глубоким анализом) Отвечай на русском языке. Будь максимально объективным и фактологичным. Не делай выводов — только собирай факты.""" ANALYST_ROOT_CAUSE_PROMPT = """Ты — эксперт по **root cause analysis** и выявлению паттернов. Твоя задача — провести глубокий диагностический анализ, выявить **корневые причины** происходящего и найти **повторяющиеся паттерны**. ## Твоя роль: - Применять 5 Whys для поиска корневых причин - Строить Fishbone (Ishikawa) диаграмму факторов - Выявлять системные паттерны и петли обратной связи - Находить leverage points (точки воздействия) ## Методологии: ### 5 Whys Для каждой проблемы задавай "почему?" 5 раз, добираясь до корня. ### Fishbone (Ishikawa) Категоризируй причины по 6M: - **Man** (люди) - **Machine** (технологии) - **Method** (процессы) - **Material** (ресурсы) - **Measurement** (метрики) - **Mother Nature** (внешняя среда) ### Systems Thinking Ищи reinforcing и balancing loops в системе. ## Формат ответа: ### 🎯 Root Cause Analysis (5 Whys) **Проблема:** (формулировка) **Why 1:** Почему это происходит? → **Ответ 1:** ... **Why 2:** Почему [Ответ 1]? → **Ответ 2:** ... **Why 3:** Почему [Ответ 2]? → **Ответ 3:** ... **Why 4:** Почему [Ответ 3]? → **Ответ 4:** ... **Why 5:** Почему [Ответ 4]? → **Корневая причина:** ... (Повтори для 2-3 ключевых проблем) ### 🐟 Fishbone Analysis | Категория | Факторы | |-----------|---------| | **People** | ... | | **Process** | ... | | **Technology** | ... | | **Resources** | ... | | **Metrics** | ... | | **Environment** | ... | ### 🔄 Системные паттерны и петли **Reinforcing loops** (усиливающие): - **R1:** [описание петли] - **R2:** ... **Balancing loops** (балансирующие): - **B1:** [описание петли] - **B2:** ... ### 🎯 Leverage Points (Где малое воздействие даёт большой эффект) 1. **High leverage:** ... 2. **Medium leverage:** ... 3. **Low leverage:** ... ### 📊 Ключевые инсайты (Что стало понятно после анализа) ### ⚠️ Системные архетипы (Какие классические системные паттерны наблюдаются) - "Limits to Growth": ... - "Shifting the Burden": ... - "Tragedy of the Commons": ... Отвечай на русском языке. Будь глубоким и системным в анализе.""" CRITIC_BLIND_SPOTS_PROMPT = """Ты — критик анализа, специализирующийся на выявлении **когнитивных искажений, слепых пятен и логических ошибок** в проведённом анализе. Твоя задача — подвергнуть сомнению выводы и найти то, что **упущено или искажено** в анализе. ## Твоя роль: - Выявлять когнитивные искажения в анализе - Находить слепые пятна и непроверенные предположения - Ставить под сомнение причинно-следственные связи - Искать альтернативные объяснения ## Когнитивные искажения для проверки: ### Confirmation Bias Ищем только подтверждающие данные? ### Survivorship Bias Анализируем только "выживших" (успешные кейсы)? ### Anchoring Привязаны к первому впечатлению/числу? ### Availability Heuristic Оцениваем по тому, что легко вспомнить? ### Correlation ≠ Causation Путаем корреляцию с причинностью? ### Hindsight Bias "Я так и знал" после события? ### Fundamental Attribution Error Приписываем поведение личности, а не ситуации? ## Формат ответа: ### 🎭 Выявленные когнитивные искажения **1. [Название искажения]** - **Где проявляется:** (в какой части анализа) - **Как искажает:** (какое влияние на выводы) - **Как исправить:** (конкретная рекомендация) **2. [Название искажения]** - ... ### 🕳️ Слепые пятна анализа (Что не было учтено) 1. **[Слепое пятно]:** (описание) - **Почему важно:** ... - **Как проверить:** ... 2. **[Слепое пятно]:** ... ### ❓ Непроверенные предположения (Что принято как данность без проверки) | Предположение | Вероятность истинности | Как проверить | |---------------|------------------------|---------------| ### 🔄 Альтернативные объяснения (Какие ещё причины могут объяснять наблюдаемое) **Объяснение 1:** [описание] - **Обоснование:** ... - **Вероятность:** X% **Объяснение 2:** ... ### ⚠️ Логические ошибки (Где нарушена логика рассуждений) ### 🎯 Рекомендации по усилению анализа (Что нужно сделать для повышения качества) Отвечай на русском языке. Будь конструктивным скептиком.""" ANALYST_FORECAST_PROMPT = """Ты — эксперт по **прогнозированию и сценарному анализу**. Твоя задача — на основе проведённого анализа построить **прогнозы развития ситуации** и разработать **сценарии будущего**. ## Твоя роль: - Строить прогнозы с указанием вероятностей - Разрабатывать сценарии (оптимистичный/базовый/пессимистичный) - Определять ранние индикаторы (leading indicators) - Оценивать риски и возможности ## Методологии: ### Scenario Planning Три сценария: оптимистичный, базовый, пессимистичный. ### Leading vs Lagging Indicators - **Leading:** предсказывают будущее - **Lagging:** отражают прошлое ### Probability-Weighted Forecasting Прогнозы с указанием вероятностей. ## Формат ответа: ### 🔮 Базовый прогноз (наиболее вероятный, ~60%) **Горизонт 1 (1-3 месяца):** - Что произойдёт: ... - Ключевые драйверы: ... - Вероятность: X% **Горизонт 2 (3-6 месяцев):** - ... **Горизонт 3 (6-12 месяцев):** - ... ### 📊 Сценарный анализ #### 🌟 Оптимистичный сценарий (вероятность ~20%) **Триггеры:** (что должно произойти) - ... **Развитие:** - Месяц 1-3: ... - Месяц 3-6: ... - Месяц 6-12: ... **Итоговый результат:** ... #### 📈 Базовый сценарий (вероятность ~60%) **Триггеры:** ... **Развитие:** ... **Итоговый результат:** ... #### ⚠️ Пессимистичный сценарий (вероятность ~20%) **Триггеры:** ... **Развитие:** ... **Итоговый результат:** ... ### 📈 Ключевые метрики для мониторинга #### Leading Indicators (предсказывают) | Метрика | Текущее значение | Порог тревоги | Что означает | |---------|------------------|---------------|--------------| #### Lagging Indicators (отражают результат) | Метрика | Текущее значение | Целевое значение | |---------|------------------|------------------| ### ⚡ Ранние индикаторы изменений (Что сигнализирует о смене тренда) 1. **[Индикатор]:** (описание) - **Что означает:** ... - **Как отслеживать:** ... ### 🎯 Риски и возможности #### ⚠️ Ключевые риски | Риск | Вероятность | Влияние | Митигация | |------|-------------|---------|-----------| #### 🌟 Возможности | Возможность | Вероятность | Потенциал | Как использовать | |-------------|-------------|-----------|------------------| ### 🔄 Точки бифуркации (Моменты, когда система может пойти по разным путям) 1. **[Точка]:** (когда и что решается) - **Вариант A:** ... (вероятность X%) - **Вариант B:** ... (вероятность Y%) ### 📅 Timeline ключевых событий (Когда ожидать изменений) Отвечай на русском языке. Будь реалистичным в прогнозах и указывай вероятности.""" WRITER_ANALYSIS_PROMPT = """Ты — технический писатель, специализирующийся на создании **аналитических отчётов** для принятия решений. Твоя задача — синтезировать весь процесс анализа в единый **профессиональный отчёт**, готовый к использованию. ## Формат итогового отчёта: ### 📋 Executive Summary (3-5 предложений: суть ситуации, ключевые findings, главная рекомендация) ### 🎯 Объект и scope анализа **Что анализировали:** ... **Границы анализа:** ... **Цель анализа:** ... ### 📊 Текущая ситуация (факты) (Объективное описание без интерпретаций) ### 🔍 Root Cause Analysis **Основная проблема:** ... **5 Whys анализ:** 1. Why? → ... 2. Why? → ... 3. Why? → ... 4. Why? → ... 5. **Корневая причина:** ... **Fishbone факторы:** - People: ... - Process: ... - Technology: ... - ... ### 🔄 Системные паттерны (Выявленные reinforcing/balancing loops) ### 🎯 Leverage Points (Где воздействие даёт максимальный эффект) 1. **High leverage:** ... 2. **Medium leverage:** ... ### 🔮 Прогнозы и сценарии #### Базовый сценарий (60% вероятность) - **Горизонт 1-3 мес:** ... - **Горизонт 3-6 мес:** ... - **Горизонт 6-12 мес:** ... #### Альтернативные сценарии - **Оптимистичный (20%):** ... - **Пессимистичный (20%):** ... ### ⚠️ Риски и возможности **Топ-3 риска:** 1. ... 2. ... 3. ... **Топ-3 возможности:** 1. ... 2. ... 3. ... ### 💡 Ключевые инсайты (Что стало понятно в результате анализа) 1. **Инсайт 1:** ... 2. **Инсайт 2:** ... 3. **Инсайт 3:** ... ### 🎯 Рекомендации к действию #### Немедленные действия (0-2 недели) 1. ... 2. ... #### Краткосрочные (1-3 месяца) 1. ... 2. ... #### Среднесрочные (3-6 месяцев) 1. ... ### 📈 Метрики для мониторинга **Leading indicators:** - ... **Lagging indicators:** - ... ### ⚠️ Ограничения анализа (Что не было учтено, где нужна дополнительная информация) ### 🔄 Следующие шаги 1. ... 2. ... ### 📚 Источники и дополнительная информация Отвечай на русском языке. Создай документ уровня executive report, готовый к использованию для принятия решений.""" # ============================================================================ # Helper функция для унифицированного запуска агентов # ============================================================================ async def _run_agent_stream( agent, agent_id: str, input_data: dict[str, Any], ) -> AsyncIterator[tuple[str, str, int, str]]: """ Запускает агента и yield'ит стандартизированные события. Yields: Кортежи (event_type, content, tokens, model) где event_type ∈ {"content", "done", "error"} """ full_output = "" total_tokens = 0 model_name = "" async for chunk in agent.run_stream(input_data): chunk_type = chunk.get("type") if chunk_type == "content": content = chunk.get("content", "") full_output += content yield ("content", content, 0, "") elif chunk_type == "done": total_tokens = chunk.get("tokens_total", 0) model_name = chunk.get("model", "") yield ("done", full_output, total_tokens, model_name) elif chunk_type == "error": error_msg = chunk.get("error", f"Unknown error in {agent_id}") yield ("error", error_msg, 0, "") return if full_output and total_tokens == 0: yield ("done", full_output, 0, "") # ============================================================================ # Main runner # ============================================================================ async def run_analyze_flow(input_data: dict[str, Any]) -> AsyncIterator[dict[str, Any]]: """ Выполняет мультиагентный analyze flow. Flow состоит из шести последовательных шагов: 1. PlannerAgent — методология анализа 2. ResearcherAgent — сбор фактов и данных 3. AnalystAgent — root cause analysis и паттерны 4. CriticAgent — проверка на искажения и слепые пятна 5. AnalystAgent (2-й) — прогнозы и сценарии 6. WriterAgent — финальный аналитический отчёт Args: input_data: Входные данные. Поддерживает поля: - 'topic' / 'input' / 'situation': что анализировать - 'context': дополнительный контекст - 'data': имеющиеся данные (опционально) - 'questions': конкретные вопросы для ответа (опционально) - 'timeframe': горизонт анализа (опционально) Yields: События SSE для UI: - agent_start: агент начал работу - agent_message: агент генерирует контент (стриминг) - agent_done: агент завершил работу - error: ошибка в одном из агентов - flow_done: финальное событие с аналитическим отчётом """ # ======================================================================== # 0. Валидация входных данных # ======================================================================== situation = ( input_data.get("topic") or input_data.get("input") or input_data.get("situation") or "" ).strip() if not situation: yield { "type": "error", "error": "Не указана ситуация для анализа (ожидается поле 'topic', 'input' или 'situation')", } return context = input_data.get("context", "").strip() data = input_data.get("data", "").strip() questions = input_data.get("questions", "").strip() timeframe = input_data.get("timeframe", "").strip() logger.info( "analyze_flow.started", situation=situation[:100], has_context=bool(context), has_data=bool(data), has_questions=bool(questions), timeframe=timeframe or "default", ) # Контейнер для накопления результатов каждого этапа stage_outputs: dict[str, Any] = { "situation": situation, "context": context, "data": data, "questions": questions, "timeframe": timeframe, } total_tokens = 0 # ======================================================================== # Шаг 1: PlannerAgent — методология анализа # ======================================================================== yield {"type": "agent_start", "agent": "planner"} planner = PlannerAgent(name="Analysis Planner") planner_input = { "topic": f"""Разработай методологию глубокого анализа ситуации: ## СИТУАЦИЯ ДЛЯ АНАЛИЗА: \"\"\" {situation} \"\"\" {f"Контекст: {context}" if context else ""} {f"Имеющиеся данные: {data}" if data else ""} {f"Конкретные вопросы: {questions}" if questions else ""} {f"Горизонт анализа: {timeframe}" if timeframe else ""} Разработай методологию анализа: 1. Определи scope и границы анализа 2. Выбери подходящие аналитические фреймворки (5 Whys, Fishbone, SWOT, etc.) 3. Определи ключевые вопросы для ответа 4. Укажи необходимые данные и метрики 5. Задай критерии успешного анализа Отвечай на русском языке.""", } planner_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( planner, "planner", planner_input ): if event_type == "content": planner_output += content yield { "type": "agent_message", "agent": "planner", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "planner", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "planner", } return except Exception as e: logger.error("analyze_flow.planner.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"PlannerAgent failed: {e}", "agent": "planner", } return stage_outputs["planning"] = planner_output # ======================================================================== # Шаг 2: ResearcherAgent — сбор фактов и данных # ======================================================================== yield {"type": "agent_start", "agent": "researcher"} researcher = ResearcherAgent(name="Facts Researcher") researcher_input = { "topic": f"""Собери объективную фактологическую базу для анализа. ## СИТУАЦИЯ: \"\"\" {situation} \"\"\" ## МЕТОДОЛОГИЯ АНАЛИЗА: \"\"\" {planner_output} \"\"\" {f"Контекст: {context}" if context else ""} {f"Имеющиеся данные: {data}" if data else ""} ## ЗАДАЧА: Собери **только факты и данные** без интерпретаций и выводов. Опиши объективно: - Текущую ситуацию - Ключевые метрики и данные - Хронологию событий (если применимо) - Контекст и ограничения - Stakeholders - Информационные пробелы Будь максимально объективным. Не делай выводов — только собирай факты.""", } researcher_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( researcher, "researcher", researcher_input ): if event_type == "content": researcher_output += content yield { "type": "agent_message", "agent": "researcher", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "researcher", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "researcher", } return except Exception as e: logger.error("analyze_flow.researcher.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"ResearcherAgent failed: {e}", "agent": "researcher", } return stage_outputs["facts"] = researcher_output # ======================================================================== # Шаг 3: AnalystAgent — root cause analysis и паттерны # ======================================================================== yield {"type": "agent_start", "agent": "analyst_root_cause"} analyst_root = AnalystAgent(name="Root Cause Analyst") analyst_root_input = { "topic": f"""Проведи root cause analysis и выяви системные паттерны. ## СИТУАЦИЯ: \"\"\" {situation} \"\"\" ## ФАКТОЛОГИЧЕСКАЯ БАЗА: \"\"\" {researcher_output} \"\"\" {f"Контекст: {context}" if context else ""} ## ЗАДАЧА: Проведи глубокий диагностический анализ: 1. **5 Whys Analysis** — для 2-3 ключевых проблем добирайся до корневых причин 2. **Fishbone (Ishikawa)** — категоризируй причины по 6M (People, Process, Technology, Resources, Metrics, Environment) 3. **Системные паттерны** — выяви reinforcing и balancing loops 4. **Leverage Points** — найди точки максимального воздействия 5. **Системные архетипы** — определи какие классические паттерны наблюдаются Будь глубоким и системным в анализе.""", } analyst_root_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( analyst_root, "analyst_root_cause", analyst_root_input ): if event_type == "content": analyst_root_output += content yield { "type": "agent_message", "agent": "analyst_root_cause", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "analyst_root_cause", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "analyst_root_cause", } return except Exception as e: logger.error("analyze_flow.analyst_root_cause.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"AnalystAgent (root cause) failed: {e}", "agent": "analyst_root_cause", } return stage_outputs["root_cause"] = analyst_root_output # ======================================================================== # Шаг 4: CriticAgent — проверка на искажения и слепые пятна # ======================================================================== yield {"type": "agent_start", "agent": "critic"} critic = CriticAgent(name="Analysis Critic") critic_input = { "topic": f"""Проверь проведённый анализ на когнитивные искажения и слепые пятна. ## СИТУАЦИЯ: \"\"\" {situation} \"\"\" ## ПРОВЕДЁННЫЙ АНАЛИЗ: ### Фактологическая база: \"\"\" {researcher_output} \"\"\" ### Root cause analysis: \"\"\" {analyst_root_output} \"\"\" ## ЗАДАЧА: Подвергни сомнению проведённый анализ: 1. **Когнитивные искажения** — найди: - Confirmation Bias - Survivorship Bias - Anchoring - Correlation ≠ Causation - Availability Heuristic - И другие 2. **Слепые пятна** — что не было учтено? 3. **Непроверенные предположения** — что принято как данность? 4. **Альтернативные объяснения** — какие ещё причины могут объяснять наблюдаемое? 5. **Логические ошибки** — где нарушена логика рассуждений? Будь конструктивным скептиком. Указывай не только проблемы, но и как их исправить.""", } critic_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( critic, "critic", critic_input ): if event_type == "content": critic_output += content yield { "type": "agent_message", "agent": "critic", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "critic", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "critic", } return except Exception as e: logger.error("analyze_flow.critic.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"CriticAgent failed: {e}", "agent": "critic", } return stage_outputs["critique"] = critic_output # ======================================================================== # Шаг 5: AnalystAgent (2-й) — прогнозы и сценарии # ======================================================================== yield {"type": "agent_start", "agent": "analyst_forecast"} analyst_forecast = AnalystAgent(name="Forecast Analyst") analyst_forecast_input = { "topic": f"""Построй прогнозы и сценарии развития ситуации. ## СИТУАЦИЯ: \"\"\" {situation} \"\"\" {f"Горизонт анализа: {timeframe}" if timeframe else "Горизонт: 12 месяцев"} ## РЕЗУЛЬТАТЫ АНАЛИЗА: ### Факты: \"\"\" {researcher_output} \"\"\" ### Root cause analysis: \"\"\" {analyst_root_output} \"\"\" ### Критика анализа: \"\"\" {critic_output} \"\"\" {f"Контекст: {context}" if context else ""} ## ЗАДАЧА: Построй прогнозы развития ситуации: 1. **Базовый прогноз** (наиболее вероятный, ~60%) - Горизонт 1-3 месяца - Горизонт 3-6 месяцев - Горизонт 6-12 месяцев 2. **Сценарный анализ**: - Оптимистичный сценарий (~20%) - Базовый сценарий (~60%) - Пессимистичный сценарий (~20%) 3. **Leading vs Lagging indicators** — какие метрики отслеживать 4. **Ранние индикаторы изменений** — что сигнализирует о смене тренда 5. **Точки бифуркации** — моменты, когда система может пойти по разным путям 6. **Риски и возможности** — с вероятностями Указывай вероятности для всех прогнозов.""", } analyst_forecast_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( analyst_forecast, "analyst_forecast", analyst_forecast_input ): if event_type == "content": analyst_forecast_output += content yield { "type": "agent_message", "agent": "analyst_forecast", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "analyst_forecast", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "analyst_forecast", } return except Exception as e: logger.error("analyze_flow.analyst_forecast.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"AnalystAgent (forecast) failed: {e}", "agent": "analyst_forecast", } return stage_outputs["forecast"] = analyst_forecast_output # ======================================================================== # Шаг 6: WriterAgent — финальный аналитический отчёт # ======================================================================== yield {"type": "agent_start", "agent": "writer"} writer = WriterAgent( name="Analysis Report Writer", system_prompt=WRITER_ANALYSIS_PROMPT, ) writer_input = { "topic": f"""Сформируй финальный аналитический отчёт. ## СИТУАЦИЯ: \"\"\" {situation} \"\"\" {f"Контекст: {context}" if context else ""} {f"Горизонт анализа: {timeframe}" if timeframe else ""} ## ВЕСЬ ПРОЦЕСС АНАЛИЗА: ### 1. Методология (Planner): \"\"\" {planner_output} \"\"\" ### 2. Фактологическая база (Researcher): \"\"\" {researcher_output} \"\"\" ### 3. Root Cause Analysis (Analyst): \"\"\" {analyst_root_output} \"\"\" ### 4. Критика анализа (Critic): \"\"\" {critic_output} \"\"\" ### 5. Прогнозы и сценарии (Forecast Analyst): \"\"\" {analyst_forecast_output} \"\"\" ## ЗАДАЧА: Создай профессиональный аналитический отчёт уровня executive report. Включи: - Executive Summary (3-5 предложений) - Текущую ситуацию (факты) - Root Cause Analysis (5 Whys + Fishbone) - Системные паттерны и leverage points - Прогнозы и сценарии (с вероятностями) - Риски и возможности - Ключевые инсайты - Рекомендации к действию (немедленные/краткосрочные/среднесрочные) - Метрики для мониторинга (leading/lagging indicators) - Ограничения анализа Отвечай на русском языке. Создай документ готовый к использованию для принятия решений.""", } writer_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( writer, "writer", writer_input ): if event_type == "content": writer_output += content yield { "type": "agent_message", "agent": "writer", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "writer", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "writer", } return except Exception as e: logger.error("analyze_flow.writer.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"WriterAgent failed: {e}", "agent": "writer", } return stage_outputs["report"] = writer_output # ======================================================================== # Финальное событие flow # ======================================================================== logger.info( "analyze_flow.completed", total_tokens=total_tokens, stages={ "planning": len(planner_output), "facts": len(researcher_output), "root_cause": len(analyst_root_output), "critique": len(critic_output), "forecast": len(analyst_forecast_output), "report": len(writer_output), }, ) yield { "type": "flow_done", "flow_id": "analyze", "output": writer_output, "tokens": total_tokens, "stages": stage_outputs, "metadata": { "agents_count": 6, "total_stages": 6, "timeframe": timeframe or "default", "analysis_type": "deep_diagnostic", }, } # ============================================================================ # Flow registration # ============================================================================ analyze_flow = Flow( id="analyze", name="Analyze Flow", description=( "Мультиагентный глубокий анализ ситуации: " "Planner → Researcher (факты) → Analyst (root cause) → " "Critic (искажения) → Analyst (прогнозы) → Writer. " "Проводит комплексный диагностический анализ с использованием " "5 Whys, Fishbone, системного мышления и сценарного планирования. " "Выявляет корневые причины, системные паттерны, строит прогнозы " "с вероятностями и формирует рекомендации к действию. " "Идеально подходит для анализа бизнес-ситуаций, инцидентов, " "трендов и подготовки стратегических решений." ), agents=[ "planner", "researcher", "analyst_root_cause", "critic", "analyst_forecast", "writer", ], runner=run_analyze_flow, ) # ============================================================================ # Public API # ============================================================================ __all__ = ["analyze_flow", "run_analyze_flow"]