6. Deklarative Pipelines mit LCEL

In den vorherigen Kapiteln haben wir imperativen Code geschrieben, um LLM-Interaktionen zu orchestrieren: einen Prompt erstellen, das Modell aufrufen, die Antwort parsen. Das funktioniert, aber wenn KI-Anwendungen komplexer werden, wird dieser Ansatz umständlich und schwerer wartbar. Sie enden mit tief verschachtelten Funktionsaufrufen, manueller Fehlerbehandlung bei jedem Schritt und Schwierigkeiten, den gesamten Datenfluss zu verstehen.

LangChain Expression Language (LCEL) löst dies, indem Sie deklarieren können, was passieren soll, nicht wie es passieren soll. Anstatt prozeduralen Code zu schreiben, der Funktionen nacheinander aufruft, komponieren Sie Komponenten mit einem einfachen Pipe-Operator (|), der wie eine Unix-Pipeline gelesen wird. Das Ergebnis ist saubererer, wartbarerer Code, der den Datenfluss durch Ihr KI-System klar ausdrückt.

Dieses Kapitel führt LCEL zum Erstellen linearer Workflows ein - Sequenzen von Operationen, bei denen Daten von Anfang bis Ende fließen, ohne Verzweigungen oder Schleifen. Wir behandeln, wann LCEL verwendet werden sollte, wie Pipelines komponiert werden und wie sie sowohl synchron als auch mit Streaming-Ausgabe ausgeführt werden.

6.1) Warum LCEL?

Das Problem, das LCEL löst

Beginnen wir mit einem konkreten Beispiel. Angenommen, Sie erstellen einen Kundensupport-Assistenten, der Folgendes tun muss:

1. Eine Benutzerfrage entgegennehmen
2. Sie in einen Prompt mit Systemanweisungen und der Benutzernachricht formatieren
3. Sie an ein LLM senden
4. Die Antwort parsen, um nur den Textinhalt zu extrahieren

So würden Sie dies imperativ schreiben (ohne 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", "Sie sind ein hilfreicher Kundensupport-Assistent."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Schritt 1: Den Prompt mit der Frage formatieren
    messages = prompt.invoke({"question": question})
    
    # Schritt 2: LLM aufrufen
    response = llm.invoke(messages)
    
    # Schritt 3: Die Ausgabe parsen, um Textinhalt zu extrahieren
    result = parser.invoke(response)
    
    return result
 
# Verwenden
result = answer_question("Wie setze ich mein Passwort zurück?")
print(result)

Ausgabe:

Um Ihr Passwort zurückzusetzen, befolgen Sie bitte diese Schritte:
1. Gehen Sie zur Login-Seite
2. Klicken Sie auf "Passwort vergessen"
3. Geben Sie Ihre E-Mail-Adresse ein
4. Überprüfen Sie Ihre E-Mail auf einen Zurücksetzungslink
5. Folgen Sie dem Link und erstellen Sie ein neues Passwort

Das funktioniert, aber beachten Sie die Probleme:

Umständlichkeit: Jeder Schritt erfordert explizite Variablenzuweisung und Funktionsaufrufe. Die eigentliche Logik (formatieren → aufrufen → parsen) ist in Boilerplate vergraben.

Starre Struktur: Wenn Sie einen Schritt hinzufügen möchten (wie die Validierung der Frage oder das Protokollieren der Antwort), müssen Sie Code in die Mitte der Funktion einfügen, was die Komplexität erhöht.

Kein eingebautes Streaming: Um Tokens zu streamen, während sie ankommen, müssten Sie die gesamte Funktion umschreiben, um llm.stream() zu verwenden und die asynchrone Iteration manuell zu handhaben.

Unklarer Datenfluss: Beim Lesen des Codes ist nicht sofort offensichtlich, dass dies eine einfache Pipeline ist. Sie müssen Variablenzuweisungen nachverfolgen, um den Fluss zu verstehen.

Jetzt sehen wir die LCEL-Version:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Die Komponenten definieren (wie zuvor)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sie sind ein hilfreicher Kundensupport-Assistent."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Sie mit dem Pipe-Operator zu einer Chain komponieren
chain = prompt | llm | parser
 
# Verwenden
result = chain.invoke({"question": "Wie setze ich mein Passwort zurück?"})
print(result)

Ausgabe:

Um Ihr Passwort zurückzusetzen, befolgen Sie bitte diese Schritte:
1. Gehen Sie zur Login-Seite
2. Klicken Sie auf "Passwort vergessen"
3. Geben Sie Ihre E-Mail-Adresse ein
4. Überprüfen Sie Ihre E-Mail auf einen Zurücksetzungslink
5. Folgen Sie dem Link und erstellen Sie ein neues Passwort

Die Ausgabe ist identisch, aber der Code ist dramatisch anders:

Deklarativ: chain = prompt | llm | parser drückt den gesamten Fluss in einer Zeile aus. Lesen Sie von links nach rechts: prompt → LLM → parser.

Komponierbar: Jede Komponente (prompt, llm, parser) ist unabhängig und wiederverwendbar. Sie können Komponenten austauschen, ohne die Pipeline neu zu schreiben.

Streaming eingebaut: Wechseln Sie zwischen chain.invoke() und chain.stream(), ohne die Pipeline-Definition zu ändern.

Klare Absicht: Der |-Operator macht den Datenfluss auf einen Blick offensichtlich.

Wo LCEL passt: Lineare Workflows

LCEL ist für lineare Workflows konzipiert - Sequenzen, bei denen Daten in eine Richtung von Anfang bis Ende fließen, ohne Verzweigungen oder Schleifen.

Dieses lineare Muster deckt viele KI-Anwendungen ab. Betrachten Sie ein Dokument-Q&A-System: Sie erhalten eine Frage → rufen relevante Dokumente ab → formatieren einen Prompt → senden an LLM → parsen die Antwort. Jeder Schritt ist eine klare Sequenz, bei der die Ausgabe eines Schritts zur Eingabe des nächsten wird.

Aber was, wenn Ihr Workflow Folgendes benötigt:

• Das LLM soll entscheiden, welches Tool aufgerufen wird, basierend auf der Frage
• Ein Tool aufrufen, das Ergebnis beobachten und dann entscheiden, was als Nächstes zu tun ist
• Fehlgeschlagene Operationen wiederholen mit unterschiedlichen Ansätzen

Diese Szenarien erfordern Schleifen und bedingte Verzweigungen - Dinge, die LCEL nicht handhaben kann. Deshalb existiert LangGraph (wir werden es in Kapitel 15 einführen).

Hier ist ein visueller Vergleich der beiden Ansätze:

LCEL ist perfekt für unkomplizierte Pipelines, bei denen jeder Schritt die Ausgabe des vorherigen Schritts verarbeitet. Daten fließen nur in eine Richtung.

LangGraph ist für Fälle, in denen Sie Entscheidungsfindung und Schleifen benötigen. Der Agent kann handeln, Ergebnisse beobachten und erneut denken.

Für dieses Kapitel konzentrieren wir uns auf LCEL. Warum zuerst lineare Pipelines meistern?

• Grundlage: Der Pipe-Operator (|) ist die Kernsyntax von LangChain. Dies zu verstehen macht alles andere einfacher.
• Voraussetzung für LangGraph: LangGraph-Agents verwenden LCEL-Chains ausgiebig in ihren Nodes
• Reales Muster: Komplexe Agents werden durch Kombination von LCEL-Chains erstellt

Der Rest dieses Kapitels zeigt Ihnen, wie Sie diese linearen Pipelines mit dem Pipe-Operator (|) erstellen.

6.2) Pipelines mit dem | Operator komponieren

