Multi-Agent
Systems

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict, List

class AgentState(TypedDict):
    messages: List
    task:     str
    plan:     str
    result:   str
    attempts: int

# Узлы графа — каждый вызывает своего агента
def planner_node(state: AgentState) -> AgentState:
    plan = planner_agent.invoke(state["task"])
    return {**state, "plan": plan}

def researcher_node(state: AgentState) -> AgentState:
    findings = researcher_agent.invoke(state["plan"])
    state["messages"].append({"role": "researcher", "content": findings})
    return state

def coder_node(state: AgentState) -> AgentState:
    code = coder_agent.invoke(state["messages"])
    return {**state, "result": code}

def critic_node(state: AgentState) -> AgentState:
    verdict = critic_agent.invoke(state["result"])
    state["attempts"] += 1
    return {**state, "messages": [*state["messages"], verdict]}

# Условный переход: повторить или завершить
def should_retry(state: AgentState) -> str:
    if "APPROVED" in state["messages"][-1]["content"]:
        return "end"
    if state["attempts"] >= 3:
        return "end"  # предохранитель
    return "retry"

# Сборка графа
graph = StateGraph(AgentState)
graph.add_node("planner",    planner_node)
graph.add_node("researcher", researcher_node)
graph.add_node("coder",      coder_node)
graph.add_node("critic",     critic_node)

graph.set_entry_point("planner")
graph.add_edge("planner",    "researcher")
graph.add_edge("researcher", "coder")
graph.add_edge("coder",      "critic")
graph.add_conditional_edges("critic", should_retry, {
    "retry": "coder",
    "end":   END
})

app = graph.compile(checkpointer=MemorySaver())

{
  "name": "CodeReviewAgent",
  "description": "Проверяет Python-код на ошибки и стиль",
  "version": "1.2.0",
  "url": "https://agents.internal/code-review",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false,
    "stateTransitionHistory": true
  },
  "skills": [
    {
      "id": "review_python",
      "name": "Review Python code",
      "inputModes":  ["text"],
      "outputModes": ["text", "data"],
      "examples": ["review this function for bugs"]
    }
  ],
  "authentication": {
    "schemes": ["Bearer"]
  }
}

import httpx
import json

class A2AClient:
    def __init__(self, agent_url: str):
        self.base = agent_url
        self.client = httpx.AsyncClient()

    async def discover(self) -> dict:
        # Получить Agent Card
        r = await self.client.get(f"{self.base}/.well-known/agent.json")
        return r.json()

    async def send_task(self, message: str,
                        session_id: str = None) -> dict:
        payload = {
            "jsonrpc": "2.0",
            "method":  "tasks/send",
            "id":      1,
            "params": {
                "sessionId": session_id,
                "message": {
                    "role": "user",
                    "parts": [{"text": message}]
                }
            }
        }
        r = await self.client.post(
            f"{self.base}/a2a",
            json=payload,
            headers={"Authorization": "Bearer <token>"}
        )
        return r.json()["result"]

# Использование: оркестратор вызывает воркер-агента
async def orchestrate():
    client = A2AClient("http://code-review-agent:8080")
    card   = await client.discover()
    print(f"Agent: {card['name']}, skills: {[s['id'] for s in card['skills']]}")
    result = await client.send_task("Review this code: def f(x): return x*x")
    print(result)

apiVersion: apps/v1
kind: Deployment
metadata:
  name: mas-orchestrator
  namespace: agents
spec:
  replicas: 2
  selector:
    matchLabels: { app: orchestrator }
  template:
    metadata:
      labels: { app: orchestrator }
      annotations:
        prometheus.io/scrape: "true"
        prometheus.io/port:   "9090"
    spec:
      containers:
      - name: orchestrator
        image: registry.internal/mas-orchestrator:v1.4.2
        env:
        - { name: LLM_BASE_URL,    value: "http://vllm-service:8000/v1" }
        - { name: REDIS_URL,       value: "redis://redis-cluster:6379" }
        - { name: VECTOR_DB_URL,  value: "http://qdrant:6333" }
        - { name: LOG_LEVEL,       value: "INFO" }
        resources:
          requests: { cpu: "500m", memory: "1Gi" }
          limits:   { cpu: "2000m", memory: "4Gi" }
        livenessProbe:
          httpGet: { path: /health, port: 8080 }
          initialDelaySeconds: 15
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: worker-hpa
  namespace: agents
spec:
  scaleTargetRef:
    apiVersion: apps/v1
    kind: Deployment
    name: mas-worker
  minReplicas: 2
  maxReplicas: 20
  metrics:
  - type: External
    external:
      metric:
        name: redis_queue_length  # кастомная метрика из Prometheus
      target: { type: Value, value: "5" } # 1 воркер на 5 задач

#!/bin/bash
# GPU VM: 2x A100 80GB, запуск vLLM как сервиса

# 1. Установка
pip install vllm --break-system-packages

# 2. Запуск модели с tensor parallelism
python -m vllm.entrypoints.openai.api_server \
  --model mistralai/Mixtral-8x7B-Instruct-v0.1 \
  --tensor-parallel-size 2 \
  --max-model-len 32768 \
  --quantization awq \
  --host 0.0.0.0 \
  --port 8000 \
  --served-model-name mixtral \
  --enable-prefix-caching \      # кешировать системный промпт
  --max-num-seqs 64             # concurrent requests

# 3. Systemd unit для автозапуска
cat > /etc/systemd/system/vllm.service <<EOF
[Unit]
Description=vLLM Inference Server
After=network.target

