Cinque pattern di function calling che hanno retto in produzione

Perché conta

L’uso dei tool è uno dei punti in cui vedo fallire più spesso i sistemi LLM in produzione. Il problema non sta solo nel reasoning dell’LLM o nella progettazione del prompt. Sta soprattutto nello spazio operativo tra “il modello sa quale tool chiamare” e “il tool viene eseguito e restituisce qualcosa di utile”.

Nei team che ho osservato emergono spesso gli stessi pattern, e anche gli stessi anti-pattern: regressioni di latenza, overflow della finestra di contesto e agenti che entrano in loop durante un’esecuzione non presidiata.

Ho costruito agenti potenziati da tool per workflow finanziari, pipeline di ricerca e sistemi multi-agent. Questo è il set di pattern che considero più utile in produzione: pattern che ho visto reggere meglio, insieme ad alternative che hanno mostrato limiti chiari.

1. Contratti tool schema-first

L’anti-pattern più comune che vedo è questo: definire una funzione Python, decorarla con @tool e fidarsi che l’LLM passi argomenti validi. Può funzionare abbastanza spesso da sembrare sicuro. Poi arrivano casi come max_results: "ten", date: "last week", query: null.

A quel punto un TypeError emerge da qualche parte nell’esecutore, viene intercettato, loggato come errore, e l’esecuzione prova ad andare avanti. In pratica, il tool è diventato inaffidabile in modo silenzioso.

La correzione è semplice: ogni tool ha uno schema Pydantic che valida gli input prima dell’esecuzione. Lo schema non serve solo alla validazione. È anche la documentazione che l’LLM legge per decidere come chiamare il tool.

# 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)
def search_documents(query: str, max_results: int = 5, date_filter: str | None = 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 fa tre cose: valida gli input prima dell’esecuzione, documenta il tool attraverso i campi description e applica type coercion quando possibile. Per type coercion intendo la conversione controllata di un valore verso il tipo atteso, per esempio max_results: "5" → 5. Con una funzione nuda, nessuno di questi tre punti è un contratto esplicito.

Tratto le descrizioni dei campi come parte della specifica del tool, non come commenti. Quando sono vaghe, spesso producono chiamate vaghe. Gli schemi migliori che ho scritto somigliano a brevi riferimenti API: spiegano cosa significa ogni campo, quale formato è atteso, quali vincoli applica e qual è il default.

Se in produzione ricevo chiamate a tool malformate, parto da lì: riscrivo le descrizioni dei campi. Solo dopo considero un cambio di modello o di prompt.

In pratica: prima rendo il contratto del tool leggibile e validabile. Poi valuto se il problema è davvero nel modello.

2. Unione di errori tipizzata, non traceback di eccezioni

Quando un tool fallisce, l’approccio ingenuo è intercettare l’eccezione e restituire il traceback come stringa. Il modello legge il traceback, prova a interpretare l’errore e tenta una chiamata diversa. A volte basta. Quando non basta, però, il fallimento diventa difficile da controllare.

Il problema è che un traceback Python passato al modello come contesto di errore è non strutturato, verboso e pieno di nomi interni. A ogni retry gonfia la finestra di contesto. Se l’errore è transitorio — rate limit, timeout di rete — il modello non riceve un segnale chiaro su cosa fare: riprovare, applicare backoff o interrompere. Da qui nascono facilmente 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.")

Un oggetto come status: rate_limited e retry_after_s: 30 dà al modello e al grafo informazione strutturata, coerente con il contratto del tool. Uno stack trace, invece, è più rumore operativo che stato applicativo. L’unione di errori tipizzata rende espliciti i modi in cui il tool può fallire e cosa ciascun fallimento significa per il chiamante.

Nel caso di rate limit, includo retry_after_s. Questo campo può aiutare il modello o il supervisore a scegliere un backoff invece di riprovare subito. Senza questo campo, nello stato rimane meno informazione e aumenta il rischio che il sistema scelga un retry immediato, amplificando il problema di rate limit.

In pratica: non passo al modello il dettaglio interno dell’eccezione se posso passargli uno stato applicativo chiaro.

3. Invio parallelo e perché il sequenziale è l’anti-pattern di default

Molti agenti nei tutorial inviano le chiamate a tool in sequenza: chiamano il tool A, aspettano il risultato, poi decidono se chiamare il tool B. È un modello semplice e spesso corretto. Per tool indipendenti, però, può introdurre latenza evitabile.

Alcuni modelli e setup di tool calling possono richiedere più chiamate a tool in un singolo response turn quando le chiamate sono indipendenti. In questi casi configuro l’esecutore perché le invii in modo 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
    ]

