Вызов функций (Tool Calling)

curl -s https://speshu.ai/api/v1/chat/completions \
  -H "Authorization: Bearer $SPESHU_API_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "'"$TEXT_MODEL"'",
    "parallel_tool_calls": true,
    "tools": [{
      "type": "function",
      "function": {
        "name": "get_weather",
        "description": "Текущая погода в городе",
        "parameters": {
          "type": "object",
          "properties": {"city": {"type": "string"}},
          "required": ["city"]
        }
      }
    }],
    "messages": [{"role": "user", "content": "Сравни погоду в Москве, Казани и Сочи."}]
  }' | jq '.choices[0].tool_calls'

import json
from typing import Any

from pydantic import BaseModel, ValidationError


class WeatherArgs(BaseModel):
    city: str
    unit: str = "celsius"


def parse_arguments(raw: str) -> dict[str, Any]:
    """Разбор аргументов вызова с валидацией по схеме."""
    try:
        payload = json.loads(raw)
    except json.JSONDecodeError as exc:
        raise ValueError(f"аргументы — не JSON: {raw!r}") from exc

    try:
        return WeatherArgs.model_validate(payload).model_dump()
    except ValidationError as exc:
        raise ValueError(f"аргументы не прошли схему: {exc.errors()}") from exc

import json


def run_tool(name: str, raw_arguments: str) -> str:
    try:
        args = parse_arguments(raw_arguments)
    except ValueError as exc:
        # модель должна увидеть, что она делает не так
        return json.dumps({"error": str(exc)})

    try:
        result = DISPATCH[name](**args)
    except TimeoutError:
        return json.dumps({"error": "источник данных не ответил вовремя"})
    except Exception as exc:  # noqa: BLE001 — ошибку показываем модели, а не логируем и теряем
        return json.dumps({"error": f"{type(exc).__name__}: {exc}"})

    return json.dumps(result, ensure_ascii=False)

import json

import requests

# буфер накапливаемых вызовов, ключ — index
buffer: dict[int, dict] = {}


def apply_delta(chunk: dict) -> None:
    for call in chunk.get("choices", [{}])[0].get("delta", {}).get("tool_calls", []) or []:
        slot = buffer.setdefault(
            call["index"],
            {"id": None, "name": "", "arguments": ""},
        )
        if call.get("id"):
            slot["id"] = call["id"]
        function = call.get("function") or {}
        if function.get("name"):
            slot["name"] += function["name"]
        if function.get("arguments"):
            slot["arguments"] += function["arguments"]


# ... внутри цикла чтения потока, после проверки на "error" и "[DONE]":
#     apply_delta(chunk)
#     if chunk["choices"][0].get("finish_reason") == "tool_calls":
#         calls = [json.loads(v["arguments"]) for v in buffer.values()]

import json
import os
from typing import Any

import requests
from pydantic import BaseModel, ValidationError

BASE = "https://speshu.ai/api/v1"
HEADERS = {
    "Authorization": f"Bearer {os.environ['SPESHU_API_KEY']}",
    "Content-Type": "application/json",
}
MODEL = os.environ["TEXT_MODEL"]
MAX_STEPS = 8


class GetWeather(BaseModel):
    city: str


def get_weather(city: str) -> dict[str, Any]:
    """Заглушка: в проде здесь ваш настоящий вызов."""
    return {"city": city, "temp_c": 14, "sky": "облачно"}


def get_exchange_rate(pair: str) -> dict[str, Any]:
    return {"pair": pair, "rate": 92.4, "as_of": "2026-10-02"}


class GetExchangeRate(BaseModel):
    pair: str


TOOLS = [
    {
        "type": "function",
        "function": {
            "name": "get_weather",
            "description": "Текущая погода в городе. Используй, когда спрашивают о погоде.",
            "parameters": {
                "type": "object",
                "properties": {"city": {"type": "string"}},
                "required": ["city"],
                "additionalProperties": False,
            },
        },
    },
    {
        "type": "function",
        "function": {
            "name": "get_exchange_rate",
            "description": "Курс валютной пары, например USD/RUB.",
            "parameters": {
                "type": "object",
                "properties": {"pair": {"type": "string"}},
                "required": ["pair"],
                "additionalProperties": False,
            },
        },
    },
]

SCHEMAS = {"get_weather": GetWeather, "get_exchange_rate": GetExchangeRate}
IMPL = {"get_weather": get_weather, "get_exchange_rate": get_exchange_rate}


def run_tool(name: str, raw_arguments: str) -> str:
    schema = SCHEMAS.get(name)
    if schema is None:
        return json.dumps({"error": f"неизвестный инструмент: {name}"})

    try:
        args = schema.model_validate_json(raw_arguments).model_dump()
    except ValidationError as exc:
        return json.dumps({"error": f"аргументы не прошли схему: {exc.errors()}"})

    try:
        return json.dumps(IMPL[name](**args), ensure_ascii=False)
    except Exception as exc:  # ошибку возвращаем модели, а не теряем
        return json.dumps({"error": f"{type(exc).__name__}: {exc}"})


def ask(question: str) -> str:
    messages: list[dict[str, Any]] = [{"role": "user", "content": question}]

    for _ in range(MAX_STEPS):
        response = requests.post(
            f"{BASE}/chat/completions",
            headers=HEADERS,
            json={"model": MODEL, "messages": messages, "tools": TOOLS},
            timeout=300,
        )
        response.raise_for_status()
        body = response.json()

        message = body["choices"][0]["message"]
        messages.append(message)

        if message.get("tool_calls"):
            for call in message["tool_calls"]:
                messages.append(
                    {
                        "role": "tool",
                        "tool_call_id": call["id"],
                        "name": call["function"]["name"],
                        "content": run_tool(
                            call["function"]["name"], call["function"]["arguments"]
                        ),
                    }
                )
            continue

        return message["content"] or ""

    raise RuntimeError(f"агент не уложился в {MAX_STEPS} шагов")


print(ask("Какая сейчас погода в Москве и какой курс USD/RUB?"))