Tools & Function Calling

Les outils donnent au LLM un pouvoir d’action : API, SQL, code, RAG. Un bon système repose sur des contrats stricts, un dispatcher fiable, des policies robustes et une observabilité fine.

Contrats JSON Schema Dispatcher & Validation Guardrails & Policies P95, coût, taux d’erreur

Définition

Décrit entrées/sorties (JSON Schema / Pydantic) + contraintes.
Types, enum, formats, plages (minimum/maximum), required.
Versionner le schéma (schema_version).

OpenAI Function Calling (ex.)

{
 "name": "weather_get",
 "description": "Météo actuelle d'une ville",
 "parameters": {
   "type": "object",
   "properties": {
     "city": {"type":"string"},
     "units":{"type":"string","enum":["metric","imperial"],"default":"metric"}
   },
   "required":["city"]
 }
}

Pydantic (validation forte)

class WeatherIn(BaseModel):
    city: str
    units: Literal["metric","imperial"] = "metric"

class WeatherOut(BaseModel):
    ok: bool
    temp: float
    desc: str

Rôle

Reçoit la prédiction LLM (nom d’outil + arguments JSON).
Valide & sanitise, exécute la fonction, gère timeouts/retry.
Retourne observation structurée (+ usage/coût/latence).

Esquisse Python

def dispatch(call: dict):
    tool = TOOL_REGISTRY[call["name"]]
    # validate inputs
    args = tool.InputModel(**call["arguments"])
    policies.pre_check(tool, args)
    # timeout + retry
    obs = retry(tool.timeout, tool.retries)(tool.fn)(**args.dict())
    out = tool.OutputModel(**obs)
    policies.post_check(tool, out)
    return {"ok": True, "data": out.dict(), "usage": {...}}

Principes

Fonctions pures et déterministes si possible.
Idempotence et lecture seule par défaut ; write = opt-in.
Déclarer coûts (cost_class) & SLA.

Registre

TOOL_REGISTRY = {
 "weather_get": Tool(fn=weather_get, InputModel=WeatherIn, OutputModel=WeatherOut,
                     timeout=5, retries=1, cost_class="low"),
 "run_sql":     Tool(fn=run_sql,     InputModel=SqlIn,      OutputModel=SqlOut,
                     timeout=8, retries=0, cost_class="mid"),
}

Convention d’output

{"ok": true, "data": {...}, "usage":{"latency_ms":132, "cost_usd":0.0007}}

Contrôles clés

PII/secrets filtering, allow-list d’hôtes, RBAC/quotas.
Limiteurs : max_calls/turn, step_timeout, global_timeout.
Journalisation des actions sensibles (audit).

Exemple de policy

{
 "pii":"mask", "allow_hosts":["api.weather.y"], "rbac":{"run_sql":["admin"]},
 "limits":{"max_calls_per_turn":3,"global_timeout_s":60}
}

Fail-safe : si policy échoue → stop + réponse partielle.

Traces & métriques

{"span":"tool.call","name":"run_sql","args_hash":"...","ok":true,
 "latency_ms":180,"cost_usd":0.0018,"error":null}

KPIs : latence P95, taux d’erreur, cost/answer.

OpenTelemetry (pseudo)

with tracer.start_as_current_span("tool.call") as sp:
    sp.set_attribute("tool", name); sp.set_attribute("ok", ok)

Interop

OpenAI Function Calling / JSON mode.
LangChain / LlamaIndex : wrappers + validation.
DSPy / Guidance : I/O déclaratifs & coercition.
NeMo Guardrails / Outlines : policies, schemas.

Ex. LangChain

@tool
def weather_get(city:str, units:str="metric") -> str:
    "Return weather JSON"
    ...

agent = initialize_agent(tools=[weather_get, run_sql], llm=llm, agent="openai-functions")

Ex. JSON mode

{"tool":"weather_get","arguments":{"city":"Paris","units":"metric"}}

class WeatherIn(BaseModel):
    city: str
    units: Literal["metric","imperial"] = "metric"

class WeatherOut(BaseModel):
    ok: bool
    temp: float
    desc: str

@tool(name="weather_get", InputModel=WeatherIn, OutputModel=WeatherOut)
def weather_get(city: str, units: str="metric") -> dict:
    url = f"https://api.weather.y/v1?city={quote(city)}&units={units}"
    resp = http_get(url, timeout=4, allow_hosts=["api.weather.y"])
    data = parse_json(resp)
    return {"ok": True, "temp": data["main"]["temp"], "desc": data["weather"][0]["description"]}

Validation stricte, allow-list d’hôtes, timeout court, output structuré.

class SqlIn(BaseModel):
    sql: constr(regex=r"^\\s*SELECT\\b.+", max_length=2000)  # read-only
    params: Optional[List[str]] = None

class SqlOut(BaseModel):
    ok: bool
    rows: List[Dict[str, Any]]

@tool(name="run_sql", InputModel=SqlIn, OutputModel=SqlOut, rbac=["admin","analyst"])
def run_sql(sql: str, params: Optional[List[str]]=None) -> dict:
    rows = db.query(sql, params=params, timeout=6)            # paramétré
    return {"ok": True, "rows": rows}

Lecture seule obligatoire, regex sur SELECT, RBAC, timeout DB.

SAFE_BUILTINS = {"len":len, "sum":sum, "min":min, "max":max, "range":range}
def py_exec(code:str, inputs:dict) -> dict:
    with sandbox(cpu_ms=200, mem_mb=128, net=False, fs=False) as sb:
        env = {"__builtins__": SAFE_BUILTINS, "inputs": inputs}
        out = sb.exec(code, env)      # pas d'import, pas d'I/O
    return {"ok": True, "result": out}

Sandbox stricte : CPU/mémoire, pas d’I/O, pas d’import dynamique.

Tools & Function Calling

Schéma d’outil (contract)

Dispatcher / Router

Outils (implementations)

Guardrails & Policies

Observabilité

Standards & écosystème

Exemple : API météo

Exemple : SQL paramétré

Exemple : Python sandbox

Stratégies, Éval & Anti-patterns