Tre schemi per l'orchestrazione di agenti che hanno superato la produzione

Il panorama dell'orchestrazione degli agenti

Negli ultimi 14 mesi, abbiamo distribuito sistemi di agenti LLM in tre domini:

1. Automazione del supporto clienti — 12.000 ticket/mese, 8 tool integrati

2. Analisi di business intelligence — 200 analisti, 15 fonti di dati

3. Elaborazione di documenti legali — 50k documenti/mese, 6 pipeline di estrazione

Ogni deployment ci ha insegnato quali schemi di orchestrazione funzionano in teoria rispetto a quali reggono sotto lo stress della produzione - timeout degli tool, limiti di frequenza delle API, query ambigue e aspettative degli utenti per risposte sotto i 3 secondi.

Questo articolo cataloga tre schemi che hanno resistito: Router, Planner-Executor e Critic.

Schema 1: Router (Dispatch semplice)

Architettura

                Query dell'utente
                    ↓
            ┌───────────────┐
            │  Router LLM   │  "Quale tool gestisce questo?"
            └───────┬───────┘
                    │
        ┌───────────┼───────────┬───────────┐
        ↓           ↓           ↓           ↓
    [Tool A]    [Tool B]    [Tool C]    [Tool D]
    Ricerca    Calcolatrice  Meteo     Calendario
        ↓           ↓           ↓           ↓
                Risposta (da un singolo tool)

Quando usarlo

• Più tool specializzati con domine chiare e non sovrapposte.

• Attività a singolo passo (una chiamata di tool → risultato).

• Applicazioni sensibili alla latenza (<1s tempo di risposta).

Implementazione

router_agent.py
from typing import Dict, Callable
class RouterAgent:
    """Agent di routing semplice — dispatcha a un singolo tool."""
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
        self.tool_descriptions = self._generate_tool_docs()
    def _generate_tool_docs(self) -> str:
        """Genera documentazione dei tool per il prompt del router."""
        docs = []
        for name, tool in self.tools.items():
            docs.append(f"- {name}: {tool.__doc__}")
        return "\n".join(docs)
    async def route(self, query: str) -> str:
        """Instrada la query al tool appropriato."""
        router_prompt = f"""
Sei un router di tool. Data una query dell'utente, seleziona il singolo migliore tool per risponderla.
Tool disponibili:
{self.tool_descriptions}
Query dell'utente: {query}
Rispondi con JSON: {{"tool": "tool_name", "reasoning": "perché questo tool"}}
"""
        routing_decision = await llm.generate(router_prompt)
        tool_name = json.loads(routing_decision)['tool']
        # Esegui il tool selezionato
        if tool_name not in self.tools:
            return f"Errore: Tool sconosciuto {tool_name}"
        return await self.toolstool_name
Utilizzo
tools = {
    "search": search_knowledge_base,
    "calculator": calculate_expression,
    "weather": get_weather_forecast,
    "calendar": check_calendar_availability
}
agent = RouterAgent(tools)
response = await agent.route("Che tempo farà domani a Parigi?")

Dati di produzione (Supporto clienti)

Metrica	Valore
Query gestite	12.000/mese
Routing corretto	94%
Latenza media	820ms
Latenza p95	1.2s
Routing ambiguo	6% (escalation a umano)

Punti di forza

✅ Bassa latenza — Singola chiamata LLM + una esecuzione di tool ✅ Prevedibile — Esecuzione lineare, facile da comprendere ✅ Debuggabile — Semplice da loggare: "Query → Decisione di routing → Tool → Risultato" ✅ Conveniente — Minime chiamate LLM

Debolezze

❌ Nessun chaining di tool — Non può combinare tool ("Cerca X, poi calcola Y") ❌ Errori di routing fatali — Selezione sbagliata del tool = risposta sbagliata ❌ Query ambigue falliscono — "Prenota un appuntamento se non piove" richiede due tool

Lezioni di produzione

Lezione 1: Costruite un classificatore di fallback

