Mit LCEL Chains erstellen
Sie verwenden den Pipe-Operator, um ein PromptTemplate, ein LLM und einen OutputParser zu einer Runnable-Sequenz zu verketten, sie synchron und asynchron aufzurufen und Zwischenausgaben zu untersuchen.
Mit LCEL Chains erstellen ist eine kostenlose AI Engineering Academy-Lektion auf CoddyKit. Dies ist Lektion 2 von 4. Du kannst die komplette Lektion unten kostenlos lesen – dann übst du sie direkt im Browser mit einem integrierten Code-Editor und einem KI-Tutor rund um die Uhr. Sie ist Teil des AI Engineering Academy-Lernpfads, und dein Fortschritt wird über Web und CoddyKit-App synchronisiert. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Was ist LCEL?
LCEL (LangChain Expression Language) ist die deklarative Möglichkeit, LangChain-Komponenten zu Pipelines zu kombinieren. Statt ausführliche Chain-Klassen zu instanziieren, verbinden Sie Runnables mit dem Operator |. LCEL-Ketten unterstützen Streaming, asynchrone Ausführung, Batch-Verarbeitung und Fallbacks automatisch – all diese Funktionen erhalten Sie allein durch die Verwendung der Pipe-Syntax.
Ihre erste LCEL-Kette
Die einfachste LCEL-Kette kombiniert ein PromptTemplate, ein ChatModel und einen OutputParser. Jede Komponente ist ein Runnable, und der Pipe-Operator verbindet sie. Wenn Sie chain.invoke() aufrufen, fließt die Eingabe nacheinander durch jeden Schritt, wobei die Ausgabe jedes Schritts zur Eingabe des nächsten wird.
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)Die Interna einer Kette untersuchen
LCEL-Ketten legen ihre Struktur transparent offen. chain.steps listet die Komponenten in ihrer Reihenfolge auf. Mit chain.input_schema und chain.output_schema können Sie untersuchen, welche Typen ein- und ausgegeben werden. Der Aufruf invoke() akzeptiert außerdem einen Parameter config, mit dem Sie Laufnamen, Tags und Callbacks an eine einzelne Ausführung übergeben können, ohne die Definition der Kette zu ändern.
# 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']}
)Synchroner und asynchroner Invoke
LCEL-Ketten stellen für alle Methoden sowohl synchrone als auch asynchrone Varianten bereit. invoke() blockiert den aufrufenden Thread. ainvoke() ist das asynchrone Gegenstück und sollte in FastAPI-Handlern sowie anderen asynchronen Kontexten verwendet werden, um eine Blockierung der Event Loop zu vermeiden. Die asynchrone Variante hat eine identische Semantik – der Unterschied an der Schnittstelle besteht lediglich 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())Tokens durch die Kette streamen
LCEL-Ketten leiten Streaming durch die gesamte Kette weiter. Rufen Sie chain.stream() auf, um Tokens zu empfangen, sobald sie vom Modell generiert werden. Der StrOutputParser unterstützt Streaming und gibt einzelne String-Chunks weiter, statt auf die vollständige Antwort zu warten. So erhalten Benutzer sofortiges Feedback anstelle einer langen Pause, auf die erst die vollständige Antwort folgt.
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)Batch-Verarbeitung mehrerer Eingaben
chain.batch() akzeptiert eine Liste von Eingabedictionaries und verarbeitet sie mit konfigurierbarer Nebenläufigkeit. Intern führt LangChain die Aufrufe in parallelen Threads aus, wodurch die Batch-Verarbeitung deutlich schneller ist, als invoke() in einer Schleife aufzurufen. Der Parameter max_concurrency begrenzt die parallelen API-Aufrufe, damit Ratenlimits eingehalten werden. Die Ergebnisse werden in derselben Reihenfolge wie die Eingaben zurückgegeben.
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 und RunnableLambda
RunnablePassthrough leitet seine Eingabe unverändert weiter – nützlich, um den ursprünglichen Kontext zusammen mit transformierten Daten weiterzugeben. RunnableLambda kapselt jede Python-Funktion als Runnable, sodass Sie beliebige Logik in eine LCEL-Kette einfügen können. Zusammen ermöglichen beide die Datenverarbeitung zwischen Kettenschritten, ohne eine vollständige eigene Runnable-Klasse schreiben zu müssen.
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)Zusätzlichen Kontext mit RunnablePassthrough übergeben
Ein häufiges Muster bei RAG besteht darin, sowohl die abgerufenen Dokumente als auch die Benutzerfrage an den abschließenden Prompt zu übergeben. Verwenden Sie RunnablePassthrough.assign(), um dem Eingabedictionary berechnete Felder hinzuzufügen, ohne die ursprünglichen Schlüssel zu verlieren. Dadurch wird das Kontextobjekt auf seinem Weg durch die Kette angereichert und alle Daten stehen dem Prompt-Template zur Verfügung.
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?'})Fallbacks zu Ketten hinzufügen
LCEL-Ketten unterstützen Fallbacks – alternative Ketten, die ausprobiert werden, wenn die primäre Kette eine Ausnahme auslöst. Rufen Sie chain.with_fallbacks([backup_chain]) auf, um eine oder mehrere Fallback-Ketten zu registrieren. LangChain versucht zuerst die primäre Kette, fängt jede Ausnahme ab und probiert anschließend die Fallbacks in der angegebenen Reihenfolge. Dies ist die Standardmethode, um Ausfälle von Anbietern oder Modellfehler kontrolliert zu behandeln.
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 AnthropicDebugging mit intermediate_steps
Das Debugging von LCEL-Ketten wird durch die Konfiguration von invoke() und die Option verbose=True erleichtert. Sie können außerdem für jede Kette .with_config({'verbose': True}) verwenden, um die Ein- und Ausgaben jedes Schritts zu protokollieren. Für strukturiertes Debugging können Sie die Kette global mit set_verbose(True) umschließen oder LangSmith Tracing verwenden, um jeden Zwischenwert in einem visuellen Trace-Explorer zu erfassen.
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'})Zur Laufzeit konfigurierbare Ketten
LCEL-Ketten unterstützen Laufzeitkonfiguration über ConfigurableField. Sie können Modellparameter – Temperatur, Modellname und max_tokens – als konfigurierbare Optionen bereitstellen, ohne sie fest zu codieren. Übergeben Sie zum Ausführungszeitpunkt eine Configurable-Konfiguration, um die Standardwerte zu überschreiben. So kann eine Kette mehrere Benutzer mit unterschiedlichen Einstellungen bedienen oder A/B-Tests von Modellkonfigurationen ermöglichen.
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}}
)Kurzer Wissenstest
Testen Sie Ihr Verständnis des Erstellens von Ketten mit LCEL.
Lektionsrückblick
In dieser Lektion haben Sie gelernt: Die LCEL-Pipe-Komposition verbindet Runnables mit | zu übersichtlichen, gut lesbaren Pipelines, RunnablePassthrough und RunnableLambda ermöglichen das Einfügen von Datenverarbeitungsschritten, ohne das LCEL-Paradigma zu verlassen, und Fallbacks machen Ketten robust, indem bei Fehlern automatisch Ersatzketten ausprobiert werden. Als Nächstes untersuchen wir Verzweigungen und parallele Ketten für komplexere Routing-Logik.
Lerne Python mit einem KI-Tutor — kostenlos
Schreibe und führe echten Code in deinem Browser aus, bekomme sofortige Hilfe von einem 24/7 KI-Tutor und setze dein Lernen im Web oder in der App fort.
- Kurse
- 30
- Lektionen
- 120
Häufig gestellte Fragen
Ist die Lektion „Mit LCEL Chains erstellen“ kostenlos?
Ja — der vollständige Text von „Mit LCEL Chains erstellen“ ist hier im Web kostenlos zu lesen. Um sie interaktiv zu üben (integrierter Code-Editor und 24/7 KI-Tutor) und den Rest des AI Engineering Academy-Kurses freizuschalten, upgrade auf CoddyKit PRO. Der AI Engineering Academy-Kurs umfasst insgesamt 4 Lektionen.
Was lerne ich in „Mit LCEL Chains erstellen“?
Sie verwenden den Pipe-Operator, um ein PromptTemplate, ein LLM und einen OutputParser zu einer Runnable-Sequenz zu verketten, sie synchron und asynchron aufzurufen und Zwischenausgaben zu untersuche… Du übst AI Engineering Academy mit praktischem Code, den du direkt im Browser ausführst, und ein 24/7 KI-Tutor beantwortet deine Fragen während du die Lektion bearbeitest.
Brauche ich Erfahrung, um AI Engineering Academy zu starten?
Keine Vorkenntnisse erforderlich. AI Engineering Academy auf CoddyKit ist für Anfänger bis fortgeschrittene Lernende strukturiert, sodass du hier starten oder von Anfang an beginnen und in deinem eigenen Tempo voranschreiten kannst. Dies ist Lektion 2 von 4.
Wie lange dauert die Lektion „Mit LCEL Chains erstellen“?
Die meisten CoddyKit-Lektionen dauern etwa 5–10 Minuten. Jede ist kompakt und interaktiv, sodass du stetig Fortschritte machst und genau dort weitermachst, wo du aufgehört hast – im Web und in der App.
Kann ich in dieser AI Engineering Academy-Lektion Code schreiben und ausführen?
Ja. Jede AI Engineering Academy-Lektion enthält einen integrierten Code-Editor, sodass du echten Code direkt in deinem Browser schreibst und ausführst und sofort KI-Feedback erhältst — ohne lokale Einrichtung erforderlich.
Alle Lektionen in diesem Kurs
- LangChain-Architektur und zentrale Abstraktionen
- Mit LCEL Chains erstellen
- Verzweigte und parallele Chains
- Ausgaben in LangChain streamen