0Pricing
AI Engineering Academy · Lezione

Creare chain con LCEL

Utilizzerà l'operatore pipe per concatenare un template di prompt, un LLM e un output parser in una sequenza Runnable, la eseguirà in modo sincrono e asincrono e analizzerà gli output intermedi.

Creare chain con LCEL è una lezione AI Engineering Academy gratuita su CoddyKit. Questa è la lezione 2 di 4. Puoi leggere la lezione completa qui gratuitamente — poi esercitati direttamente nel browser con un editor di codice integrato e un tutor IA disponibile 24/7. Fa parte del percorso di apprendimento AI Engineering Academy, e i tuoi progressi si sincronizzano tra il web e l'app CoddyKit. Il corso AI Engineering Academy include 4 lezioni in totale.

Che cos'è LCEL?

LCEL (LangChain Expression Language) è il modo dichiarativo per comporre i componenti LangChain in pipeline. Invece di istanziare classi Chain verbose, si collegano i Runnable con l'operatore |. Le catene LCEL supportano automaticamente streaming, operazioni asincrone, elaborazione batch e fallback: basta usare la sintassi pipe per ottenere tutte queste funzionalità.

La prima catena LCEL

La catena LCEL più semplice combina un PromptTemplate, un ChatModel e un OutputParser. Ogni componente è un Runnable e l'operatore pipe li collega. Quando si chiama chain.invoke(), l'input attraversa ogni passaggio in sequenza e l'output di ciascuno diventa l'input del successivo.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template(
    'Write a product description for: {product}'
)
model = ChatOpenAI(model='gpt-4o-mini', temperature=0.7)
parser = StrOutputParser()

chain = prompt | model | parser

result = chain.invoke({'product': 'noise-cancelling headphones'})
print(result)

Esaminare gli interni della catena

Le catene LCEL espongono la propria struttura in modo trasparente. chain.steps elenca i componenti nell'ordine in cui vengono eseguiti. È possibile chiamare chain.input_schema e chain.output_schema per esaminare i tipi che entrano ed escono. La chiamata invoke() accetta anche un parametro config per passare nomi di esecuzione, tag e callback a una singola invocazione senza modificare la definizione della catena.

# Inspect chain structure
print('Steps:', chain.steps)
print('Input schema:', chain.input_schema.schema())
print('Output schema:', chain.output_schema.schema())

# Named run for observability
result = chain.invoke(
    {'product': 'coffee maker'},
    config={'run_name': 'product-desc-run', 'tags': ['production']}
)

Invoke sincrono e asincrono

Le catene LCEL espongono versioni sincrone e asincrone di tutti i metodi. invoke() blocca il thread chiamante. ainvoke() è l'equivalente asincrono e dovrebbe essere usato nei gestori FastAPI e in altri contesti asincroni per evitare di bloccare l'event loop. La versione asincrona ha semantica identica: la differenza dell'interfaccia consiste solo in await.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = ChatPromptTemplate.from_template('Explain {topic}') | ChatOpenAI() | StrOutputParser()

async def main():
    # Async invocation
    result = await chain.ainvoke({'topic': 'quantum computing'})
    print(result)

asyncio.run(main())

Trasmettere i token attraverso la catena

Le catene LCEL propagano lo streaming lungo tutta la catena. Chiami chain.stream() per ricevere i token man mano che il modello li genera. StrOutputParser è consapevole dello streaming e inoltra singoli chunk di stringa invece di attendere la risposta completa. In questo modo gli utenti ricevono un feedback immediato anziché dover attendere a lungo prima di visualizzare la risposta completa.

chain = (
    ChatPromptTemplate.from_template('Tell me about {topic} in detail.')
    | ChatOpenAI(model='gpt-4o-mini')
    | StrOutputParser()
)

# Print each token as it arrives
for chunk in chain.stream({'topic': 'neural networks'}):
    print(chunk, end='', flush=True)
print()  # newline at end

# Async streaming
async def stream_async():
    async for chunk in chain.astream({'topic': 'transformers'}):
        print(chunk, end='', flush=True)

Elaborare più input in batch

chain.batch() accetta un elenco di dizionari di input e li elabora con concorrenza configurabile. Internamente LangChain esegue le chiamate in thread paralleli, rendendo batch molto più veloce rispetto a chiamare invoke() in un ciclo. Il parametro max_concurrency limita il numero di chiamate API parallele per rispettare i limiti di frequenza. I risultati vengono restituiti nello stesso ordine degli input.

products = [
    {'product': 'laptop'},
    {'product': 'tablet'},
    {'product': 'smartwatch'},
    {'product': 'earbuds'},
]

# Process all 4 concurrently
results = chain.batch(products, config={'max_concurrency': 4})

for product, description in zip(products, results):
    print(f'{product["product"]}: {description[:80]}...')

RunnablePassthrough e RunnableLambda

RunnablePassthrough inoltra il proprio input senza modificarlo, utile per passare il contesto originale insieme ai dati trasformati. RunnableLambda avvolge qualsiasi funzione Python come Runnable, consentendo di inserire logica arbitraria in una catena LCEL. Insieme permettono di manipolare i dati tra i passaggi della catena senza dover scrivere una classe Runnable personalizzata completa.

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

def word_count(text: str) -> str:
    count = len(text.split())
    return f'{text}\n\n[Word count: {count}]'

chain = (
    ChatPromptTemplate.from_template('Write about {topic}')
    | ChatOpenAI(model='gpt-4o-mini')
    | StrOutputParser()
    | RunnableLambda(word_count)  # Add word count to output
)

result = chain.invoke({'topic': 'climate change'})
print(result)

Passare contesto aggiuntivo con RunnablePassthrough

Un pattern comune nel RAG consiste nel passare al prompt finale sia i documenti recuperati sia la domanda dell'utente. Usi RunnablePassthrough.assign() per aggiungere campi calcolati al dizionario di input senza perdere le chiavi originali. In questo modo l'oggetto contesto si arricchisce mentre attraversa la catena e il template del prompt può accedere a tutti i dati.

from langchain_core.runnables import RunnablePassthrough

# Fake retriever for illustration
def retrieve_docs(question: str):
    return ['Doc 1 content', 'Doc 2 content']

retriever = RunnableLambda(retrieve_docs)

rag_chain = (
    RunnablePassthrough.assign(context=retriever)  # adds 'context' key
    | ChatPromptTemplate.from_template(
        'Context: {context}\n\nAnswer: {question}'
    )
    | ChatOpenAI()
    | StrOutputParser()
)

result = rag_chain.invoke({'question': 'What is machine learning?'})

Aggiungere fallback alle catene

Le catene LCEL supportano i fallback, ovvero catene alternative da provare se la catena principale genera un'eccezione. Chiami chain.with_fallbacks([backup_chain]) per registrare una o più catene di fallback. LangChain prova prima quella principale, intercetta eventuali eccezioni e poi prova ogni fallback nell'ordine. Questo è il modo standard per gestire con affidabilità le interruzioni dei provider o gli errori dei modelli.

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary_chain = (
    ChatPromptTemplate.from_template('Answer: {question}')
    | ChatOpenAI(model='gpt-4o')
    | StrOutputParser()
)

fallback_chain = (
    ChatPromptTemplate.from_template('Answer: {question}')
    | ChatAnthropic(model='claude-3-haiku-20240307')
    | StrOutputParser()
)

robust_chain = primary_chain.with_fallbacks([fallback_chain])
# If OpenAI fails, automatically tries Anthropic

Eseguire il debug con intermediate_steps

Il debug delle catene LCEL è più semplice usando la configurazione di invoke() e l'opzione verbose=True. È inoltre possibile usare .with_config({'verbose': True}) su qualsiasi catena per registrare l'input e l'output di ogni passaggio. Per un debug strutturato, avvolga globalmente la catena con set_verbose(True) oppure usi il tracing di LangSmith per acquisire ogni valore intermedio in un esploratore visivo delle tracce.

