515 lines
21 KiB
Python
515 lines
21 KiB
Python
#!/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()
|