LangChain LCEL in der Praxis: Vom klassischen Chain zum Streaming – der moderne Ansatz

Letztes Jahr habe ich ein älteres Projekt übernommen – beim Öffnen des Quellcodes erstarrte ich: Eine einzige Dialog-Chain umfasste über 200 Zeilen. PromptTemplate initialisieren, LLMChain konfigurieren, Ein-/Ausgabe-Mapping von Hand, eigene Callbacks für Streaming. Schlimmer: Niemand im Team wagte es anzufassen. „Läuft doch“ wurde zur ungeschriebenen Regel.

Das ist Erbe aus LangChains Frühphase. LLMChain und SequentialChain waren 2023 noch Standard, heute offiziell als veraltet markiert. Online-Tutorials nutzen die alte Schreibweise oft noch.

Dieser Artikel ist Teil 13 der Serie KI-Entwicklung in der Praxis. Anhand von Code-Vergleichen zeigen wir, warum LCEL (LangChain Expression Language) gleiche Funktion mit bis zu 70 % weniger Code liefert – und Streaming sowie Async ohne Boilerplate.

Wenn Sie RAG-Systeme oder Agent-Anwendungen bauen, passt der Text zu RAG-System-Optimierung in der Praxis und LangGraph-Zustandsverwaltung – zentrale Bausteine im LangChain-Ökosystem.

Kapitel 1: Was ist LCEL – und warum lohnt es sich?

Wer LangChain mit Tutorials von 2023 gelernt hat, kennt Code in dieser Form:

# Traditional LLMChain approach (deprecated)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

# 1. Initialize model
llm = OpenAI(temperature=0.7)

# 2. Define prompt template
template = """You are a senior {role}.
User question: {question}
Please provide a professional answer:"""
prompt = PromptTemplate(
    template=template,
    input_variables=["role", "question"]
)

# 3. Create chain
chain = LLMChain(llm=llm, prompt=prompt)

# 4. Invoke chain (note the parameter passing method)
result = chain.run(role="frontend engineer", question="React vs Vue?")
print(result)

Wirkt noch überschaubar? Sobald Streaming, Batch-Verarbeitung oder mehrere verkettete Chains dazukommen, wächst der Code exponentiell. Klassische Chains haben drei zentrale Schwächen:

Erstens: schwaches Streaming. LLMChain unterstützt Streaming standardmäßig nicht. Eigene Callbacks für Token-Events sind nötig – Code wird aufgebläht, Async fehleranfällig.

Zweitens: umständliche Komposition. Zwei Chains verketten? SequentialChain. Parallel? wieder eine andere API. Jedes Muster bedeutet neue Schnittstellen – hohe kognitive Last.

Drittens: explizites, mühsames Ein-/Ausgabe-Mapping. Jede Chain braucht input_variables und output_variables; Feldnamen zwischen Schritten müssen Sie manuell angleichen.

LCEL adressiert genau das. Dieselbe Funktion in LCEL:

# LCEL approach (recommended for LangChain v0.3+)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# 1. Define model
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 2. Define prompt
prompt = ChatPromptTemplate.from_template(
    "You are a senior {role}.\nUser question: {question}\nPlease provide a professional answer:"
)

# 3. Connect components with pipe operator
chain = prompt | model

# 4. Invoke chain (automatically handles input-output mapping)
result = chain.invoke({"role": "frontend engineer", "question": "React vs Vue?"})
print(result.content)

Von 15 auf 9 Zeilen – entscheidender sind die Nebenwirkungen:

• Streaming: invoke durch stream ersetzen – sonst nichts ändern
• Async: ainvoke oder astream – eine Zeile
• Batch: batch mit einer Liste – parallele Ausführung

Der Pipe-Operator | folgt Linux-Pipes: cat log.txt | grep error | wc -l – Ausgabe wird zur Eingabe des nächsten Schritts. In LangChain: prompt | model | output_parser, Daten fließen von links nach rechts, der Code liest sich wie ein Satz.

Beim ersten Anblick irritierte mich | – in Python doch bitweises OR? Tatsächlich Syntax-Zucker ab Python 3.10 über __or__ – elegant gelöst.

Kapitel 2: Wie der Pipe-Operator funktioniert