Come esempio illustrativo, se tre chiamate a tool indipendenti impiegano circa 300 ms ciascuna, un invio sequenziale arriva a circa 900 ms. Un invio parallelo può ridurre il tempo effettivo verso la durata della chiamata più lenta, più overhead. Non è una costante generale: dipende da tool, rete, runtime e carico. Il punto pratico è che, quando le chiamate sono davvero indipendenti, il tempo effettivo non deve essere la somma delle latenze.

Mantengo due cautele. Primo, asyncio.gather va usato con return_exceptions=True: un singolo tool che fallisce non deve far fallire automaticamente l’intera raccolta dei risultati. Le eccezioni vanno gestite per singolo risultato. Secondo, non tutte le chiamate a tool sono indipendenti. Se la chiamata B dipende dal risultato della chiamata A, il sequenziale è corretto.

Quando vedo chiamate sequenziali nello stesso turn per tool che dovrebbero essere indipendenti, controllo prima le descrizioni dei tool: spesso non spiegano abbastanza bene cosa restituisce ciascun tool.

In pratica: parallelizzo solo quando le dipendenze tra tool sono assenti o esplicite. Se c’è dipendenza dati, il sequenziale non è un problema: è il comportamento corretto.

4. Rilevamento dei loop tramite fingerprint delle chiamate

Il loop infinito dei tool è un fallimento ricorrente in produzione: il modello chiama più volte lo stesso tool con gli stessi argomenti perché non riceve un risultato soddisfacente oppure perché è entrato in un loop di reasoning. Se non lo fermo, consuma il token budget e non converge.

Il principio operativo è bloccare la ripetizione prima che diventi parte dello stato conversazionale. Per implementarlo, aggiungo un controllo di fingerprint prima di ogni invio del tool. Qui fingerprint significa una firma stabile della chiamata, calcolata a partire dal nome del tool e dagli argomenti normalizzati.

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)

A livello di routing, tratto loop_detected come un cambio di strategia, non come un altro errore da riprovare nello stesso nodo. Quando l’esecutore restituisce status: loop_detected, instrado il conditional edge del grafo verso un nodo di escalation. Per conditional edge intendo un arco del grafo che sceglie il nodo successivo in base a un valore nello stato. Il nodo di escalation è di solito un agente supervisore o un interrupt HITL, cioè Human-in-the-loop: un punto in cui l’esecuzione passa a una revisione o decisione umana.

Non torno allo stesso agente che sta chiamando i tool. In questo modo il modello e il grafo ricevono un segnale strutturato: la strategia corrente non sta convergendo e serve un percorso diverso.

Il fingerprint viene calcolato da (tool_name, sorted args) dopo normalizzazione JSON. max_results: 5 e max_results: 5 hanno lo stesso fingerprint. query: "revenue Q3" e query: "revenue Q4" sono diversi. Di solito imposto max_repeats: 2 come punto di partenza: due chiamate identiche possono ancora indicare un retry su fallimento transitorio; alla terza chiamata identica tratto la sequenza come un possibile blocco e faccio escalation.

In pratica: non aspetto che il modello si accorga del loop. Metto un controllo deterministico prima dell’esecuzione ripetuta.

5. Routing basato su status tra chiamate a tool

Una volta definita un’unione di errori tipizzata, il grafo può essere instradato in base allo status invece di fare parsing del contenuto del tool. Il principio è lo stesso dei conditional edge di LangGraph: la logica di routing vive nello stato, non nelle funzioni degli edge.

L’implementazione resta semplice. Dopo ogni risultato di tool, inserisco un nodo router leggero. Il nodo legge result.status e scrive nello stato un segnale di routing. Il conditional edge legge quel segnale.

La logica diventa: rate_limited → attendi e riprova; auth_error → escalation a HITL; empty → prova un tool alternativo; ok → continua. Questo rende il comportamento più facile da testare, da leggere nelle state trace e da mantenere.

La distinzione importante è tra stato applicativo e testo generato. L’alternativa è fare parsing del contenuto del tool dentro la funzione dell’edge per inferire cosa sia successo. La evito perché è fragile. Il formato dell’output del tool può cambiare con le versioni del modello e con i system prompt.

Una decisione di routing basata su “l’output contiene la parola Error” può rompersi quando il formato cambia. Una decisione basata su status: Literal["ok", "rate_limited", "auth_error", "empty"] resta legata a un contratto tipizzato.

La conseguenza pratica è che posso scrivere unit test per la logica di routing senza mockare l’LLM. Creo un SearchResult(status="rate_limited", retry_after_s=30), lo passo nel router e verifico che il nodo successivo sia "wait_and_retry". Quel test è veloce, deterministico e copre la modalità di fallimento che mi interessa.

In pratica: se una decisione del grafo dipende da un risultato di tool, la rappresento come stato tipizzato, non come testo da interpretare.

6. Scelta del tool e igiene del prompt

Un segnale che le definizioni dei tool hanno bisogno di lavoro è l’uso frequente di tool_choice={"type": "function", "function": {"name": "specific_tool"}} per forzare il modello a chiamare un tool specifico. La scelta forzata del tool è un escape hatch valido. Come pattern di default, però, la considero un segnale di debolezza architetturale.

Quando ricorro spesso alla scelta forzata del tool, di solito sto compensando un prompt o uno schema che non rende abbastanza chiaro al modello quando ogni tool dovrebbe essere usato. La correzione passa prima dalle descrizioni dei tool e dal system prompt, finché tool_choice="auto" instrada in modo adeguato sulla distribuzione reale di query. Questo riduce la dipendenza da vincoli hard-coded e rende più esplicito il contratto tra modello e livello dei tool.

L’eccezione è l’ultimo step di una pipeline di structured output. Per structured output intendo un output vincolato da uno schema, invece di testo libero. Se voglio che il modello chiami sempre un tool "finalize" e produca il proprio output come schema tipizzato, forzare la scelta del tool è corretto. In quel caso è un vincolo strutturale, non una compensazione per descrizioni poco chiare.

Per l’igiene del prompt negli agenti con molti tool, mantengo il system prompt focalizzato su ruolo e scope dell’agente, non sulla 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: probabilmente la descrizione del tool non lo dice abbastanza chiaramente. La prima correzione, quindi, è sulla descrizione del tool.

In pratica: uso tool_choice forzato quando è parte del contratto della pipeline, non per compensare definizioni ambigue.

Il principio comune

Questi pattern hanno un filo comune: rendere l’interfaccia tra l’LLM e il livello dei tool esplicita, tipizzata e ispezionabile.

Il confronto operativo è questo: contratto di schema invece di dizionario non validato. Unione di errori tipizzata invece di traceback di eccezioni. Invio concorrente invece di sequenziale quando le chiamate sono indipendenti. Rilevamento dei loop invece di sperare che il modello si autocorregga. Segnale di stato invece di parsing del contenuto nella logica di routing.

Gli anti-pattern sono spesso versioni di “lascia che sia l’LLM a capirlo”. Può bastare a livello demo. In produzione, con distribuzioni reali di query e modalità di fallimento reali, i contratti impliciti tendono a rompersi presto.

Se stai costruendo o revisionando un agente che usa tool e incontri problemi di affidabilità o latenza, scrivimi. Queste modalità di fallimento hanno correzioni praticabili una volta che sai cosa cercare.

FAQ

Perché usare uno schema Pydantic per i tool LLM?

Uno schema Pydantic valida gli input prima dell’esecuzione, documenta il tool tramite le descrizioni dei campi e applica type coercion quando possibile. Per type coercion intendo una conversione controllata verso il tipo atteso, per esempio da stringa numerica a intero. Le descrizioni fanno parte della specifica: formato atteso, vincoli e default devono essere espliciti.

Come gestisco gli errori dei tool senza passare traceback al modello?

La soluzione che uso è un’unione di errori tipizzata nel return schema, per esempio con status come ok, empty, rate_limited o auth_error. Uno stack trace è verboso, non strutturato e gonfia la finestra di contesto; uno status tipizzato rende chiaro il tipo di fallimento e il suo significato per il chiamante.

Quando conviene eseguire chiamate a tool in parallelo?

Le chiamate a tool conviene eseguirle in parallelo quando sono davvero indipendenti. In quel caso il tempo effettivo non deve essere la somma delle latenze. Se invece la chiamata B dipende dal risultato della chiamata A, il sequenziale resta corretto.

Come rilevo un loop di chiamate a tool identiche?

Prima di ogni invio calcolo un fingerprint da tool_name e dagli argomenti ordinati dopo normalizzazione JSON. Il fingerprint è una firma stabile della chiamata. Se la stessa chiamata supera max_repeats, restituisco status: loop_detected e instrado verso escalation, invece di tornare allo stesso agente che sta chiamando i tool.

Come instrado un grafo dopo il risultato di un tool?

Uso lo status del risultato come segnale di routing, non il parsing del contenuto. Un nodo router legge result.status e decide il passo successivo, per esempio attendere e riprovare su rate_limited, fare escalation su auth_error, provare un tool alternativo su empty o continuare su ok.