Salta al contenuto
Torna alla home
2024-11-259 min read
LLMFunction CallingOpenAIProduction

Quattro pattern di function calling che hanno retto in produzione

>-

Perché è importante

L’uso dei tool è uno dei punti in cui vedo fallire più spesso i sistemi LLM in produzione. Non tanto nel reasoning dell’LLM o nel prompt design, ma nello spazio tra “il modello sa quale tool chiamare” e “il tool viene eseguito e restituisce qualcosa di utile”. Nei team che ho visto, ricorrono spesso versioni degli stessi pattern. In molti casi ho visto anche gli stessi anti-pattern: regressioni di latenza, overflow della context window e agent che entrano in loop durante un’esecuzione non presidiata.

Ho costruito agent potenziati da tool per workflow finanziari, pipeline di ricerca e sistemi multi-agent. Quello che segue è il catalogo che avrei voluto avere due anni fa: pattern che ho visto reggere meglio il contatto con la produzione, e l’aspetto delle alternative che non hanno retto.

1. Contratti tool schema-first

L’anti-pattern più comune che incontro è 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 i casi come max_results: "ten", date: "last week", query: null. A quel punto un TypeError emerge da qualche parte nell’executor, viene intercettato, loggato come errore, e l’esecuzione prova ad andare avanti. In pratica, il tool è diventato inaffidabile in modo silenzioso.

La correzione che preferisco è semplice: ogni tool ha uno schema Pydantic che valida gli input prima dell’esecuzione. Il punto critico è che 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, 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 fa tre cose: valida gli input prima che l’executor venga eseguito, documenta il tool attraverso i campi description e applica type coercion quando possibile, per esempio max_results: "5"5. Con una funzione nuda, io non ho nessuno di questi tre punti come contratto esplicito.

Tratto le descrizioni dei campi come parte della specifica del tool, non come commenti. Descrizioni vaghe producono spesso chiamate vaghe. Gli schemi migliori che ho scritto somigliano a brevi riferimenti API: cosa significa ogni campo, quale formato si aspetta, quali vincoli ha, qual è il default. Se in produzione ricevo tool call malformate, prima riscrivo le descrizioni dei campi. Solo dopo considero un cambio di modello o di prompt.

2. Typed error union, non exception trace

Quando un tool fallisce, l’approccio ingenuo è intercettare l’eccezione e restituire il traceback come stringa. Il modello legge il traceback, prova a “capire” 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 context window. Se l’errore è transitorio — rate limit, network timeout — il modello non riceve un segnale chiaro su cosa fare: retry, backoff o abort. Questo aumenta il rischio di retry immediati e di 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.")

Io preferisco dare al modello e al graph un oggetto come status: rate_limited e retry_after_s: 30: è informazione strutturata e coerente con il contratto del tool. Uno stack trace, invece, è rumore operativo più che stato applicativo. La typed error union rende anche espliciti i modi in cui il tool può fallire e cosa ciascun fallimento significa per il caller.

Nel caso di rate limit, includo retry_after_s. La presenza di questo campo può aiutare il modello o il supervisor a scegliere un backoff invece di riprovare subito. Senza questo campo, io lascio meno informazione nello stato e aumento il rischio che il sistema scelga un retry immediato, amplificando il problema di rate limit.

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

Molti agent nei tutorial dispatchano le tool call 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.

Gli LLM moderni (GPT-4o, Claude 3.5 Sonnet) possono richiedere più tool call in un singolo response turn quando le chiamate sono indipendenti. In questi casi io imposto l’executor perché le dispatchi 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 tool call indipendenti impiegano circa 300 ms ciascuna, una dispatch sequenziale arriva a circa 900 ms. Una dispatch parallela può stare intorno al tempo della chiamata più lenta, per esempio circa 320 ms con un po’ di network jitter. Non è una costante generale: dipende da tool, rete, runtime e carico. Il punto è che, quando le chiamate sono davvero indipendenti, il wall-clock time non deve essere la somma delle latenze.

Tengo due cautele. Primo, uso asyncio.gather con return_exceptions=True: un singolo tool che fallisce non deve far fallire automaticamente l’intero gather. Gestisco le eccezioni per singolo risultato. Secondo, non tutte le tool call 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.

4. Loop detection tramite call fingerprinting

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 reasoning loop. Se non lo fermo, consuma il token budget e non converge.

La correzione che uso è un controllo di 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, instrado il conditional edge del graph verso un nodo di escalation, di solito un supervisor agent o un interrupt HITL. Non torno allo stesso agent che sta chiamando i tool. In questo modo il modello riceve un segnale chiaro e strutturato: serve una strategia diversa.

Calcolo il fingerprint 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 preferisco trattare la sequenza come un possibile blocco e fare escalation.

5. Routing basato su status tra tool call

Una volta che ho una typed error union, posso instradare il graph in base allo status invece di fare parsing del contenuto del tool. È lo stesso principio dei conditional edge di LangGraph: la routing logic vive nello stato, non nelle edge function.

Dopo ogni risultato di tool, uso un router node leggero. Il nodo legge result.status e scrive nello stato un routing signal. Il conditional edge legge quel segnale. La routing logic diventa: rate_limited → attendi e riprova; auth_error → escalation a HITL; empty → prova un tool alternativo; ok → continua. È più facile da testare, da leggere nelle state trace e da mantenere.

L’alternativa è fare parsing del contenuto del tool dentro la edge function per inferire cosa sia successo. Io 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 routing logic 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.

6. Tool choice 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 forced tool choice è un escape hatch valido. Come pattern di default, però, io la considero un odore architetturale.

Quando forzo la tool choice, spesso sto compensando un prompt o uno schema che non rende abbastanza chiaro al modello quando ogni tool dovrebbe essere usato. La correzione che preferisco è migliorare le descrizioni dei tool e il 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 tool layer.

L’eccezione è l’ultimo step di una pipeline di structured output. Se voglio che il modello chiami sempre un tool "finalize" e produca il proprio output come typed schema, forzare la tool choice è corretto. In quel caso è un vincolo strutturale, non una compensazione per descrizioni poco chiare.

Per l’igiene del prompt negli agent ad alta densità di tool, io mantengo il system prompt focalizzato su ruolo e scope dell’agent, 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. Correggo prima la descrizione del tool.

Il category frame

Questi pattern hanno un filo comune: rendere l’interfaccia tra l’LLM e il tool layer esplicita, tipizzata e ispezionabile. Schema contract invece di raw dict. Typed error union invece di exception trace. Dispatch concorrente invece di sequenziale quando le chiamate sono indipendenti. Loop detection invece di sperare che il modello si autocorregga. Status signal invece di content parsing nella routing logic.

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 tool-use agent e incontri problemi di affidabilità o latenza, scrivimi. Queste modalità di fallimento hanno correzioni praticabili una volta che sai cosa cercare.

Condividi: