/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/flows/debate_flow.py
772 строки
30 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
# core/engine/src/flows/debate_flow.py """Debate flow — мультиагентные дебаты для принятия взвешенного решения. Последовательность агентов: 1. PlannerAgent — формулирует чёткий тезис и структуру дебатов 2. DebaterAgent PRO — приводит сильные аргументы ЗА тезис 3. DebaterAgent CON — опровергает PRO и приводит аргументы ПРОТИВ 4. AnalystAgent — объективно анализирует силу аргументов обеих сторон 5. JudgeAgent — выносит финальный вердикт с практической рекомендацией Философия flow: Дебаты позволяют рассмотреть сложную проблему с противоположных сторон, выявить скрытые предположения и принять обоснованное решение. Такой подход особенно полезен для: - Архитектурных решений (монолит vs микросервисы) - Стратегических выборов (build vs buy) - Оценки рисков (внедрять технологию X или нет) - Этических дилемм - Приоритизации задач События 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, DebaterAgent, AnalystAgent, JudgeAgent, ) from src.primitives.flow import Flow logger = structlog.get_logger() # ============================================================================ # System prompts для дебатёров (разные роли) # ============================================================================ DEBATER_PRO_SYSTEM_PROMPT = """Ты — опытный дебатёр, защищающий позицию ЗА в дебатах. Твоя задача — привести максимально сильные, убедительные аргументы в пользу тезиса. ## Принципы работы: 1. **Steel Manning** — представляй тезис в наиболее сильной форме 2. **Evidence-Based** — подкрепляй каждое утверждение фактами, статистикой, примерами 3. **Логическая строгость** — строй чёткие причинно-следственные цепочки 4. **Практичность** — показывай конкретные выгоды и применение 5. **Предвосхищение возражений** — заранее отвечай на возможные контраргументы ## Формат ответа: ### 🎯 Главный тезис (1-2 предложения, ядро позиции) ### 💪 Основные аргументы (5-7 штук) Каждый аргумент в формате: - **Название аргумента** (сила: ⭐⭐⭐⭐⭐) - Суть: ... - Доказательства: ... - Практическое значение: ... ### 🛡️ Предвосхищение контраргументов (3-4 возможных возражения и ответы на них) ### 📊 Эмпирические данные (Статистика, исследования, case studies) ### 🎯 Призыв к действию (Почему стоит принять эту позицию) Отвечай на русском языке. Будь убедительным, но честным.""" DEBATER_CON_SYSTEM_PROMPT = """Ты — опытный дебатёр, защищающий позицию ПРОТИВ в дебатах. Твоя задача — выявить слабые места тезиса, опровергнуть аргументы оппонента и показать альтернативные подходы. ## Принципы работы: 1. **Principle of Charity** — интерпретируй аргументы ПРО в наиболее сильной форме 2. **Critical Thinking** — выявляй скрытые предположения и логические ошибки 3. **Reductio ad absurdum** — показывай, к чему ведёт тезис в крайних случаях 4. **Альтернативы** — предлагай лучшие решения той же проблемы 5. **Risk Awareness** — фокусируйся на рисках и негативных последствиях ## Формат ответа: ### 🎯 Контр-тезис (1-2 предложения, ядро твоей позиции) ### ⚔️ Опровержение аргументов ПРО Для каждого сильного аргумента оппонента: - **Аргумент ПРО:** [цитата] - **Опровержение:** [почему он слабый или неверный] - **Альтернатива:** [что делать вместо этого] ### 💪 Основные контраргументы (5-7 штук) Каждый в формате: - **Название** (сила: ⭐⭐⭐⭐⭐) - Суть: ... - Доказательства: ... - Почему перевешивает аргументы ПРО: ... ### ⚠️ Риски и негативные последствия (Что пойдёт не так, если принять тезис) ### 🔄 Альтернативные подходы (Как решить ту же проблему лучше) ### 🎯 Итоговая позиция (Почему стоит отвергнуть исходный тезис) Отвечай на русском языке. Будь критичным, но конструктивным.""" ANALYST_SYSTEM_PROMPT = """Ты — объективный аналитик, оценивающий силу аргументов в дебатах. Твоя задача — беспристрастно проанализировать аргументы обеих сторон, выявить логические ошибки, оценить качество доказательств и определить победителя по каждому пункту. ## Критерии оценки: 1. **Логическая корректность** (0-10): нет ли fallacies 2. **Эмпирическая обоснованность** (0-10): есть ли факты/данные 3. **Практическая применимость** (0-10): можно ли реализовать 4. **Полнота покрытия** (0-10): учтены ли все аспекты 5. **Устойчивость к критике** (0-10): выдерживает ли контраргументы ## Формат анализа: ### 📊 Матрица оценок | Критерий | ПРО (1-10) | КОН (1-10) | Комментарий | |----------|------------|------------|-------------| | ... | ... | ... | ... | ### 🏆 Победители по пунктам Для каждого ключевого аспекта — кто выиграл и почему ### 🔍 Логические ошибки - **ПРО:** [ошибки в аргументации] - **КОН:** [ошибки в аргументации] ### 🤝 Точки согласия (Где обе стороны согласны) ### ⚖️ Баланс сил - ПРО: X сильных, Y средних аргументов - КОН: X сильных, Y средних аргументов - **Перевес:** ПРО / КОН / Равновесие Отвечай на русском языке. Будь максимально объективным.""" JUDGE_SYSTEM_PROMPT = """Ты — беспристрастный судья, выносящий финальный вердикт в дебатах. Твоя задача — на основе всех аргументов и анализа принять обоснованное решение и дать практическую рекомендацию. ## Принципы судейства: 1. **Объективность** — решение на основе фактов, не эмоций 2. **Практичность** — рекомендация должна быть применима в реальности 3. **Нюансированность** — учитывать контекст и условия 4. **Решительность** — давать чёткий ответ, не уходить от него 5. **Intellectual Honesty** — признавать неопределённость там, где она есть ## Формат вердикта: ### ⚖️ Финальный вердикт **Решение:** [ЗА / ПРОТИВ / КОМПРОМИСС] **Обоснование:** (развёрнутое объяснение) ### 🏆 Победитель дебатов [ПРО / КОН / Ничья] — с объяснением почему ### 🎯 Ключевые решающие аргументы (2-3 аргумента, которые перевесили чашу весов) ### 💡 Практическая рекомендация Конкретный план действий на основе дебатов: 1. Шаг 1 2. Шаг 2 3. Шаг 3 ### ⚠️ Условия пересмотра При каких обстоятельствах вердикт может измениться ### 🎓 Главный урок (Что мы узнали из этих дебатов) Отвечай на русском языке. Будь решительным и практичным.""" # ============================================================================ # 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 # Если агент не отправил done — финализируем сами if full_output and total_tokens == 0: yield ("done", full_output, 0, "") # ============================================================================ # Основной runner # ============================================================================ async def run_debate_flow(input_data: dict[str, Any]) -> AsyncIterator[dict[str, Any]]: """ Выполняет мультиагентный debate flow. Args: input_data: Входные данные. Поддерживает поля: - 'topic' / 'input' / 'question': тема для дебатов (обязательно) - 'context': дополнительный контекст (опционально) - 'stance': изначальная склонность "pro" / "con" / "neutral" (опционально) Yields: События SSE для UI: - agent_start: {type, agent} - agent_message: {type, agent, content} - agent_done: {type, agent, output, tokens, model} - error: {type, error, agent?} - flow_done: {type, flow_id, output, tokens, stages} """ # ======================================================================== # 0. Валидация входных данных # ======================================================================== topic = ( input_data.get("topic") or input_data.get("input") or input_data.get("question") or "" ).strip() if not topic: yield { "type": "error", "error": "Не указана тема для дебатов (ожидается поле 'topic', 'input' или 'question')", } return context = input_data.get("context", "").strip() stance = input_data.get("stance", "neutral") logger.info( "debate_flow.started", topic=topic[:100], has_context=bool(context), stance=stance, ) # Контейнер для накопления результатов каждого этапа stage_outputs: dict[str, str] = { "topic": topic, "context": context, "stance": stance, } total_tokens = 0 # ======================================================================== # Шаг 1: PlannerAgent — формулировка тезиса # ======================================================================== yield {"type": "agent_start", "agent": "planner"} planner = PlannerAgent() planner_input = { "topic": f"""Сформулируй чёткий, дискуссионный тезис для дебатов по теме: "{topic}" {f"Дополнительный контекст: {context}" if context else ""} {f"Изначальная склонность: {stance}" if stance != "neutral" else ""} ## Требования к тезису: 1. **Конкретность** — чёткое утверждение, а не размытый вопрос 2. **Дискуссионность** — должны быть сильные аргументы как ЗА, так и ПРОТИВ 3. **Практическая значимость** — решение влияет на реальные действия 4. **Формулируемость** — можно сформулировать как "Следует/Не следует X" 5. **Актуальность** — тема важна здесь и сейчас ## Предоставь: ### 🎯 Основной тезис (1-2 предложения — ядро дебатов) ### 📋 Ключевые аспекты для обсуждения (3-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("debate_flow.planner.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"PlannerAgent failed: {e}", "agent": "planner", } return stage_outputs["thesis"] = planner_output # ======================================================================== # Шаг 2: DebaterAgent (PRO) — аргументация ЗА # ======================================================================== yield {"type": "agent_start", "agent": "debater_pro"} debater_pro = DebaterAgent( name="Debater PRO", system_prompt=DEBATER_PRO_SYSTEM_PROMPT, ) debater_pro_input = { "topic": f"""Защити позицию ЗА в дебатах. ## ТЕЗИС: \"\"\" {planner_output} \"\"\" ## ЗАДАЧА: Приведи максимально сильные аргументы ЗА этот тезис. Используй все доступные средства убеждения: - Логические рассуждения - Эмпирические данные и статистику - Экспертные мнения - Практические примеры и case studies - Аналогии и метафоры - Предвосхищение контраргументов Помни: твоя цель — убедить аудиторию принять этот тезис. Будь настойчивым, но честным. Не выдумывай факты.""", } debater_pro_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( debater_pro, "debater_pro", debater_pro_input ): if event_type == "content": debater_pro_output += content yield { "type": "agent_message", "agent": "debater_pro", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "debater_pro", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "debater_pro", } return except Exception as e: logger.error("debate_flow.debater_pro.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"DebaterAgent (PRO) failed: {e}", "agent": "debater_pro", } return stage_outputs["arguments_pro"] = debater_pro_output # ======================================================================== # Шаг 3: DebaterAgent (CON) — аргументация ПРОТИВ # ======================================================================== yield {"type": "agent_start", "agent": "debater_con"} debater_con = DebaterAgent( name="Debater CON", system_prompt=DEBATER_CON_SYSTEM_PROMPT, ) debater_con_input = { "topic": f"""Защити позицию ПРОТИВ в дебатах. ## ТЕЗИС: \"\"\" {planner_output} \"\"\" ## АРГУМЕНТЫ ОППОНЕНТА (PRO): \"\"\" {debater_pro_output} \"\"\" ## ЗАДАЧА: 1. **Опровергни** каждый сильный аргумент оппонента 2. **Приведи** собственные сильные аргументы ПРОТИВ тезиса 3. **Выяви** скрытые предположения и логические ошибки PRO 4. **Предложи** альтернативные решения той же проблемы 5. **Покажи** риски и негативные последствия принятия тезиса Используй: - Critical thinking для анализа аргументов PRO - Эмпирические данные в поддержку своей позиции - Reductio ad absurdum там, где это уместно - Конструктивные альтернативы вместо просто критики Помни: твоя цель — убедить аудиторию отвергнуть тезис. Будь критичным, но честным.""", } debater_con_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( debater_con, "debater_con", debater_con_input ): if event_type == "content": debater_con_output += content yield { "type": "agent_message", "agent": "debater_con", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "debater_con", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "debater_con", } return except Exception as e: logger.error("debate_flow.debater_con.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"DebaterAgent (CON) failed: {e}", "agent": "debater_con", } return stage_outputs["arguments_con"] = debater_con_output # ======================================================================== # Шаг 4: AnalystAgent — анализ аргументов # ======================================================================== yield {"type": "agent_start", "agent": "analyst"} analyst = AnalystAgent( name="Debate Analyst", system_prompt=ANALYST_SYSTEM_PROMPT, ) analyst_input = { "topic": f"""Проведи объективный анализ дебатов. ## ТЕЗИС: \"\"\" {planner_output} \"\"\" ## АРГУМЕНТЫ ЗА (PRO): \"\"\" {debater_pro_output} \"\"\" ## АРГУМЕНТЫ ПРОТИВ (CON): \"\"\" {debater_con_output} \"\"\" ## ЗАДАЧА: Проведи беспристрастный анализ по следующим критериям: 1. **Логическая корректность** (0-10) 2. **Эмпирическая обоснованность** (0-10) 3. **Практическая применимость** (0-10) 4. **Полнота покрытия** (0-10) 5. **Устойчивость к критике** (0-10) Для каждой стороны: - Оцени по каждому критерию - Выяви логические ошибки (fallacies) - Найди непроверенные предположения - Определи, какие аргументы остались без ответа Найди: - Точки согласия между сторонами - Ключевые разногласия - Баланс сил (у кого больше сильных аргументов) Будь максимально объективным. Признай, если обе стороны правы в чём-то.""", } analyst_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( analyst, "analyst", analyst_input ): if event_type == "content": analyst_output += content yield { "type": "agent_message", "agent": "analyst", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "analyst", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "analyst", } return except Exception as e: logger.error("debate_flow.analyst.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"AnalystAgent failed: {e}", "agent": "analyst", } return stage_outputs["analysis"] = analyst_output # ======================================================================== # Шаг 5: JudgeAgent — финальный вердикт # ======================================================================== yield {"type": "agent_start", "agent": "judge"} judge = JudgeAgent( name="Debate Judge", system_prompt=JUDGE_SYSTEM_PROMPT, ) judge_input = { "topic": f"""Вынеси финальный вердикт по результатам дебатов. ## ТЕЗИС: \"\"\" {planner_output} \"\"\" ## АРГУМЕНТЫ ЗА (PRO): \"\"\" {debater_pro_output} \"\"\" ## АРГУМЕНТЫ ПРОТИВ (CON): \"\"\" {debater_con_output} \"\"\" ## АНАЛИЗ АРГУМЕНТОВ: \"\"\" {analyst_output} \"\"\" {f"Контекст: {context}" if context else ""} ## ЗАДАЧА: Как беспристрастный судья, вынеси окончательный вердикт: ### ⚖️ Финальный вердикт **Решение:** [ЗА / ПРОТИВ / КОМПРОМИСС] **Обоснование:** (почему именно такое решение) ### 🏆 Победитель дебатов [ПРО / КОН / Ничья] — с объяснением ### 🎯 Ключевые решающие аргументы (2-3 аргумента, которые перевесили чашу весов) ### 💡 Практическая рекомендация Конкретный план действий: 1. Шаг 1 2. Шаг 2 3. Шаг 3 ### ⚠️ Условия пересмотра (При каких обстоятельствах вердикт может измениться) ### 🎓 Главный урок (Что мы узнали из этих дебатов) Будь решительным. Дай практическую, применимую рекомендацию. Если есть неопределённость — честно об этом скажи.""", } judge_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( judge, "judge", judge_input ): if event_type == "content": judge_output += content yield { "type": "agent_message", "agent": "judge", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "judge", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "judge", } return except Exception as e: logger.error("debate_flow.judge.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"JudgeAgent failed: {e}", "agent": "judge", } return stage_outputs["verdict"] = judge_output # ======================================================================== # Финальное событие flow # ======================================================================== logger.info( "debate_flow.completed", total_tokens=total_tokens, stages={ "thesis": len(planner_output), "pro": len(debater_pro_output), "con": len(debater_con_output), "analysis": len(analyst_output), "verdict": len(judge_output), }, ) yield { "type": "flow_done", "flow_id": "debate", "output": judge_output, "tokens": total_tokens, "stages": { "thesis": planner_output, "arguments_pro": debater_pro_output, "arguments_con": debater_con_output, "analysis": analyst_output, "verdict": judge_output, }, "metadata": { "topic": topic, "agents_count": 5, "total_stages": 5, }, } # ============================================================================ # Регистрация flow # ============================================================================ debate_flow = Flow( id="debate", name="Debate Flow", description=( "Мультиагентные дебаты для принятия взвешенных решений: " "Planner → Debater PRO + Debater CON → Analyst → Judge. " "Рассматривает проблему с противоположных сторон, анализирует " "аргументы и выносит обоснованный вердикт с практической рекомендацией." ), agents=["planner", "debater_pro", "debater_con", "analyst", "judge"], runner=run_debate_flow, ) # ============================================================================ # Public API # ============================================================================ __all__ = ["debate_flow", "run_debate_flow"]