Quando la confidenza del routing è bassa (<70%), escala a umano:

routing_confidence = routing_decision['confidence']
if routing_confidence < 0.70:
    return escalate_to_human(query, reason="routing ambiguo")

Lezione 2: Cachate le decisioni di routing

Le query comuni ("Controlla stato ordine") si routano allo stesso modo ogni volta:

@cache(ttl=3600)
def route_query(query: str):
    # Cache del routing per 1 ora
    return router.route(query)

Lezione 3: Monitorate la precisione del routing

Tracciate quali tool vengono selezionati rispetto a cosa gli utenti avevano effettivamente bisogno:

Logga le decisioni di routing
log_routing_decision(
    query=query,
    selected_tool=tool_name,
    user_satisfaction=feedback  # Raccogli dopo l'interazione
)
Analisi settimanale
routing_errors = query_logs.filter(user_satisfaction < 3)
print(f"Top query errate: {routing_errors.most_common(10)}")

Risultato: Abbiamo migliorato la precisione del routing dall'87% al 94% riaddestrando sulle query errate.

Schema 2: Planner-Executor (Reasoning multi-passo)

Architettura

            Query dell'utente: "Confronta ricavi Q1 vs Q2"
                    ↓
            ┌───────────────┐
            │  Planner LLM  │  Genera piano di esecuzione
            └───────┬───────┘
                    ↓
            Piano: [Passo 1, Passo 2, Passo 3]
            1. Recupera ricavi Q1 dal DB
            2. Recupera ricavi Q2 dal DB
            3. Calcola la differenza
                    ↓
            ┌───────────────┐
            │   Executor    │  Esegui il piano sequenzialmente
            └───────┬───────┘
                    ↓
        ┌───────────┼───────────┐
        ↓           ↓           ↓
    [Interroga DB] [Interroga DB] [Calcola]
        ↓           ↓           ↓
    $120K       $145K        +$25K (+21%)
                    ↓
            Risposta finale

Quando usarlo

• Workflow multi-passo che richiedono composizione di tool.

• Selezione dinamica dei tool (non si può prevedere la sequenza in anticipo).

• Attività strutturate (analisi dei dati, generazione di report).

Implementazione

planner_executor_agent.py
from typing import List, Dict
import json
class PlannerExecutorAgent:
    """Agent che pianifica prima dell'esecuzione."""
    def __init__(self, tools: Dict[str, Callable]):
        self.tools = tools
    async def plan(self, query: str) -> List[Dict]:
        """Genera il piano di esecuzione."""
        planner_prompt = f"""
Sei un pianificatore di attività. Suddividi questa query in passi eseguibili usando i tool disponibili.
Tool disponibili:
{self._tool_docs()}
Query dell'utente: {query}
Genera un piano come array JSON:
[
  {{"step": 1, "tool": "tool_name", "input": "...", "output_var": "var1"}},
  {{"step": 2, "tool": "tool_name", "input": "usa {{var1}}", "output_var": "var2"}},
  ...
]
"""
        plan_json = await llm.generate(planner_prompt)
        return json.loads(plan_json)
    async def execute(self, plan: List[Dict]) -> Dict:
        """Esegui il piano passo dopo passo."""
        context = {}  # Memorizza risultati intermedi
        for step in plan:
            tool_name = step['tool']
            tool_input = step['input']
            # Sostituisci le variabili dal contesto
            for var, value in context.items():
                tool_input = tool_input.replace(f"{{{var}}}", str(value))
            # Esegui il tool
            result = await self.toolstool_name
            # Memorizza il risultato nel contesto
            output_var = step.get('output_var', f"step_{step['step']}")
            context[output_var] = result
            print(f"Passo {step['step']}: {tool_name}({tool_input}) → {result}")
        return context
    async def run(self, query: str) -> str:
        """Pianifica ed esegui."""
        plan = await self.plan(query)
        context = await self.execute(plan)
        # Genera la risposta finale usando il contesto
        final_prompt = f"""
Query dell'utente: {query}
Risultati dell'esecuzione:
{json.dumps(context, indent=2)}
Fornisci una risposta in linguaggio naturale all'utente.
"""
        return await llm.generate(final_prompt)
