6. Декларативные пайплайны с LCEL

В предыдущих главах мы писали императивный код для оркестрации взаимодействий с LLM: создавали промпт, вызывали модель, парсили ответ. Это работает, но по мере усложнения AI-приложений такой подход становится многословным и труднее поддерживается. Вы получаете глубоко вложенные вызовы функций, ручную обработку ошибок на каждом шаге и сложность понимания общего потока данных.

LangChain Expression Language (LCEL) решает эту проблему, позволяя вам декларировать, что вы хотите сделать, а не как это сделать. Вместо написания процедурного кода, который последовательно вызывает функции, вы компонуете компоненты с помощью простого оператора pipe (|), который читается как Unix-пайплайн. Результат — более чистый, легче поддерживаемый код, который ясно выражает поток данных через вашу AI-систему.

Эта глава знакомит с LCEL для построения линейных воркфлоу — последовательностей операций, где данные текут от начала до конца без ветвлений или циклов. Мы рассмотрим, когда использовать LCEL, как компоновать пайплайны и как выполнять их как синхронно, так и с потоковым выводом.

6.1) Зачем нужен LCEL?

Проблема, которую решает LCEL

Начнём с конкретного примера. Предположим, вы создаёте ассистента службы поддержки клиентов, который должен:

Принять вопрос пользователя
Отформатировать его в промпт с системными инструкциями и сообщением пользователя
Отправить его в LLM
Распарсить ответ, чтобы извлечь только текстовое содержимое

Вот как вы могли бы написать это императивно (без LCEL):

python

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", "Вы полезный ассистент службы поддержки клиентов."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Шаг 1: Форматируем промпт с вопросом
    messages = prompt.invoke({"question": question})
    
    # Шаг 2: Вызываем LLM
    response = llm.invoke(messages)
    
    # Шаг 3: Парсим вывод для извлечения текстового содержимого
    result = parser.invoke(response)
    
    return result
 
# Используем
result = answer_question("Как мне сбросить пароль?")
print(result)

Вывод:

Чтобы сбросить пароль, выполните следующие шаги:
1. Перейдите на страницу входа
2. Нажмите "Забыли пароль"
3. Введите ваш адрес электронной почты
4. Проверьте почту на наличие ссылки для сброса
5. Перейдите по ссылке и создайте новый пароль

Это работает, но обратите внимание на проблемы:

Многословность: Каждый шаг требует явного присваивания переменных и вызовов функций. Фактическая логика (форматирование → вызов → парсинг) погребена в шаблонном коде.

Жёсткая структура: Если вы хотите добавить шаг (например, валидацию вопроса или логирование ответа), вам нужно вставить код в середину функции, увеличивая сложность.

Нет встроенного стриминга: Чтобы стримить токены по мере их поступления, вам нужно переписать всю функцию для использования llm.stream() и вручную обрабатывать асинхронную итерацию.

Неясный поток данных: Читая код, не сразу очевидно, что это простой пайплайн. Вам нужно отслеживать присваивания переменных, чтобы понять поток.

Теперь посмотрим на версию с LCEL:

python

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", "Вы полезный ассистент службы поддержки клиентов."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Компонуем их в цепочку с помощью оператора pipe
chain = prompt | llm | parser
 
# Используем
result = chain.invoke({"question": "Как мне сбросить пароль?"})
print(result)

Вывод:

Чтобы сбросить пароль, выполните следующие шаги:
1. Перейдите на страницу входа
2. Нажмите "Забыли пароль"
3. Введите ваш адрес электронной почты
4. Проверьте почту на наличие ссылки для сброса
5. Перейдите по ссылке и создайте новый пароль

Вывод идентичен, но код драматически отличается:

Декларативность: chain = prompt | llm | parser выражает весь поток в одной строке. Читайте слева направо: промпт → LLM → парсер.

Компонуемость: Каждый компонент (prompt, llm, parser) независим и переиспользуем. Вы можете заменять компоненты без переписывания пайплайна.

Встроенный стриминг: Переключайтесь между chain.invoke() и chain.stream() без изменения определения пайплайна.

