Cinque pattern di function calling che hanno retto in produzione

Perché conta

L'uso dei tool è un punto di errore frequente nei sistemi LLM che costruisco o audito. Il problema non dipende solo dal ragionamento del modello o dal disegno del prompt. Spesso nasce nello spazio tra "il modello sa quale tool chiamare" e "il tool esegue e restituisce qualcosa di utile". In questi sistemi ricorrono cinque pattern. Ricorrono anche tre anti-pattern che portano a regressioni di latenza, overflow della context window e agent che entrano in loop indefiniti durante esecuzioni non presidiate.

Ho lavorato su agent potenziati con tool per workflow finanziari, pipeline di ricerca e sistemi multi-agent. I pattern sotto sono quelli che uso in produzione: cinque approcci che hanno retto in sistemi che ho costruito o auditato, più le alternative che hanno fallito.

1. Contratti dei tool schema-first

Un anti-pattern comune è definire una funzione Python, decorarla con @tool e fidarsi che l'LLM passi argomenti validi. Funziona finché smette di funzionare. Input come max_results: "ten", date: "last week" o query: null producono un TypeError da qualche parte nell'executor. Se mi limito a catturare quell'errore, loggarlo e continuare, ho reso il tool silenziosamente inaffidabile.

Lo evito dando a ogni tool uno schema che valida gli input prima dell'esecuzione: un modello Pydantic nell'agent framework, un JSON Schema nella definizione del tool del provider, oppure una modalità "strict" / structured outputs del provider che rifiuta argomenti non conformi al boundary dell'API. Lo schema documenta anche il tool: è ciò che il modello legge quando decide come chiamarlo. Le API dei framework cambiano (args_schema=…, inferenza da type annotation, oggetti parameters compatibili con OpenAI); l'invariante no.

# Anti-pattern: raw function with no input schema
@tool
def search_documents(query: str, max_results: int = 5):
    # LLM can pass anything — no validation before executor runs
    return document_store.search(query, limit=max_results)

# Pattern: explicit schema with field constraints and descriptions
class SearchInput(BaseModel):
    query: str = Field(description="Search terms for semantic retrieval. No boolean operators.")
    max_results: int = Field(default=5, ge=1, le=20, description="Number of documents to return.")
    date_filter: str | None = Field(default=None, description="ISO 8601 date prefix, e.g. '2024-Q1'.")

@tool(args_schema=SearchInput)  # or the current framework equivalent
def search_documents(query: str, max_results: int, date_filter: str | None):
    results = document_store.search(query, limit=max_results)
    if date_filter:
        results = [r for r in results if r.date.startswith(date_filter)]
    return results

Lo schema svolge tre compiti: validazione, documentazione e type coercion. Intercetta gli input non validi prima che l'executor venga eseguito. Espone i campi description che il modello usa per costruire chiamate valide. Può anche convertire valori come max_results: "5" in 5, quando il runtime lo consente. Una funzione nuda non fa nulla di tutto questo. Quando il provider supporta l'aderenza strict allo schema sugli argomenti dei tool, mantengo comunque un controllo a livello applicativo: il wire può rifiutare forme errate, ma le business rule, come range, enum e auth scope, restano nel mio codice.

Le descrizioni dei campi non sono commenti. Sono parte della specifica del tool che l'LLM riceve. Descrizioni poco chiare producono chiamate poco chiare. Gli schemi più utili che ho scritto somigliano a brevi riferimenti API: cosa significa ogni campo, quale formato si aspetta, quali vincoli si applicano e qual è il default. Quando vedo chiamate tool malformate in produzione, riscrivo le descrizioni dei campi prima di cambiare modello o prompt.

2. Union di errori tipizzati, non traceback

Quando un tool fallisce, l'approccio ingenuo è catturare l'eccezione e restituire il traceback come stringa. Il modello legge il traceback, può inferire l'errore e prova una chiamata diversa. Nella mia esperienza, a volte funziona. Il problema inizia quando il traceback diventa l'interfaccia di recovery.

