AI Engineering Academy · Урок

Создание цепочек с LCEL

«Создание цепочек с LCEL» — бесплатный урок AI Engineering Academy на CoddyKit. Это урок 2 из 4. Ты можешь прочитать весь урок бесплатно ниже — а потом практиковать его прямо в браузере с встроенным редактором кода и ИИ-репетитором 24/7. Это часть пути обучения AI Engineering Academy, и твой прогресс синхронизируется между веб-версией и приложением CoddyKit. Курс AI Engineering Academy содержит 4 уроков всего.

Части этого урока еще не переведены и отображаются на английском.

What Is LCEL?

LCEL (LangChain Expression Language) is the declarative way to compose LangChain components into pipelines. Instead of instantiating verbose Chain classes, you connect Runnables with the | operator. LCEL chains support streaming, async, batching, and fallbacks automatically — you get all these capabilities for free just by using the pipe syntax.

Your First LCEL Chain

The simplest LCEL chain combines a PromptTemplate, a ChatModel, and an OutputParser. Each component is a Runnable, and the pipe operator connects them. When you call chain.invoke(), the input flows through each step in sequence, with the output of each becoming the input of the next.

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)

Inspecting Chain Internals

LCEL chains expose their structure transparently. chain.steps lists the components in order. You can call chain.input_schema and chain.output_schema to inspect what types flow in and out. The invoke() call also accepts a config parameter for passing run names, tags, and callbacks to a single invocation without changing the chain definition.

# 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']}
)

Synchronous vs Asynchronous Invoke

LCEL chains expose both synchronous and asynchronous versions of all methods. invoke() blocks the calling thread. ainvoke() is the async equivalent and should be used in FastAPI handlers and other async contexts to avoid blocking the event loop. The async version has identical semantics — the interface difference is just 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())

Streaming Tokens Through the Chain

LCEL chains propagate streaming all the way through. Call chain.stream() to receive tokens as they are generated by the model. The StrOutputParser is streaming-aware and passes individual string chunks through rather than waiting for the full response. This gives users immediate feedback instead of a long pause followed by the complete answer.

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 Processing Multiple Inputs

chain.batch() accepts a list of input dictionaries and processes them with configurable concurrency. Internally LangChain runs the calls in parallel threads, making batch far faster than calling invoke() in a loop. The max_concurrency parameter caps parallel API calls to stay within rate limits. Results come back in the same order as inputs.

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 and RunnableLambda

RunnablePassthrough forwards its input unchanged — useful for passing original context alongside transformed data. RunnableLambda wraps any Python function as a Runnable, letting you insert arbitrary logic into an LCEL chain. Together they allow data manipulation between chain steps without writing a full custom Runnable class.

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)

Passing Extra Context with RunnablePassthrough

A common pattern in RAG is to pass both the retrieved documents and the user question to the final prompt. Use RunnablePassthrough.assign() to add computed fields to the input dictionary without losing the original keys. This enriches the context object as it flows through the chain, making all data available to the prompt template.

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?'})

Adding Fallbacks to Chains

LCEL chains support fallbacks — alternative chains to try if the primary chain raises an exception. Call chain.with_fallbacks([backup_chain]) to register one or more fallback chains. LangChain tries the primary first, catches any exception, then tries each fallback in order. This is the standard way to handle provider outages or model errors gracefully.

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

Debugging with intermediate_steps

Debugging LCEL chains is easier with invoke()'s config and the verbose=True option. You can also use .with_config({'verbose': True}) on any chain to log every step's input and output. For structured debugging, wrap the chain with set_verbose(True) globally or use LangSmith tracing to capture every intermediate value in a visual trace explorer.

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'})

Configurable Chains at Runtime

LCEL chains support runtime configuration via ConfigurableField. You can expose model parameters — temperature, model name, max_tokens — as configurable options without hardcoding them. At invocation time, pass a Configurable config to override the defaults. This lets one chain serve multiple users with different settings or A/B test model configurations.

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}}
)

Quick Check

Test your understanding of building chains with LCEL.

Lesson Recap

In this lesson you learned: LCEL pipe composition connects Runnables with | into clean, readable pipelines, RunnablePassthrough and RunnableLambda let you inject data manipulation steps without leaving the LCEL paradigm, and fallbacks make chains resilient by automatically trying backup chains on failure. Next up we explore branching and parallel chains for more complex routing logic.

Часто задаваемые вопросы

Урок «Создание цепочек с LCEL» бесплатный?

Да — полный текст урока «Создание цепочек с LCEL» бесплатно доступен здесь в веб-версии. Чтобы практиковать его интерактивно (встроенный редактор кода и ИИ-репетитор 24/7) и разблокировать остальной курс AI Engineering Academy, подпишись на CoddyKit PRO. Курс AI Engineering Academy содержит 4 уроков всего.

Чему я научусь в уроке «Создание цепочек с LCEL»?

Используйте оператор вертикальной черты, чтобы объединить шаблон запроса, LLM и анализатор вывода в последовательность Runnable, вызывайте её синхронно и асинхронно и изучайте промежуточный вывод. Ты практикуешь AI Engineering Academy с помощью реального кода, который запускаешь прямо в браузере, и ИИ-репетитор 24/7 отвечает на твои вопросы во время урока.

Нужен ли мне опыт, чтобы начать AI Engineering Academy?

Предыдущий опыт не требуется. AI Engineering Academy на CoddyKit структурирован для всех уровней — от новичков до продвинутых, поэтому ты можешь начать отсюда или с самого начала и учиться в своем темпе. Это урок 2 из 4.

Сколько времени занимает урок «Создание цепочек с LCEL»?

Большинство уроков CoddyKit занимают около 5–10 минут. Каждый из них компактный и интерактивный, поэтому ты постоянно делаешь прогресс и продолжаешь с того же места в веб-версии и приложении.

Можно ли писать и запускать код в этом уроке AI Engineering Academy?

Да. Каждый урок AI Engineering Academy включает встроенный редактор кода, поэтому ты пишешь и запускаешь реальный код прямо в браузере и получаешь моментальную обратную связь от AI — локальная установка не требуется.

Все уроки этого курса

Архитектура LangChain и основные абстракции
Создание цепочек с LCEL
Разветвлённые и параллельные цепочки
Потоковый вывод в LangChain

← Назад к AI Engineering Academy