Инструменты и реестр: как LLM выбирает и вызывает функции

async def search_web(query: str, max_results: int = 5) -> list[dict[str, str]]:
    results = await asyncio.to_thread(_ddg_search, query, max_results)
    return results

def _ddg_search(query: str, max_results: int) -> list[dict]:
    try:
        raw = DDGS().text(query, max_results=max_results) or []
    except (RatelimitException, TimeoutException, DDGSException) as e:
        raise ToolError(str(e), tool_name="search_web") from e

    return [{"url": r["href"], "title": r["title"], "snippet": r["body"]} for r in raw]

async def fetch_pages(urls: list[str]) -> list[dict[str, str]]:
    async with httpx.AsyncClient(timeout=30, follow_redirects=True) as client:
        raw_results = await asyncio.gather(
            *[_fetch_one(client, url) for url in urls],
            return_exceptions=True,   # ← один упавший URL не ломает остальные
        )

    output = []
    for url, result in zip(urls, raw_results):
        if isinstance(result, Exception):
            output.append({"url": url, "error": str(result)})   # LLM увидит ошибку
        else:
            output.append(result)
    return output

async def _fetch_one(client: httpx.AsyncClient, url: str) -> dict:
    response = await client.get(url)
    response.raise_for_status()

    soup = BeautifulSoup(response.text, "html.parser")

    # Удаляем навигацию и шумовые теги
    for tag in soup(["script", "style", "nav", "header", "footer", "aside"]):
        tag.decompose()

    text = soup.get_text(separator="\n", strip=True)

    # Обрезаем до 3000 символов — достаточно для анализа, не перегружает контекст
    if len(text) > 3_000:
        text = text[:3_000] + "\n... [truncated]"

    title = soup.find("title")
    return {"url": url, "title": title.text if title else url, "content": text}

SUMMARIZE_PROMPT = """\
Summarize the following content, focusing on: {focus}
Extract the most important facts, statistics, and insights.
Be concise — aim for 200-400 words.

Content:
{content}
"""

async def summarize_page(content: str, focus: str = "key findings") -> str:
    client = anthropic.AsyncAnthropic(api_key=settings.ANTHROPIC_API_KEY)
    prompt = SUMMARIZE_PROMPT.format(focus=focus, content=content[:8_000])

    message = await client.messages.create(
        model=settings.DEFAULT_MODEL,
        max_tokens=600,
        messages=[{"role": "user", "content": prompt}],
    )
    return message.content[0].text

@dataclass
class ReportResult:
    title: str
    content: str
    sources: list[dict[str, str]]
    word_count: int

async def write_report(
    title: str,
    content: str,
    sources: list[dict[str, str]] | None = None
) -> ReportResult:
    sources = sources or []

    # Добавляем раздел References если есть источники
    if sources:
        refs = "\n\n## References\n\n"
        for i, src in enumerate(sources, 1):
            refs += f"{i}. [{src['title']}]({src['url']})\n"
        content = content.rstrip() + refs

    return ReportResult(
        title=title,
        content=content,
        sources=sources,
        word_count=len(content.split())
    )

class ToolRegistry:
    def __init__(self):
        self._schemas = {s["name"]: s for s in TOOL_SCHEMAS}
        self._dispatch = TOOL_DISPATCH   # dict: name → async function

    def get_schemas(self) -> list[dict]:
        """Вернуть все схемы для передачи в LLM."""
        return list(self._schemas.values())

    async def dispatch(self, tool_name: str, **kwargs) -> Any:
        """Найти и вызвать инструмент по имени."""
        if tool_name not in self._dispatch:
            raise ToolError(f"Unknown tool: {tool_name}")

        kwargs = _normalize_arg_names(tool_name, kwargs)  # Шаг 1
        kwargs = _coerce_args(kwargs, self._schemas[tool_name]["input_schema"])  # Шаг 2
        return await self._dispatch[tool_name](**kwargs)  # Шаг 3

_ARG_ALIASES = {
    "fetch_pages": {
        "url_list":  "urls",     # Llama3 называет так
        "url":       "urls",     # единственное число
        "page_urls": "urls",
        "links":     "urls",
    },
    "search_web": {
        "search_query": "query", # частое отклонение
        "q":            "query",
        "keywords":     "query",
        "num_results":  "max_results",
    },
    ...
}

def _normalize_arg_names(tool_name: str, kwargs: dict) -> dict:
    aliases = _ARG_ALIASES.get(tool_name, {})
    return {aliases.get(k, k): v for k, v in kwargs.items()}