Der Pipe-Operator wirkt simpel, dahinter steckt ein durchdachtes Design. Ein Experiment:

from langchain_core.runnables import RunnableLambda

# Create two simple Runnables
def add_one(x: int) -> int:
    return x + 1

def multiply_two(x: int) -> int:
    return x * 2

# Wrap regular functions with RunnableLambda
add_one_runnable = RunnableLambda(add_one)
multiply_two_runnable = RunnableLambda(multiply_two)

# Connect with pipe operator
chain = add_one_runnable | multiply_two_runnable

# Execute
result = chain.invoke(3)  # 3 -> 4 -> 8
print(result)  # Output: 8

Bei chain = add_one_runnable | multiply_two_runnable ruft Python add_one_runnable.__or__(multiply_two_runnable) auf.

Vultr - Hochleistungs-NVMe-Cloud-Server mit 32 Standorten weltweit, Docker per Klick

Preise ansehen →Anzeige

Die Runnable-Klasse implementiert __or__ und liefert ein RunnableSequence-Objekt mit allen verketteten Runnables. Bei invoke laufen die Komponenten der Reihe nach, Ausgabe wird jeweils weitergegeben.

Runnable ist die Kernabstraktion von LCEL – vier Methoden, einheitlich für alle Komponenten:

Async-Pendants: astream, abatch, abatch_as_completed.

Damit sind PromptTemplate, ChatModel, OutputParser und eigene RunnableLambda gleich aufrufbar.

# Einheitliche Aufrufe
chain.invoke({"input": "hello"})        # Einzelaufruf
chain.stream({"input": "hello"})        # Streaming
chain.batch([{"input": "a"}, {"input": "b"}])  # Batch

# Async
await chain.ainvoke({"input": "hello"})
async for chunk in chain.astream({"input": "hello"}):
    print(chunk, end="", flush=True)

Wie fließen Daten in der Pipe? Ein etwas komplexeres Beispiel:

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

# Three components
prompt = ChatPromptTemplate.from_template("Translate to English: {text}")
model = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()

# Compose
chain = prompt | model | parser

# Invoke
result = chain.invoke({"text": "Hello World in Chinese"})
print(result)  # Output: Hello World

Der Datenfluss:

{"text": "你好世界"}
       ↓
   [prompt] → ChatPromptValue(messages=[HumanMessage("Translate to English: 你好世界")])
       ↓
   [model]  → AIMessage(content="Hello World")
       ↓
   [parser] → "Hello World" (str)

Jede Komponente hat Ein-/Ausgabetypen: Prompt nimmt dict, liefert ChatPromptValue; Model nimmt PromptValue, liefert AIMessage; Parser nimmt Message, liefert str.

Falsche Reihenfolge (z. B. model | prompt) führt zur Laufzeit zu Typfehlern – die IDE warnt oft schon vorher.

Kapitel 3: Streaming in der Praxis

Streaming ist die LCEL-Funktion, die mich am meisten überzeugt hat.

Stellen Sie sich einen Support-Bot vor. Die Nutzerfrage: „Analysieren Sie Vor- und Nachteile dieses Produkts und vergleichen Sie mit Wettbewerbern.“ GPT-4o-mini braucht 5–8 Sekunden für die komplette Antwort.

Ohne Streaming starrt der Nutzer 8 Sekunden auf einen leeren Bildschirm – Absturz? Netzwerk? Seite neu laden? Die Unruhe steigt.

Mit Streaming erscheint sofort das erste Zeichen, dann Wort für Wort – wie Live-Tippen. Die subjektive Wartezeit sinkt stark.

Klassisches LangChain braucht viel Code für Streaming:

# Traditional streaming implementation (LLMChain era)
from langchain.chains import LLMChain
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

llm = OpenAI(
    temperature=0.7,
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()]
)
chain = LLMChain(llm=llm, prompt=prompt)
chain.run(role="customer service", question="...")

Drei Probleme:

1. Callbacks sind mühsam – eigene Logik (z. B. Tokens ans Frontend) erfordert BaseCallbackHandler.
2. Kein Umschalten zur Laufzeit – streaming=True/False nur bei der Initialisierung.
3. Async-Streaming noch komplexer – AsyncCallbackHandler, doppelter Code.

Bei LCEL ist Streaming eingebaut:

# LCEL streaming implementation
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "You are a professional {role}. Please answer the user's question: {question}"
)
chain = prompt | model

# Non-streaming call
result = chain.invoke({"role": "customer service", "question": "Help me analyze the pros and cons of this product"})
print(result.content)

# Streaming call (just change method name)
for chunk in chain.stream({"role": "customer service", "question": "Help me analyze the pros and cons of this product"}):
    print(chunk.content, end="", flush=True)

So einfach: invoke → stream, sonst nichts.

Ein vollständiges Echtzeit-Chat-Beispiel:

import asyncio
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

# 1. Define model and prompt
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a friendly AI assistant skilled at answering technical questions."),
    ("human", "{input}")
])
parser = StrOutputParser()

# 2. Build base chain
chain = prompt | model | parser

# 3. Add conversation history (independent memory for each user)
memory = ChatMessageHistory()

chain_with_history = RunnableWithMessageHistory(
    chain,
    get_session_history=lambda session_id: memory,
    input_messages_key="input",
    history_messages_key="chat_history"
)

# 4. Streaming conversation function
async def chat_stream(user_input: str):
    """Stream conversation response"""
    print("AI: ", end="", flush=True)

    async for chunk in chain_with_history.astream(
        {"input": user_input},
        config={"configurable": {"session_id": "demo"}}
    ):
        print(chunk, end="", flush=True)

    print("\n")  # Newline

# 5. Run conversation
async def main():
    print("=== AI Assistant (Streaming Response Demo) ===")

    await chat_stream("What is LangChain?")
    await chat_stream("What can it be used for?")
    await chat_stream("What's the relationship with LCEL?")

if __name__ == "__main__":
    asyncio.run(main())

Ausgabe:

=== AI Assistant (Streaming Response Demo) ===
AI: LangChain is an open-source framework for building applications based on large language models...
AI: It can be used to build chatbots, RAG systems, Agent applications, etc...
AI: LCEL is LangChain Expression Language, a core component of LangChain...

Jedes Zeichen erscheint live – ohne Warten auf die Gesamtantwort.

Vergleich aus Messungen:

Ohne Streaming entspricht die First-Token-Latenz der Gesamtgenerierungszeit – 8 Sekunden ohne Feedback. Mit Streaming nur die Zeit bis zum ersten Token – meist unter 1 Sekunde.

🔥 Verschwende keine Monate für ungenutzte Produkte! Teste deine SaaS-Nutzenversprechen mit Adsterra bei kleinem Budget →

Jetzt testen →Anzeige

Technisch ruft stream rekursiv die stream-Methode jeder Pipe-Komponente auf. Das Modell nutzt die Streaming-API von OpenAI; Prompt und Parser liefern oft einmalig das vollständige Zwischenergebnis. Die Pipe koordiniert das automatisch.

Sie müssen nicht manuell prüfen, welche Komponente streamt. Nicht-streamfähige Teile werden als Einmal-Rückgabe behandelt – das Gesamt-Streaming bleibt erhalten.

Kapitel 4: Runnable-Komponenten im Detail

Der Pipe-Operator löst Verkettung – in Projekten brauchen Sie oft Parallelität, Zwischenergebnisse und eigene Transformationen. Dafür gibt es weitere Runnable-Typen.

RunnableParallel: Parallel ausführen

In RAG-Systemen typisch: gleichzeitig Vektor-DB, Keyword-Suche und Wissensgraph. RunnableParallel führt die Zweige parallel aus:

from langchain_core.runnables import RunnableParallel

# Define three retrievers (using RunnableLambda for demo)
def vector_search(query: str) -> str:
    return f"Vector search results: 3 relevant documents for {query}"

def keyword_search(query: str) -> str:
    return f"Keyword search results: 5 matching records for {query}"

def graph_search(query: str) -> str:
    return f"Graph search results: 2 related entities for {query}"

# Create parallel retrieval chain
retrievers = RunnableParallel(
    vector=RunnableLambda(vector_search),
    keyword=RunnableLambda(keyword_search),
    graph=RunnableLambda(graph_search)
)

