Teraonious/MultiAgent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

вторая итерация

Browse Source
This commit is contained in:
ki.sagidullin
2026-04-12 11:48:37 +05:00
parent 8f16a6d669
commit f5677013cc
7 changed files with 190 additions and 234 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
agent/__init__.py
View File

@@ -1,5 +1,7 @@
from .developer import DeveloperAgent
from .reviewer import ReviewerAgent
from .planner import PlannerAgent
from .executor import ExecutorAgent
from .plan_reviewer import PlanReviewerAgent
__all__ = ["DeveloperAgent", "ReviewerAgent", "PlannerAgent"]
__all__ = ["DeveloperAgent", "ReviewerAgent", "PlannerAgent", "ExecutorAgent", "PlanReviewerAgent"]

23
agent/executor.py
View File

@@ -1,14 +1,19 @@
# executor.py
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_classic.agents import AgentExecutor
from langchain.agents import create_agent
from langchain_ollama import ChatOllama
from langchain_core.prompts import ChatPromptTemplate
from config import LLM_HOST, MODEL_NAME
class ToolExecutor:
from typing import List, Dict
class ExecutorAgent:
def __init__(self, tools):
self.tools = tools
self.llm = llm or ChatOllama(
self.llm = ChatOllama(
model=MODEL_NAME,
temperature=0.0
temperature=0.0,
base_url=LLM_HOST
)
self.agent = self._create_agent()
@@ -18,9 +23,15 @@ class ToolExecutor:
("human", "{input}"),
("placeholder", "{agent_scratchpad}")
])
agent = create_tool_calling_agent(self.llm, self.tools, prompt)
agent = create_agent(self.llm, self.tools, prompt)
return AgentExecutor(agent=agent, tools=self.tools, verbose=True)
"""Устаревший агент-исполнитель. Теперь используется DeveloperAgent напрямую."""
def __init__(self, tools=None):
# Этот класс больше не используется, оставлен для обратной совместимости
pass
def execute_plan(self, plan: List[Dict]) -> List[Dict]:
results = []
for step in plan:
@@ -35,3 +46,5 @@ class ToolExecutor:
"success": True # можно анализировать ошибки
})
return results
"""Устаревший метод - теперь выполняется напрямую через DeveloperAgent"""
raise NotImplementedError("Этот метод больше не используется. Используйте DeveloperAgent.develop() напрямую.")

6
agent/plan_reviewer.py
View File

@@ -1,12 +1,16 @@
# reviewer.py
from langchain_ollama import ChatOllama
from langchain_core.messages import SystemMessage, HumanMessage
from typing import List, Dict
from config import MODEL_NAME, LLM_HOST
import json
class PlanReviewerAgent:
def __init__(self, llm=None):
self.llm = llm or ChatOllama(
model=MODEL_NAME,
temperature=0.0
temperature=0.0,
base_url=LLM_HOST
)
def review(self, original_task: str, plan: List[Dict], execution_results: List[Dict]) -> dict:

5
agent/planner.py
View File

@@ -3,13 +3,14 @@ from langchain_ollama import ChatOllama
from langchain_core.messages import SystemMessage, HumanMessage
import json
from typing import List, Dict
from config import MODEL_NAME
from config import MODEL_NAME, LLM_HOST
class PlannerAgent:
def __init__(self, llm=None):
self.llm = llm or ChatOllama(
model=MODEL_NAME,
temperature=0.0
temperature=0.0,
base_url=LLM_HOST
)
def create_plan(self, task: str, tools_description: List[Dict]) -> List[Dict]:

1
models.py
View File

@@ -11,6 +11,7 @@ class TaskStatus(str, Enum):
APPROVED = "approved"
REJECTED = "rejected"
FAILED = "failed"
DEVELOPMENT = "development"
class ReviewComment(BaseModel):
line: Optional[int] = None

365
orchestrator.py
View File