def _coerce_args(kwargs: dict, input_schema: dict) -> dict:
    """Привести типы аргументов к ожидаемым по схеме."""
    properties = input_schema.get("properties", {})
    result = dict(kwargs)

    for key, value in list(result.items()):
        prop_type = properties.get(key, {}).get("type")

        if prop_type == "integer" and not isinstance(value, int):
            result[key] = int(value)           # "10" → 10

        elif prop_type == "array":
            if isinstance(value, str):
                try:
                    parsed = json.loads(value) # '["a","b"]' → ["a", "b"]
                    if isinstance(parsed, list):
                        result[key] = parsed
                except json.JSONDecodeError:
                    result[key] = [value]      # одна строка → список
            elif not isinstance(value, list):
                result[key] = [value]          # любой объект → список

    return result

from tools.registry import ToolError

# ❌ НЕПРАВИЛЬНО — оркестратор упадёт с необработанным исключением
async def bad_tool(query: str) -> list:
    if not query:
        raise ValueError("Query is empty")  # оркестратор не знает об этом

# ✅ ПРАВИЛЬНО — оркестратор перехватит, LLM увидит ошибку и адаптируется
async def good_tool(query: str) -> list:
    if not query:
        raise ToolError("Query is empty", tool_name="good_tool")

# В оркестраторе (invariant 3):
try:
    result = await self.registry.dispatch(tool_name, **tool_input)
    return json.dumps(result, ...)
except ToolError as e:
    log.warning("tool_error", tool=tool_name, error=str(e))
    return f"Error executing {tool_name}: {e}"  # ← строка уходит в контекст LLM

# tools/extract_keywords.py
from __future__ import annotations
import re
from collections import Counter
import structlog
from tools.registry import ToolError

log = structlog.get_logger(__name__)

async def extract_keywords(text: str, max_keywords: int = 10) -> list[str]:
    """Извлечь ключевые слова из текста по частоте.

    Args:
        text: Текст для анализа.
        max_keywords: Количество ключевых слов (1-20). Default: 10.

    Returns:
        Список ключевых слов по убыванию частоты.
    """
    if not text or len(text.strip()) < 10:
        raise ToolError("Text is too short", tool_name="extract_keywords")

    # Только слова длиннее 3 символов, без стоп-слов
    words = re.findall(r'\b[a-zA-Zа-яА-Я]{4,}\b', text.lower())
    stop_words = {"this", "that", "with", "from", "have", "been", "they"}
    filtered = [w for w in words if w not in stop_words]

    if not filtered:
        raise ToolError("No keywords found", tool_name="extract_keywords")

    keywords = [word for word, _ in Counter(filtered).most_common(max_keywords)]
    log.info("keywords_extracted", count=len(keywords))
    return keywords

# В tools/registry.py, добавить в список TOOL_SCHEMAS:
{
    "name": "extract_keywords",
    "description": (
        "Extract key terms and concepts from text content. "
        "Use this to identify main topics before writing the report."
    ),
    "input_schema": {
        "type": "object",
        "properties": {
            "text": {
                "type": "string",
                "description": "The text content to analyze.",
            },
            "max_keywords": {
                "type": "integer",
                "description": "Max keywords to return (1-20). Default: 10.",
                "default": 10,
            },
        },
        "required": ["text"],
    },
},

# В tools/registry.py, в функции _register_tools():
def _register_tools() -> None:
    from tools.search import search_web
    from tools.fetch import fetch_pages
    from tools.summarize import summarize_page
    from tools.report import write_report
    from tools.extract_keywords import extract_keywords   # ← добавить

    TOOL_DISPATCH.update({
        "search_web":       search_web,
        "fetch_pages":      fetch_pages,
        "summarize_page":   summarize_page,
        "write_report":     write_report,
        "extract_keywords": extract_keywords,             # ← добавить
    })

# В tests/test_tools.py:
import pytest
from tools.registry import ToolRegistry, ToolError

@pytest.mark.anyio
async def test_extract_keywords_returns_list():
    from tools.extract_keywords import extract_keywords
    result = await extract_keywords(
        "Retrieval-Augmented Generation combines retrieval with generation models.",
        max_keywords=5
    )
    assert isinstance(result, list)
    assert len(result) <= 5
    assert all(isinstance(k, str) for k in result)

@pytest.mark.anyio
async def test_extract_keywords_raises_on_empty():
    from tools.extract_keywords import extract_keywords
    with pytest.raises(ToolError):
        await extract_keywords("")

@pytest.mark.anyio
async def test_extract_keywords_via_registry():
    registry = ToolRegistry()
    result = await registry.dispatch("extract_keywords", text="Machine learning models.")
    assert isinstance(result, list)

python3 -c "from tools.registry import ToolRegistry; print(ToolRegistry().list_tools())"
# ['search_web', 'fetch_pages', 'summarize_page', 'write_report', 'extract_keywords']

pytest tests/test_tools.py -v -k "keyword"
# test_extract_keywords_returns_list PASSED
# test_extract_keywords_raises_on_empty PASSED
# test_extract_keywords_via_registry PASSED