# Execute (all three retrievals happen simultaneously)
results = retrievers.invoke("LangChain LCEL")
print(results)
# Output: {
#   'vector': 'Vector search results: 3 relevant documents for LangChain LCEL',
#   'keyword': 'Keyword search results: 5 matching records for LangChain LCEL',
#   'graph': 'Graph search results: 2 related entities for LangChain LCEL'
# }

RunnableParallel liefert ein Dictionary: Schlüssel wie bei der Definition, Werte die Zweig-Ergebnisse – weiterverarbeitbar in der nächsten Stufe.

RunnablePassthrough: Eingabe durchreichen

Manchmal muss die Original-Eingabe in der Pipe erhalten bleiben – im RAG braucht der Retriever die Query, der Generator Kontext plus Originalfrage:

from langchain_core.runnables import RunnablePassthrough

# Simulate retriever
def retrieve(query: dict) -> str:
    return "Retrieved document content..."

# Build chain: preserve original query while retrieving
chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
)

result = chain.invoke({"query": "What is LCEL?"})
print(result)
# Output: {
#   'retrieved_docs': 'Retrieved document content...',
#   'original_query': {'query': 'What is LCEL?'}
# }

RunnablePassthrough ändert nichts – reicht die Eingabe unverändert weiter. In komplexen Pipes unverzichtbar.

RunnableLambda: Eigene Funktionen

Nicht alles ist vorgefertigt. RunnableLambda packt normale Python-Funktionen in Runnable und bindet sie in die Pipe ein:

from langchain_core.runnables import RunnableLambda

# Define a formatting function
def format_output(result: dict) -> str:
    """Format retrieval results into prompt input"""
    docs = result["retrieved_docs"]
    query = result["original_query"]["query"]
    return f"Reference material: {docs}\nUser question: {query}\nPlease answer based on the material:"

# Usage
chain = RunnableParallel(
    retrieved_docs=RunnableLambda(retrieve),
    original_query=RunnablePassthrough()
) | RunnableLambda(format_output)

formatted = chain.invoke({"query": "What is LCEL?"})
print(formatted)
# Output: Reference material: Retrieved document content...
#         User question: What is LCEL?
#         Please answer based on the material:

RunnableLambda ist das „Universalkleber“-Element – jede Python-Funktion passt in die Pipe.

Vollständige RAG-Pipe

Kombiniert entsteht eine komplette RAG-Pipe:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableParallel, RunnableLambda
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings

# 1. Initialize vector database (example)
embeddings = OpenAIEmbeddings()
# In real projects, this would load actual document vectors
vectorstore = FAISS.from_texts(
    ["LCEL is LangChain's expression language",
     "Pipe operator is used for component chaining",
     "Runnable is LCEL's core abstraction"],
    embeddings
)
retriever = vectorstore.as_retriever()

# 2. Define prompt
rag_prompt = ChatPromptTemplate.from_template(
    """Answer the user's question based on the following reference material.

Reference material:
{context}

User question: {question}

Please provide an accurate, detailed answer:"""
)

# 3. Define formatting function (convert retrieval results to string)
def format_docs(docs) -> str:
    return "\n".join(doc.page_content for doc in docs)

# 4. Build complete RAG chain
rag_chain = (
    # Parallel execution: retrieval + pass original question
    RunnableParallel(
        context=retriever | RunnableLambda(format_docs),
        question=RunnablePassthrough()
    )
    # Assemble prompt
    | rag_prompt
    # Call model
    | ChatOpenAI(model="gpt-4o-mini")
    # Parse output
    | StrOutputParser()
)

# 5. Usage
answer = rag_chain.invoke("What is LCEL?")
print(answer)
# Output: LCEL is LangChain Expression Language, LangChain's expression language...

Struktur der RAG-Chain:

{"question": "What is LCEL?"}
              ↓
    ┌─────────┴─────────┐
    ↓                   ↓
[retriever]        [Passthrough]
    ↓                   ↓
format_docs        question
    ↓                   ↓
    └─────────┬─────────┘
              ↓
        {"context": "...", "question": "..."}
              ↓
         [rag_prompt]
              ↓
           [model]
              ↓
          [parser]
              ↓
        "Answer content..."

Diese RAG-Implementierung passt zu RAG-System-Optimierung in der Praxis – Reranking, Multi-Path-Recall usw. lassen sich direkt auf diese LCEL-Struktur legen.

