From eb84fb96afc8205d3cb6762f120361b1e3dd6c50 Mon Sep 17 00:00:00 2001 From: Zakhar Telyatnikov <1@zaek.eu> Date: Sat, 19 Sep 2026 13:18:49 +0000 Subject: [PATCH] old intents --- CoderAgent.py | 116 +++++++++++ PlannerAgent.py | 192 ++++++++++++++++++ TesterAgent.py | 101 ++++++++++ analyzer.py | 131 ++++++++++++ start.py | 514 ++++++++++++++++++++++++++++++++++++++++++++++++ 5 files changed, 1054 insertions(+) create mode 100644 CoderAgent.py create mode 100644 PlannerAgent.py create mode 100644 TesterAgent.py create mode 100644 analyzer.py create mode 100644 start.py diff --git a/CoderAgent.py b/CoderAgent.py new file mode 100644 index 0000000..40ee545 --- /dev/null +++ b/CoderAgent.py @@ -0,0 +1,116 @@ +class CoderAgent: + """Агент-исполнитель: выполняет подзадачу, генерирует код и применяет изменения.""" + + @staticmethod + def execute(project: Project, subtask: str, files: List[str], error_feedback: str = "") -> bool: + print(f"💻 Кодер выполняет: {subtask}") + + # 1. Начинаем с переданных файлов + context_files = set(files) + + # 2. Извлекаем пути из error_feedback (если есть) + if error_feedback: + import re + # Ищем пути к файлам (с расширениями .ts, .tsx, .js, .jsx, .html, .css, .py) + # Паттерн ищет строки, содержащие путь с расширением, например "src/App.tsx" + path_pattern = r'[^\s"\']+\.(tsx?|jsx?|html|css|py)\b' + found_paths = re.findall(path_pattern, error_feedback, re.IGNORECASE) + # Также ищем полные URL типа file:// или http:// с путем + url_pattern = r'(?:file://|https?://[^\s]+)(/[^\s"\']+)' + url_matches = re.findall(url_pattern, error_feedback) + for match in url_matches: + # обрезаем до относительного пути относительно корня проекта? + # Пока просто добавляем как есть, но лучше нормализовать + found_paths.append(match) + + # Добавляем найденные пути, если они не в списке + for p in found_paths: + # Удаляем лишние символы, например, :line:col или суффиксы + # Часто путь может быть вида "file:///home/project/src/App.tsx:12:34" + # или просто "src/App.tsx:12:34" + p = p.split(':')[0] # отбрасываем строку/колонку, если есть + p = p.strip() + # Если путь начинается с /, то это абсолютный путь, пытаемся сделать относительным + if p.startswith('/'): + p = p.lstrip('/') + # Если это абсолютный путь внутри проекта, то можно обрезать корень + # Упрощённо: ищем вхождение папки проекта + proj_root = str(project.path.resolve()) + if p.startswith(proj_root): + p = os.path.relpath(p, proj_root) + else: + # иначе оставляем как есть (может не сработать) + pass + if p and p not in context_files: + context_files.add(p) + + # 3. Читаем содержимое всех файлов + file_context = [] + for rel_path in context_files: + full_path = project.path / rel_path + if not full_path.exists(): + print(f" ⚠️ Файл не найден: {rel_path}") + continue + content = project.read_file_content(full_path) + if content: + file_context.append(f"=== {rel_path} ===\n{content}") + else: + print(f" ⚠️ Не удалось прочитать {rel_path}") + + context = "\n".join(file_context) if file_context else "(нет файлов для контекста)" + + # 4. Формируем промпт + system_prompt = "Ты — опытный разработчик. Отвечай только кодом, без лишнего текста. Если нужно изменить несколько файлов, выдавай блоки вида: [filename]\n```language\nкод\n```" + user_prompt = f""" +Текущая задача: {subtask} +{f"При предыдущей попытке возникли ошибки: {error_feedback}" if error_feedback else ""} + +Текущий код указанных файлов: +{context} + +Внеси изменения в соответствующие файлы. Если создаёшь новый файл — укажи его путь относительно корня проекта. +Выдай изменения в формате: +[путь/к/файлу1] +```язык +код +``` +[путь/к/файлу2] +```язык +код +``` +""" + messages = [{"role": "system", "content": system_prompt}, + {"role": "user", "content": user_prompt}] + response = project.ollama_client.chat(messages, temperature=0.2) + if not response: + print("⚠️ Ответ от Ollama пустой.") + return False + + try: + pattern = r'\[(.*?)\]\s*```(?:\w+)?\s*(.*?)```' + matches = re.findall(pattern, response, re.DOTALL) + if not matches: + print("⚠️ Не удалось распарсить ответ, пытаемся сохранить как один файл") + if context_files: + first_file = next(iter(context_files)) + full_path = project.path / first_file + project.write_file(full_path, response) + print(f" Записан {first_file} (весь ответ)") + return True + if files: + first_file = files[0] + full_path = project.path / first_file + project.write_file(full_path, response) + print(f" Записан {first_file} (весь ответ)") + return True + return False + + for file_path_str, content in matches: + file_path_str = file_path_str.strip().strip('"').strip("'") + target_path = project.path / file_path_str + project.write_file(target_path, content.strip()) + print(f" Обновлён {target_path.relative_to(project.path)}") + return True + except Exception as e: + print(f" Ошибка при парсинге ответа кодера: {e}") + return False \ No newline at end of file diff --git a/PlannerAgent.py b/PlannerAgent.py new file mode 100644 index 0000000..d6cc321 --- /dev/null +++ b/PlannerAgent.py @@ -0,0 +1,192 @@ +class PlannerAgent: + """Агент-планировщик: анализирует проект и создаёт план подзадач.""" + + @staticmethod + def plan(project: Project, user_task: str, errors: List[str] = None) -> List[Dict[str, str]]: + # Получаем суммари из состояния + summaries = project.state.get("file_summaries", {}) + if not summaries: + print("⚠️ Нет описаний файлов. Сначала сгенерируйте их.") + return [{"type": "error", "description": "Нет описаний файлов"}] + + # Формируем текстовое представление суммари + items = [] + for file_path, desc in summaries.items(): + items.append(f"- {file_path}: {desc}") + summaries_text = "\n".join(items) + + # Берём последние ошибки из состояния (если есть) + errors_context = "" + if errors: + errors_context = "Актуальные ошибки в браузере:\n" + "\n".join(errors[:5]) + else: + errors_context = "Ошибок в браузере не зафиксировано." + # last_errors = project.state.get("last_errors", []) + # errors_context = "" + # if last_errors: + # errors_context = "\n".join(last_errors[:3]) + + prompt = f""" +Ты — главный архитектор. Получена задача: {user_task} + +Вот краткое описание каждого важного файла проекта: +{summaries_text} + +{errors_context} + +Твоя задача — составить план действий. План должен быть списком задач. +Для каждой задачи укажи тип агента: +- "test" — запустить браузер и проверить консоль на ошибки +- "code" — написать или изменить код. Для этой задачи обязательно укажи поле "files" — массив относительных путей к файлам, которые нужно передать кодеру для контекста. +- "browser" — открыть браузер для визуальной проверки (не обязательно) + +План должен быть логичным: сначала обычно проверяют текущее состояние (test), затем исправляют ошибки (code), снова проверяют (test) и т.д. +Но ты можешь предложить любую последовательность, которая, по твоему мнению, решит задачу. + +Ответ дай строго в формате JSON-списка объектов: +[ + {{"type": "test", "description": "описание задачи для тестировщика"}}, + {{"type": "code", "description": "описание задачи для кодера", "files": ["путь/к/файлу1", "путь/к/файлу2"]}} +] +Никаких пояснений, только JSON. + """ + messages = [{"role": "system", "content": "Ты — архитектор, выдаёшь только JSON."}, + {"role": "user", "content": prompt}] + response = project.ollama_client.chat(messages, temperature=0.3) + try: + start = response.find('[') + end = response.rfind(']') + 1 + if start != -1 and end != -1: + json_str = response[start:end] + plan = json.loads(json_str) + if isinstance(plan, list) and all( + isinstance(item, dict) and "type" in item and "description" in item for item in plan): + for item in plan: + if item["type"] == "code" and "files" not in item: + item["files"] = [] # если не указано, ставим пустой массив + return plan + except Exception as e: + print(f"Ошибка парсинга плана: {e}") + # fallback + return [{"type": "code", "description": "Не удалось создать план, попробуйте вручную.", "files": []}] + + @staticmethod + def replan(project: Project, user_task: str, errors: List[str]) -> List[Dict[str, str]]: + """Анализирует ошибки и возвращает список новых задач для их исправления.""" + summaries = project.state.get("file_summaries", {}) + summaries_text = "\n".join([f"- {p}: {d}" for p, d in summaries.items()]) + errors_text = "\n".join(errors) # ограничим для контекста + + prompt = f""" + Ты — главный архитектор. Получена задача: {user_task} + + Вот описание файлов проекта: + {summaries_text} + + При выполнении теста были обнаружены следующие ошибки: + {errors_text} + + Проанализируй ошибки. Определи, какие изменения в коде нужно внести, чтобы их исправить. + Составь список задач для кодера (тип "code") с конкретными инструкциями по исправлению. + Если ошибки связаны с отсутствием файлов или неправильными путями, добавь задачу по созданию или перемещению. + Ответ дай строго в формате JSON-списка объектов: [{{"type": "code", "description": "описание"}}] + Никаких пояснений, только JSON. + """ + messages = [{"role": "system", "content": "Ты — архитектор, выдаёшь только JSON."}, + {"role": "user", "content": prompt}] + response = project.ollama_client.chat(messages, temperature=0.3) + try: + start = response.find('[') + end = response.rfind(']') + 1 + if start != -1 and end != -1: + json_str = response[start:end] + tasks = json.loads(json_str) + if isinstance(tasks, list) and all( + isinstance(item, dict) and "type" in item and "description" in item for item in tasks): + return tasks + except Exception: + pass + return [{"type": "code", "description": "Исправить ошибки в коде (ручное вмешательство)"}] + + @staticmethod + def next_task(project: Project, user_task: str) -> dict: + summaries = project.state.get("file_summaries", {}) + if not summaries: + return {"type": "error", "description": "Нет описаний файлов"} + + completed = project.state.get("completed", []) + last_errors = project.state.get("last_errors", []) + + errors_text = "" + if last_errors: + for err in last_errors[:5]: + msg = err.get("message", "") + loc = err.get("location") + if loc: + file_name = loc.get("url", "неизвестный файл") + line = loc.get("line", "?") + col = loc.get("column", "?") + errors_text += f"- {msg} (в {file_name}:{line}:{col})\n" + else: + errors_text += f"- {msg}\n" + if err.get("stack"): + errors_text += f" Стек: {err['stack']}\n" + else: + errors_text = "Ошибок не зафиксировано." + + summaries_text = "\n".join([f"- {path}: {desc}" for path, desc in list(summaries.items())[:10]]) + completed_text = "\n".join([f"- {task['type']}: {task['description']}" for task in completed]) or "нет" + + prompt = f""" + Ты — главный архитектор. Получена задача: {user_task} + + Описание файлов проекта: + {summaries_text} + + Уже выполненные задачи: + {completed_text} + + Актуальные ошибки (с указанием файлов и строк): + {errors_text} + + Твоя задача — решить, что делать дальше. + ВНИМАНИЕ: НЕ возвращай finish, пока не будут выполнены как минимум 2 задачи и не будет ошибок. + Если ошибки есть — ты должен предложить задачу типа "code" для их исправления. + Если ошибок нет и уже было выполнено хотя бы 2 задачи, тогда можешь вернуть finish. + Иначе верни одну задачу для следующего шага. Типы задач: + - "test" — запустить браузер и проверить консоль на ошибки (это нужно делать, если неизвестно состояние) + - "code" — написать или изменить код. Укажи поле "files" — массив относительных путей к файлам, которые нужно передать кодеру. Старайся указывать файлы на основе информации об ошибках. + - "browser" — открыть браузер для визуальной проверки + + Ответ дай строго в формате JSON-объекта: + {{"type": "test", "description": "описание"}} + или + {{"type": "code", "description": "описание", "files": ["путь/к/файлу"]}} + Никаких пояснений, только JSON. + """ + messages = [{"role": "system", "content": "Ты — архитектор, выдаёшь только JSON."}, + {"role": "user", "content": prompt}] + response = project.ollama_client.chat(messages, temperature=0.3) + print(f"Ответ планировщика: {response}") # отладка + + if not response: + return {"type": "test", "description": "Проверить состояние"} + + try: + start = response.find('{') + end = response.rfind('}') + 1 + if start != -1 and end != -1: + json_str = response[start:end] + task = json.loads(json_str) + if "type" in task and "description" in task: + if task["type"] == "code" and "files" not in task: + task["files"] = [] + # Принудительно не даём finish, если мало задач выполнено + if task.get("type") == "finish" and len(completed) < 2: + return {"type": "test", "description": "Проверить состояние"} + return task + except Exception as e: + print(f"Парсинг JSON не удался: {e}") + + # Если не удалось распарсить, возвращаем тест по умолчанию + return {"type": "test", "description": "Проверить состояние"} \ No newline at end of file diff --git a/TesterAgent.py b/TesterAgent.py new file mode 100644 index 0000000..608d01c --- /dev/null +++ b/TesterAgent.py @@ -0,0 +1,101 @@ +from start import Project + + +class TesterAgent: + """Агент-тестировщик: открывает браузер, проверяет консоль на ошибки.""" + + @staticmethod + def test(project: Project) -> List[Dict[str, Any]]: + print(f"🌐 Тестировщик проверяет {project.url} ...") + errors = [] + try: + with sync_playwright() as p: + browser = p.chromium.launch(headless=True) + context = browser.new_context() + page = context.new_page() + console_errors = [] + + # Перехват ошибок страницы (даёт объект Error) + def handle_pageerror(error): + error_info = { + "message": str(error), + "type": "pageerror", + "location": None, + "stack": error.stack if hasattr(error, 'stack') else None + } + # Извлекаем location из стека, если есть + if error_info["stack"]: + import re + # Ищем строки вида at ... (file:line:col) + match = re.search(r'at .*?\((.*?):(\d+):(\d+)\)', error_info["stack"]) + if match: + file_path = match.group(1) + line = int(match.group(2)) + col = int(match.group(3)) + error_info["location"] = {"url": file_path, "line": line, "column": col} + console_errors.append(error_info) + + page.on("pageerror", handle_pageerror) + + # Обработка console сообщений + def handle_console(msg): + if msg.type in ("error", "warning"): + error_info = { + "message": msg.text, + "type": msg.type, + "location": None, + "stack": None + } + # Местоположение (уже есть) + if hasattr(msg, 'location') and msg.location: + error_info["location"] = { + "url": msg.location.get("url", ""), + "line": msg.location.get("lineNumber", 0), + "column": msg.location.get("columnNumber", 0) + } + # Попытка извлечь стек из сообщения (часто в конце сообщения есть строки at) + if error_info["message"]: + import re + # Ищем строки, начинающиеся с 'at ' (стек) + stack_lines = re.findall(r'at .*', error_info["message"]) + if stack_lines: + error_info["stack"] = "\n".join(stack_lines) + else: + # В некоторых браузерах стек может быть в конце сообщения после \n + parts = error_info["message"].split('\n') + if len(parts) > 1: + # Проверяем, есть ли строки, начинающиеся с 'at' + stack_candidate = "\n".join( + [line for line in parts if line.strip().startswith('at')]) + if stack_candidate: + error_info["stack"] = stack_candidate + + console_errors.append(error_info) + + page.on("console", handle_console) + + # Внедряем скрипт для перехвата необработанных ошибок и rejection (чтобы они попали в console) + page.add_init_script(""" + window.addEventListener('error', function(e) { + console.error(e.message + '\\n' + e.stack); + }); + window.addEventListener('unhandledrejection', function(e) { + console.error(e.reason.message || e.reason); + }); + """) + + page.goto(project.url, timeout=30000, wait_until='load') + time.sleep(2) + browser.close() + if console_errors: + errors = console_errors + else: + print(" ✅ Ошибок в консоли не обнаружено.") + except Exception as e: + errors.append({ + "message": f"Ошибка запуска браузера: {e}", + "type": "error", + "location": None, + "stack": None + }) + return errors \ No newline at end of file diff --git a/analyzer.py b/analyzer.py new file mode 100644 index 0000000..274dff8 --- /dev/null +++ b/analyzer.py @@ -0,0 +1,131 @@ +import json +import tree_sitter +from tree_sitter import Language, Parser +import tree_sitter_python as tspython +import tree_sitter_javascript as tsjavascript +import tree_sitter_typescript as tstypescript +import tree_sitter_html as tshtml +from pathlib import Path +from typing import Dict, List, Any, Optional + +class CodeAnalyzer: + """Анализирует код с помощью tree-sitter и извлекает структуру.""" + LANGUAGES = { + '.py': Language(tspython.language()), + '.js': Language(tsjavascript.language()), + '.ts': Language(tstypescript.language_typescript()), + '.tsx': Language(tstypescript.language_tsx()), + '.html': Language(tshtml.language()), + } + + def __init__(self): + self.parsers = {} + for ext, language in self.LANGUAGES.items(): + # Язык теперь передаётся сразу при создании парсера + self.parsers[ext] = Parser(language) + + def analyze_file(self, content: str, extension: str) -> Dict[str, Any]: + """Анализирует содержимое файла и возвращает структурированную информацию.""" + parser = self.parsers.get(extension) + if not parser: + # Для неподдерживаемых расширений возвращаем сырой текст (обрезанный до разумного) + return {"raw": content[:1000] + "..." if len(content) > 1000 else content} + + tree = parser.parse(bytes(content, 'utf-8')) + root = tree.root_node + + info = { + "imports": [], + "exports": [], + "functions": [], + "classes": [], + "variables": [], + "has_side_effects": False, + "jsx_elements": [] if extension in ('.tsx', '.jsx') else None + } + + # Обходим все узлы + self._traverse(root, info, extension) + + return info + + def _traverse(self, node, info, extension): + """Рекурсивно обходит AST и заполняет info.""" + if node.type == 'function_definition' or node.type == 'function_declaration': + func_name = self._get_node_text(node.child_by_field_name('name')) + if func_name: + params = [] + params_node = node.child_by_field_name('parameters') + if params_node: + for p in params_node.children: + if p.type == 'identifier' or p.type == 'formal_parameter': + params.append(self._get_node_text(p)) + info['functions'].append({ + "name": func_name, + "params": params + }) + + elif node.type == 'class_definition' or node.type == 'class_declaration': + class_name = self._get_node_text(node.child_by_field_name('name')) + if class_name: + methods = [] + body = node.child_by_field_name('body') + if body: + for child in body.children: + if child.type == 'method_definition' or child.type == 'function_definition': + mname = self._get_node_text(child.child_by_field_name('name')) + if mname: + methods.append(mname) + info['classes'].append({ + "name": class_name, + "methods": methods + }) + + elif node.type == 'import_statement' or node.type == 'import_from_statement': + # Для Python и JS/TS импорты выглядят по-разному, но мы упростим + import_text = self._get_node_text(node) + info['imports'].append(import_text.strip()) + + elif node.type == 'export_statement' or node.type == 'export_named_statement': + export_text = self._get_node_text(node) + info['exports'].append(export_text.strip()) + + elif node.type == 'assignment' or node.type == 'variable_declaration': + # Попытка извлечь имена переменных верхнего уровня + var_names = [] + if node.type == 'assignment': + left = node.child_by_field_name('left') + if left and left.type == 'identifier': + var_names.append(self._get_node_text(left)) + elif node.type == 'variable_declaration': + for child in node.children: + if child.type == 'variable_declarator': + name_node = child.child_by_field_name('name') + if name_node: + var_names.append(self._get_node_text(name_node)) + info['variables'].extend(var_names) + + # Побочные эффекты: выражение верхнего уровня, не являющееся импортом/экспортом/декларацией + if node.type == 'expression_statement': + # Исключаем вызовы, которые являются частью модуля (например, вызов ReactDOM.render) + # Считаем, что любое выражение верхнего уровня — потенциальный side-effect + # Можно уточнить позже + info['has_side_effects'] = True + + if extension in ('.tsx', '.jsx') and node.type == 'jsx_element': + # Извлекаем название компонента + opening = node.child_by_field_name('opening_element') + if opening: + name_node = opening.child_by_field_name('name') + if name_node: + info['jsx_elements'].append(self._get_node_text(name_node)) + + # Рекурсивно обходим детей + for child in node.children: + self._traverse(child, info, extension) + + @staticmethod + def _get_node_text(node) -> str: + if node is None: + return "" + return node.text.decode('utf-8') \ No newline at end of file diff --git a/start.py b/start.py new file mode 100644 index 0000000..3621255 --- /dev/null +++ b/start.py @@ -0,0 +1,514 @@ +#!/usr/bin/env python3 +""" +Multi-Agent Development Workflow +- Planner: анализирует проект и создаёт план подзадач +- Coder: выполняет подзадачи, пишет код +- Tester: открывает браузер, проверяет ошибки в консоли +Поддерживает несколько проектов, состояние хранится в папке проекта. +""" +import json +import tiktoken +import os +import pathlib +import re +import sys +import time +from pathlib import Path +from typing import List, Dict, Any, Optional +import requests +from playwright.sync_api import sync_playwright +from analyzer import CodeAnalyzer + +# ---------- Конфигурация ---------- +CONFIG_FILE = "config.json" + + +def load_config(): + """Загружает конфигурацию из файла.""" + if not os.path.exists(CONFIG_FILE): + default_config = { + "ollama_base_url": "http://10.6.3.2:11434", + "default_model": "qwen3:4b-instruct-2507-q4_K_M", + "projects": [ + { + "name": "example_project", + "path": "./projects/example", + "url": "http://localhost:3000", + "model": "qwen3:4b-instruct-2507-q4_K_M" + } + ] + } + with open(CONFIG_FILE, "w") as f: + json.dump(default_config, f, indent=2) + print(f"Создан файл {CONFIG_FILE}. Отредактируйте его и запустите снова.") + sys.exit(0) + with open(CONFIG_FILE, "r") as f: + return json.load(f) + + +# ---------- Клиент для Ollama ---------- +class OllamaClient: + def __init__(self, base_url: str, model: str): + self.base_url = base_url.rstrip('/') + self.model = model + self.api_url = f"{self.base_url}/api/chat" + + def chat(self, messages: List[Dict[str, str]], temperature: float = 0.2) -> str: + """Отправляет запрос к Ollama и возвращает ответ.""" + payload = { + "model": self.model, + "messages": messages, + "stream": False, + "temperature": temperature, + "options": { + "num_ctx": 16000 + } + } + try: + response = requests.post(self.api_url, json=payload, timeout=600) + response.raise_for_status() + data = response.json() + return data.get("message", {}).get("content", "") + except Exception as e: + print(f"Ошибка при запросе к Ollama: {e}") + return "" + + +# ---------- Управление проектами ---------- +class Project: + def __init__(self, name: str, path: str, url: str, model: str, file_masks, ollama_base: str): + self.name = name + self.path = Path(path) + self.url = url + self.model = model + self.file_masks = file_masks + self.ollama_client = OllamaClient(ollama_base, model) + self.state_file = self.path / ".workflow_state.json" + self.state = self.load_state() + + self.tokenizer = tiktoken.get_encoding("gpt2") + self.max_summary_tokens = 3500 + self.analyzer = CodeAnalyzer() + + def load_state(self) -> Dict[str, Any]: + """Загружает состояние проекта (план, выполненные подзадачи, история ошибок).""" + if self.state_file.exists(): + with open(self.state_file, "r") as f: + return json.load(f) + return { + "plan": [], + "current_step": 0, + "completed": [], + "errors": [], + "history": [] + } + + def save_state(self): + """Сохраняет состояние проекта.""" + self.state_file.parent.mkdir(parents=True, exist_ok=True) + with open(self.state_file, "w") as f: + json.dump(self.state, f, indent=2) + + def get_project_files(self) -> List[Path]: + """Возвращает список файлов проекта (рекурсивно все файлы, кроме системных).""" + exclude = {".git", "__pycache__", "node_modules", ".idea", ".vscode", "venv"} + files = [] + for root, dirs, filenames in os.walk(self.path): + dirs[:] = [d for d in dirs if d not in exclude] + for fname in filenames: + if fname.endswith(('.py', '.js', '.html', '.css', '.json', '.txt')): + files.append(Path(root) / fname) + return files + + def read_file_content(self, file_path: Path) -> str: + """Читает содержимое файла (с обработкой кодировок).""" + try: + with open(file_path, 'r', encoding='utf-8') as f: + return f.read() + except Exception: + return "" + + def write_file(self, file_path: Path, content: str): + """Записывает содержимое в файл (создавая папки при необходимости).""" + file_path.parent.mkdir(parents=True, exist_ok=True) + with open(file_path, 'w', encoding='utf-8') as f: + f.write(content) + + def get_project_structure(self) -> Dict[str, Any]: + """ + Возвращает дерево папок и файлов, отфильтрованных по маскам. + Возвращает список путей и текстовое представление дерева. + """ + masks = self.file_masks + include_masks = masks.get("include", []) + exclude_masks = masks.get("exclude", []) + + """Возвращает дерево папок и файлов, отфильтрованных по маскам.""" + base = self.path.resolve() + all_paths = [] + for root, dirs, files in os.walk(base): + rel_root = Path(root).relative_to(base) + for f in files: + rel_path = str(rel_root / f).replace('\\', '/') + # Пропускаем файлы, которые не проходят фильтр + if not self._matches_masks(rel_path, include_masks, exclude_masks): + continue + all_paths.append(rel_path) + all_paths.sort() + + tree_lines = self._build_tree(all_paths) + + return { + "paths": all_paths, + "tree": "\n".join(tree_lines) + } + + def _build_tree(self, paths: List[str]) -> List[str]: + """Строит ASCII-дерево из списка путей.""" + if not paths: + return ["(пусто)"] + # Простая реализация: разбиваем по "/" и строим вложенный словарь + tree = {} + for p in paths: + parts = p.split('/') + node = tree + for part in parts: + if part not in node: + node[part] = {} + node = node[part] + + lines = [] + + def walk(node, prefix=""): + items = sorted(node.items()) + for i, (name, subnode) in enumerate(items): + is_last = (i == len(items) - 1) + lines.append(f"{prefix}{'└── ' if is_last else '├── '}{name}") + if subnode: + walk(subnode, prefix + (" " if is_last else "│ ")) + + walk(tree) + return lines + + @staticmethod + def _mask_to_regex(mask: str) -> str: + """ + Преобразует glob-подобную маску в регулярное выражение. + Поддерживает: + - ** : любое количество любых символов (включая /) + - * : любое количество любых символов, кроме / + - {a,b,c} : выбор из нескольких вариантов + """ + # Убираем ведущий /, если он есть (пути передаются без него) + if mask.startswith('/'): + mask = mask[1:] + + parts = [] + i = 0 + n = len(mask) + while i < n: + ch = mask[i] + if ch == '*': + if i + 1 < n and mask[i + 1] == '*': + # ** + parts.append('.*') + i += 2 + else: + # * + parts.append('[^/]*') + i += 1 + elif ch == '{': + # Ищем закрывающую } + j = mask.find('}', i + 1) + if j == -1: + # Нет закрывающей – экранируем как обычный символ + parts.append(re.escape(ch)) + i += 1 + else: + inner = mask[i + 1:j] + # Разбиваем по запятым, экранируем каждую опцию + options = [re.escape(opt.strip()) for opt in inner.split(',')] + parts.append('(?:' + '|'.join(options) + ')') + i = j + 1 + else: + # Экранируем все остальные спецсимволы + parts.append(re.escape(ch)) + i += 1 + + regex = ''.join(parts) + return '^' + regex + '$' + + def _matches_masks(self, rel_path: str, include_masks: List[str], exclude_masks: List[str]) -> bool: + """Проверяет, подходит ли файл под маски включения/исключения.""" + # Если нет include – считаем, что включено + included = False + if not include_masks: + included = True + else: + for mask in include_masks: + regex = self._mask_to_regex(mask) + if re.match(regex, rel_path): + included = True + break + if not included: + return False + + # Проверяем исключения + for mask in exclude_masks: + regex = self._mask_to_regex(mask) + if re.match(regex, rel_path): + return False + return True + + def _count_tokens(self, text: str) -> int: + return len(self.tokenizer.encode(text)) + + def generate_summaries(self, file_paths: List[str]) -> Dict[str, str]: + """ + Генерирует описания для файлов, пока общее количество токенов не превысит лимит. + Сохраняет результат в state и возвращает словарь {путь: описание}. + """ + # Если уже есть в состоянии, возвращаем его (можно добавить проверку актуальности позже) + if "file_summaries" in self.state and self.state["file_summaries"]: + return self.state["file_summaries"] + + file_paths = sorted(file_paths, key=lambda p: (self.path / p).stat().st_size, reverse=True) + + summaries = {} + total_tokens = 0 + + for rel_path in file_paths: + full_path = self.path / rel_path + content = self.read_file_content(full_path) + if not content: + continue + + # Получаем структуру через tree-sitter (как раньше) + ext = full_path.suffix + signature = self.analyzer.analyze_file(content, ext) + signature_str = json.dumps(signature, ensure_ascii=False, indent=2) + + # Генерируем описание + prompt = f""" +Файл: {rel_path} +Структура: +{signature_str} + +Опиши кратко (1–2 предложения), что делает этот файл, перечисли все компоненты (классы, функции). +Не пиши код, только текст. +""" + response = self.ollama_client.chat([{"role": "user", "content": prompt}]) + if not response: + response = "(описание не получено)" + + # Считаем токены для этого описания (плюс небольшой запас на разделители) + desc_tokens = self._count_tokens(response) + + summaries[rel_path] = response + total_tokens += desc_tokens + print(f" ✅ {rel_path} – {desc_tokens} токенов, всего {total_tokens}") + + # Сохраняем в состоянии + self.state["file_summaries"] = summaries + self.save_state() + return summaries + + +# ---------- Основной цикл ---------- + +def run_workflow(project: Project, user_task: str, max_iterations: int = 5): + print(f"\n🚀 Начинаем работу над проектом '{project.name}'") + + # Получаем структуру и суммари + file_paths = project.state.get('paths') + print(f"📁 Найдено {len(file_paths)} файлов по маскам.") + + summaries = project.generate_summaries(file_paths) + if not summaries: + print("❌ Не удалось получить описания файлов.") + return + + # Инициализация состояния (если нет) + if "completed" not in project.state: + project.state["completed"] = [] + if "errors_history" not in project.state: + project.state["errors_history"] = [] + if "last_errors" not in project.state: + project.state["last_errors"] = [] + + iteration = 0 + while iteration < max_iterations: + iteration += 1 + print(f"\n--- Итерация {iteration} ---") + + # Планировщик решает, что делать дальше + task = PlannerAgent.next_task(project, user_task) + print(json.dumps(task)) + if not task or task.get("type") == "finish": + print("✅ Планировщик завершил работу.") + break + + task_type = task["type"] + description = task["description"] + files = task.get("files", []) + print(f"📌 Следующая задача: {task_type.upper()} — {description}") + if files: + print(f" Файлы: {', '.join(files)}") + + if task_type == "test": + errors = TesterAgent.test(project) + if errors: + print(f" ❌ Обнаружены ошибки ({len(errors)}):") + for err in errors[:3]: + print(f" {err}") + project.state["last_errors"] = errors + project.state["errors_history"].extend(errors) + else: + print(" ✅ Ошибок не обнаружено.") + project.state["last_errors"] = [] + project.save_state() + + elif task_type == "code": + error_feedback = "" + if project.state.get("last_errors"): + # Преобразуем структурированные ошибки в текст + error_lines = [] + for err in project.state["last_errors"]: + msg = err.get("message", "") + loc = err.get("location") + if loc: + file_name = loc.get("url", "неизвестный файл") + line = loc.get("line", "?") + col = loc.get("column", "?") + error_lines.append(f"{msg} (в {file_name}:{line}:{col})") + else: + error_lines.append(msg) + error_feedback = "\n".join(error_lines) + success = CoderAgent.execute(project, description, files, error_feedback) + if success: + print(" ✅ Код применён.") + # После кода сбрасываем last_errors, так как они могут быть исправлены + project.state["last_errors"] = [] + else: + print(" ⚠️ Кодер не смог применить изменения.") + project.save_state() + + elif task_type == "browser": + print(" 🌐 Открываем браузер для визуальной проверки...") + # Можно реализовать отдельный метод + else: + print(f" ⚠️ Неизвестный тип задачи: {task_type}") + + # Запоминаем выполненную задачу + project.state["completed"].append(task) + project.save_state() + + if iteration >= max_iterations: + print(f"⛔ Достигнут лимит итераций ({max_iterations}).") + else: + print("\n✅ Работа завершена успешно.") + +def replan(project: Project, user_task: str, errors: List[str]) -> bool: + """Перепланирование на основе ошибок.""" + print("🔄 Перепланирование с учетом ошибок...") + new_plan = PlannerAgent.plan(project, user_task) + if not new_plan or (len(new_plan) == 1 and new_plan[0].get("type") == "error"): + print("❌ Не удалось создать новый план.") + return False + project.state["plan"] = new_plan + project.state["current_step"] = 0 + project.state["completed"] = [] + project.state["errors_history"] = [] + project.save_state() + print("📋 Новый план:") + for i, task in enumerate(new_plan, 1): + files_str = f" files: {task.get('files', [])}" if task.get("type") == "code" else "" + print(f"{i}. [{task['type']}] {task['description']}{files_str}") + return True +# ---------- Точка входа - --------- +def main(): + import argparse + parser = argparse.ArgumentParser() + parser.add_argument("--auto", action="store_true", + help="Автоматический режим без вопросов (выбирает первый проект)") + parser.add_argument("--task", type=str, + help="Задача для выполнения (в автоматическом режиме или при повторном запуске)") + parser.add_argument("--repeat", action="store_true", + help="Перезапустить выполнение последней задачи (сброс состояния)") + args = parser.parse_args() + + config = load_config() + ollama_base = config.get("ollama_base_url", "http://10.6.3.2:11434") + projects = config.get("projects", []) + if not projects: + print("Нет проектов в конфигурации. Добавьте их в config.json.") + sys.exit(1) + + if args.auto: + # Берём первый проект + selected = projects[0] + else: + print("Доступные проекты:") + for idx, proj in enumerate(projects, 1): + print(f" {idx}. {proj['name']} (путь: {proj['path']})") + + try: + choice = int(input("Выберите номер проекта: ")) - 1 + if choice < 0 or choice >= len(projects): + raise ValueError + except ValueError: + print("Некорректный ввод.") + sys.exit(1) + + selected = projects[choice] + + project = Project( + name=selected["name"], + path=selected["path"], + url=selected.get("url", "http://localhost:3000"), + model=selected.get("model", config.get("default_model", "llama3.1")), + file_masks=selected['file_masks'], + ollama_base=ollama_base, + ) + + if not project.path.exists(): + print(f"Папка проекта {project.path} не существует. Создаю.") + project.path.mkdir(parents=True, exist_ok=True) + + # Задача из аргументов или дефолтная + task_arg = args.task + if not task_arg: + # Если не задана, пробуем взять из состояния проекта (последняя задача) + # Но у нас нет такого поля, поэтому используем заглушку + task_arg = input("Введите описание задачи для этого проекта: ").strip() + if not task_arg and not project.state['plan']: + print("Задача не введена. Завершение.") + sys.exit(1) + + if not args.auto: + structure = project.get_project_structure() + print("\n📁 Структура проекта (используемые файлы):") + print(structure["tree"]) + if input("Ok? [n/Y]: ").strip() == "n": + sys.exit(1) + + project.state['tree'] = structure["tree"] + project.state['paths'] = structure["paths"] + project.save_state() + + # Если repeat, сбрасываем состояние выполнения + if args.repeat: + print("🔄 Состояние сброшено для перезапуска задачи.") + + try: + run_workflow(project, task_arg) + except KeyboardInterrupt: + print("\nПрервано пользователем.") + except Exception as e: + print(f"Критическая ошибка: {e}") + import traceback + traceback.print_exc() + + +if __name__ == "__main__": + main()