Skills: декларация того, что умеет агент, для роутинга запросов

from dataclasses import dataclass, field
from typing import List, Optional

@dataclass
class AgentSkill:
    id: str
    name: str
    description: str
    examples: List[str]
    tags: List[str] = field(default_factory=list)
    input_modes: List[str] = field(default_factory=lambda: ["text"])
    output_modes: List[str] = field(default_factory=lambda: ["text"])
    constraints: dict = field(default_factory=dict)

# Skill аналитического агента
analyze_data_skill = AgentSkill(
    id="analyze-data",
    name="Анализ и визуализация данных",
    description=(
        "Агент анализирует структурированные данные: CSV, JSON, таблицы. "
        "Строит графики (линейные, столбчатые, тепловые карты), вычисляет "
        "агрегаты (сумма, среднее, медиана), выявляет тренды и аномалии. "
        "Работает с временными рядами, категориальными и числовыми переменными."
    ),
    examples=[
        "Построй график продаж за последние 6 месяцев",
        "Найди аномалии в этом датасете",
        "Какой месяц был самым прибыльным?",
        "Сравни выручку по регионам",
        "Покажи тренд посещаемости по неделям",
        "Вычисли среднее и медиану по колонке revenue",
    ],
    tags=["data", "analytics", "visualization", "statistics"],
    input_modes=["text", "file/csv", "file/json", "data/dataframe"],
    output_modes=["text", "chart", "data/json"],
    constraints={
        "max_file_size_mb": 50,
        "max_rows": 1_000_000,
    }
)

import re
from collections import Counter

class KeywordRouter:
    def __init__(self, registry: "SkillRegistry"):
        self.registry = registry
        self._build_index()

    def _tokenize(self, text: str) -> set[str]:
        # Простая токенизация: строчные слова от 3 символов
        return {w for w in re.findall(r'\b\w+\b', text.lower()) if len(w) >= 3}

    def _build_index(self):
        """Индексируем все skills заранее."""
        self._skill_tokens: dict[str, set[str]] = {}
        for agent_name, skills in self.registry.skills.items():
            for skill in skills:
                tokens = self._tokenize(skill.description)
                for ex in skill.examples:
                    tokens |= self._tokenize(ex)
                tokens |= {t.lower() for t in skill.tags}
                self._skill_tokens[f"{agent_name}::{skill.id}"] = tokens

    def route(self, task: str) -> tuple[str, str, float]:
        """Возвращает (agent_name, skill_id, score)."""
        task_tokens = self._tokenize(task)
        best_key, best_score = None, 0.0

        for key, skill_tokens in self._skill_tokens.items():
            intersection = task_tokens & skill_tokens
            union = task_tokens | skill_tokens
            score = len(intersection) / len(union) if union else 0.0
            if score > best_score:
                best_score = score
                best_key = key

        if best_key:
            agent_name, skill_id = best_key.split("::")
            return agent_name, skill_id, best_score
        return None, None, 0.0

import anthropic
import json

class LLMRouter:
    def __init__(self, registry: "SkillRegistry", model: str = "claude-haiku-4-5-20251001"):
        self.registry = registry
        self.client = anthropic.Anthropic()
        self.model = model  # Используем быструю/дешёвую модель для роутинга

    def _build_skills_summary(self) -> str:
        lines = []
        for agent_name, skills in self.registry.skills.items():
            for skill in skills:
                lines.append(
                    f'- agent="{agent_name}", skill_id="{skill.id}": {skill.description}'
                )
        return "\n".join(lines)

    def route(self, task: str) -> tuple[str, str, float]:
        skills_summary = self._build_skills_summary()

        prompt = f"""Выбери наиболее подходящего агента и skill для задачи.

Доступные агенты и skills:
{skills_summary}

Задача пользователя: {task}

Ответь строго в формате JSON:
{{"agent": "имя_агента", "skill_id": "id_skill", "confidence": 0.0-1.0, "reason": "краткое объяснение"}}

Если ни один skill не подходит, верни {{"agent": null, "skill_id": null, "confidence": 0.0}}"""

        response = self.client.messages.create(
            model=self.model,
            max_tokens=200,
            messages=[{"role": "user", "content": prompt}]
        )

        result = json.loads(response.content[0].text)
        return result["agent"], result["skill_id"], result["confidence"]

import numpy as np
import anthropic

class EmbeddingRouter:
    def __init__(self, registry: "SkillRegistry"):
        self.registry = registry
        self.client = anthropic.Anthropic()
        self._index: list[dict] = []  # [{agent, skill_id, vector}]
        self._build_index()

    def _embed(self, texts: list[str]) -> np.ndarray:
        """Получаем эмбеддинги через Voyage AI (встроен в Anthropic SDK)."""
        response = self.client.embeddings.create(
            model="voyage-3",
            input=texts,
        )
        return np.array([e.embedding for e in response.data])

    def _build_index(self):
        """Строим векторный индекс один раз при запуске."""
        texts_to_embed = []
        metadata = []

        for agent_name, skills in self.registry.skills.items():
            for skill in skills:
                # Конкатенируем description + примеры в один текст для эмбеддинга
                skill_text = skill.description + " " + " ".join(skill.examples)
                texts_to_embed.append(skill_text)
                metadata.append({"agent": agent_name, "skill_id": skill.id})

                # Каждый пример также индексируем отдельно для точности
                for example in skill.examples:
                    texts_to_embed.append(example)
                    metadata.append({"agent": agent_name, "skill_id": skill.id})

        vectors = self._embed(texts_to_embed)
        self._index = [
            {"agent": m["agent"], "skill_id": m["skill_id"], "vector": v}
            for m, v in zip(metadata, vectors)
        ]

    def route(self, task: str) -> tuple[str, str, float]:
        task_vec = self._embed([task])[0]

        best_agent, best_skill_id, best_score = None, None, -1.0
        scores: dict[tuple, float] = {}

        for entry in self._index:
            # Cosine similarity
            score = float(np.dot(task_vec, entry["vector"]) /
                         (np.linalg.norm(task_vec) * np.linalg.norm(entry["vector"]) + 1e-9))
            key = (entry["agent"], entry["skill_id"])
            # Берём максимальный score среди всех текстов skill
            scores[key] = max(scores.get(key, -1.0), score)

        if scores:
            best_key = max(scores, key=scores.__getitem__)
            return best_key[0], best_key[1], scores[best_key]

        return None, None, 0.0