Kapitel 5: Migration von Legacy-Chains

Läuft Ihr Projekt noch mit LLMChain, ist die Umstellung machbar. Bei einem E-Commerce-Kundenservice-Bot habe ich letztes Jahr zwei Tage für das Modul gebraucht. Typische Muster:

LLMChain → Pipe-Syntax

Kern: Prompt + Model – nach der Migration direkt per Pipe:

# Old code (LLMChain)
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import OpenAI

llm = OpenAI(temperature=0.7)
prompt = PromptTemplate(
    template="User question: {question}\nPlease answer:",
    input_variables=["question"]
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run(question="...")

# New code (LCEL)
from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

model = ChatOpenAI(temperature=0.7)
prompt = ChatPromptTemplate.from_template("User question: {question}\nPlease answer:")
chain = prompt | model
result = chain.invoke({"question": "..."})

Wichtige Punkte:

1. Modellklasse: früher OpenAI (Completion API), empfohlen ChatOpenAI (Chat API).
2. Prompt: PromptTemplate geht noch, ChatPromptTemplate unterstützt System-Messages und Rollen.
3. Aufruf: chain.run() → chain.invoke(), Rückgabe Message statt String – .content oder StrOutputParser.

SequentialChain → RunnableParallel

Früher SequentialChain für mehrere Schritte – nach der Migration Pipe oder RunnableParallel:

# Old code (SequentialChain)
from langchain.chains import SequentialChain, LLMChain

# First step: generate title
title_chain = LLMChain(
    llm=llm, prompt=title_prompt,
    output_key="title"
)

# Second step: generate content
content_chain = LLMChain(
    llm=llm, prompt=content_prompt,
    output_key="content"
)

# Chain
full_chain = SequentialChain(
    chains=[title_chain, content_chain],
    input_variables=["topic"],
    output_variables=["title", "content"]
)
result = full_chain({"topic": "AI Development"})
print(result["title"], result["content"])

# New code (LCEL)
from langchain_core.runnables import RunnableParallel

# Define two branches
title_chain = title_prompt | model
content_chain = content_prompt | model

# Parallel execution (if you need serial, connect directly with |)
full_chain = RunnableParallel(
    title=title_chain,
    content=content_chain
)
result = full_chain.invoke({"topic": "AI Development"})
print(result["title"].content, result["content"].content)

SequentialChain ist seriell – jede Chain wartet auf die vorherige. RunnableParallel parallelisiert – schneller. Brauchen Sie echte Seriell-Abhängigkeit (Schritt 2 nutzt Ausgabe von Schritt 1), verketten Sie mit |:

# Seriell: Ausgabe von Schritt 1 an Schritt 2
chain = (
    title_prompt | model | StrOutputParser()
    | (lambda title: {"topic": topic, "title": title})  # Zwischenergebnis
    | content_prompt | model
)

TransformChain → RunnableLambda

TransformChain für eigene Logik in der Chain – Ersatz: RunnableLambda:

# Old code (TransformChain)
from langchain.chains import TransformChain

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    processed = text.upper()  # Some processing
    return {"processed_text": processed}

transform_chain = TransformChain(
    input_variables=["text"],
    output_variables=["processed_text"],
    transform=transform_func
)

# New code (RunnableLambda)
from langchain_core.runnables import RunnableLambda

def transform_func(inputs: dict) -> dict:
    text = inputs["text"]
    processed = text.upper()
    return {"processed_text": processed}

transform_chain = RunnableLambda(transform_func)

RunnableLambda ist flexibler – keine expliziten input_variables/output_variables, direkt in der Pipe.

Typische Migrations-Fallen

Falle 1: Rückgabetyp

KiloClaw - Managed OpenClaw für Unternehmen: AI-Agent-Orchestrierung und smartes Modell-Routing

Jetzt starten →Anzeige

LLMChain.run() liefert String, LCEL.invoke() eine Message.

# Old: get string directly
result = chain.run(...)  # str

# New: need to get content
result = chain.invoke(...)  # AIMessage
text = result.content  # str

Lösung: StrOutputParser am Pipe-Ende.

chain = prompt | model | StrOutputParser()
result = chain.invoke(...)  # Returns str directly

Falle 2: Memory

Früher ConversationChain mit eingebautem Memory:

# Old code
from langchain.chains import ConversationChain

chain = ConversationChain(llm=llm, memory=memory)

Neu: RunnableWithMessageHistory:

from langchain_core.runnables import RunnableWithMessageHistory

chain = prompt | model
chain_with_memory = RunnableWithMessageHistory(
    chain,
    get_session_history=get_history,
    input_messages_key="input",
    history_messages_key="chat_history"
)

Mehr Parameter – input_messages_key und history_messages_key explizit setzen. Details: Agent-Tool-Aufrufe in der Praxis mit vollständigem Dialog-Beispiel.

Falle 3: Import-Pfade in v0.3

Viele Module von langchain nach langchain_core oder langchain_community:

# Old imports
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# New imports
from langchain_core.runnables import RunnableLambda, RunnableParallel
from langchain_core.output_parsers import StrOutputParser
from langchain_community.chat_message_histories import ChatMessageHistory

Die IDE meldet fehlende Imports – den Vorschlägen folgen.

Produktions-Migration

Der E-Commerce-Bot: ursprünglich ~300 Zeilen mit LLMChain + SequentialChain + TransformChain. Kernlogik nach der Migration:

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# Initialize
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.5)