Utilizzo
tools = {
    "sql_query": execute_sql,
    "calculator": calculate,
    "search_docs": search_documentation,
    "send_email": send_email
}
agent = PlannerExecutorAgent(tools)
response = await agent.run("Confronta ricavi Q1 vs Q2 e invia riepilogo via email al CFO")

Dati di produzione (Business Intelligence)

Metrica	Valore
Query gestite	1.200/mese
Completamenti riusciti	89%
Latenza media	3.2s
Latenza p95	8.4s
Errori di piano	11% (tool non valido, ordine sbagliato)

Punti di forza

✅ Gestisce workflow complessi — Composizione multi-tool ✅ Flessibile — Si adatta dinamicamente alla complessità della query ✅ Trasparente — Il piano è leggibile dagli umani, debuggabile ✅ Ripristinabile — Può riprovare singoli passi in caso di errore

Debolezze

❌ Latenza più elevata — N+1 chiamate LLM (1 per la pianificazione, N per l'esecuzione) ❌ I piani possono essere sbagliati — Selezione tool non valida, ordine errato, passi mancanti ❌ Propagazione degli errori — Un errore nei primi passi rompe l'intero piano ❌ I costi scalano con i passi — Piano a 5 passi = 6 chiamate LLM

Lezioni di produzione

Lezione 1: Validare i piani prima dell'esecuzione

Non fidatevi ciecamente dei piani generati da LLM:

def validate_plan(plan: List[Dict]) -> bool:
    """Controlla il piano per errori comuni."""
    for step in plan:
        # Controlla che il tool esista
        if step['tool'] not in self.tools:
            raise PlanError(f"Tool sconosciuto: {step['tool']}")
        # Controlla le dipendenze delle variabili
        required_vars = extract_variables(step['input'])
        available_vars = [s['output_var'] for s in plan[:step['step']-1]]
        for var in required_vars:
            if var not in available_vars:
                raise PlanError(f"Variabile {var} non disponibile al passo {step['step']}")
    return True

Lezione 2: Aggiungere tentativi a livello di passo

Errori di rete e limiti di frequenza accadono. Riprovate i singoli passi:

from tenacity import retry, stop_after_attempt, wait_exponential
@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10))
async def execute_step(tool_name: str, tool_input: str):
    return await self.toolstool_name

Lezione 3: Implementare la cache dei piani per query simili

Query come "Confronta ricavi Q1 vs Q2" hanno piani simili:

Cache dei modelli di piano
plan_template = cached_plans.get(query_category)
if plan_template:
    plan = instantiate_template(plan_template, query_params)
else:
    plan = await self.plan(query)

Risultato: Latenza di pianificazione ridotta del 40% per pattern di query ricorrenti.

Schema 3: Critic (Raffinamento iterativo)

Architettura

            Query dell'utente: "Bozza un'email di scuse professionale"
                    ↓
            ┌───────────────┐
            │ Generator LLM │  Genera risposta iniziale
            └───────┬───────┘
                    ↓
            Bozza v1: "Ci scusiamo per il problema..."
                    ↓
            ┌───────────────┐
            │  Critic LLM   │  Valuta la qualità
            └───────┬───────┘
                    ↓
        [Passato: punteggio ≥ 8/10] ────→ Restituisci risposta
                    │
        [Non passato: punteggio < 8/10]
                    ↓
            Feedback: "Troppo informale. Aggiungi dettagli specifici."
                    ↓
            ┌───────────────┐
            │  Generator    │  Rigenera con feedback
            └───────┬───────┘
                    ↓
            Bozza v2: "Ci scusiamo sinceramente per [problema specifico]..."
                    ↓
            [Ripeti fino a max_iterations=3]

Quando usarlo

• Output a qualità critica (documenti legali, comunicazioni con i clienti).

• Raffinamento iterativo necessario.