from typing import Optional
import json

class SkillRegistry:
    def __init__(self):
        self.skills: dict[str, list[AgentSkill]] = {}  # agent_name → [skills]
        self._agents: dict[str, any] = {}  # agent_name → agent instance

    def register(self, agent_name: str, agent_instance, skills: list[AgentSkill]):
        """Регистрирует агента вместе с его skills."""
        self.skills[agent_name] = skills
        self._agents[agent_name] = agent_instance

    def unregister(self, agent_name: str):
        self.skills.pop(agent_name, None)
        self._agents.pop(agent_name, None)

    def get_agent(self, agent_name: str):
        return self._agents.get(agent_name)

    def find_by_tag(self, tag: str) -> list[tuple[str, AgentSkill]]:
        """Быстрая фильтрация по тегу перед семантическим поиском."""
        result = []
        for agent_name, skills in self.skills.items():
            for skill in skills:
                if tag in skill.tags:
                    result.append((agent_name, skill))
        return result

    def to_dict(self) -> dict:
        """Сериализация для кэширования и отладки."""
        return {
            agent: [
                {
                    "id": s.id, "name": s.name,
                    "description": s.description,
                    "examples": s.examples,
                    "tags": s.tags,
                }
                for s in skills
            ]
            for agent, skills in self.skills.items()
        }


# Инициализация системы
registry = SkillRegistry()

# Регистрируем агентов (каждый агент знает свои skills)
registry.register("DataAgent", data_agent_instance, [
    AgentSkill(
        id="analyze-data",
        name="Анализ и визуализация данных",
        description="Анализирует CSV/JSON данные, строит графики, вычисляет статистику.",
        examples=["Построй график продаж", "Найди аномалии в данных", "Какой месяц лучший?"],
        tags=["data", "analytics"],
    ),
    AgentSkill(
        id="forecast",
        name="Прогнозирование временных рядов",
        description="Строит прогнозы по историческим данным с помощью статистических моделей.",
        examples=["Спрогнозируй продажи на следующий квартал", "Какой будет трафик в пятницу?"],
        tags=["data", "forecasting"],
    ),
])

registry.register("CodeAgent", code_agent_instance, [
    AgentSkill(
        id="write-code",
        name="Написание кода",
        description="Пишет, рефакторит и отлаживает код на Python, JavaScript, SQL.",
        examples=["Напиши функцию для парсинга JSON", "Отрефакторь этот класс"],
        tags=["code", "development"],
    ),
])

# Инициализируем роутер
router = EmbeddingRouter(registry)

# Роутим задачу
agent_name, skill_id, score = router.route("Покажи динамику выручки за 2024 год")
if score >= 0.65:
    agent = registry.get_agent(agent_name)
    result = await agent.execute(task, skill_id=skill_id)
else:
    result = "Не могу найти подходящего агента для этой задачи"

class HybridRouter:
    """
    Двухэтапный роутинг:
    1. Быстрая фильтрация по тегам (keyword-based)
    2. Семантический поиск среди прошедших фильтр
    """
    def __init__(self, registry: SkillRegistry, threshold: float = 0.65):
        self.registry = registry
        self.threshold = threshold
        self.embedding_router = EmbeddingRouter(registry)

    def _detect_tags(self, task: str) -> list[str]:
        """Быстрое определение домена по ключевым словам."""
        task_lower = task.lower()
        tags = []
        tag_keywords = {
            "data": ["график", "данные", "csv", "таблица", "статистика",
                     "chart", "plot", "анализ", "тренд"],
            "code": ["код", "функция", "класс", "баг", "ошибка",
                     "python", "javascript", "sql", "рефакторинг"],
            "search": ["найди", "поищи", "загрузи", "скачай",
                       "интернет", "сайт", "статья", "новости"],
        }
        for tag, keywords in tag_keywords.items():
            if any(kw in task_lower for kw in keywords):
                tags.append(tag)
        return tags

    async def route(self, task: str) -> tuple[str, str, float]:
        # Шаг 1: определяем теги
        detected_tags = self._detect_tags(task)

        # Шаг 2: если тег найден — ограничиваем пространство поиска
        if detected_tags:
            candidate_agents = set()
            for tag in detected_tags:
                for agent_name, skill in self.registry.find_by_tag(tag):
                    candidate_agents.add(agent_name)
        else:
            candidate_agents = set(self.registry.skills.keys())

        # Шаг 3: embedding-роутинг только среди кандидатов
        # (в реальной системе — отдельный индекс per tag group)
        agent_name, skill_id, score = self.embedding_router.route(task)

        if agent_name in candidate_agents and score >= self.threshold:
            return agent_name, skill_id, score

        # Fallback: LLM routing если embedding не уверен
        if score < self.threshold:
            llm_router = LLMRouter(self.registry)
            return llm_router.route(task)

        return None, None, 0.0