Come costruire un sistema diprompt chainingper automatizzare flussi di lavoro complessi e ottenere output strutturati su LLM multipli

Domanda chiave:Come posso collegare più prompt, gestire lo stato e garantire coerenza quando utilizzo diversi Large Language Model (LLM) in una pipeline di produzione?In questo articolo troverai la risposta completa, con architetture, pattern, esempi di codice e suggerimenti per ottimizzare costi e latenza.

Indice

• Introduzione al prompt chaining
• Architetture di chaining: sequenziare i prompt
• Gestione degli stati e memoria contestuale
• Pattern di branching e fallback
• Ottimizzazione di performance e costi
• Esempio pratico: workflow di generazione di report
• Conclusioni e prossimi passi

Introduzione al prompt chaining

Ilprompt chainingè la tecnica di collegare più richieste (prompt) a uno o più LLM, facendo passare il risultato di uno step come input per il successivo. Questo approccio permette di:

• Dividere compiti complessi in micro'task più gestibili;
• Mantenere la coerenza semantica grazie al passaggio esplicito del contesto;
• Inserire punti di validazione, correzione o riformulazione prima di proseguire;
• Sfruttare le specificità di LLM diversi (es. un modello ottimizzato per codice, uno per ragionamento logico).

Il risultato è unworkflow automatizzatoche può generare outputstrutturati(JSON, CSV, XML) ecoerenti, ideale per applicazioni aziendali, data'pipeline o chatbot avanzati.

Architetture di chaining: come sequenziare prompt multipli

Esistono tre architetture di base per il chaining:

1. Linear chaining

Una catena semplice in cui ogni step riceve l'output del precedente. Ideale per processi lineari (es. estrazione ' pulizia ' formattazione).

# Pseudo'code per Linear chaining con OpenAI API
import openai

def call_llm(prompt, model="gpt-4"):
    response = openai.ChatCompletion.create(
        model=model,
        messages=[{"role": "user", "content": prompt}]
    )
    return response.choices[0].message.content.strip()

# Step 1: estrazione informazioni da testo grezzo
step1 = call_llm("Estrai tutti i dati anagrafici dal seguente testo: ...")
# Step 2: pulizia e normalizzazione
step2 = call_llm(f"Normalizza i dati in formato JSON: {step1}")
# Step 3: validazione schema
step3 = call_llm(f"Convalida il JSON rispetto allo schema fornito: {step2}")
print(step3)

2. Branching chaining

Quando un output può generare percorsi diversi, si crea un albero di decisione. Ogni ramo può invocare LLM specializzati.

# Esempio di branching basato su tipo di documento
doc_type = call_llm("Classifica il documento come 'fattura', 'contratto' o 'altro': ...")
if doc_type == "fattura":
    result = call_llm("Estrai importo, data e partita IVA dalla fattura: ...", model="gpt-4o-mini")
elif doc_type == "contratto":
    result = call_llm("Riassumi le clausole chiave del contratto: ...", model="gpt-4")
else:
    result = call_llm("Genera un abstract generico: ...")
print(result)

3. Looping chaining con validazione

Un ciclo che ripete un passo finché non è soddisfatta una condizione (es. validazione JSON). Questo riduce errori di sintassi.

def validate_json(json_str, schema):
    # funzione di validazione semplificata
    import jsonschema, json
    try:
        jsonschema.validate(instance=json.loads(json_str), schema=schema)
        return True
    except jsonschema.ValidationError:
        return False

schema = {"type": "object", "properties": {"nome": {"type": "string"}}, "required": ["nome"]}

output = ""
while not validate_json(output, schema):
    output = call_llm(f"Genera JSON conforme allo schema {schema}: ...")
print("JSON valido:", output)

Gestione degli stati: tecniche per mantenere coerenza e memoria contestuale

In catene lunghe, ilcontestopuò perdersi. Ecco le tecniche più efficaci:

• State object centralizzato: un dizionario Python (o una struttura JSON) che raccoglie tutti gli output intermedi e le variabili di controllo.
• Prompt templating: inserire placeholder come{{previous_output}}nel prompt successivo.
• Chunking intelligente: quando l'output supera il limite di token, suddividerlo in blocchi e riassumere prima di passarlo avanti.
• Memory cache(Redis, Memcached): utile per pipeline distribuite dove più worker devono accedere allo stesso stato.

Esempio distate objectin Python:

state = {
    "raw_text": "...",
    "extracted": None,
    "normalized": None,
    "validated": False,
}

state["extracted"] = call_llm(f"Estrai dati: {state['raw_text']}")
state["normalized"] = call_llm(f"Normalizza: {state['extracted']}")
state["validated"] = validate_json(state["normalized"], schema)

Pattern di branching e fallback: logica condizionale basata su output intermedi

Un workflow robusto deve prevederefallbackquando un modello fallisce o restituisce risultati ambigui.

Pattern "Try'Catch" per LLM

try:
    result = call_llm(prompt, model="gpt-4", temperature=0.0)
    if "error" in result.lower():
        raise ValueError("LLM ha segnalato errore")