• Tolleranza alla latenza (gli utenti si aspettano 3-10s per attività complesse).

Implementazione

critic_agent.py
from typing import Tuple
class CriticAgent:
    """Agent con ciclo di auto-critica."""
    def __init__(self, max_iterations: int = 3):
        self.max_iterations = max_iterations
    async def generate(self, query: str, feedback: str = None) -> str:
        """Genera risposta (con feedback opzionale)."""
        if feedback:
            prompt = f"""
Richiesta dell'utente: {query}
Il tentativo precedente ha ricevuto questo feedback:
{feedback}
Genera una risposta migliorata che affronti il feedback.
"""
        else:
            prompt = f"Richiesta dell'utente: {query}\n\nGenera una risposta."
        return await llm.generate(prompt)
    async def critique(self, query: str, response: str) -> Tuple[float, str]:
        """Critica la qualità della risposta (punteggio 0-10, feedback)."""
        critic_prompt = f"""
Valuta questa risposta per qualità, accuratezza e professionalità.
Richiesta dell'utente: {query}
Risposta: {response}
Fornisci:
1. Punteggio (0-10)
2. Feedback specifico per il miglioramento
Formato: {{"score": X, "feedback": "..."}}
"""
        critique = await llm.generate(critic_prompt)
        result = json.loads(critique)
        return result['score'], result['feedback']
    async def run(self, query: str, min_score: float = 8.0) -> Dict:
        """Genera con raffinamento iterativo."""
        history = []
        for iteration in range(self.max_iterations):
            # Genera risposta (con feedback dell'iterazione precedente)
            feedback = history[-1]['feedback'] if history else None
            response = await self.generate(query, feedback)
            # Critica la risposta
            score, feedback = await self.critique(query, response)
            history.append({
                "iteration": iteration + 1,
                "response": response,
                "score": score,
                "feedback": feedback
            })
            # Controlla se la soglia di qualità è raggiunta
            if score >= min_score:
                return {
                    "response": response,
                    "iterations": iteration + 1,
                    "final_score": score,
                    "history": history
                }
        # Raggiunto il massimo delle iterazioni, restituisci il miglior tentativo
        best = max(history, key=lambda x: x['score'])
        return {
            "response": best['response'],
            "iterations": self.max_iterations,
            "final_score": best['score'],
            "history": history,
            "warning": "Raggiunto il massimo delle iterazioni senza raggiungere la soglia di qualità"
        }
Utilizzo
agent = CriticAgent(max_iterations=3)
result = await agent.run("Bozza una scusa professionale per spedizione in ritardo")
print(f"Risposta finale (Punteggio: {result['final_score']}):\n{result['response']}")

Dati di produzione (Generazione documenti legali)

Metrica	Valore
Documenti generati	800/mese
Successo al primo tentativo	62% (punteggio ≥ 8/10)
Successo al secondo tentativo	89%
Successo al terzo tentativo	96%
Latenza media	4.2s
Latenza p95	11.8s

Punti di forza

✅ Output di qualità superiore — L'autocorrezione cattura gli errori ✅ Adattabile — Impara dai propri errori all'interno della sessione ✅ Trasparente — Il feedback della critica spiega i problemi di qualità ✅ Degradazione graziosa — Restituisce il miglior tentativo se la soglia non è raggiunta

Debolezze

❌ Alta latenza — 2-6 chiamate LLM (2x per iterazione) ❌ Costoso — I costi scalano con le iterazioni ❌ Può ciclare all'infinito — Bisogna impostare max_iterations ❌ Il critic può sbagliare — Falsi negativi (buona risposta punteggiata bassa)

Lezioni di produzione

Lezione 1: Impostare un limite aggressivo di max_iterations

Il nostro limite iniziale era 5. Il 12% delle query lo raggiungeva (sprecando 10 chiamate LLM). Ridotto a 3:

Analisi dei costi
avg_cost_per_llm_call = $0.02
max_iterations = 5 → avg_cost = $0.20 (10 chiamate)
max_iterations = 3 → avg_cost = $0.12 (6 chiamate)
Riduzione dei costi del 40% con impatto minimo sulla qualità

