# main.py import sys import argparse import json from pathlib import Path from datetime import datetime from typing import Optional from orchestrator import Orchestrator from models import TaskStatus def load_requirement_from_file(filepath: str) -> str: """Загружает требование из файла""" with open(filepath, 'r', encoding='utf-8') as f: return f.read() def save_report(task, output_file: Optional[str] = None): """Сохраняет детальный отчет с историей всех итераций""" report = { "task_id": task.id, "timestamp": datetime.now().isoformat(), "requirement": task.requirement, "status": task.status.value, "iterations": task.iteration, "final_review": None, "error_log": task.error_log, "files_generated": [], "history": [] # <-- добавляем историю } # Добавляем историю итераций for record in task.history: report["history"].append({ "iteration": record.iteration, "timestamp": record.timestamp, "agent": record.agent, "action": record.action, "input_summary": record.input_summary, "output_summary": record.output_summary, "details": record.details, "error": record.error }) if task.last_review: report["final_review"] = { "status": task.last_review.status, "summary": task.last_review.summary, "comments_count": len(task.last_review.comments), "comments": [ { "file": c.file, "line": c.line, "severity": c.severity, "text": c.text } for c in task.last_review.comments ] } if task.solution: report["files_generated"] = list(task.solution.files.keys()) report["tests_generated"] = list(task.solution.tests.keys()) report["description"] = task.solution.description if not output_file: output_file = f"report_{task.id}.json" with open(output_file, 'w', encoding='utf-8') as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"\n📄 Report saved: {output_file}") return report def interactive_mode(): """Интерактивный режим для ввода задач""" print("\n" + "="*60) print("🤖 Мультиагентная система разработки") print("="*60) print("\nВведите задачу для разработки (для завершения введите 'exit'):") print("-"*60) tasks = [] orchestrator = Orchestrator() while True: print("\n📝 Описание задачи (Ctrl+D или 'exit' для завершения):") lines = [] try: while True: line = input() if line.lower() == 'exit': break lines.append(line) except EOFError: break if not lines or lines[0].lower() == 'exit': break requirement = "\n".join(lines) if requirement.strip(): print("\n🚀 Запускаем разработку...") result = orchestrator.run(requirement) task = result["task"] save_report(task) tasks.append(task) # Показываем результат if task.status == TaskStatus.APPROVED: print("\n✨ Код успешно создан и прошел ревью!") if task.solution: print(f"📁 Сгенерировано файлов: {len(task.solution.files)}") print(f"🧪 Сгенерировано тестов: {len(task.solution.tests)}") else: print(f"\n❌ Неудача. Статус: {task.status}") if task.last_review: print(f"💬 {task.last_review.summary}") print("\n" + "-"*60) print("Введите следующую задачу или 'exit' для выхода:") print(f"\n📊 Всего выполнено задач: {len(tasks)}") print(f"✅ Успешно: {sum(1 for t in tasks if t.status == TaskStatus.APPROVED)}") print(f"❌ Провалено: {sum(1 for t in tasks if t.status != TaskStatus.APPROVED)}") def batch_mode(tasks_file: str): """Пакетный режим для выполнения задач из JSON файла""" with open(tasks_file, 'r', encoding='utf-8') as f: tasks_config = json.load(f) requirements = tasks_config.get('tasks', []) orchestrator = Orchestrator() results = [] print(f"\n📦 Запуск пакетной обработки: {len(requirements)} задач") for i, req in enumerate(requirements, 1): print(f"\n{'='*60}") print(f"Задача {i}/{len(requirements)}") print(f"{'='*60}") result = orchestrator.run(req) task = result["task"] save_report(task, f"report_task_{i}_{task.id}.json") results.append(task) # Небольшая задержка между задачами import time time.sleep(1) # Итоговый отчет print("\n" + "="*60) print("📊 ИТОГОВАЯ СТАТИСТИКА") print("="*60) print(f"Всего задач: {len(results)}") print(f"✅ Approved: {sum(1 for t in results if t.status == TaskStatus.APPROVED)}") print(f"🔄 Rework (лимит): {sum(1 for t in results if t.status == TaskStatus.REWORK)}") print(f"❌ Rejected: {sum(1 for t in results if t.status == TaskStatus.REJECTED)}") print(f"💀 Failed: {sum(1 for t in results if t.status == TaskStatus.FAILED)}") avg_iterations = sum(t.iteration for t in results) / len(results) print(f"📈 Среднее количество итераций: {avg_iterations:.2f}") def preview_mode(): """Режим предпросмотра - показывает примеры задач""" examples = [ { "name": "Калькулятор", "task": """ Реализуй класс Calculator с методами: - add(a, b) -> сумма - subtract(a, b) -> разность - multiply(a, b) -> произведение - divide(a, b) -> частное (обработать деление на ноль) Требования: - Все методы должны возвращать float - Добавь обработку исключений - Напиши юнит-тесты для всех методов - Добавь аннотации типов """ }, { "name": "Кеш с TTL", "task": """ Реализуй класс TTLCache с поддержкой: - set(key, value, ttl_seconds) - установка значения с TTL - get(key) - получение значения (возвращает None, если истекло) - delete(key) - удаление ключа - clear() - очистка всего кеша Особенности: - Автоматическая очистка просроченных записей - Потокобезопасность - Эффективная память (удалять просроченные при get) Напиши тесты, проверяющие TTL и конкурентный доступ. """ } ] print("\n📚 ПРИМЕРЫ ЗАДАЧ") print("="*60) for i, example in enumerate(examples, 1): print(f"\n{i}. {example['name']}:") print("-"*40) print(example['task'].strip()) print("-"*40) print("\n💡 Используйте эти задачи для тестирования системы") print("Запустите интерактивный режим и скопируйте одну из задач") def main(): parser = argparse.ArgumentParser( description="Мультиагентная система разработки с оркестратором, разработчиком и ревьювером", formatter_class=argparse.RawDescriptionHelpFormatter ) parser.add_argument( '--mode', '-m', choices=['interactive', 'batch', 'preview', 'single'], default='interactive', help='Режим работы: interactive (по умолч.), batch, preview, single' ) parser.add_argument( '--task', '-t', type=str, help='Одиночная задача (текст или путь к файлу)' ) parser.add_argument( '--tasks-file', '-f', type=str, help='JSON файл с задачами для пакетного режима' ) parser.add_argument( '--output', '-o', type=str, help='Файл для сохранения отчета' ) args = parser.parse_args() if args.mode == 'preview': preview_mode() return if args.mode == 'single': if not args.task: print("❌ Ошибка: для режима 'single' укажите задачу через --task") sys.exit(1) # Проверяем, является ли аргумент файлом if Path(args.task).exists(): requirement = load_requirement_from_file(args.task) else: requirement = args.task orchestrator = Orchestrator() result = orchestrator.run(requirement) task = result["task"] save_report(task, args.output) # Показываем результат if task.status == TaskStatus.APPROVED: print("\n✅ Успех! Код сгенерирован и прошел ревью.") if task.solution: print(f"📁 Файлы сохранены в папке 'generated_code/'") else: print(f"\n❌ Неудача. Статус: {task.status}") if task.last_review: print(f"Причина: {task.last_review.summary}") return if args.mode == 'batch': if not args.tasks_file: print("❌ Ошибка: для пакетного режима укажите файл с задачами через --tasks-file") print("Пример: python main.py --mode batch --tasks-file tasks.json") sys.exit(1) if not Path(args.tasks_file).exists(): print(f"❌ Ошибка: файл {args.tasks_file} не найден") sys.exit(1) batch_mode(args.tasks_file) return # Режим по умолчанию - интерактивный interactive_mode() # tasks.json - пример файла для пакетного режима EXAMPLE_TASKS_JSON = { "tasks": [ "Реализуй функцию is_palindrome(s) для проверки палиндрома", "Создай класс Stack с методами push, pop, peek, is_empty", "Напиши декоратор @timer, который замеряет время выполнения функции" ] } if __name__ == "__main__": # Создаем пример конфига, если его нет if not Path("tasks.json").exists(): with open("tasks.json", "w", encoding='utf-8') as f: json.dump(EXAMPLE_TASKS_JSON, f, indent=2, ensure_ascii=False) print("📝 Создан пример файла tasks.json с тестовыми задачами") main()