Un traceback Python è un contesto di errore debole per un modello. È non strutturato, verboso e pieno di nomi interni che il modello potrebbe interpretare male. Inoltre gonfia la context window a ogni retry. Se l'errore è transitorio, come un rate limit o un network timeout, il modello potrebbe non sapere se fare retry, back off o abort. Nei sistemi che ho osservato, questa ambiguità può causare retry immediati e cascate.

# Anti-pattern: exception trace as error recovery context
try:
    result = search_documents(query=query, max_results=max_results)
    return result
except Exception as e:
    return f"Error: {traceback.format_exc()}"  # 20 lines of internal stack trace

# Pattern: typed error union in the return schema
class SearchResult(BaseModel):
    status: Literal["ok", "empty", "rate_limited", "auth_error"]
    documents: list[Document] = []
    retry_after_s: int | None = None  # set when status == "rate_limited"
    error_detail: str | None = None   # set on auth_error

def search_documents(query: str, max_results: int) -> SearchResult:
    try:
        docs = document_store.search(query, limit=max_results)
        if not docs:
            return SearchResult(status="empty")
        return SearchResult(status="ok", documents=docs)
    except RateLimitError as e:
        return SearchResult(status="rate_limited", retry_after_s=e.retry_after)
    except AuthError:
        return SearchResult(status="auth_error", error_detail="API key invalid or expired.")

Il modello può ragionare su status: rate_limited e retry_after_s: 30 perché quella struttura corrisponde a pattern API comuni. Uno stack trace gli dà materiale meno affidabile. La typed error union rende anche esplicito il contratto del tool: queste sono le modalità di failure note, e questo è ciò che ciascuna significa per il caller.

Includo retry_after_s per i rate limit. I modelli con supporto a function calling possono usare questo segnale per fare back off o escalation, in base all'agent loop e alla logica dell'executor. Senza questo campo, un agent loop può ritentare subito e amplificare i problemi di rate limit.

3. Dispatch parallelo e perché il sequenziale è l'anti-pattern di default

La maggior parte degli agent tutorial invia le chiamate tool in modo sequenziale: chiama il tool A, aspetta il risultato, poi decide se chiamare il tool B. Funziona. Per tool indipendenti, però, può essere più lento del necessario.

La maggior parte delle principali API di tool calling ora emette più chiamate tool in un singolo turn di risposta quando il modello decide che il lavoro è indipendente. Non è più un caso esotico, ma l'executor deve rispettarlo. Se serializzo quelle chiamate una per una, perdo il vantaggio di latenza che il modello ha già richiesto. Quando non c'è dipendenza tra le chiamate, le mando in esecuzione concorrente.

import asyncio

# Anti-pattern: sequential dispatch
async def run_tools_sequential(tool_calls: list[ToolCall]) -> list[ToolResult]:
    results = []
    for call in tool_calls:
        result = await execute_tool(call)  # wait for each before starting next
        results.append(result)
    return results

# Pattern: concurrent dispatch
async def run_tools_parallel(tool_calls: list[ToolCall]) -> list[ToolResult]:
    tasks = [execute_tool(call) for call in tool_calls]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    return [
        ToolResult(error=str(r)) if isinstance(r, Exception) else r
        for r in results
    ]

Il guadagno concreto dipende dalla latenza dei tool, dall'overhead dell'executor, dall'overhead di scheduling, dall'overhead di rete e dal fatto che le chiamate siano davvero indipendenti. Tre chiamate tool indipendenti con media di 300ms ciascuna richiedono circa 900ms in sequenza. Se domina la latenza dei tool e l'executor può eseguirle in modo concorrente, il parallel dispatch può ridurre il wall-clock time verso la durata della chiamata più lenta, più overhead. In un agent loop con più step tool indipendenti, questo può ridurre il wall-clock time nelle stesse condizioni.

