6. Pipeline Dichiarative con LCEL

Nei capitoli precedenti, abbiamo scritto codice imperativo per orchestrare le interazioni con gli LLM: creare un prompt, invocare il modello, analizzare la risposta. Questo funziona, ma man mano che le applicazioni AI diventano più complesse, questo approccio diventa verboso e più difficile da mantenere. Si finisce con chiamate di funzioni profondamente annidate, gestione manuale degli errori ad ogni passo e difficoltà nel comprendere il flusso complessivo dei dati.

LangChain Expression Language (LCEL) risolve questo problema permettendoti di dichiarare cosa vuoi che accada, non come farlo accadere. Invece di scrivere codice procedurale che chiama funzioni in sequenza, componi componenti usando un semplice operatore pipe (|) che si legge come una pipeline Unix. Il risultato è un codice più pulito e manutenibile che esprime chiaramente il flusso dei dati attraverso il tuo sistema AI.

Questo capitolo introduce LCEL per costruire flussi di lavoro lineari - sequenze di operazioni in cui i dati fluiscono dall'inizio alla fine senza ramificazioni o cicli. Tratteremo quando usare LCEL, come comporre pipeline e come eseguirle sia in modo sincrono che con output in streaming.

6.1) Perché LCEL?

Il Problema che LCEL Risolve

Iniziamo con un esempio concreto. Supponiamo che tu stia costruendo un assistente di supporto clienti che deve:

1. Prendere una domanda dell'utente
2. Formattarla in un prompt con istruzioni di sistema e il messaggio dell'utente
3. Inviarla a un LLM
4. Analizzare la risposta per estrarre solo il contenuto testuale

Ecco come potresti scriverlo in modo imperativo (senza LCEL):

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sei un assistente di supporto clienti utile."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Passo 1: Formatta il prompt con la domanda
    messages = prompt.invoke({"question": question})
    
    # Passo 2: Invoca l'LLM
    response = llm.invoke(messages)
    
    # Passo 3: Analizza l'output per estrarre il contenuto testuale
    result = parser.invoke(response)
    
    return result
 
# Usalo
result = answer_question("Come posso reimpostare la mia password?")
print(result)

Output:

Per reimpostare la tua password, segui questi passaggi:
1. Vai alla pagina di login
2. Clicca su "Password Dimenticata"
3. Inserisci il tuo indirizzo email
4. Controlla la tua email per un link di reimpostazione
5. Segui il link e crea una nuova password

Questo funziona, ma nota i problemi:

Verbosità: Ogni passo richiede assegnazioni di variabili esplicite e chiamate di funzioni. La logica effettiva (formatta → invoca → analizza) è sepolta nel boilerplate.

Struttura Rigida: Se vuoi aggiungere un passo (come validare la domanda o registrare la risposta), devi inserire codice nel mezzo della funzione, aumentando la complessità.

Nessuno Streaming Integrato: Per fare streaming dei token man mano che arrivano, dovresti riscrivere l'intera funzione per usare llm.stream() e gestire manualmente l'iterazione asincrona.

Flusso di Dati Poco Chiaro: Leggendo il codice, non è immediatamente ovvio che questa sia una semplice pipeline. Devi tracciare le assegnazioni di variabili per comprendere il flusso.

Ora vediamo la versione LCEL:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Definisci i componenti (come prima)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sei un assistente di supporto clienti utile."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Componili in una catena usando l'operatore pipe
chain = prompt | llm | parser
 
# Usalo
result = chain.invoke({"question": "Come posso reimpostare la mia password?"})
print(result)

Output:

Per reimpostare la tua password, segui questi passaggi:
1. Vai alla pagina di login
2. Clicca su "Password Dimenticata"
3. Inserisci il tuo indirizzo email
4. Controlla la tua email per un link di reimpostazione
5. Segui il link e crea una nuova password

L'output è identico, ma il codice è drammaticamente diverso:

Dichiarativo: chain = prompt | llm | parser esprime l'intero flusso in una riga. Leggilo da sinistra a destra: prompt → LLM → parser.