Lezione 2: Usare modelli rapidi per la critica

Il critic non ha bisogno dell'intelligenza di un modello frontier. Usiamo GPT-4 per la generazione, GPT-3.5-turbo per la critica:

async def critique(self, query: str, response: str):
    # Usa un modello più economico e veloce per la critica
    critique = await llm.generate(critic_prompt, model="gpt-3.5-turbo")
    # ...

Risultato: Latenza della critica ridotta del 60% (600ms → 240ms) con la stessa accuratezza.

Lezione 3: Aggiungere early stopping su punteggi "perfetti"

Se il primo tentativo punteggia 9.5/10, saltate le iterazioni successive:

if score >= 9.5:  # Soglia "perfetta"
    return early_with_success(response, score)

Confronto latenza: Dati reali di produzione

Schema	Latenza media	Latenza p95	Latenza p99	Chiamate LLM
Router	820ms	1.2s	1.8s	1
Planner-Executor (3 passi)	3.2s	8.4s	14.1s	4
Critic (media 1.8 iterazioni)	4.2s	11.8s	18.5s	3.6

Confronto costi

Ipotesi:

• Input GPT-4: $0.01/1K token.

• Output GPT-4: $0.03/1K token.

• Query media: 200 token di input.

• Risposta media: 500 token di output.

Schema	Chiamate LLM	Costo medio
Router	1	$0.017
Planner-Executor	4	$0.068
Critic	3.6	$0.061

Matrice decisionale: Quale schema usare?

Scegliete Router se:

✅ Il dispatch a singolo tool è sufficiente ✅ Serve latenza <1s ✅ Il routing delle query è univoco ✅ Il costo per query è importante

Scegliete Planner-Executor se:

✅ Servono workflow multi-passo ✅ Serve composizione di tool ✅ Latenza <5s è accettabile ✅ La trasparenza (piano visibile) è preziosa

Scegliete Critic se:

✅ La qualità dell'output è missione-critica ✅ Latenza <10s è accettabile ✅ L'autocorrezione crea valore ✅ La qualità della prima bozza è insufficiente

Schemi ibridi che abbiamo testato

Schema 4: Router + Planner-Executor

Instradate query semplici a tool singoli, quelle complesse al planner:

if query_complexity(query) < 0.5:
    return router.route(query)  # Percorso veloce
else:
    return planner_executor.run(query)  # Percorso lento

Risultato: Il 70% delle query prende il percorso veloce (media 850ms), il 30% il percorso lento (media 3.5s). Media totale: 1.6s.

Schema 5: Planner-Executor + Critic

Pianificate, eseguite, poi criticate l'output finale:

context = await planner_executor.execute(query)
final_response = await generate_response(context)
score, feedback = await critic.critique(query, final_response)
if score < 8.0:
    final_response = await regenerate_with_feedback(context, feedback)

Risultato: Usato per report ad alto rischio. Latenza: 8-12s. Qualità: 98% di soddisfazione degli utenti.

Conclusione

Dopo 18+ mesi in produzione:

1. Router gestisce l'80% delle query con eccellente latenza

2. Planner-Executor eccelle per i workflow multi-passo ma richiede validazione del piano

3. Critic migliora la qualità del 15-20% ma raddoppia costi e latenza

La nostra raccomandazione predefinita:

• Iniziate con Router per MVP.

• Aggiungete Planner-Executor quando gli utenti richiedono attività multi-passo.

• Riservate Critic per output a qualità critica (legali, finanze, medicina).

Il miglior schema dipende dal vostro budget di latenza, dai requisiti di qualità e dai vincoli di costo. Non fate over-engineering — distribuite semplice, aumentate la complessità secondo necessità.

---

State costruendo sistemi di agenti? Contattateci per discutere la vostra architettura. Offriamo consulenza sul design di agenti, supporto all'implementazione e servizi di ottimizzazione della produzione.