Tengo presenti due cautele. Primo, asyncio.gather con return_exceptions=True è importante: un singolo tool che fallisce non dovrebbe far crashare l'intero gather. Gestisco le eccezioni per risultato. Secondo, non tutte le chiamate tool sono indipendenti. Se la chiamata B dipende dal risultato della chiamata A, l'esecuzione sequenziale è corretta. L'LLM può rappresentarlo mettendo le chiamate dipendenti in turn separati. Se vedo chiamate sequenziali nello stesso turn per tool che dovrebbero essere indipendenti, ispeziono prima le descrizioni dei tool. Potrebbero non essere abbastanza chiare su cosa restituisce ciascun tool.

4. Rilevamento dei loop tramite fingerprint delle chiamate

Il loop infinito dei tool è una failure di produzione che ho visto più volte. Il modello chiama ripetutamente lo stesso tool con gli stessi argomenti, perché non riceve un risultato soddisfacente o perché è entrato in un reasoning loop. Se nulla lo ferma, il loop esaurisce il budget di token e comunque non risolve il task.

Controllo un fingerprint prima di ogni dispatch del tool:

from hashlib import sha256
import json

class LoopDetector:
    def __init__(self, max_repeats: int = 2):
        self.call_counts: dict[str, int] = {}
        self.max_repeats = max_repeats

    def is_looping(self, tool_name: str, args: dict) -> bool:
        key = sha256(
            json.dumps({"tool": tool_name, "args": args}, sort_keys=True).encode()
        ).hexdigest()[:16]
        self.call_counts[key] = self.call_counts.get(key, 0) + 1
        return self.call_counts[key] > self.max_repeats

# In the executor:
detector = LoopDetector(max_repeats=2)
for call in tool_calls:
    if detector.is_looping(call.name, call.arguments):
        return ToolResult(
            status="loop_detected",
            message=f"Tool '{call.name}' called with identical arguments {detector.max_repeats + 1} times."
        )
    result = await execute_tool(call)

Quando l'executor restituisce status: loop_detected, indirizzo il conditional edge del graph verso un nodo di escalation, di solito un supervisor agent o un interrupt HITL. Non lo rimando al tool-calling agent. Il modello riceve un segnale chiaro e strutturato: deve usare una strategia diversa.

Il fingerprint è calcolato da (tool_name, sorted args) dopo normalizzazione JSON. max_results: 5 e max_results: 5 sono lo stesso fingerprint. query: "revenue Q3" e query: "revenue Q4" sono diversi. Uso max_repeats: 2 come punto di partenza. Due chiamate identiche possono significare "failure transitorio, sto ritentando". Nei sistemi che ho ispezionato, tre chiamate identiche sono spesso un segno che l'agent è bloccato.

5. Routing basato sullo status tra chiamate tool

Una volta che ho typed error unions, posso indirizzare il graph in base allo status invece di fare parsing del contenuto del tool. Il principio è lo stesso dei LangGraph conditional edges: la logica di routing vive nello stato, non nelle edge function.

Dopo ogni risultato tool, un router node leggero legge result.status e scrive un segnale di routing nello stato. Il conditional edge legge quel segnale. La routing map diventa diretta: rate_limited → aspetta e ritenta; auth_error → escalation a HITL; empty → prova un tool alternativo; ok → continua. Il comportamento resta pulito, testabile e visibile nelle trace di stato.

Fare parsing del contenuto del tool dentro la edge function è più fragile. Il formato dell'output del tool può variare con versioni del modello e system prompt. Una decisione di routing basata su "l'output contiene la parola Error" può rompersi quando cambia il formato dell'output. Una decisione di routing basata su status: Literal["ok", "rate_limited", "auth_error", "empty"] è più stabile.

La conseguenza pratica è semplice: posso scrivere unit test per la logica di routing senza fare mocking dell'LLM. Posso creare un SearchResult(status="rate_limited", retry_after_s=30), passarlo attraverso il router e verificare che il prossimo nodo sia "wait_and_retry". Quel test è veloce, deterministico e copre la modalità di failure che mi interessa.

6. Tool choice e igiene del prompt