Ясное намерение: Оператор | делает поток данных очевидным с первого взгляда.

Где подходит LCEL: Линейные воркфлоу

LCEL разработан для линейных воркфлоу — последовательностей, где данные текут в одном направлении от начала до конца без ветвлений или циклов.

Этот линейный паттерн охватывает многие AI-приложения. Рассмотрим систему вопросов-ответов по документам: вы получаете вопрос → извлекаете релевантные документы → форматируете промпт → отправляете в LLM → парсите ответ. Каждый шаг — это чёткая последовательность, где вывод одного шага становится входом следующего.

Но что, если вашему воркфлоу нужно:

Чтобы LLM решал, какой инструмент вызвать на основе вопроса
Вызвать инструмент, наблюдать результат, затем решить, что делать дальше
Повторять неудачные операции с разными подходами

Эти сценарии требуют циклов и условного ветвления — вещей, которые LCEL не может обработать. Вот почему существует LangGraph (мы познакомимся с ним в главе 15).

Вот визуальное сравнение двух подходов:

LCEL идеален для простых пайплайнов, где каждый шаг обрабатывает вывод предыдущего шага. Данные текут только в одном направлении.

LangGraph предназначен для случаев, когда нужно принятие решений и циклы. Агент может действовать, наблюдать результаты и думать снова.

В этой главе мы фокусируемся на LCEL. Почему сначала освоить линейные пайплайны?

Основа: Оператор pipe (|) — это основной синтаксис LangChain. Понимание этого делает всё остальное проще.
Предпосылка для LangGraph: Агенты LangGraph активно используют LCEL-цепочки внутри своих узлов
Реальный паттерн: Сложные агенты строятся путём комбинирования LCEL-цепочек

Остальная часть этой главы покажет вам, как строить эти линейные пайплайны с помощью оператора pipe (|).

6.2) Композиция пайплайнов с оператором |

Как работают пайпы LCEL

Оператор pipe (|) в LCEL работает благодаря интерфейсу Runnable.

Что такое Runnable?

Runnable — это стандартный интерфейс LangChain. Когда компонент реализует интерфейс Runnable, его можно соединять с другими компонентами с помощью оператора |.

Каждый Runnable предоставляет эти методы:

.invoke(input) — Выполнить один раз и получить полный результат
.stream(input) — Выполнить и получать каждое слово по мере генерации LLM
.batch(inputs) — Выполнить несколько раз с разными входами и получить все результаты

В этой главе мы сосредоточимся на .invoke() и .stream() (мы рассмотрим .batch() позже, когда это понадобится).

Почему работает |?

Потому что класс Runnable реализует оператор | с помощью перегрузки операторов Python. Когда вы пишете prompt | llm, создаётся новый Runnable, который последовательно выполняет два компонента.

Большинство компонентов LangChain — это Runnable

Вот почему вы можете соединять так много разных компонентов:

ChatPromptTemplate — это Runnable
ChatOpenAI — это Runnable
StrOutputParser — это Runnable
Даже пользовательские цепочки, которые вы создаёте с помощью |, сами являются Runnable!

Это означает, что вы можете строить сложные пайплайны, комбинируя более простые.

Соединение компонентов: Типы входа/выхода

При соединении компонентов с помощью | вам нужно убедиться, что тип выхода одного компонента соответствует типу входа следующего.

Ключевые сигнатуры компонентов:

Компонент	Тип входа	Тип выхода
`ChatPromptTemplate`	`dict`	`list[BaseMessage]`
`ChatOpenAI` (LLM)	`list[BaseMessage]`	`AIMessage`
`StrOutputParser`	`AIMessage`	`str`

Пример потока:

python

chain = prompt | llm | parser

Вот как тип данных трансформируется при прохождении через каждый компонент:

prompt: Принимает dict на входе и преобразует в list[BaseMessage]
llm: Принимает list[BaseMessage] на входе и преобразует в AIMessage
parser: Принимает AIMessage на входе и преобразует в str