Wie LCEL-Pipes funktionieren

Der Pipe-Operator (|) von LCEL funktioniert aufgrund der Runnable-Schnittstelle.

Was ist ein Runnable?

Runnable ist die Standardschnittstelle von LangChain. Wenn eine Komponente die Runnable-Schnittstelle implementiert, kann sie mit anderen Komponenten über den |-Operator verkettet werden.

Jedes Runnable bietet diese Methoden:

• .invoke(input) - Einmal ausführen und das vollständige Ergebnis erhalten
• .stream(input) - Ausführen und jedes Wort empfangen, während das LLM es generiert
• .batch(inputs) - Mehrmals mit unterschiedlichen Eingaben ausführen und alle Ergebnisse erhalten

Für dieses Kapitel konzentrieren wir uns auf .invoke() und .stream() (wir behandeln .batch() später bei Bedarf).

Warum funktioniert |?

Weil die Runnable-Klasse den |-Operator mithilfe von Pythons Operator-Überladung implementiert. Wenn Sie prompt | llm schreiben, erstellt es ein neues Runnable, das die beiden Komponenten sequenziell ausführt.

Die meisten LangChain-Komponenten sind Runnables

Deshalb können Sie so viele verschiedene Komponenten verketten:

• ChatPromptTemplate ist ein Runnable
• ChatOpenAI ist ein Runnable
• StrOutputParser ist ein Runnable
• Sogar benutzerdefinierte Chains, die Sie mit | erstellen, sind selbst Runnables!

Das bedeutet, Sie können komplexe Pipelines erstellen, indem Sie einfachere kombinieren.

Komponenten verbinden: Eingabe-/Ausgabetypen

Beim Verbinden von Komponenten mit | müssen Sie sicherstellen, dass der Ausgabetyp einer Komponente mit dem Eingabetyp der nächsten übereinstimmt.

Wichtige Komponenten-Signaturen:

Beispielfluss:

chain = prompt | llm | parser

So transformiert sich der Datentyp, während er durch jede Komponente läuft:

• prompt: Nimmt dict als Eingabe und konvertiert zu list[BaseMessage]
• llm: Nimmt list[BaseMessage] als Eingabe und konvertiert zu AIMessage
• parser: Nimmt AIMessage als Eingabe und konvertiert zu str

Sehen wir uns das in Aktion an:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sie sind ein hilfreicher Assistent."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "Was ist 2+2?"})
print(result)  # "2+2 ergibt 4."

Was bei jedem Schritt passiert:

Was passiert, wenn Typen nicht übereinstimmen?

Wenn Sie versuchen, inkompatible Komponenten zu verbinden, erhalten Sie einen Fehler:

# FEHLER: Das funktioniert nicht
chain = llm | prompt  # LLM gibt AIMessage aus, aber prompt erfordert dict als Eingabe

Die Fehlermeldung teilt Ihnen mit, welchen Eingabetyp die nächste Komponente erwartet im Vergleich zu dem Eingabetyp, den sie tatsächlich erhalten hat.

6.3) Eine Chain ausführen: .invoke() und .stream()

Eine Pipeline ausführen

Sobald Sie eine Chain komponiert haben, führen Sie sie mit der Methode .invoke() aus. Dies ist die synchrone Methode zum Ausführen einer Pipeline - sie wartet auf die vollständige Antwort, bevor sie das Ergebnis zurückgibt.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sie sind ein hilfreicher Assistent."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Synchroner Aufruf
result = chain.invoke({"question": "Was ist 2+2?"})
print(result)

Ausgabe:

2+2 ergibt 4.

Die Methode .invoke() ist unkompliziert: Übergeben Sie den Eingabeparameter, den die erste Komponente erwartet, und erhalten Sie die Ausgabe zurück, die von der letzten Komponente produziert wird.

Tokens von derselben Chain streamen

Während .invoke() einfach ist, hat es eine Einschränkung für benutzerseitige Anwendungen: Benutzer sehen nichts, bis die gesamte Antwort vollständig ist. Bei langen Antworten (10-20 Sekunden) entsteht eine schlechte Benutzererfahrung.

Streaming zeigt Tokens sofort an, während sie generiert werden, anstatt auf die vollständige Antwort zu warten. Es ist der Tippeffekt, den Sie in ChatGPT sehen.

Sehen wir uns Streaming in Aktion an:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Sie sind ein hilfreicher Assistent."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Chunks streamen, während sie ankommen
for chunk in chain.stream({"question": "Erklären Sie, was Python ist, in einem Satz"}):
    print(chunk, end="", flush=True)

Ausgabe (in Echtzeit angezeigt, Token für Token):

Python ist eine vielseitige, hochrangige Programmiersprache, die für ihre Einfachheit und Lesbarkeit bekannt ist und weit verbreitet in Webentwicklung, Data Science und Automatisierung verwendet wird.

Beachten Sie, dass die Chain-Definition identisch mit dem .invoke()-Beispiel oben ist. Wir mussten sie nicht neu erstellen - wir haben nur .stream() anstelle von .invoke() aufgerufen.

Die Methode .stream() gibt Ergebnisse als Chunks zurück, während das LLM Tokens generiert. Das print(chunk, end="", flush=True) zeigt jeden Chunk sofort auf dem Bildschirm an und erzeugt den Live-Tippeffekt.

Wann .invoke() vs .stream() verwenden

Verwenden Sie .invoke(), wenn:

• Sie nur das Endergebnis benötigen (Analyse, Übersetzung, Klassifizierung)
• Die Antwort kurz ist und die Wartezeit kein Problem darstellt
• Sie die vollständige Ausgabe benötigen, bevor Sie mit dem nächsten Schritt fortfahren

Verwenden Sie .stream(), wenn:

• Benutzer Fortschritt sehen müssen (Chat, Content-Generierung)
• Die Antwort lang ist und die Wartezeit spürbar wäre
• Sie eine UI erstellen, bei der Echtzeit-Feedback wichtig ist

Beide Methoden funktionieren auf derselben Chain. Definieren Sie die Chain einmal und wählen Sie dann den Ausführungsmodus basierend auf Ihren Anforderungen.

In diesem Kapitel haben Sie LCEL gelernt - LangChains deklarative Pipeline-Syntax:

• Pipelines erstellen mit dem Pipe-Operator: prompt | llm | parser
• Flexibel ausführen: Verwenden Sie .invoke() für vollständige Ergebnisse oder .stream() für Echtzeit-Ausgabe
• Typsicherheit: Passen Sie Ausgabetypen an Eingabetypen an, wenn Sie Komponenten verketten

Dieselbe Chain funktioniert für beide Ausführungsmodi - einmal definieren, überall verwenden.

Weiter: Kapitel 7 behandelt strukturierte Ausgabe mit Pydantic und ermöglicht es Ihnen, validierte JSON-Daten aus LLM-Antworten zu extrahieren.