[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu
ExecStart=/home/ubuntu/.local/bin/python -m vllm.entrypoints.openai.api_server \
  --model /models/mixtral --tensor-parallel-size 2 --port 8000
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target
EOF

systemctl enable --now vllm

ORCHESTRATOR_SYSTEM = """
Ты — Orchestrator в multi-agent системе.
Твоя роль: планирование и делегирование. Никогда не решай задачи напрямую.

ДОСТУПНЫЕ АГЕНТЫ:
- researcher: поиск информации, веб-поиск, анализ документов
- coder: написание и исправление кода на Python/JS/Go
- critic: проверка кода, поиск ошибок, code review
- executor: запуск команд, работа с файловой системой
- writer: написание текстов, документации

ПРОЦЕСС:
1. Проанализируй задачу. Ответь себе: что нужно на входе? что на выходе?
2. Разбей на шаги. Определи зависимости.
3. Создай план в JSON формате с полями:
   - task_id, agent, input, depends_on[], priority
4. Передай план через tool: execute_plan(plan)
5. После получения результатов — агрегируй их.

ПРАВИЛА:
- Не превышай 7 подзадач в одном плане
- Параллели всё, что можно (depends_on: [])
- При ошибке воркера — попробуй другого или уточни задачу
- Всегда запрашивай critic после coder
- Устанавливай timeout на каждую задачу
"""

DECOMPOSE_PROMPT = """
Задача: {task}

Создай план выполнения. Формат JSON:
{{
  "plan": [
    {{
      "task_id": "t1",
      "agent": "researcher",
      "input": "найди информацию о ...",
      "depends_on": [],
      "priority": 1,
      "timeout_seconds": 60
    }},
    ...
  ],
  "expected_output": "описание финального результата"
}}
"""

CODER_SYSTEM = """
Ты — Coder Agent. Специализируешься на написании Python-кода.

ВХОД: Описание задачи + контекст от researcher

ВЫХОД (строго JSON):
{
  "code": "...",          // готовый Python код
  "explanation": "...",   // что делает код (1-3 предложения)
  "tests": "...",         // pytest тесты
  "dependencies": [],     // pip пакеты
  "confidence": 0.0-1.0   // уверенность в решении
}

ИНСТРУМЕНТЫ:
- execute_python(code: str) → stdout, stderr — запустить и проверить
- search_docs(query: str) → str — найти в документации

ПРАВИЛА:
1. ВСЕГДА запускай код через execute_python перед отправкой
2. Если confidence < 0.6 — верни {"error": "недостаточно контекста", "needs": [...]}
3. Максимум 5 попыток выполнения
4. Никогда не используй subprocess.run с shell=True

ПРИМЕР хорошего ответа:
{
  "code": "def parse_csv(path: str) -> list[dict]:\\n    ...",
  "explanation": "Читает CSV файл и возвращает список словарей",
  "tests": "def test_parse_csv():\\n    ...",
  "dependencies": ["pandas"],
  "confidence": 0.95
}
"""

from langchain.agents import create_react_agent
from langchain_core.tools import tool

@tool
def execute_python(code: str) -> str:
    """Выполнить Python код в sandbox и вернуть вывод"""
    import subprocess
    result = subprocess.run(
        ["python", "-c", code],
        capture_output=True, text=True, timeout=30
    )
    return result.stdout + (f"\nERROR: {result.stderr}" if result.stderr else "")

coder_agent = create_react_agent(
    model=llm,
    tools=[execute_python, search_docs],
    prompt=ChatPromptTemplate.from_messages([
        ("system", CODER_SYSTEM),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
)

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
import time, uuid

tracer = trace.get_tracer("mas.agent")

class TracedAgent:
    def __init__(self, name: str, agent):
        self.name  = name
        self.agent = agent

    async def invoke(self, task: str, parent_trace_id: str = None):
        task_id = str(uuid.uuid4())[:8]

        with tracer.start_as_current_span(
            f"agent.{self.name}",
            attributes={
                "agent.name":         self.name,
                "task.id":            task_id,
                "task.input_length":  len(task),
                "parent.trace_id":    parent_trace_id or "root",
            }
        ) as span:
            start = time.time()
            try:
                result = await self.agent.ainvoke({"input": task})
                span.set_attribute("task.status", "success")
                span.set_attribute("task.output_length", len(str(result)))
                return result
            except Exception as e:
                span.set_attribute("task.status", "error")
                span.set_attribute("error.message", str(e))
                span.record_exception(e)
                raise
            finally:
                duration = time.time() - start
                span.set_attribute("task.duration_s", duration)
                task_duration.observe(duration)  # prometheus

import asyncio, time
from enum import Enum

class State(Enum):
    CLOSED = "closed"  # норма
    OPEN   = "open"    # блокируем вызовы
    HALF   = "half"    # пробный вызов

class LLMCircuitBreaker:
    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = State.CLOSED
        self.failures   = 0
        self.threshold  = failure_threshold
        self.timeout    = recovery_timeout
        self.opened_at  = None
        self.fallback_url = "https://api.openai.com/v1"  # облачный fallback

    async def call(self, llm_fn, *args, **kwargs):
        if self.state == State.OPEN:
            if time.time() - self.opened_at > self.timeout:
                self.state = State.HALF
            else:
                # Переключиться на fallback LLM
                return await self._fallback_call(*args, **kwargs)
        try:
            result = await asyncio.wait_for(llm_fn(*args, **kwargs), timeout=90)
            self._on_success()
            return result
        except (Exception, asyncio.TimeoutError) as e:
            self._on_failure()
            raise

    def _on_success(self):
        self.failures = 0
        self.state = State.CLOSED

    def _on_failure(self):
        self.failures += 1
        if self.failures >= self.threshold:
            self.state = State.OPEN
            self.opened_at = time.time()