@@ -1,18 +1,19 @@
# orchestrator.py
from typing import Dict, Any, Literal, TypedDict, List
from typing import Dict, Any, Literal, TypedDict, List, Optional
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
import uuid
from datetime import datetime
from models import Task, TaskStatus, CodeSolution, ReviewResult, IterationRecord
from agent import DeveloperAgent, ReviewerAgent
from agent.developer import DeveloperAgent
from agent.reviewer import ReviewerAgent
from config import MAX_REVIEW_RETRIES
class AgentState(TypedDict):
task: str
plan: List[Dict]
task: Task
plan: Dict
execution_results: List[Dict]
review: Dict
review: Optional[ReviewResult]
iteration: int
max_iterations: int
@@ -70,204 +71,140 @@ class Orchestrator:
return workflow.compile()
def _plan_node(self, state: AgentState):
print(f"\n🔧 [Iteration {state.get('iteration', 0)}] Планировщик начал работу...")
# Сначала получим описания инструментов для планировщика
tools_desc = [{"name": t.name, "description": t.description, "args": t.args} for t in self.tools]
plan = self.planner.create_plan(state["task"], tools_desc)
task = state["task"]
print(f"\n🔧 [Iteration {task.iteration}] Планировщик начал работу...")
# Создаем запись в истории
task.history.append(IterationRecord(
iteration=task.iteration,
timestamp=datetime.now().isoformat(),
agent="developer",
action="start",
input_summary=f"Requirement: {task.requirement[:100]}...",
details={}
))
# План просто содержит задачу - разработчик выполнит её напрямую
plan = {"task": task.requirement, "iteration": task.iteration}
state["plan"] = plan
return state
def _exec_node(self, state: AgentState):
"""Узел разработчика"""
print(f"\n🔧 [Iteration {state.get('iteration', 0)}] Исполнитель начал работу...")
task = state["task"]
print(f"\n🔧 [Iteration {task.iteration}] Исполнитель начал работу...")
# Логируем начало работы разработчика
results = self.executor.execute_plan(state["plan"])
state["execution_results"] = results
# Обновляем статус
task.status = TaskStatus.DEVELOPMENT
# Разработчик выполняет задачу напрямую
solution = self.developer.develop(task, [])
task.solution = solution
task.status = TaskStatus.REVIEW
# Записываем результат выполнения
state["execution_results"] = [{"status": "success", "solution": solution}]
# Добавляем запись в историю
task.history.append(IterationRecord(
iteration=task.iteration,
timestamp=datetime.now().isoformat(),
agent="developer",
action="complete",
output_summary=f"Generated {len(solution.files)} file(s), {len(solution.tests)} test(s)",
details={
"files": list(solution.files.keys()),
"tests": list(solution.tests.keys())
}
))
print(f"✅ Разработчик создал решение: {len(solution.files)} файлов")
return state
def _review_node(self, state: AgentState):
print(f"\n🔧 [Iteration {state.get('iteration', 0)}] Ревьювер начал работу...")
review = self.reviewer.review(state["task"], state["plan"], state["execution_results"])
state["review"] = review
state["iteration"] = state.get("iteration", 0) + 1
return state
def _after_review(self, state: AgentState):
status = state["review"].get("status")
if status == "approved":
return "approved"
elif status == "changes_requested":
if state["iteration"] >= state.get("max_iterations", 3):
return "reject"
# пробуем перевыполнить те же шаги (может, ошибка временная)
return "rework"
elif status == "rejected":
if state["iteration"] >= state.get("max_iterations", 3):
return "reject"
# нужен новый план
return "replan"
else:
return "reject"
def _developer_node(self, state: Dict) -> Dict:
"""Узел разработчика"""
task = state["task"]
print(f"\n🔧 [Iteration {task.iteration}] Разработчик начал работу...")
print(f"\n🔍 [Iteration {task.iteration}] Ревьювер начал работу...")
# Логируем начало работы разработчика
self._add_history(
task, "developer", "start",
input_summary=f"Requirement: {task.requirement[:100]}...",
details={"iteration": task.iteration, "previous_comments_count": len(task.last_review.comments) if task.last_review else 0}
)
# Получаем замечания с предыдущего ревью
previous_comments = []
if task.last_review and task.last_review.comments:
previous_comments = [
{"file": c.file, "line": c.line, "text": c.text}
for c in task.last_review.comments
]
try:
# Генерируем код
solution = self.developer.develop(task, previous_comments)
task.solution = solution
task.status = TaskStatus.REVIEW
# Логируем успех
self._add_history(
task, "developer", "complete",
output_summary=f"Generated {len(solution.files)} file(s), {len(solution.tests)} test(s)",
details={"files": list(solution.files.keys()), "tests": list(solution.tests.keys())}
)
print(f"✅ Разработчик создал решение: {len(solution.files)} файлов")
for filename in solution.files:
print(f" - {filename}")
except Exception as e:
task.status = TaskStatus.FAILED
task.error_log.append(f"Developer error: {str(e)}")
# Логируем ошибку
self._add_history(
task, "developer", "error",
error=error_msg,
details={"exception": str(e)}
)
print(f"❌ Ошибка разработчика: {e}")
return {"task": task}
def _reviewer_node(self, state: Dict) -> Dict:
"""Узел ревьювера"""
task = state["task"]
print(f"\n🔍 [Iteration {task.iteration}] Ревьювер проверяет код...")
# Логируем начало ревью
self._add_history(
task, "reviewer", "start",
input_summary=f"Reviewing {len(task.solution.files)} file(s)" if task.solution else "No solution",
details={"iteration": task.iteration}
)
# Добавляем запись о начале ревью
task.history.append(IterationRecord(
iteration=task.iteration,
timestamp=datetime.now().isoformat(),
agent="reviewer",
action="start",
input_summary=f"Reviewing {len(task.solution.files) if task.solution else 0} file(s)",
details={}
))
if not task.solution:
task.status = TaskStatus.FAILED
error_msg = "No solution to review"
task.error_log.append(error_msg)
self._add_history(task, "reviewer", "error", error=error_msg)
return {"task": task}
task.history.append(IterationRecord(
iteration=task.iteration,
timestamp=datetime.now().isoformat(),
agent="reviewer",
action="error",
error=error_msg
))
return state
try:
# Проводим ревью
review = self.reviewer.review(task, task.solution)
task.last_review = review
task.iteration += 1
# Проводим ревью
review = self.reviewer.review(task, task.solution)
task.last_review = review
task.iteration += 1
# Логируем результат ревью
self._add_history(
task, "reviewer", "complete",
output_summary=f"Verdict: {review.status}, Comments: {len(review.comments)}",
details={
"verdict": review.status,
"comments_count": len(review.comments),
"summary": review.summary
}
)
# Добавляем запись в историю
task.history.append(IterationRecord(
iteration=task.iteration - 1, # предыдущая итерация
timestamp=datetime.now().isoformat(),
agent="reviewer",
action="complete",
output_summary=f"Verdict: {review.status}, Comments: {len(review.comments)}",
details={
"verdict": review.status,
"comments_count": len(review.comments)
}
))
print(f"📊 Вердикт: {review.status}")
if review.comments:
print(f" Найдено замечаний: {len(review.comments)}")
for comment in review.comments[:3]: # Показываем первые 3
print(f" - [{comment.severity}] {comment.text[:100]}")
# Сохраняем результат ревью
state["review"] = review
# Обновляем статус
if review.status == "approved":
task.status = TaskStatus.APPROVED
print("✅ Код одобрен!")
elif review.status == "rejected":
task.status = TaskStatus.REJECTED
print("❌ Код отклонен архитектурно")
else:
task.status = TaskStatus.REWORK
print("🔄 Требуются доработки")
print(f"📊 Вердикт: {review.status}")
if review.comments:
print(f" Найдено замечаний: {len(review.comments)}")
except Exception as e:
task.status = TaskStatus.FAILED
error_msg = f"Reviewer error: {str(e)}"
task.error_log.append(error_msg)
self._add_history(task, "reviewer", "error", error=error_msg, details={"exception": str(e)})
return state
return {"task": task}
def _decide_next_step(self, state: Dict) -> Literal["approved", "rework", "rejected", "failed"]:
"""Решает, что делать дальше"""
def _after_review(self, state: AgentState):
task = state["task"]
review = state["review"]
if task.status == TaskStatus.APPROVED:
return "approved"
elif task.status == TaskStatus.REJECTED:
return "rejected"
elif task.status == TaskStatus.FAILED:
return "failed"
elif task.iteration >= self.max_retries:
print(f"\n⚠️ Достигнут лимит итераций ({self.max_retries}). Останавливаемся.")
task.status = TaskStatus.FAILED
return "failed"
# Определяем статус
if isinstance(review, ReviewResult):
status = review.status
elif isinstance(review, dict):
status = review.get("status")
else:
status = None
if status == "approved":
task.status = TaskStatus.APPROVED
return "approved"
elif status == "changes_requested":
if task.iteration >= state.get("max_iterations", 3):
task.status = TaskStatus.FAILED
return "reject"
task.status = TaskStatus.REWORK
return "rework"
def _finalize_node(self, state: Dict) -> Dict:
"""Финальный узел - подведение итогов"""
task = state["task"]
print("\n" + "="*50)
print("📋 ИТОГИ РАБОТЫ")
print("="*50)
print(f"Задача: {task.requirement[:100]}...")
print(f"Статус: {task.status}")
print(f"Итераций: {task.iteration}")
if task.status == TaskStatus.APPROVED and task.solution:
print(f"\n✅ Решение принято!")
print(f" Файлов: {len(task.solution.files)}")
print(f" Тестов: {len(task.solution.tests)}")
# Сохраняем решение в файлы
self._save_solution(task.solution)
elif task.last_review:
print(f"\n📝 Финальный вердикт: {task.last_review.summary}")
if task.error_log:
print(f"\n❌ Ошибки: {len(task.error_log)}")
for error in task.error_log:
print(f" - {error}")
return {"task": task, "finalized": True}
elif status == "rejected":
if task.iteration >= state.get("max_iterations", 3):
task.status = TaskStatus.FAILED
return "reject"
task.status = TaskStatus.REWORK
return "replan"
else:
task.status = TaskStatus.FAILED
return "reject"
def _save_solution(self, solution: CodeSolution):
"""Сохраняет решение в файлы"""
@@ -287,62 +224,60 @@ class Orchestrator:
f.write(test_content)
print(f" 💾 Сохранен тест: {testpath}")
def run(self, task: str, max_iterations:int = None) -> Dict:
def run(self, task_text: str, max_iterations: int = 3) -> Dict:
"""Запускает процесс разработки"""
task_id = str(uuid.uuid4())
# Создаем объект Task
task = Task(
id=task_id,
requirement=task_text,
status=TaskStatus.PENDING,
iteration=0,
solution=None,
last_review=None,
error_log=[],
history=[]
)
initial_state = {
"task": task,
"plan": [],
"plan": {},
"execution_results": [],
"review": {},
"review": None,
"iteration": 0,
"max_iterations": max_iterations
}
print(f"\n🚀 Запуск мультиагентной системы")
print(f"📝 Задача: {task}")
print(f"📝 Задача: {task_text[:100]}...")
print(f"🆔 ID: {task_id}")
# Запускаем граф
final_state = self.workflow.invoke(initial_state)
return final_state
# Сохраняем решение если успешно
final_task = final_state["task"]
if final_task.status == TaskStatus.APPROVED and final_task.solution:
self._save_solution(final_task.solution)
# main.py
def main():
"""Пример использования"""
print("\n" + "="*50)
print("📋 ИТОГИ РАБОТЫ")
print("="*50)
print(f"Статус: {final_task.status.value}")
print(f"Итераций: {final_task.iteration}")
# Создаем оркестратора
orchestrator = Orchestrator()
if final_task.status == TaskStatus.APPROVED:
print(f"\n✅ Решение принято!")
print(f" Файлов: {len(final_task.solution.files)}")
print(f" Тестов: {len(final_task.solution.tests)}")
elif final_task.last_review:
print(f"\n📝 Финальный вердикт: {final_task.last_review.summary}")
# Пример задачи
requirement = """
Реализуй класс BankAccount с методами:
- deposit(amount): пополнение счета
- withdraw(amount): снятие средств (нельзя снять больше, чем есть)
- get_balance(): получение баланса
- add_interest(rate): добавление процентов (rate в процентах)
if final_task.error_log:
print(f"\n❌ Ошибки: {len(final_task.error_log)}")
for error in final_task.error_log:
print(f" - {error}")
Требования:
- Нельзя создать счет с отрицательным балансом
- Все операции должны быть потокобезопасными (используй threading.Lock)
- Напиши юнит-тесты (pytest) для всех методов, включая граничные случаи
- Добавь docstring для всех методов
"""
# Запускаем процесс
result = orchestrator.run(requirement)
# Анализируем результат
task = result["task"]
if task.status == TaskStatus.APPROVED:
print("\n🎉 Успех! Код принят и сохранен в папке 'generated_code'")
else:
print(f"\n💔 Неудача. Статус: {task.status}")
if task.last_review:
print(f"Причина: {task.last_review.summary}")
if __name__ == "__main__":
main()
return {"task": final_task}

2
requirements.txt
View File

@@ -1,7 +1,7 @@
# requirements.txt
langgraph>=0.3.0
langchain>=0.2.0
langchain>=0.3.0
langchain-community>=0.0.16
langchain-ollama
python-dotenv>=1.0.0