Давайте посмотрим это в действии:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Вы полезный ассистент."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "Сколько будет 2+2?"})
print(result)  # "2+2 равно 4."

Что происходит на каждом шаге:

Шаг	Вход	Компонент	Выход
1	`- dict -` `{"question": "Сколько будет 2+2?"}`	→ prompt →	`- list[BaseMessage] -` `[SystemMessage(...), HumanMessage(...)]`
2	`- list[BaseMessage] -` `[SystemMessage(...), HumanMessage(...)]`	→ llm →	`- AIMessage -` `AIMessage(content="2+2 равно 4.")`
3	`- AIMessage -` `AIMessage(content="2+2 равно 4.")`	→ parser →	`- str -` `"2+2 равно 4."`

Что происходит, если типы не совпадают?

Если вы попытаетесь соединить несовместимые компоненты, вы получите ошибку:

python

# ОШИБКА: Это не сработает
chain = llm | prompt  # LLM выводит AIMessage, но prompt требует dict на входе

Сообщение об ошибке скажет вам, какой тип входа ожидает следующий компонент, и какой тип входа он фактически получил.

6.3) Выполнение цепочки: .invoke() и .stream()

Запуск пайплайна

После того как вы скомпоновали цепочку, вы выполняете её с помощью метода .invoke(). Это синхронный способ запуска пайплайна — он ждёт полного ответа перед возвратом результата.

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Вы полезный ассистент."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Синхронный вызов
result = chain.invoke({"question": "Сколько будет 2+2?"})
print(result)

Вывод:

2+2 равно 4.

Метод .invoke() прост: передайте входной параметр, который ожидает первый компонент, и получите выход, произведённый последним компонентом.

Стриминг токенов из той же цепочки

Хотя .invoke() прост, у него есть ограничение для пользовательских приложений: пользователи ничего не видят, пока не будет готов полный ответ. Для длинных ответов (10-20 секунд) это создаёт плохой пользовательский опыт.

Стриминг показывает токены немедленно по мере их генерации, а не ждёт полного ответа. Это эффект печати, который вы видите в ChatGPT.

Давайте посмотрим стриминг в действии:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Вы полезный ассистент."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Стримим чанки по мере их поступления
for chunk in chain.stream({"question": "Объясните, что такое Python, одним предложением"}):
    print(chunk, end="", flush=True)

Вывод (отображается в реальном времени, токен за токеном):

Python — это универсальный высокоуровневый язык программирования, известный своей простотой и читаемостью, широко используемый в веб-разработке, науке о данных и автоматизации.

Обратите внимание, что определение цепочки идентично примеру с .invoke() выше. Нам не нужно было перестраивать её — мы просто вызвали .stream() вместо .invoke().

Метод .stream() возвращает результаты в виде чанков, пока LLM генерирует токены. print(chunk, end="", flush=True) отображает каждый чанк немедленно на экране, создавая эффект живой печати.

Когда использовать .invoke() vs .stream()

Используйте .invoke(), когда:

Вам нужен только финальный результат (анализ, перевод, классификация)
Ответ короткий и время ожидания не является проблемой
Вам нужен полный вывод перед переходом к следующему шагу

Используйте .stream(), когда:

Пользователям нужно видеть прогресс (чат, генерация контента)
Ответ длинный и время ожидания будет заметным
Вы создаёте UI, где важна обратная связь в реальном времени

Оба метода работают на одной и той же цепочке. Определите цепочку один раз, затем выберите режим выполнения в зависимости от ваших потребностей.

В этой главе вы изучили LCEL — декларативный синтаксис пайплайнов LangChain:

Строить пайплайны с помощью оператора pipe: prompt | llm | parser
Гибко выполнять: Используйте .invoke() для полных результатов или .stream() для вывода в реальном времени
Типобезопасность: Сопоставляйте типы выхода с типами входа при соединении компонентов

Одна и та же цепочка работает для обоих режимов выполнения — определите один раз, используйте везде.

Далее: Глава 7 охватывает структурированный вывод с Pydantic, позволяя вам извлекать валидированные JSON-данные из ответов LLM.