from langchain.globals import set_verbose, set_debug

# Verbose: logs LLM inputs/outputs
set_verbose(True)

# Debug: logs everything including prompts, parsers, and runnables
set_debug(True)

result = chain.invoke({'product': 'smart watch'})

# Or per-chain verbose config
chain_debug = chain.with_config({'verbose': True})
result = chain_debug.invoke({'product': 'laptop'})

Catene configurabili in fase di esecuzione

Le catene LCEL supportano la configurazione in fase di esecuzione tramite ConfigurableField. È possibile esporre come opzioni configurabili i parametri del modello, come temperatura, nome del modello e max_tokens, senza codificarli direttamente. Al momento dell'invocazione, passi una configurazione Configurable per sovrascrivere i valori predefiniti. In questo modo una sola catena può servire più utenti con impostazioni diverse oppure consentire test A/B delle configurazioni del modello.

from langchain_core.runnables import ConfigurableField

llm = ChatOpenAI(model='gpt-4o-mini').configurable_fields(
    temperature=ConfigurableField(
        id='temperature',
        name='LLM Temperature',
        description='Controls output creativity'
    )
)

chain = ChatPromptTemplate.from_template('Write about {topic}') | llm | StrOutputParser()

# Low temperature for factual output
factual = chain.invoke(
    {'topic': 'history of Rome'},
    config={'configurable': {'temperature': 0.0}}
)

Verifica rapida

Verifichi la Sua comprensione della creazione di catene con LCEL.

Riepilogo della lezione

In questa lezione ha imparato che: la composizione pipe di LCEL collega i Runnable con | creando pipeline chiare e leggibili; RunnablePassthrough e RunnableLambda consentono di inserire passaggi di manipolazione dei dati senza uscire dal paradigma LCEL; e i fallback rendono le catene più resilienti, provando automaticamente catene di riserva in caso di errore. Ora esploreremo le catene con diramazioni e parallele per gestire logiche di instradamento più complesse.

Domande Frequenti

La lezione «Creare chain con LCEL» è gratuita?

Sì — il testo completo di «Creare chain con LCEL» è gratuito qui sul web. Per esercitarvi in modo interattivo (un editor di codice integrato e un tutor IA 24/7) e sbloccare il resto del corso AI Engineering Academy, passa a CoddyKit PRO. Il corso AI Engineering Academy include 4 lezioni in totale.

Cosa imparerò in «Creare chain con LCEL»?

Utilizzerà l'operatore pipe per concatenare un template di prompt, un LLM e un output parser in una sequenza Runnable, la eseguirà in modo sincrono e asincrono e analizzerà gli output intermedi. Eserciti AI Engineering Academy con codice pratico che esegui direttamente nel browser, e un tutor IA 24/7 risponde alle tue domande mentre lavori sulla lezione.

Ho bisogno di esperienza per iniziare AI Engineering Academy?

Non è richiesta alcuna esperienza precedente. AI Engineering Academy su CoddyKit è strutturato per principianti e studenti avanzati, quindi puoi iniziare da qui o dall'inizio e procedere al tuo ritmo. Questa è la lezione 2 di 4.

Quanto tempo richiede la lezione «Creare chain con LCEL»?

La maggior parte delle lezioni CoddyKit richiede circa 5–10 minuti. Ogni lezione è breve e interattiva, quindi fai progressi costanti e riprendi esattamente da dove hai lasciato su web e app.

Posso scrivere ed eseguire codice in questa lezione AI Engineering Academy?

Sì. Ogni lezione AI Engineering Academy include un editor di codice integrato, quindi scrivi ed esegui codice reale direttamente nel tuo browser e ricevi feedback istantaneo dall'IA — nessuna configurazione locale necessaria.

Tutte le lezioni di questo corso

  1. Architettura di LangChain e astrazioni fondamentali
  2. Creare chain con LCEL
  3. Chain ramificate e parallele
  4. Streaming dell'output in LangChain
← Torna a AI Engineering Academy