Un segnale che le definizioni dei tool richiedono lavoro è la necessità di forzare un tool nominato tramite tool_choice (in stile OpenAI {"type": "function", "function": {"name": "…"}}, o l'equivalente del provider per "required" / selezione nominata). Forzare la tool choice è una escape hatch valida. Come pattern di default, però, è un cattivo odore.

Quando forzo la tool choice, spesso sto compensando un prompt o uno schema che non rende abbastanza chiara la selezione del tool. Preferisco chiarire le descrizioni dei tool e il system prompt finché tool_choice="auto" (o il default del provider) instrada correttamente sulla distribuzione reale delle query. Questo riduce la logica di controllo a inference time e rende il comportamento meno dipendente da una singola forma di query ristretta.

C'è un'eccezione. Per l'ultimo step di una pipeline di structured output, dove voglio che il modello chiami sempre un tool "finalize" — oppure emetta uno schema JSON fisso tramite structured outputs / strict mode — forzare il canale è corretto. È un vincolo strutturale, non una compensazione per descrizioni poco chiare.

Per agent con molti tool, tengo il system prompt focalizzato sul ruolo e sullo scope dell'agent. Evito di metterci logica di selezione dei tool. Se mi trovo a scrivere "usa il search tool quando l'utente chiede informazioni sui documenti" nel system prompt, lo tratto come un segnale che la descrizione del tool non lo dice abbastanza chiaramente. Correggo la descrizione del tool.

Il quadro di categoria

I cinque pattern condividono un principio: rendere l'interfaccia tra l'LLM e il layer dei tool esplicita, tipizzata e ispezionabile. Uso schema contracts invece di raw dict. Uso typed error unions invece di exception traces. Uso concurrent dispatch invece di sequential dispatch quando le chiamate sono indipendenti e domina la latenza dei tool. Uso loop detection invece di sperare che il modello si auto-corregga. Uso status signals invece del content parsing nella logica di routing.

Gli anti-pattern sono versioni di "lascia che sia l'LLM a capirlo". Può funzionare a qualità demo. In condizioni simili alla produzione, con distribuzioni di query reali e modalità di failure reali, ho visto rompersi per primi i contratti impliciti.

Quando costruisco o revisiono un tool-use agent e vedo problemi di affidabilità o latenza, cerco queste modalità di failure e applico le correzioni corrispondenti.

FAQ

Come dovrei validare gli argomenti dei tool LLM prima dell'esecuzione?

Uso uno schema Pydantic per ogni tool e lo passo come schema degli argomenti del tool. Lo schema valida gli input prima che l'executor venga eseguito, documenta ogni campo per il modello e gestisce type coercion, per esempio la conversione di max_results: "5" in 5.

Perché evitare di restituire traceback Python al modello?

Evito i traceback perché sono non strutturati, verbosi e pieni di nomi interni che il modello potrebbe interpretare male. Gonfiano anche la context window durante i retry. Una typed error union dà al modello status espliciti come rate_limited, auth_error, empty e ok.

Quando le chiamate tool dovrebbero essere eseguite in parallelo invece che in sequenza?

Eseguo il dispatch delle chiamate tool in modo concorrente quando sono indipendenti. L'esecuzione sequenziale è corretta quando la chiamata B dipende dalla chiamata A. Le chiamate indipendenti nello stesso turn possono essere raccolte in modo concorrente e gestite per risultato, se l'executor lo supporta e se domina la latenza dei tool.

Come posso rilevare un loop infinito dei tool LLM?

Calcolo il fingerprint di ogni chiamata a partire dal nome del tool e dagli argomenti normalizzati in JSON e ordinati prima del dispatch. Se lo stesso fingerprint supera il limite di ripetizione configurato, l'executor restituisce status: loop_detected e indirizza verso escalation invece di rimandare l'agent nello stesso loop.

Come dovrebbe funzionare il routing dopo un risultato tool?

Eseguo il routing da valori di status strutturati, non facendo parsing del contenuto del tool. Un router legge result.status e scrive un segnale di routing nello stato. In questo modo casi come rate_limited, auth_error, empty e ok possono mappare a prossimi step chiari, visibili nelle trace di stato e facili da testare con unit test.