Паттерны AI-агентов

import requests
import json

OLLAMA = "http://localhost:11434/api/chat"

def react_agent(task: str, tools: list, max_steps: int = 5):
    """
    ReAct агент: чередует рассуждение и действия.
    """
    messages = [
        {"role": "system", "content": (
            "Ты — ReAct-агент. Отвечай на запросы пользователя. "
            "Если нужно найти информацию — используй инструменты. "
            "Отвечай только когда у тебя достаточно данных."
        )},
        {"role": "user", "content": task}
    ]
    
    for step in range(max_steps):
        response = requests.post(OLLAMA, json={
            "model": "qwen3.5:4b",
            "messages": messages,
            "tools": tools,
            "stream": False
        })
        msg = response.json()["message"]
        messages.append(msg)
        
        # Если модель вызывает инструмент
        if msg.get("tool_calls"):
            for tc in msg["tool_calls"]:
                name = tc["function"]["name"]
                args = tc["function"]["arguments"]
                print(f"  [Step {step+1}] {name}({json.dumps(args)})")
                
                # Выполняем (в реальности — lookup)
                result = {"result": f"выполнено {name} с {args}"}
                
                messages.append({
                    "role": "tool",
                    "name": name,
                    "content": json.dumps(result)
                })
        else:
            return msg["content"]
    
    return "Не удалось выполнить задачу за отведённое число шагов"

def plan_and_execute(task: str, tools: list):
    """Plan-and-Execute: сначала план, потом выполнение."""
    
    # Фаза 1: создаём план
    plan_prompt = (
        f"Составь подробный план для задачи: {task}\n"
        "Верни план как список шагов в формате JSON.\n"
        "Каждый шаг: {\"step\": \"описание\", \"tool\": \"имя_инструмента\"}"
    )
    
    plan_response = requests.post(OLLAMA, json={
        "model": "qwen3.5:4b",
        "messages": [{"role": "user", "content": plan_prompt}],
        "format": {
            "type": "object",
            "properties": {
                "steps": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "step": {"type": "string"},
                            "tool": {"type": "string"}
                        }
                    }
                }
            }
        },
        "stream": False
    })
    
    plan = json.loads(plan_response.json()["message"]["content"])
    print(f"📋 План: {len(plan['steps'])} шагов")
    
    # Фаза 2: выполняем шаги
    results = []
    for i, step in enumerate(plan["steps"]):
        print(f"  ▶ Шаг {i+1}: {step['step']}")
        
        # Выполняем инструмент или просто спрашиваем модель
        response = requests.post(OLLAMA, json={
            "model": "qwen3.5:4b",
            "messages": [
                {"role": "system", "content": (
                    f"Ты выполняешь шаг {i+1} плана. "
                    f"Контекст: {step['step']}. "
                    f"Предыдущие результаты: {json.dumps(results)}"
                )},
                {"role": "user", "content": step['step']}
            ],
            "tools": tools,
            "stream": False
        })
        result = response.json()["message"]["content"]
        results.append({"step": i+1, "result": result})
    
    return results

def reflection_agent(task: str, iterations: int = 2):
    """Reflection агент: генерирует, критикует, улучшает."""
    
    messages = [
        {"role": "system", "content": "Ты — эксперт. Отвечай на вопросы."},
        {"role": "user", "content": task}
    ]
    
    # Первичная генерация
    response = requests.post(OLLAMA, json={
        "model": "qwen3.5:4b",
        "messages": messages,
        "stream": False
    })
    answer = response.json()["message"]["content"]
    
    for i in range(iterations):
        # Критика
        critique = requests.post(OLLAMA, json={
            "model": "qwen3.5:4b",
            "messages": [
                {"role": "system", "content": (
                    "Ты — строгий критик. Найди ошибки, неточности и "
                    "пропущенные детали в ответе. Будь придирчивым."
                )},
                {"role": "user", "content": f"Задача: {task}\n\nОтвет: {answer}"}
            ],
            "stream": False
        })
        feedback = critique.json()["message"]["content"]
        
        # Улучшение
        improved = requests.post(OLLAMA, json={
            "model": "qwen3.5:4b",
            "messages": [
                {"role": "system", "content": (
                    "Улучши ответ на основе критики. Исправь все ошибки, "
                    "добавь детали. Верни только финальную версию."
                )},
                {"role": "user", "content": (
                    f"Исходный ответ: {answer}\n\n"
                    f"Критика: {feedback}\n\n"
                    f"Улучшенный ответ:"
                )}
            ],
            "stream": False
        })
        answer = improved.json()["message"]["content"]
    
    return answer

def tool_call(model: str, user_input: str, tools: list):
    """Одиночный вызов инструмента — без цикла."""
    response = requests.post(f"{OLLAMA}", json={
        "model": model,
        "messages": [{"role": "user", "content": user_input}],
        "tools": tools,
        "stream": False
    })
    return response.json()["message"]

class Agent:
    """Простой агент с ролью."""
    
    def __init__(self, name: str, role: str, model: str = "qwen3.5:4b"):
        self.name = name
        self.role = role
        self.model = model
    
    def run(self, task: str, context: str = "") -> str:
        response = requests.post(OLLAMA, json={
            "model": self.model,
            "messages": [
                {"role": "system", "content": (
                    f"Ты — {self.name}. Твоя роль: {self.role}."
                )},
                {"role": "user", "content": f"{context}\n\nЗадача: {task}"}
            ],
            "stream": False
        })
        return response.json()["message"]["content"]


class Supervisor:
    """Управляет командой агентов."""
    
    def __init__(self, agents: list[Agent]):
        self.agents = agents
    
    def run(self, project_task: str):
        context = ""
        for agent in self.agents:
            print(f"\n  ▶ {agent.name} работает...")
            result = agent.run(project_task, context)
            print(f"    Результат: {result[:100]}...")
            context += f"\n{agent.name}: {result}"
        return context


# Пример: команда для анализа проекта
pm = Agent("PM", "Менеджер проекта. Ставишь задачи и проверяешь сроки.")
analyst = Agent("Аналитик", "Анализируешь требования и пишешь спецификации.")
dev = Agent("Разработчик", "Пишешь код по спецификации.")

team = Supervisor([pm, analyst, dev])
result = team.run("Создать REST API для управления задачами")

def choose_pattern(task: str):
    """
    Эвристика: какой паттерн подойдёт для задачи.
    """
    if len(task.split()) < 10:
        return "Tool Use"
    elif "проверь" in task.lower() or "улучши" in task.lower():
        return "Reflection"
    elif any(w in task.lower() for w in ["найди", "поищи", "сколько", "какой"]):
        return "ReAct"
    elif any(w in task.lower() for w in ["спланируй", "сделай проект", "разработай"]):
        return "Plan-and-Execute"
    elif any(w in task.lower() for w in ["команда", "агенты", "распредели"]):
        return "Multi-Agent"
    else:
        return "ReAct"