/
alexefan136
/
flowstack
Обзор
Документация
Войти
/
alexefan136
/
flowstack
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
CI/CD
Аналитика
Безопасность
main
core/engine/src/flows/code_review_flow.py
952 строки
39 KB
Alexander Efanov
Обновление репозитория
15 июл 2026, 12:19
15 июл 2026, 12:19
76704c6
Код
Авторство
О чём код?
# core/engine/src/flows/code_review_flow.py """Code Review flow — мультиагентное профессиональное ревью кода. Последовательность агентов: 1. ResearcherAgent — изучает код, определяет язык, структуру, зависимости 2. AnalystAgent — анализирует архитектуру, паттерны, метрики качества 3. ReviewerAgent — проводит детальный code review с приоритизацией 4. CriticAgent — критическая оценка: упущения, альтернативы, tech debt 5. WriterAgent — формирует итоговый actionable отчёт в markdown Философия flow: Каждое ревью рассматривается через пять различных "линз" экспертизы, что позволяет получить всестороннюю оценку кода: - **Researcher** видит общую картину (что делает код) - **Analyst** видит архитектуру (как код устроен) - **Reviewer** видит детали (что не так в коде) - **Critic** видит стратегию (почему это проблема в долгосрочной перспективе) - **Writer** синтезирует всё в конкретный план действий Применение: - Pull Request ревью - Pre-production code audit - Technical debt assessment - Onboarding (понимание legacy кода) - Обучение junior разработчиков События 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 ( ResearcherAgent, AnalystAgent, ReviewerAgent, CriticAgent, WriterAgent, ) from src.primitives.flow import Flow logger = structlog.get_logger() # ============================================================================ # Specialized system prompts для каждого агента # ============================================================================ RESEARCHER_SYSTEM_PROMPT = """Ты — опытный code researcher, специализирующийся на первичном анализе кодовой базы. Твоя задача — изучить предоставленный код и создать исчерпывающее описание для последующих экспертов, которым нужно быстро понять, что это за код. ## Что ты должен определить: 1. **Язык программирования** и использованные фреймворки 2. **Назначение кода** — какую задачу он решает 3. **Структура** — основные модули, классы, функции 4. **Зависимости** — импорты, внешние библиотеки, API 5. **Архитектурный стиль** — MVC, микросервисы, event-driven и т.д. 6. **Область применения** — web, data processing, CLI, библиотека и т.д. ## Формат ответа: ### 📋 Паспорт кода - **Язык:** [язык + версия] - **Фреймворки:** [список] - **Назначение:** [1-2 предложения] - **Размер:** [LOC, количество функций/классов] ### 🏗️ Структура - Перечень ключевых компонентов - Взаимосвязи между ними - Точки входа (entry points) ### 🔗 Зависимости - Внешние библиотеки - Внутренние модули - API endpoints (если есть) ### ⚠️ Области внимания - Что стоит проверить в первую очередь - Потенциально сложные участки Отвечай на русском языке. Будь фактологичным и структурированным.""" ANALYST_SYSTEM_PROMPT = """Ты — senior software architect, специализирующийся на анализе архитектуры и качества кода. Твоя задача — провести глубокий архитектурный анализ предоставленного кода с использованием признанных инженерных принципов и метрик. ## Критерии анализа: 1. **SOLID принципы** — соответствие каждому из 5 принципов 2. **DRY / KISS / YAGNI** — соблюдение принципов простоты 3. **Coupling & Cohesion** — связанность и связность модулей 4. **Цикломатическая сложность** — сложность логики 5. **Тестируемость** — насколько легко писать unit-тесты 6. **Производительность** — потенциальные узкие места 7. **Безопасность** — распространённые уязвимости 8. **Concurrency** — потокобезопасность, race conditions ## Формат ответа: ### 🏛️ Архитектурная оценка - **Паттерны:** использованные и упущенные - **SOLID compliance:** оценка по каждому принципу (1-10) - **Coupling score:** низкий/средний/высокий - **Cohesion score:** низкий/средний/высокий ### 📊 Метрики качества - **Сложность (Cyclomatic):** оценка + примеры - **Дублирование (DRY):** найденные повторы - **Testability score:** 1-10 ### ⚡ Производительность - Big-O для критических операций - Потенциальные bottlenecks - Memory usage concerns ### 🔐 Безопасность - OWASP Top 10 проверки - Injection риски - Authentication/authorization issues ### 📈 Масштабируемость - Как поведёт себя код при росте нагрузки - Где нужны оптимизации Отвечай на русском языке. Используй конкретные примеры из кода.""" REVIEWER_SYSTEM_PROMPT = """Ты — опытный code reviewer с 10+ годами опыта в индустрии. Твоя задача — провести детальный, профессиональный code review с приоритизацией замечаний по серьёзности. ## Категории замечаний: - **P0 (Critical):** баги, security holes, data loss risks - **P1 (Major):** серьёзные проблемы качества, архитектурные flaws - **P2 (Minor):** style issues, мелкие улучшения - **P3 (Nitpick):** субъективные предпочтения ## Что проверять: 1. **Code style** — соответствие best practices языка 2. **Naming conventions** — имена переменных, функций, классов 3. **Error handling** — полнота и корректность обработки ошибок 4. **Documentation** — docstrings, комментарии, README 5. **Type safety** — использование типов (там где применимо) 6. **Edge cases** — обработка граничных случаев 7. **Logging** — правильность и полнота логирования 8. **Configuration** — hardcoded values, magic numbers ## Формат ответа: ### 🔴 P0 — Критические проблемы Для каждого issue: ```[language] // Line X: problematic code ``` **Проблема:** ... **Решение:** ... **Влияние:** ... ### 🟡 P1 — Важные замечания Аналогичный формат ### 🔵 P2 — Улучшения Краткий список ### ✅ Что сделано хорошо (3-5 пунктов, важно для баланса) ### 📝 Рекомендации по стилю Общие наблюдения Отвечай на русском языке. Ссылайся на конкретные строки кода.""" CRITIC_SYSTEM_PROMPT = """Ты — критически мыслящий senior engineer, ищущий то, что упустили другие эксперты. Твоя задача — взглянуть на код и предыдущие анализы свежим взглядом, найти стратегические проблемы и предложить альтернативные подходы. ## Фокус критики: - Упущенные проблемы — что не заметили другие эксперты - Стратегические риски — долгосрочные последствия - Альтернативные архитектуры — как можно было сделать лучше - Technical debt — что накапливается и когда "взорвётся" - Maintainability — насколько легко развивать код через год - Business alignment — решает ли код реальную бизнес-задачу ## Формат ответа: ### 🎯 Ключевые упущения Проблема 1: что упустили → почему это важно Проблема 2: ... ### 🔮 Прогноз на будущее Через 6 месяцев: ... Через 2 года: ... При масштабировании: ... ### 💡 Альтернативные подходы Вариант A: [Название] Плюсы: ... Минусы: ... Когда применять: ... Вариант B: [Название] ... ### 📉 Технический долг Что рефакторить в первую очередь Оценка стоимости рефакторинга (в человеко-днях) ROI рефакторинга ### 🎓 Чему учит этот код Главный урок для команды Отвечай на русском языке. Будь конструктивным критиком.""" WRITER_SYSTEM_PROMPT = """Ты — технический писатель, специализирующийся на синтезе экспертных мнений в actionable отчёты. Твоя задача — объединить результаты всех предыдущих экспертов в единый, структурированный, практичный отчёт, который команда может немедленно использовать для улучшения кода. ## Принципы работы: - Убрать дублирование — разные эксперты могли сказать одно и то же - Разрешить противоречия — если эксперты не согласны, выбрать лучшее - Приоритизировать — что делать в первую очередь - Дать конкретные примеры — не "улучшить naming", а "переименовать X в Y" - Оценить effort — сколько времени займёт каждое улучшение ## Формат итогового отчёта: ### 📋 Executive Summary (3-5 предложений: общее качество, ключевые findings, приоритеты) ### ✅ Сильные стороны кода (Что сделано хорошо — важно для мотивации команды) ### 🔴 Критические проблемы (P0) — исправить немедленно Для каждой: - Проблема: ... - Влияние: ... - Решение: ... - Effort: X часов - Пример кода: (было → стало) ### 🟡 Важные улучшения (P1) — в текущем спринте Аналогичный формат ### 🔵 Рекомендации (P2) — в следующих итерациях Краткий список с оценкой effort ### 🎯 План действий (Roadmap) День 1-2: ... Неделя 1: ... Месяц 1: ... ### 📊 Общая оценка качества - Code Quality Score: X/10 - Architecture Score: X/10 - Maintainability Score: X/10 - Security Score: X/10 - Performance Score: X/10 - Overall: X/10 с обоснованием ### 📝 Примеры рефакторинга 3-5 конкретных примеров "было → стало" для самых важных проблем Отвечай на русском языке. Будь практичным и конкретным.""" # ============================================================================ # 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, "") # ============================================================================ # Вспомогательная функция: определение языка кода # ============================================================================ def _detect_language(code: str, declared_language: str) -> str: """ Определяет язык программирования по коду или возвращает указанный. Используется для форматирования code blocks в markdown. """ if declared_language and declared_language != "auto-detect": return declared_language # Простая эвристика по характерным признакам if "def " in code and ":" in code and "import " in code: return "python" if "function " in code or "const " in code or "=>" in code: return "javascript" if "public class " in code or "public static void main" in code: return "java" if "func " in code and "package " in code: return "go" if "fn " in code and ("let mut" in code or "impl " in code): return "rust" if "<?php" in code: return "php" if "using System" in code or "namespace " in code: return "csharp" if "#include " in code or "int main(" in code: return "cpp" return "" # ============================================================================ # Основной runner # ============================================================================ async def run_code_review_flow(input_data: dict[str, Any]) -> AsyncIterator[dict[str, Any]]: """ Выполняет мультиагентный code review flow. Flow состоит из пяти последовательных шагов: 1. ResearcherAgent — изучение кода и контекста 2. AnalystAgent — анализ архитектуры и метрик 3. ReviewerAgent — детальный code review с приоритизацией 4. CriticAgent — критическая оценка и альтернативы 5. WriterAgent — формирование итогового отчёта Args: input_data: Входные данные. Поддерживает поля: - 'code' / 'input' / 'content': код для ревью (обязательно) - 'language': язык программирования (опционально, auto-detect) - 'context': контекст проекта (опционально) - 'focus': фокус ревью — 'security' | 'performance' | 'architecture' | 'style' | 'comprehensive' (опционально, default: comprehensive) - 'project_type': тип проекта (web/api/cli/library) — опционально 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, metadata} """ # ======================================================================== # 0. Валидация входных данных # ======================================================================== code = ( input_data.get("code") or input_data.get("input") or input_data.get("content") or "" ) if not code.strip(): yield { "type": "error", "error": "Не указан код для ревью (ожидается поле 'code', 'input' или 'content')", } return language = input_data.get("language", "auto-detect") detected_language = _detect_language(code, language) context = input_data.get("context", "").strip() focus = input_data.get("focus", "comprehensive") project_type = input_data.get("project_type", "") code_length = len(code) lines_count = code.count("\n") + 1 logger.info( "code_review_flow.started", code_length=code_length, lines_count=lines_count, language=detected_language or "auto", focus=focus, has_context=bool(context), ) # Контейнер для накопления результатов каждого этапа stage_outputs: dict[str, str] = { "code": code, "language": detected_language, "context": context, "focus": focus, } total_tokens = 0 # Общий блок с кодом для использования во всех промптах code_block = f"```{detected_language}\n{code}\n```" context_block = f"\n\n## Контекст проекта:\n{context}" if context else "" focus_block = f"\n\n## Фокус ревью: {focus}" if focus != "comprehensive" else "" project_block = f"\n## Тип проекта: {project_type}" if project_type else "" # ======================================================================== # Шаг 1: ResearcherAgent — изучение кода # ======================================================================== yield {"type": "agent_start", "agent": "researcher"} researcher = ResearcherAgent( name="Code Researcher", system_prompt=RESEARCHER_SYSTEM_PROMPT, ) researcher_input = { "topic": f'''Изучи следующий код и подготовь исчерпывающее описание для последующих экспертов (архитектора, ревьюера, критика). Код для изучения: {code_block} {context_block}{project_block} Задача: Создай "паспорт кода" — структурированное описание, которое позволит другим экспертам быстро понять, с чем они работают, не читая весь код заново. Определи: - Язык программирования и версию - Используемые фреймворки и библиотеки - Основное назначение кода (что он делает) - Структуру (ключевые модули, классы, функции) - Точки входа (entry points) - Внешние зависимости и API - Архитектурный стиль - Области, заслуживающие особого внимания Будь максимально фактологичным — твоё описание станет базой для всех последующих анализов.''', } 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("code_review_flow.researcher.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"ResearcherAgent failed: {e}", "agent": "researcher", } return stage_outputs["research"] = researcher_output # ======================================================================== # Шаг 2: AnalystAgent — анализ архитектуры # ======================================================================== yield {"type": "agent_start", "agent": "analyst"} analyst = AnalystAgent( name="Code Architect", system_prompt=ANALYST_SYSTEM_PROMPT, ) analyst_input = { "topic": f'''Проведи глубокий архитектурный анализ кода. Код: {code_block} Предварительное исследование: """ {researcher_output} """ {context_block}{focus_block} Задача: Проанализируй код с точки зрения software architecture и engineering best practices. Оцени по следующим критериям: - SOLID принципы (1-10 по каждому): - Single Responsibility - Open-Closed - Liskov Substitution - Interface Segregation - Dependency Inversion - DRY / KISS / YAGNI — нарушения и примеры - Coupling & Cohesion — насколько модули связаны и сфокусированы - Цикломатическая сложность — укажи самые сложные функции - Тестируемость — насколько легко писать unit-тесты - Производительность — Big-O для критических операций, bottlenecks - Безопасность — OWASP Top 10, injection, auth issues - Масштабируемость — как поведёт себя код при росте нагрузки Приводи конкретные примеры из кода для каждого пункта.''', } 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("code_review_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 # ======================================================================== # Шаг 3: ReviewerAgent — детальный code review # ======================================================================== yield {"type": "agent_start", "agent": "reviewer"} reviewer = ReviewerAgent( name="Code Reviewer", system_prompt=REVIEWER_SYSTEM_PROMPT, ) reviewer_input = { "topic": f'''Проведи детальный code review. Код: {code_block} Предыдущие экспертизы: """ Исследование: {researcher_output} Архитектурный анализ: {analyst_output} """ {context_block}{focus_block} Задача: Проведи профессиональный code review с приоритизацией по серьёзности. Категории замечаний: 🔴 P0 (Critical): баги, security holes, data loss risks — исправить немедленно 🟡 P1 (Major): серьёзные проблемы качества — исправить в этом спринте 🔵 P2 (Minor): style issues, мелкие улучшения — в следующей итерации ⚪ P3 (Nitpick): субъективные предпочтения — по желанию Что проверять: - Code style — соответствие best practices {detected_language or 'языка'} - Naming conventions — имена переменных, функций, классов - Error handling — полнота и корректность обработки ошибок - Documentation — docstrings, комментарии - Type safety — использование типов - Edge cases — граничные случаи, null/undefined - Logging — правильность и полнота логирования - Configuration — hardcoded values, magic numbers Для каждого P0/P1 issue укажи: - Номер строки (если можно определить) - Проблематичный код - Объяснение проблемы - Конкретное решение с примером кода - Влияние на систему В конце обязательно укажи 3-5 вещей, которые сделаны хорошо — это важно для баланса и мотивации команды.''', } reviewer_output = "" try: async for event_type, content, tokens, model in _run_agent_stream( reviewer, "reviewer", reviewer_input ): if event_type == "content": reviewer_output += content yield { "type": "agent_message", "agent": "reviewer", "content": content, } elif event_type == "done": total_tokens += tokens yield { "type": "agent_done", "agent": "reviewer", "output": content, "tokens": tokens, "model": model, } elif event_type == "error": yield { "type": "error", "error": content, "agent": "reviewer", } return except Exception as e: logger.error("code_review_flow.reviewer.failed", error=str(e), exc_info=True) yield { "type": "error", "error": f"ReviewerAgent failed: {e}", "agent": "reviewer", } return stage_outputs["review"] = reviewer_output # ======================================================================== # Шаг 4: CriticAgent — критическая оценка # ======================================================================== yield {"type": "agent_start", "agent": "critic"} critic = CriticAgent( name="Code Critic", system_prompt=CRITIC_SYSTEM_PROMPT, ) critic_input = { "topic": f'''Проведи критическую оценку кода и предыдущих анализов. Код: {code_block} Все предыдущие экспертизы: """ Исследование: {researcher_output} Архитектурный анализ: {analyst_output} Code review: {reviewer_output} """ {context_block} Задача: Взгляни на код свежим взглядом и найди то, что упустили другие эксперты. Сфокусируйся на: - Упущенные проблемы — что не заметили исследователь, архитектор и ревьюер - Стратегические риски — долгосрочные последствия текущих решений - Альтернативные архитектуры — как можно было решить задачу принципиально иначе - Technical debt — что накапливается и когда "взорвётся" - Maintainability через год — сможет ли новый разработчик разобраться - Business alignment — решает ли код реальную бизнес-задачу Предоставь: - 3-5 ключевых упущений с объяснением, почему они важны - Прогноз развития кода (6 месяцев / 2 года / при масштабировании) - 2-3 альтернативных архитектурных подхода с плюсами/минусами - Оценку технического долга и стоимость его погашения - Главный урок для команды Будь конструктивным — указывай не только проблемы, но и решения.''', } 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("code_review_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: WriterAgent — итоговый отчёт # ======================================================================== yield {"type": "agent_start", "agent": "writer"} writer = WriterAgent( name="Review Report Writer", system_prompt=WRITER_SYSTEM_PROMPT, ) writer_input = { "topic": f'''Сформируй итоговый отчёт по code review. Код: {code_block} {context_block}{project_block} Все экспертные заключения: """ Исследование (Researcher): {researcher_output} Архитектурный анализ (Analyst): {analyst_output} Детальный code review (Reviewer): {reviewer_output} Критическая оценка (Critic): {critic_output} """ Задача: Создай профессиональный, actionable отчёт в markdown формате, который команда может немедленно использовать для улучшения кода. Структура отчёта: ### 📋 Executive Summary (3-5 предложений) Общее качество, ключевые findings, приоритеты ### ✅ Сильные стороны кода Что сделано хорошо — важно для мотивации команды ### 🔴 Критические проблемы (P0) — исправить немедленно Для каждой: - Проблема - Влияние на систему - Конкретное решение - Effort в часах - Пример кода "было → стало" ### 🟡 Важные улучшения (P1) — в текущем спринте Аналогичный формат ### 🔵 Рекомендации (P2) — в следующих итерациях Краткий список с оценкой effort ### 🎯 План действий (Roadmap) День 1-2: что делаем Неделя 1: что делаем Месяц 1: что делаем ### 📊 Общая оценка качества (каждая метрика 1-10): - Code Quality Score - Architecture Score - Maintainability Score - Security Score - Performance Score - Overall Score с обоснованием ### 📝 Топ-5 примеров рефакторинга Конкретные примеры "было → стало" с объяснением, почему стало лучше Важные принципы: - Убирай дублирование между экспертами - Разрешай противоречия (если эксперты не согласны — выбирай лучшее решение) - Всё должно быть конкретно и применимо - Указывай effort для каждого действия - Пиши на русском языке''', } 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("code_review_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( "code_review_flow.completed", total_tokens=total_tokens, stages={ "research": len(researcher_output), "analysis": len(analyst_output), "review": len(reviewer_output), "critique": len(critic_output), "report": len(writer_output), }, ) yield { "type": "flow_done", "flow_id": "code_review", "output": writer_output, "tokens": total_tokens, "stages": { "research": researcher_output, "analysis": analyst_output, "review": reviewer_output, "critique": critic_output, "report": writer_output, }, "metadata": { "language": detected_language, "code_length": code_length, "lines_count": lines_count, "focus": focus, "project_type": project_type, "agents_count": 5, "total_stages": 5, }, } # ============================================================================ # Регистрация flow # ============================================================================ code_review_flow = Flow( id="code_review", name="Code Review Flow", description=( "Мультиагентное профессиональное ревью кода: " "Researcher → Analyst → Reviewer → Critic → Writer. " "Проводит всесторонний анализ кода через 5 экспертных линз " "и формирует actionable отчёт с конкретными рекомендациями, " "приоритизированными по P0/P1/P2 и оценкой effort." ), agents=["researcher", "analyst", "reviewer", "critic", "writer"], runner=run_code_review_flow, ) # ============================================================================ # Public API # ============================================================================ __all__ = ["code_review_flow", "run_code_review_flow"]