except Exception as e:
    # Fallback al modello più piccolo
    result = call_llm(prompt, model="gpt-4o-mini", temperature=0.2)
    log_warning(f"Fallback attivato: {e}")

Branching basato su punteggio di confidenza

Alcune API (es. Azure OpenAI) restituiscono unlogprobmedio. Puoi usarlo per decidere se accettare o ricalcolare.

response = openai.ChatCompletion.create(..., logprobs=True)
conf = sum(response.choices[0].logprobs.token_logprobs) / len(response.choices[0].logprobs.token_logprobs)
if conf < -1.5:  # soglia arbitraria
    # Riprocessa con temperature più alta
    result = call_llm(prompt, temperature=0.8)

Ottimizzazione delle performance: ridurre latenza e costi

In produzione, la velocità e il budget sono cruciali. Ecco le best practice:

• Batching dei prompt: raggruppa richieste simili in un unico chiamata quando il modello lo supporta.
• Modelli specializzati per task: usagpt-4o-miniper compiti semplici (estrazione, formattazione) e riservagpt-4per ragionamento complesso.
• Cache dei risultati: salva output già calcolati (hash del prompt) per evitare chiamate duplicate.
• Limiti di token: mantieni i prompt sotto i 2 000 token per ridurre il tempo di elaborazione.
• Parallelismo controllato: utilizzaasyncioo thread pool per eseguire più rami indipendenti contemporaneamente, ma monitorando il rate'limit dell'API.

Esempio di caching confunctools.lru_cache:

from functools import lru_cache

@lru_cache(maxsize=1024)
def cached_call(prompt, model="gpt-4"):
    return call_llm(prompt, model)

# Uso
result = cached_call("Qual è la capitale d'Italia?")

Esempio pratico: workflow di generazione di report finanziario

Supponiamo di dover creare un report mensile a partire da dati CSV grezzi. Il flusso include:

1. Caricamento CSV ' estrazione insight
2. Riformulazione in linguaggio naturale
3. Creazione di un executive summary
4. Formattazione finale in JSON per il dashboard

Diagramma della catena

CSV ' [LLM'1 (analisi dati)] ' insight_raw
insight_raw ' [LLM'2 (riformulazione)] ' insight_text
insight_text ' [LLM'3 (summary)] ' summary
summary + insight_raw ' [LLM'4 (JSON formatter)] ' report_json

Implementazione in Python (FastAPI + OpenAI)

from fastapi import FastAPI, UploadFile
import pandas as pd, json, openai, asyncio
app = FastAPI()

async def llm(prompt, model="gpt-4o-mini"):
    resp = await openai.ChatCompletion.acreate(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.0,
    )
    return resp.choices[0].message.content.strip()

@app.post("/report")
async def generate_report(file: UploadFile):
    df = pd.read_csv(file.file)
    # Step 1: insight grezzo
    prompt1 = f"Analizza questi dati e fornisci gli insight chiave in 5 punti:\n{df.head().to_csv(index=False)}"
    insight_raw = await llm(prompt1, model="gpt-4o-mini")

    # Step 2: traduzione in linguaggio naturale
    prompt2 = f"Riformula i seguenti insight in un paragrafo fluido:\n{insight_raw}"
    insight_text = await llm(prompt2)

    # Step 3: executive summary
    prompt3 = f"Crea un executive summary di massimo 3 frasi basato su:\n{insight_text}"
    summary = await llm(prompt3)

    # Step 4: formato JSON per dashboard
    prompt4 = f"Restituisci in JSON le chiavi: summary, details (lista di insight). Usa questo template:\n{{\"summary\": \"...\", \"details\": [\"...\"]}}. Dati: {insight_raw}"
    report_json = await llm(prompt4, model="gpt-4")

    return json.loads(report_json)

Il codice dimostra:

• Uso diasyncioper ridurre latenza;
• Passaggio esplicito dello stato (insight_raw,insight_text);
• Validazione finale conjson.loadsper assicurare output strutturato.

Conclusioni e prossimi passi

Costruire unsistema di prompt chainingefficace richiede una progettazione attenta dell'architettura, della gestione dello stato e dei meccanismi di fallback. Seguendo i pattern descritti " linear, branching, looping " e applicando le tecniche di ottimizzazione (caching, batching, modelli specializzati) è possibile realizzare pipeline robuste, veloci e a costi contenuti.

Per approfondire, considera di sperimentare con:

• Framework dedicati comeLangChainoSemantic Kernel;
• Strumenti di monitoraggio (Prometheus, Grafana) per tracciare latenza e token usage;
• Modelli open'source (Llama 3, Mistral) per ridurre dipendenza da API cloud.

Inizia subito a mappare il tuo caso d'uso, definisci i punti di validazione e implementa il primoprompt chaincon il codice di esempio. Il risultato sarà un’automazione intelligente capace di trasformare dati grezzi in insight azionabili, con la qualità e la coerenza garantite da LLM multipli.