Componibile: Ogni componente (prompt, llm, parser) è indipendente e riutilizzabile. Puoi scambiare componenti senza riscrivere la pipeline.

Streaming Integrato: Passa da chain.invoke() a chain.stream() senza cambiare la definizione della pipeline.

Intento Chiaro: L'operatore | rende il flusso dei dati ovvio a colpo d'occhio.

Dove si Colloca LCEL: Flussi di Lavoro Lineari

LCEL è progettato per flussi di lavoro lineari - sequenze in cui i dati fluiscono in una direzione dall'inizio alla fine senza ramificazioni o cicli.

Questo pattern lineare copre molte applicazioni AI. Considera un sistema di Q&A su documenti: ricevi una domanda → recuperi documenti rilevanti → formatti un prompt → invii all'LLM → analizzi la risposta. Ogni passo è una sequenza chiara in cui l'output di un passo diventa l'input del successivo.

Ma cosa succede se il tuo flusso di lavoro deve:

• Far decidere all'LLM quale strumento chiamare in base alla domanda
• Chiamare uno strumento, osservare il risultato, poi decidere cosa fare dopo
• Ritentare operazioni fallite con approcci diversi

Questi scenari richiedono cicli e ramificazioni condizionali - cose che LCEL non può gestire. Questo è il motivo per cui esiste LangGraph (lo introdurremo nel Capitolo 15).

Ecco un confronto visivo dei due approcci:

LCEL è perfetto per pipeline semplici in cui ogni passo elabora l'output del passo precedente. I dati fluiscono in una sola direzione.

LangGraph è per quando hai bisogno di prendere decisioni e cicli. L'agente può agire, osservare i risultati e pensare di nuovo.

Per questo capitolo, ci concentriamo su LCEL. Perché padroneggiare prima le pipeline lineari?

• Fondamento: L'operatore pipe (|) è la sintassi centrale di LangChain. Comprendere questo rende tutto il resto più facile.
• Prerequisito per LangGraph: Gli agenti LangGraph usano estensivamente catene LCEL all'interno dei loro nodi
• Pattern del mondo reale: Gli agenti complessi sono costruiti combinando catene LCEL

Il resto di questo capitolo ti mostrerà come costruire queste pipeline lineari con l'operatore pipe (|).

6.2) Comporre Pipeline con l'Operatore |

Come Funzionano le Pipe LCEL

L'operatore pipe (|) di LCEL funziona grazie all'interfaccia Runnable.

Cos'è un Runnable?

Runnable è l'interfaccia standard di LangChain. Quando un componente implementa l'interfaccia Runnable, può essere concatenato con altri componenti usando l'operatore |.

Ogni Runnable fornisce questi metodi:

• .invoke(input) - Esegui una volta e ottieni il risultato completo
• .stream(input) - Esegui e ricevi ogni parola mentre l'LLM la genera
• .batch(inputs) - Esegui più volte con input diversi e ottieni tutti i risultati

Per questo capitolo, ci concentreremo su .invoke() e .stream() (tratteremo .batch() più avanti quando necessario).

Perché | funziona?

Perché la classe Runnable implementa l'operatore | usando l'overloading degli operatori di Python. Quando scrivi prompt | llm, crea un nuovo Runnable che esegue i due componenti in sequenza.

La maggior parte dei componenti LangChain sono Runnable

Questo è il motivo per cui puoi concatenare così tanti componenti diversi:

• ChatPromptTemplate è un Runnable
• ChatOpenAI è un Runnable
• StrOutputParser è un Runnable
• Anche le catene personalizzate che crei con | sono esse stesse Runnable!

Questo significa che puoi costruire pipeline complesse combinando quelle più semplici.

Connettere Componenti: Tipi di Input/Output

Quando connetti componenti con |, devi assicurarti che il tipo di output di un componente corrisponda al tipo di input del successivo.

Firme dei componenti chiave:

Esempio di flusso:

chain = prompt | llm | parser

Ecco come il tipo di dato si trasforma mentre passa attraverso ogni componente:

• prompt: Prende dict come input e converte in list[BaseMessage]
• llm: Prende list[BaseMessage] come input e converte in AIMessage
• parser: Prende AIMessage come input e converte in str

Vediamolo in azione:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sei un assistente utile."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "Quanto fa 2+2?"})
print(result)  # "2+2 fa 4."

Cosa succede ad ogni passo:

Cosa succede se i tipi non corrispondono?

Se provi a connettere componenti incompatibili, otterrai un errore:

# ERRORE: Questo non funzionerà
chain = llm | prompt  # L'LLM produce AIMessage, ma prompt richiede dict come input

Il messaggio di errore ti dirà quale tipo di input si aspetta il componente successivo rispetto a quale tipo di input ha effettivamente ricevuto.

6.3) Eseguire una Catena: .invoke() e .stream()

Eseguire una Pipeline

Una volta composta una catena, la esegui usando il metodo .invoke(). Questo è il modo sincrono per eseguire una pipeline - attende l'intera risposta prima di restituire il risultato.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sei un assistente utile."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Invocazione sincrona
result = chain.invoke({"question": "Quanto fa 2+2?"})
print(result)

Output:

2+2 fa 4.

Il metodo .invoke() è semplice: passa il parametro di input che il primo componente si aspetta e ottieni l'output prodotto dall'ultimo componente.

Streaming di Token dalla Stessa Catena

Mentre .invoke() è semplice, ha una limitazione per le applicazioni rivolte agli utenti: gli utenti non vedono nulla fino a quando la risposta completa non è pronta. Per risposte lunghe (10-20 secondi), questo crea una scarsa esperienza utente.

Lo streaming mostra i token immediatamente mentre vengono generati, invece di aspettare la risposta completa. È l'effetto di digitazione che vedi in ChatGPT.

Vediamo lo streaming in azione:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sei un assistente utile."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Fai streaming dei chunk man mano che arrivano
for chunk in chain.stream({"question": "Spiega cos'è Python in una frase"}):
    print(chunk, end="", flush=True)

Output (visualizzato in tempo reale, token per token):

Python è un linguaggio di programmazione versatile e di alto livello noto per la sua semplicità e leggibilità, ampiamente utilizzato nello sviluppo web, nella data science e nell'automazione.

Nota che la definizione della catena è identica all'esempio .invoke() sopra. Non abbiamo dovuto ricostruirla - abbiamo solo chiamato .stream() invece di .invoke().

Il metodo .stream() restituisce i risultati come chunk mentre l'LLM genera i token. Il print(chunk, end="", flush=True) visualizza ogni chunk immediatamente sullo schermo, creando l'effetto di digitazione dal vivo.

Quando Usare .invoke() vs .stream()

Usa .invoke() quando:

• Hai bisogno solo del risultato finale (analisi, traduzione, classificazione)
• La risposta è breve e il tempo di attesa non è un problema
• Hai bisogno dell'output completo prima di procedere al passo successivo

Usa .stream() quando:

• Gli utenti devono vedere il progresso (chat, generazione di contenuti)
• La risposta è lunga e il tempo di attesa sarebbe evidente
• Stai costruendo un'interfaccia utente dove il feedback in tempo reale è importante

Entrambi i metodi funzionano sulla stessa catena. Definisci la catena una volta, poi scegli la modalità di esecuzione in base alle tue esigenze.

In questo capitolo, hai imparato LCEL - la sintassi dichiarativa delle pipeline di LangChain:

• Costruire pipeline con l'operatore pipe: prompt | llm | parser
• Eseguire in modo flessibile: Usa .invoke() per risultati completi o .stream() per output in tempo reale
• Sicurezza dei tipi: Abbina i tipi di output ai tipi di input quando concateni componenti

La stessa catena funziona per entrambe le modalità di esecuzione - definisci una volta, usa ovunque.

Prossimo: Il Capitolo 7 tratta l'output strutturato con Pydantic, permettendoti di estrarre dati JSON validati dalle risposte degli LLM.