# Intent classification prompt
intent_prompt = ChatPromptTemplate.from_template(
    """Analyze user intent and return one of the following categories:
    - product_query (product inquiry)
    - order_status (order lookup)
    - complaint (complaint/feedback)
    - other (other)

    User message: {message}
    Intent category:"""
)

# Prompts for each intent
product_prompt = ChatPromptTemplate.from_template(
    "User inquiring about product: {message}\nPlease retrieve from product database and answer:"
)
order_prompt = ChatPromptTemplate.from_template(
    "User checking order: {message}\nPlease query order status and reply:"
)

# Build branching logic
def route_by_intent(result):
    intent = result.content.strip().lower()
    if "product" in intent:
        return "product"
    elif "order" in intent:
        return "order"
    else:
        return "default"

# Complete chain
intent_chain = intent_prompt | model | StrOutputParser() | RunnableLambda(route_by_intent)

# Branch routing (pseudocode, actual implementation needs RunnableBranch)
full_chain = (
    {"message": RunnablePassthrough()}
    | RunnableParallel(
        intent=intent_chain,
        original=RunnablePassthrough()
    )
    # Select different processing branches based on intent
    # ... actual code uses RunnableBranch implementation
)

# Streaming output
async for chunk in full_chain.astream({"message": "I want to check order 12345"}):
    print(chunk, end="", flush=True)

Nach der Migration ~150 Zeilen – halbiert. Streaming und Async, früher Extra-Entwicklung, jetzt oft eine Zeile.

Zusammenfassung

LCEL ist die empfohlene Architektur für LangChain v0.3+: Pipe-Operator, einheitliche Runnable-Schnittstelle, natives Streaming.

Die größte Hürde ist nicht die Syntax, sondern das Denken: klassische Chains setzen auf explizite Deklaration; LCEL auf impliziten Fluss in der Pipe – Typkonventionen stecken in den Komponenten.

Alte LLMChain-Projekte schrittweise migrieren: zuerst einfache Dialog-Chains, dann komplexe Komposition. LangSmith hilft bei Typ-Mismatches.

Als Nächstes: LangGraph-Zustandsverwaltung in der Praxis – Agent-Framework des LangChain-Teams; mit LCEL kombiniert für komplexe Agenten. Einfache Ketten: LCEL; komplexe Zustände: LangGraph – ein etabliertes Muster.

Navigation: Serie KI-Entwicklung in der Praxis

• Teil 1: Claude API – vom Auth bis zum Mehr-Runden-Dialog
• Teil 2: Prompt Engineering – Praxis
• Teil 8: RAG-System-Optimierung: Retrieval und Generierung
• Teil 11: LangGraph-Zustandsverwaltung – Agent-Architektur 2026
• Teil 13: LangChain LCEL in der Praxis: Vom klassischen Chain zum Streaming (dieser Artikel)
• Teil 13: Agent-Tool-Aufrufe: externe APIs und Dienste