spb/uqo-chat

1"""Tool registry: JSON schemas (OpenAI function-calling format), validation, dispatch."""23from __future__ import annotations45import json6import time7from collections.abc import Awaitable, Callable8from dataclasses import dataclass, field9from pathlib import Path10from typing import Any1112from pydantic import BaseModel, ValidationError1314from app.core.config import Settings15from app.core.logging import get_logger16from app.llm.schemas import ToolResult1718log = get_logger("tools")19SCHEMAS_DIR = Path(__file__).parent / "schemas"202122@dataclass23class ToolContext:24    user_id: str25    user_id_hash: str26    conversation_id: str27    course_code: str28    settings: Settings29    role: str = "student"30    file_ids: list[str] = field(default_factory=list)  # files attached in the conversation31    progress: Callable[[str, str], Awaitable[None]] | None = None  # (status, detail)3233    async def report(self, status: str, detail: str = "") -> None:34        if self.progress:35            await self.progress(status, detail)363738ToolFn = Callable[[dict[str, Any], ToolContext], Awaitable[ToolResult]]394041@dataclass42class ToolSpec:43    name: str44    fn: ToolFn45    args_model: type[BaseModel]46    schema: dict[str, Any]47    heavy: bool = False484950def _schema_hint(schema: dict[str, Any]) -> str:51    props = (schema.get("parameters") or {}).get("properties") or {}52    req = set((schema.get("parameters") or {}).get("required") or [])53    parts = []54    for k, v in props.items():55        t = v.get("type", "any")56        if v.get("enum"):57            t = "|".join(map(str, v["enum"]))58        parts.append(f"{k}{'*' if k in req else ''}: {t}")59    return ", ".join(parts)606162class ToolRegistry:63    def __init__(self) -> None:64        self._tools: dict[str, ToolSpec] = {}6566    def register(self, name: str, fn: ToolFn, args_model: type[BaseModel],67                 heavy: bool = False) -> None:68        schema_path = SCHEMAS_DIR / f"{name}.json"69        schema = json.loads(schema_path.read_text(encoding="utf-8"))70        self._tools[name] = ToolSpec(name, fn, args_model, schema, heavy)7172    def names(self) -> list[str]:73        return list(self._tools)7475    def openai_tools(self, enabled: set[str] | None = None) -> list[dict[str, Any]]:76        return [77            {"type": "function", "function": t.schema}78            for n, t in self._tools.items()79            if enabled is None or n in enabled80        ]8182    def is_heavy(self, name: str) -> bool:83        t = self._tools.get(name)84        return bool(t and t.heavy)8586    async def run(self, name: str, arguments: dict[str, Any], ctx: ToolContext) -> ToolResult:87        spec = self._tools.get(name)88        if spec is None:89            return ToolResult(content=f"Outil inconnu : {name}. Outils disponibles : "90                              f"{', '.join(self._tools)}.", error=True)91        if "__invalid_json__" in arguments:92            return ToolResult(93                content=f"Les arguments de {name} n'étaient pas du JSON valide et n'ont pas pu être "94                        "réparés (souvent : sortie trop longue ou guillemets non échappés dans du "95                        f"code). Attendu : {{{_schema_hint(spec.schema)}}}. Renvoie un appel plus "96                        "court : pour create_excel, utilise un `template` ou moins de lignes ; pour "97                        "execute_python, un code plus concis, ou découpe en plusieurs appels.",98                error=True)99        repaired = bool(arguments.pop("__repaired__", False))100        # drop unknown keys instead of failing (models add helpful extras)101        allowed = set(spec.args_model.model_fields)102        extra = {k: v for k, v in arguments.items() if k not in allowed}103        clean = {k: v for k, v in arguments.items() if k in allowed}104        try:105            args = spec.args_model.model_validate(clean)106        except ValidationError as exc:107            errs = "; ".join(f"{'.'.join(map(str, e['loc']))}: {e['msg']}"108                             for e in exc.errors(include_url=False))109            return ToolResult(110                content=f"Arguments invalides pour {name} — {errs}. Attendu : "111                        f"{{{_schema_hint(spec.schema)}}}. Corrige et réessaie.",112                error=True)113        payload = args.model_dump()114        if extra:115            payload["__extra__"] = extra116        t0 = time.perf_counter()117        try:118            result = await spec.fn(payload, ctx)119        except Exception as exc:  # noqa: BLE001 — error goes back to the model120            log.exception("tool_failed", tool=name)121            result = ToolResult(content=f"L'outil {name} a échoué : {type(exc).__name__}: {exc}. "122                                "Ajuste les arguments (types, cellules, formats) et réessaie ; "123                                "si l'erreur persiste, explique le calcul en texte.", error=True)124        result.meta["duration_ms"] = int((time.perf_counter() - t0) * 1000)125        if repaired:126            result.content += ("\n\n(Note : les arguments JSON étaient tronqués ou malformés et ont "127                               "été réparés automatiquement — vérifie que le résultat est complet.)")128        return result129130131registry = ToolRegistry()132