old intents
This commit is contained in:
+116
@@ -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
|
||||
+192
@@ -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": "Проверить состояние"}
|
||||
+101
@@ -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
|
||||
+131
@@ -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')
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user