Создание конвейеров LangChain с помощью LCEL: линейные, параллельные и разветвленные.

Полезно разделять пайплайны LangChain на четыре типа. Строго говоря, это не отдельные классы в современном LangChain; это шаблоны, которые создаются на основе одних и тех же примитивов Runnable. Названия всё равно остаются удобным словарем:

• Простая цепочка: одна прямая линия от запроса к парсеру.
• Последовательная цепочка: несколько таких линий, соединённых друг с другом.
• Параллельная цепочка: несколько независимых подцепочек, работающих с одним и тем же входным данным одновременно.
• Условная цепочка: точка принятия решения, которая определяет, какая подцепочка будет обрабатывать входные данные.

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

Простые цепочки: один прямой проход

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

Input
  ↓
Prompt Template
  ↓
LLM
  ↓
Output Parser
  ↓
Output

Приведённый пример служит инструментом для изучения: при указании темы модель должна предоставить несколько простых и понятных фактов, не углубляясь в сложные технические детали. Для этого используется максимально короткая цепочка компонентов:

prompt → llm → parser

Если говорить простыми словами, значение topic попадает в шаблон запроса, форматированный запрос отправляется модели, а ответ модели проходит через парсер вывода перед тем, как быть возвращённым. С технической точки зрения парсер является необязательным, но его наличие обеспечивает предсказуемый тип возвращаемых данных.

В конкретном примере в качестве темы указывается крикет, и запрашиваются пять интересных фактов. Данные передаются следующим образом:

{'topic': 'cricket'}
        ↓
PromptTemplate
        ↓
LLM
        ↓
StrOutputParser
        ↓
Five interesting facts about cricket

Версия Gemini

Вот полная программа с использованием Gemini от Google через интеграцию langchain_google_genai:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()

llm = GoogleGenerativeAI(model="gemini-2.5-flash")

prompt = PromptTemplate(
    template='Generate 5 interesting facts about {topic}',
    input_variables=['topic']
)

parser = StrOutputParser()

chain = prompt | llm | parser

result = chain.invoke({'topic':'cricket'})
# print(result)

chain.get_graph().print_ascii()

В ней создается модель GoogleGenerativeAI на основе gemini-2.5-flash, используется одномерный промпт и класс StrOutputParser; затем все эти элементы объединяются по схеме prompt | llm | parser. Последняя строка выводит ASCII-диаграмму этой цепочки, которая рассматривается далее. Обратите внимание, что строка print(result) в этой версии закомментирована, поэтому скрипт отображает только диаграмму, если вы её снова не включите.

Разбор кода Gemini

Импорты включают четыре компонента, необходимые скрипту:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

GoogleGenerativeAI представляет собой обертку вокруг модели Gemini. PromptTemplate определяет шаблон запроса с полем {topic}, которое заполняется во время выполнения. StrOutputParser преобразует вывод модели в обычную строку на Python. load_dotenv читает локальный файл .env и загружает его содержимое в переменные окружения.

Вызов load_dotenv обеспечивает доступ к ключу API и другим настройкам из файла .env для процесса до создания любого клиента:

load_dotenv()

Интеграция с Gemini заимствует свой ключ из переменных окружения, поэтому этот вызов должен выполняться до инициализации модели. Обычной практикой является хранение ключей в файле .env (а не в системе контроля версий).

Затем создается сама модель, указанная как gemini-2.5-flash:

llm = GoogleGenerativeAI(model="gemini-2.5-flash")

Шаблон запроса указывает как текст, так и переменные, которые он ожидает:

prompt = PromptTemplate(
    template='Generate 5 interesting facts about {topic}',
    input_variables=['topic']
)

Во время вызова местоопределение {topic} заполняется из словаря входных данных. При передаче этих данных:

{'topic': 'cricket'}

формируется следующий конкретный запрос:

Generate 5 interesting facts about cricket

Парсер создается без аргументов:

parser = StrOutputParser()

Его задача — вернуть ответ модели в виде строки. Для моделей типа завершения текста, таких как GoogleGenerativeAI, которые уже возвращают текст, это почти полное пропускание данных. Для чат-моделей, возвращающих объект сообщения, он извлекает текстовое содержимое. В любом случае следующий потребитель получает обычную строку str.

Затем следует строка, определяющая всю цепочку обработки:

chain = prompt | llm | parser

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

PromptTemplate → Gemini 2.5 Flash → StrOutputParser

В конце цепочка запускается с использованием словаря, содержащего тему:

result = chain.invoke({'topic':'cricket'})

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

cricket
   ↓
PromptTemplate
   ↓
Gemini 2.5 Flash
   ↓
StrOutputParser
   ↓
Final Result

Что возвращается

Значение, возвращаемое функцией, — это строка с пятью фактами о крикете примерно в таком виде:

1. Cricket originated in England and has been played for several centuries.
2. A cricket team consists of 11 players.
3. Test cricket matches can last up to five days.
4. The Cricket World Cup is one of the major international cricket tournaments.
5. Cricket is especially popular in countries such as India, Australia, England, and New Zeakand.

Поскольку текст генерируется, формулировки и конкретные факты меняются при каждом запуске, а пример вывода даже содержит опечатку (“Zeakand”). Рассматривайте примеры вывода как иллюстрации формата, а не как фиксированные результаты. Если вам нужен детерминистичный формат, это задача более строгого парсера или структурированного вывода, а не ожидания того, что модель будет повторять одно и то же.

Тот же пайплайн на OpenAI

Теперь тот же инструмент для анализа данных, работающий на OpenAI. Примечательно то, насколько мало изменений:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini")
prompt = PromptTemplate(
    template='Generate 5 interesting facts about {topic}',
    input_variables=['topic']
)
parser = StrOutputParser()
chain = prompt | llm | parser
result = chain.invoke({'topic':'cricket'})
print(result)
chain.get_graph().print_ascii()

Теперь модель — ChatOpenAI с gpt-4o-mini. Запрос, парсер и процесс формирования результата остались точно такими же, как и раньше; эта версия выводит как результат, так и график.

Единственное нововведение — объект модели:

llm = ChatOpenAI(model="gpt-4o-mini")

ChatOpenAI из библиотеки langchain_openai заменяет GoogleGenerativeAI. Это модель для чатов, поэтому она возвращает объект сообщения, а не простую строку, и именно поэтому сохранение парсера в пайплайне имеет смысл.

Шаблон запроса не изменился:

prompt = PromptTemplate(
    template='Generate 5 interesting facts about {topic}',
    input_variables=['topic']
)

Парсер тоже не изменился:

parser = StrOutputParser()

Процесс формирования результата также остался прежним:

chain = prompt | llm | parser

Структура остаётся той же — трёхэтапная цепочка, с OpenAI посередине:

PromptTemplate → OpenAI → StrOutputParser

Вызов цепочки и вывод результата выглядят так:

result = chain.invoke({'topic':'cricket'})
print(result)

Запуск OpenAI также возвращает строку с пятью фактами о крикете, например:

1. Cricket is believed to have originated in England.
2. Each cricket team has 11 players.
3. A Test match can be played over a maximum of five days.
4. The Cricket World Cup is one of the most important international cricket tournaments.
5. Cricket has a particularly large following in India and several other countries.

Как и раньше, точные формулировки будут отличаться в зависимости от запусков и моделей.

Вывод структуры цепочки

Оба скрипта заканчиваются вызовом, который отображает конвейер в виде текстовой диаграммы:

chain.get_graph().print_ascii()

get_graph() возвращает графическое представление объекта Runnable, а print_ascii() отрисовывает его в терминале. Для цепочки Gemini получается что-то вроде этого:

PromptInput
     *
     *
PromptTemplate
     *
     *
GoogleGenerativeAI
     *
     *
StrOutputParser
     *
     *
StrOutputParserOutput

Узлы PromptInput и StrOutputParserOutput являются точками входа и выхода, определенными схемой; три узла между ними представляют собой ваши компоненты. Это быстрая проверка на то, что цепочка соединена так, как вы себе представляете. В большинстве сред print_ascii() зависит от необязательного пакета grandalf, поэтому установите его, если вызов приводит к ошибке импорта.

Изменился только модель

Разместите обе цепочки рядом — разница заключается в одном узле. С Gemini:

Input
  ↓
PromptTemplate
  ↓
GoogleGenerativeAI
  ↓
StrOutputParser
  ↓
Output

С OpenAI:

Input
  ↓
PromptTemplate
  ↓
ChatOpenAI
  ↓
StrOutputParser
  ↓
Output

В коде вся разница сводится к тому, какой объект модели вы создаете:

# Google Gemini
llm = GoogleGenerativeAI(model="gemini-2.5-flash")

# OpenAI
llm = ChatOpenAI(model="gpt-4o-mini")

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

Полный путь обработки данных

В общих чертах простая цепочка выглядит следующим образом:

Input
  ↓
PromptTemplate
  ↓
LLM
  ↓
StrOutputParser
  ↓
Output

При отслеживании с использованием входных данных в формате cricket каждое промежуточное значение становится видимым — от словаря до готового запроса, от сырого ответа модели до окончательной строки:

{'topic': 'cricket'}
        ↓
PromptTemplate
        ↓
"Generate 5 interesting facts about cricket"
        ↓
LLM
        ↓
Raw LLM Response
        ↓
StrOutputParser
        ↓
Final String Output

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

Последовательные цепочки: передача результатов одного вызова модели в следующий

Последовательная цепь распространяет ту же идею на несколько этапов. Простая цепь обычно содержит один такой блок:

Prompt → LLM → Output Parser

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

Input
  ↓
Prompt Template 1
  ↓
LLM 1
  ↓
Output Parser
  ↓
Prompt Template 2
  ↓
LLM 2
  ↓
Output Parser
  ↓
Output

Пример: подробный отчет, затем краткое резюме

В данном примере задача состоит из двух фаз. Сначала генерируется подробный отчет по определенной теме. Затем этот отчет сокращается до резюме из пяти пунктов. На второй фазе никогда не рассматривается сама исходная тема; она используется только на основе отчета, сгенерированного на первой фазе:

Input
  ↓
Explanation Prompt
  ↓
LLM
  ↓
StrOutputParser
  ↓
Detailed Report
  ↓
Summary Prompt
  ↓
LLM
  ↓
StrOutputParser
  ↓
5-Point Summary

Для выполнения задачи используются два шаблона запросов. explaination_prompt (ошибка в написании присутствует в коде, поэтому сохраняйте её при копировании фрагментов) запрашивает подробный отчёт. summary_prompt принимает этот отчёт в качестве входных данных и запрашивает пять ключевых моментов для краткого изложения.

Версия Gemini

Вот полная программа для работы с Gemini:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

from dotenv import load_dotenv
load_dotenv()
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
# Prompt 1 : Explanation
explaination_prompt = PromptTemplate(
    template='Generate a detailed report on {topic}',
    input_variables=['topic']
)
# Prompt 2 : Summary
summary_prompt = PromptTemplate(
    template='Generate a 5 pointer summary from the following text \n {text}',
    input_variables=['text']
)
parser = StrOutputParser()
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
result = chain.invoke({"topic": "Generative AI"})
# print(result)
chain.get_graph().print_ascii()

Один экземпляр GoogleGenerativeAI обслуживает обе стадии, причём вся схема представляет собой одно выражение, состоящее из шести компонентов: двух запросов, модели дважды и парсера также дважды.

Разбор кода Gemini

Импорты соответствуют простой цепочке:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from dotenv import load_dotenv

GoogleGenerativeAI — для работы с моделью, PromptTemplate — для двух повторно используемых шаблонов запросов, StrOutputParser — для преобразования каждого ответа в строку, а load_dotenv — для настройки параметров.

Переменные среды загружаются в первую очередь, точно как и раньше:

load_dotenv()

Модель создается один раз:

llm = GoogleGenerativeAI(model="gemini-2.5-flash")

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

Первый запрос принимает параметр topic и требует подробного отчета:

explaination_prompt = PromptTemplate(
    template='Generate a detailed report on {topic}',
    input_variables=['topic']
)

При наличии такого входного данных:

{"topic": "Generative AI"}

результат будет выглядеть следующим образом:

Generate a detailed report on Generative AI

Интересная передача данных происходит при втором запросе:

summary_prompt = PromptTemplate(
    template='Generate a 5 pointer summary from the following text \n {text}',
    input_variables=['text']
)

Он ожидает переменную с именем text, но в словаре входных данных нет элемента с таким именем. Значение поступает со второй стадии: в эту ячейку записывается отчет, сгенерированный при первом вызове модели.

Это работает благодаря парсеру, находящемуся между этапами:

parser = StrOutputParser()

После первого вызова модели парсер генерирует обычную строку. Когда PromptTemplate с ровно одной переменной входных данных получает просто строку вместо словаря, LCEL сопоставляет эту строку с единственной переменной. Поэтому отчет оказывается в формате {text} без какого-либо явного сопоставления. Если бы второй промпт содержал две переменные, вам пришлось бы самостоятельно создать словарь, например с помощью небольшого словаря Runnables или метода RunnablePassthrough.assign.

Вся цепочка обработки представлена одной строкой:

chain = explaination_prompt | llm | parser | summary_prompt | llm | parser

Чтение её слева направо показывает порядок выполнения:

Explanation Prompt
        ↓
      Gemini
        ↓
 StrOutputParser
        ↓
  Summary Prompt
        ↓
      Gemini
        ↓
 StrOutputParser
        ↓
     Final Output

Каждый компонент получает результат работы предыдущего компонента. В вашем коде ничто не влияет на промежуточный отчет.

Для его запуска требуется всего одна тема:

result = chain.invoke({"topic": "Generative AI"})

За этим вызовом обе стадии выполняются одна за другой:

"Generative AI"
       ↓
Explanation Prompt
       ↓
Gemini
       ↓
StrOutputParser
       ↓
Detailed Report
       ↓
Summary Prompt
       ↓
Gemini
       ↓
StrOutputParser
       ↓
Final 5-Point Summary

Что возвращается

Конечное значение — это пятипунктный обзор генеративного ИИ, например:

1. Generative AI is a branch of artificial intelligence that creates new content.
2. It can generate text, images, audio, video, and code.
3. Foundation models are an important part of modern Generative AI systems.
4. Generative AI is used across areas such as education, software development, and customer support.
5. It is enabling applications that can automate and simplify many creative and knowledge-based tasks.

Следует отметить, что этот обзор не формируется непосредственно на основе темы. Gemini сначала создает полный отчет, и только этот отчет отправляется в запрос на формирование обзора. Такой двухэтапный подход позволяет использовать более обширный материал для сжатия, но при этом требуется дополнительный вызов модели, а также увеличивается задержка и количество токенов. Как всегда, конкретные формулировки меняются при каждом запуске.

Тот же процесс на OpenAI

Вариант от OpenAI сохраняет исходную структуру работы:

from langchain_openai import ChatOpenAI
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

from dotenv import load_dotenv
load_dotenv()
llm = ChatOpenAI()
explaination_prompt = PromptTemplate(
    template="Generate a detailed report on {topic}",
    input_variables=["topic"]
)
summary_prompt = PromptTemplate(
    template="Generate a 5 pointer summary from the following text \n {text}",
    input_variables=["text"],
)
parser = StrOutputParser()
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
result = chain.invoke({"topic": "Generative AI"})
print(result)
chain.get_graph().print_ascii()

Это та же двухэтапная структура: запрос на отчет, модель, парсер, запрос на краткое резюме, модель, парсер. В скрипте также дважды импортируется функция load_dotenv, что не влияет на работу, но можно устранить.

На этот раз модель создается без аргументов:

llm = ChatOpenAI()

При отсутствии аргумента model функция ChatOpenAI использует стандартную модель библиотеки, которая менялась в разных версиях langchain_openai. В реальных проектах необходимо явно указать имя модели, чтобы обновление не приводило к скрытым изменениям в работе или стоимости.

Запрос на отчет идентичен версии Gemini:

explaination_prompt = PromptTemplate(
    template="Generate a detailed report on {topic}",
    input_variables=["topic"]
)

То же самое касается запроса на резюме, который также ожидает отчет в формате {text}:

summary_prompt = PromptTemplate(
    template="Generate a 5 pointer summary from the following text \n {text}",
    input_variables=["text"],
)

Именно здесь проявляется преимущество последовательного подхода: второй запрос автоматически формируется на основе результата первого этапа.

Парсер остается тем же — StrOutputParser:

parser = StrOutputParser()

И составление происходит символ за символом так же:

chain = explaination_prompt | llm | parser | summary_prompt | llm | parser

С точки зрения потока данные обрабатываются следующим образом:

Prompt 1
   ↓
OpenAI
   ↓
StrOutputParser
   ↓
Prompt 2
   ↓
OpenAI
   ↓
StrOutputParser
   ↓
Final Output

Только узел модели отличается от версии Gemini.

При вызове тема передаётся снова. Обратите внимание, что приведённый фрагмент не содержит закрывающей скобки; в полной версии программы есть правильная строка:

result = chain.invoke({"topic": "Generative AI"}

Затем цепочка выполняет обе стадии самостоятельно.

Что возвращается

Результатом снова является краткое резюме из пяти пунктов, полученное на основе промежуточного отчёта:

1. Generative AI creates new content by learning patterns from existing data.
2. It is powered by large foundation models trained on diverse datasets.
3. Generative AI can produce text, images, audio, video, and code.
4. It is being used in many industries and applications.
5. It is becoming an important part of modern AI-powered software.

Детали будут отличаться при разных запусках.

Вывод структуры цепочки

Помощник для графа работает также с более длинными цепочками:

chain.get_graph().print_ascii()

Для последовательной цепочки Gemini на диаграмме показаны обе стадии, расположенные вертикально друг над другом:

PromptInput
     *
     *
PromptTemplate
     *
     *
GoogleGenerativeAI
     *
     *
StrOutputParser
     *
     *
PromptTemplate
     *
     *
GoogleGenerativeAI
     *
     *
StrOutputParser
     *
     *
StrOutputParserOutput

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

Сравнение двух реализаций

Обе версии имеют одну и ту же последовательную структуру; единственная реальная разница — это поставщик. Конвейер Gemini:

Input
  ↓
Explanation Prompt
  ↓
GoogleGenerativeAI
  ↓
StrOutputParser
  ↓
Summary Prompt
  ↓
GoogleGenerativeAI
  ↓
StrOutputParser
  ↓
Output

Конвейер OpenAI:

Input
  ↓
Explanation Prompt
  ↓
ChatOpenAI
  ↓
StrOutputParser
  ↓
Summary Prompt
  ↓
ChatOpenAI
  ↓
StrOutputParser
  ↓
Output

Пromptы, парсер и структура цепочки совпадают. Единственное, что меняется — это создание модели:

# Google Gemini
llm = GoogleGenerativeAI(model="gemini-2.5-flash")

# OpenAI
llm = ChatOpenAI()

Сама составленная цепочка остается идентичной у всех поставщиков:

chain = explaination_prompt | llm | parser | summary_prompt | llm | parser

В абстрактном смысле последовательная цепочка выглядит так:

Prompt 1
   ↓
LLM 1
   ↓
Parser
   ↓
Prompt 2
   ↓
LLM 2
   ↓
Parser
   ↓
Output

Разница с простой цепочкой заключается в том, что здесь есть несколько этапов, причем каждый следующий этап зависит от предыдущего.

Полный путь обработки данных

В общем виде последовательная цепочка передает данные следующим образом:

Input
  ↓
PromptTemplate 1
  ↓
LLM 1
  ↓
StrOutputParser
  ↓
Detailed Report
  ↓
PromptTemplate 2
  ↓
LLM 2
  ↓
StrOutputParser
  ↓
Final Output

Отслеживается с использованием входных данных генеративного ИИ, включая обработанные запросы:

{"topic": "Generative AI"}
            ↓
    Explanation Prompt
            ↓
      "Generate a detailed
       report on Generative AI"
            ↓
           LLM
            ↓
    StrOutputParser
            ↓
      Detailed Report
            ↓
      Summary Prompt
            ↓
    "Generate a 5 pointer
     summary from the following text"
            ↓
           LLM
            ↓
    StrOutputParser
            ↓
     5-Point Summary

Первый вызов модели является промежуточным шагом, а не ответом. Его результат становится сырьем для следующей операции. Можно продолжать добавлять этапы, если в рабочем процессе их требуется больше:

Prompt
  ↓
LLM
  ↓
Parser
  ↓
Prompt
  ↓
LLM
  ↓
Parser
  ↓
Prompt
  ↓
LLM
  ↓
Output

Как бы длинной ни становилась цепочка, порядок остается неизменным, и каждый этап ждет предыдущего. Это также означает, что задержки накапливаются: три последовательных вызова моделей стоят примерно столько же, сколько сумма времени их откликов. Также целесообразно добавить защиту от чрезмерного роста промежуточных результатов, поскольку очень длинный отчет попадает непосредственно в следующий запрос, увеличивая его количество токенов.

Параллельные цепочки: расширение на основе одних и тех же входных данных

Параллельная цепочка выполняет несколько подцепочек одновременно. В то время как последовательная цепочка ждет завершения каждого этапа перед началом следующего, параллельная цепочка запускает независимые подцепочки одновременно:

                    ┌──→ Sub-Chain 1 ──→ Output 1
                    │
Input ──────────────┼──→ Sub-Chain 2 ──→ Output 2
                    │
                    └──→ Sub-Chain 3 ──→ Output 3
                                      ↓
                              Combined Output

Поскольку подцепочки независимы, ни одна из них не видит результатов работы другой. Механизм довольно прост:

1. Каждая подцепочка получает один и тот же входной данные и генерирует свой собственный результат.
2. Все подцепочки запускаются одновременно.
3. Параллельный этап ждет, пока все подцепочки не завершат работу.
4. Результаты всех подцепочек собираются в одно объединенное значение.

Это объединенное значение представляет собой словарь. Каждый ключ — это имя, которое вы задали подцепочке, а каждое значение — результат работы этой подцепочки:

{
    "facts": "...",
    "definition": "...",
    "use_case": "..."
}

Пример: три вида отображения одной темы

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

                         Input
                           ↓
                    RunnableParallel
                     /      |      \
                    /       |       \
                   ↓        ↓        ↓
               Facts    Definition  Use Case
                 ↓         ↓          ↓
                LLM       LLM        LLM
                 ↓         ↓          ↓
               Parser    Parser     Parser
                    \       |       /
                     \      |      /
                      ↓     ↓     ↓
                     Combined Result

Работу выполняют три под-цепочки: facts — для пяти фактов, definition — для краткого определения, и use_case — для примера практического применения. Они объединяются с помощью RunnableParallel. В первой реализации для всех трех задач используется Gemini; во второй применяются OpenAI и Anthropic с дополнительным шагом объединения результатов.

Почему обрабатывать под-цепочки параллельно

Существует две причины для использования этого подхода.

Первая причина — скорость. В процессе вызова моделей основное время занимает передача данных по сети и генерация ответа, поэтому независимые вызовы в значительной степени выигрывают от одновременного выполнения. Представьте три подзадачи, каждая из которых занимает примерно две секунды. Если выполнять их по очереди, общее время составит сумму времени каждой из них:

Chain 1 → 2 seconds
Chain 2 → 2 seconds
Chain 3 → 2 seconds

Total ≈ 6 seconds

Если выполнять их одновременно, общее время будет примерно равно времени выполнения самой медленной из них:

Chain 1 ─┐
Chain 2 ─┼→ execute together → ≈ 2 seconds
Chain 3 ─┘

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

                    Document
                       ↓
               ┌───────┼───────┐
               ↓       ↓       ↓
            Summary  Topics  Questions

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

Версия Gemini

Вот трехсторонняя структура обработки в Gemini:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

from dotenv import load_dotenv
load_dotenv()

llm = GoogleGenerativeAI(model="gemini-2.5-flash")
parser = StrOutputParser()

# Prompt 1: Facts
fact_prompt = PromptTemplate(
    input_variables=["topic"],
    template="Give five interesting short facts about: {topic}"
)

# Prompt 2: Definition
definition_prompt = PromptTemplate(
    input_variables=["topic"],
    template="Give a concise definition of: {topic}"
)

# Prompt 3: Use Case
usecase_prompt = PromptTemplate(
    input_variables=["topic"],
    template="What is a real-world use case of: {topic}?"
)

# Run all chains in parallel
parallel_chain = RunnableParallel({
    "facts": fact_prompt | llm | parser,
    "definition": definition_prompt | llm | parser,
    "use_case": usecase_prompt | llm | parser
})

# Invoke parallel chain
result = parallel_chain.invoke({"topic": "Artificial Intelligence"})

# print(type(result))
# print(result)
# print("\n📌 Facts:\n", result["facts"])
# print("\n📘 Definition:\n", result["definition"])
# print("\n🛠️ Use Case:\n", result["use_case"])

parallel_chain.get_graph().print_ascii()

Три шаблона запросов, одна модель и один парсер объединяются в три небольшие цепочки, которые затем RunnableParallel выполняет одновременно на входных данных Artificial Intelligence. Несколько строк с командой print закомментированы; их активация позволяет увидеть тип результата и каждое поле.

Разбор кода Gemini

Импорты добавляют один новый компонент:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel

Помимо обёртки модели, PromptTemplate и StrOutputParser, скрипт импортирует RunnableParallel из модуля langchain_core.runnables, который отвечает за одновременную работу с подцепями.

Создаётся одна модель Gemini 2.5 Flash:

llm = GoogleGenerativeAI(model="gemini-2.5-flash")

Все три подцепи используют эту модель. Общее использование одного объекта модели между параллельными ветвями безопасно, поскольку каждый вызов является независимым.

Также общается один парсер:

parser = StrOutputParser()

В первом запросе требуются пять кратких фактов:

fact_prompt = PromptTemplate(
    input_variables=["topic"],
    template="Give five interesting short facts about: {topic}"
)

Во втором запросе требуется краткое определение:

definition_prompt = PromptTemplate(
    input_variables=["topic"],
    template="Give a concise definition of: {topic}"
)

В третьем запросе требуется пример применения в реальном мире:

usecase_prompt = PromptTemplate(
    input_variables=["topic"],
    template="What is a real-world use case of: {topic}?"
)

Все три запроса используют одинаковое имя переменной:

{topic}

Именно это общее имя позволяет одному словарю входных данных подавать данные всем ветвям. Сами задачи остаются полностью разделенными.

Теперь центральная часть — параллельная цепочка:

parallel_chain = RunnableParallel({
    "facts": fact_prompt | llm | parser,
    "definition": definition_prompt | llm | parser,
    "use_case": usecase_prompt | llm | parser
})

RunnableParallel принимает словарь, значения которого являются объектами типа Runnable. Ключи этого словаря становятся ключами результата:

facts
definition
use_case

Каждое значение представляет собой полностью самостоятельную небольшую цепочку:

fact_prompt → LLM → parser

definition_prompt → LLM → parser

usecase_prompt → LLM → parser

Поскольку ни одна из них не зависит от другой, все они могут выполняться одновременно. С использованием синхронного метода invoke LangChain запускает ветви в пуле потоков; с использованием метода ainvoke они выполняются как конкурирующие асинхронные задачи.

Цепочка вызывается один раз:

result = parallel_chain.invoke({"topic": "Artificial Intelligence"})

с использованием одного входного значения:

{"topic": "Artificial Intelligence"}

Этот словарь передается каждой ветви. Концептуально:

Artificial Intelligence
          ↓
    RunnableParallel
     /      |       \
    ↓       ↓        ↓
  Facts  Definition  Use Case
    ↓       ↓        ↓
   LLM     LLM      LLM
    ↓       ↓        ↓
 Parser   Parser   Parser
     \      |       /
      \     |      /
       Combined Result

Что возвращается

Результатом является словарь, в котором по одной записи на каждую подцепочку:

{
    "facts": "1. Artificial Intelligence enables machines to perform tasks that normally require human intelligence.\n2. ...",
    "definition": "Artificial Intelligence is the field of creating systems that can perform tasks requiring human-like intelligence.",
    "use_case": "A real-world use case of AI is customer support chatbots that can understand and respond to user queries."
}

Текст внутри каждого поля будет отличаться при разных запусках, но его структура остаётся неизменной, и именно на этом основан ваш код:

result["facts"]
result["definition"]
result["use_case"]

Каждый ключ возвращает результат соответствующей ветви, что позволяет коду ниже получить именно ту информацию, которая ему нужна.

Смешивание поставщиков и объединение результатов

Вторая реализация демонстрирует ещё две идеи: разные ветви могут использовать разных поставщиков, а параллельный шаг сам может быть одним из этапов более крупной последовательной обработки. Здесь одна ветвь использует OpenAI, другая — Anthropic, а последний вызов OpenAI объединяет оба результата в один документ:

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from dotenv import load_dotenv
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.schema.runnable import RunnableParallel

load_dotenv()

model1 = ChatOpenAI()

model2 = ChatAnthropic(model_name='claude-3-7-sonnet-20250219')

prompt1 = PromptTemplate(
    template='Generate short and simple notes from the following text \n {text}',
    input_variables=['text']
)

prompt2 = PromptTemplate(
    template='Generate 5 short question answers from the following text \n {text}',
    input_variables=['text']
)

prompt3 = PromptTemplate(
    template='Merge the provided notes and quiz into a single document \n notes -> {notes} and quiz -> {quiz}',
    input_variables=['notes', 'quiz']
)

parser = StrOutputParser()

parallel_chain = RunnableParallel({
    'notes': prompt1 | model1 | parser,
    'quiz': prompt2 | model2 | parser
})

merge_chain = prompt3 | model1 | parser

chain = parallel_chain | merge_chain

text = """
Support vector machines (SVMs) are a set of supervised learning methods used for classification, regression and outliers detection.

The advantages of support vector machines are:

Effective in high dimensional spaces.

Still effective in cases where number of dimensions is greater than the number of samples.

Uses a subset of training points in the decision function (called support vectors), so it is also memory efficient.

Versatile: different Kernel functions can be specified for the decision function. Common kernels are provided, but it is also possible to specify custom kernels.

The disadvantages of support vector machines include:

If the number of features is much greater than the number of samples, avoid over-fitting in choosing Kernel functions and regularization term is crucial.

SVMs do not directly provide probability estimates, these are calculated using an expensive five-fold cross-validation (see Scores and probabilities, below).

The support vector machines in scikit-learn support both dense (numpy.ndarray and convertible to that by numpy.asarray) and sparse (any scipy.sparse) sample vectors as input. However, to use an SVM to make predictions for sparse data, it must have been fit on such data. For optimal performance, use C-ordered numpy.ndarray (dense) or scipy.sparse.csr_matrix (sparse) with dtype=float64.
"""

result = chain.invoke({'text': text})

print(result)

chain.get_graph().print_ascii()

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

Два момента в этом описании требуют уточнения. В коде импортируется класс RunnableParallel из пути langchain.schema.runnable, который является устаревшим путем ещё до разделения проекта на langchain_core; в текущих версиях этот класс доступен по пути langchain_core.runnables, как показано в примере с Gemini, и устаревший путь может отсутствовать в новых версиях. Также задан идентификатор модели claude-3-7-sonnet-20250219, который устарел; поставщики со временем удаляют старые версии моделей, поэтому перед повторным использованием необходимо проверить актуальный список моделей Anthropic.

Создаются два объекта модели:

model1 = ChatOpenAI()

model2 = ChatAnthropic(model_name='claude-3-7-sonnet-20250219')

ChatOpenAI записывает заметки, а затем выполняет слияние. ChatAnthropic создаёт тестовый вопросник. В классе RunnableParallel нет требований к тому, чтобы все ветви использовали одну и ту же модель, что удобно, когда для определённой подзадачи один поставщик обслуживания лучше или дешевле.

Подсказка для записи заметок преобразует введённый текст в краткие, простые заметки:

prompt1 = PromptTemplate(
    template='Generate short and simple notes from the following text \n {text}',
    input_variables=['text']
)

Подсказка для тестового вопросника генерирует из одного и того же текста пять кратких пар вопросов и ответов:

prompt2 = PromptTemplate(
    template='Generate 5 short question answers from the following text \n {text}',
    input_variables=['text']
)

Ни одна из задач не нуждается в результатах другой, поэтому они являются естественными параллельными ветвями.

Третья подсказка отличается: она выполняется после разветвления и требует обоих результатов:

prompt3 = PromptTemplate(
    template='Merge the provided notes and quiz into a single document \n notes -> {notes} and quiz -> {quiz}',
    input_variables=['notes', 'quiz']
)

В ней объявляются две переменные входных данных:

notes
quiz

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

Параллельный шаг определяет две ветви:

parallel_chain = RunnableParallel({
    'notes': prompt1 | model1 | parser,
    'quiz': prompt2 | model2 | parser
})

Его структура представляет собой двусторонний раздел, который объединяется в словарь:

                Input Text
                    ↓
             RunnableParallel
                /        \
               ↓          ↓
          Notes Chain   Quiz Chain
               ↓          ↓
          OpenAI        Anthropic
               ↓          ↓
            Parser      Parser
                \        /
                 ↓      ↓
                Dictionary

Концептуально промежуточное значение выглядит так:

{
    "notes": "...",
    "quiz": "..."
}

Цепочка объединения — это обычная простая цепочка:

merge_chain = prompt3 | model1 | parser

Она берет словарь из параллельного шага, заполняет {notes} и {quiz}, а затем просит OpenAI собрать один документ.

В конце два этапа соединяются с помощью ещё одного «трубопровода»:

chain = parallel_chain | merge_chain

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

                    Input Text
                        ↓
                 RunnableParallel
                  /            \
                 ↓              ↓
              Notes            Quiz
                 ↓              ↓
              OpenAI         Anthropic
                 ↓              ↓
              Parser          Parser
                  \            /
                   ↓          ↓
                 Combined Dictionary
                        ↓
                  Merge Prompt
                        ↓
                      OpenAI
                        ↓
                      Parser
                        ↓
                  Final Output

Ключевой аспект поведения заключается в том, что обе ветви выполняются независимо, но этап слияния не может начаться, пока обе не завершат свою работу. Поэтому самая медленная ветвь определяет темп всей структуры параллельных операций. Если одна из ветвей терпит неудачу, параллельный процесс прерывается, и слияние так и не происходит; поэтому при важности этого фактора необходимо обернуть ненадежные компоненты в механизмы повторных попыток или резервных решений.

Весь поток обработки запускается с использованием следующего текста:

result = chain.invoke({'text': text})

Ветви для записей и тестов обрабатывают текст одновременно. Как только обе из них завершат работу, их результаты поступают в цепочку слияния, а строковый результат этой цепочки и возвращается функцией invoke.

Вывод структуры цепочки

Вспомогательный инструмент для визуализации графа также позволяет отобразить структуру параллельных операций:

parallel_chain.get_graph().print_ascii()

Для трехветвной цепочки Gemini диаграмма показывает процесс разделения и объединения данных:

                 Parallel<...>Input
                   /      |      \
                  /       |       \
          PromptTemplate PromptTemplate PromptTemplate
               ↓              ↓              ↓
              LLM            LLM            LLM
               ↓              ↓              ↓
          StrOutputParser StrOutputParser StrOutputParser
                  \          |          /
                   \         |         /
                    Parallel<...>Output

Узлы Parallel<...>Input и Parallel<...>Output обозначают место, где происходит распределение входных данных, и место, где результаты собираются обратно в словарь.

Сравнение двух реализаций

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

Input
  ↓
RunnableParallel
  ├── Facts → Gemini → Parser
  ├── Definition → Gemini → Parser
  └── Use Case → Gemini → Parser
              ↓
       Dictionary Output

Версия с смешанными поставщиками разделяется на две ветви, работающие с разными поставщиками, после чего добавляется этап объединения:

Input
  ↓
RunnableParallel
  ├── Notes → OpenAI → Parser
  └── Quiz → Anthropic → Parser
              ↓
       Dictionary Output
              ↓
         Merge Chain
              ↓
            OpenAI
              ↓
            Output

В основе обеих версий лежит одна и та же структура:

RunnableParallel({
    "chain_1": chain_1,
    "chain_2": chain_2,
    "chain_3": chain_3
})

Ключи обозначают подцепочки; значения — сами эти подцепочки. В качестве удобства, когда в выражении с трубкой появляется обычный словарь объектов типа Runnable, LCEL автоматически преобразует его в RunnableParallel, поэтому часто можно увидеть этот словарь без указания конкретного класса.

Полный путь к данным

В общем случае параллельная цепочка распределяет один входной данные и формирует словарь:

                         Input
                           ↓
                    RunnableParallel
                    /       |       \
                   ↓        ↓        ↓
               Sub-chain Sub-chain Sub-chain
                   ↓        ↓        ↓
                 Output   Output   Output
                    \       |       /
                     \      |      /
                      ↓     ↓     ↓
                        Dictionary

Пример использования искусственного интеллекта на Gemini:

{"topic": "Artificial Intelligence"}
                ↓
        RunnableParallel
          /      |      \
         ↓       ↓       ↓
      Facts  Definition  Use Case
         ↓       ↓       ↓
       Gemini  Gemini   Gemini
         ↓       ↓       ↓
      Parser  Parser   Parser
          \      |      /
           \     |     /
             Dictionary

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

{
    "facts": "...",
    "definition": "...",
    "use_case": "..."
}

В этом и заключается основное отличие от последовательной цепочки. В последовательной цепочке каждый этап использует результат предыдущего этапа:

A → B → C → D

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

       ┌→ A ─┐
Input ─┼→ B ─┼→ Combined Output
       └→ C ─┘

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

Условные цепочки: направление входных данных к соответствующей подцепочке

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

Input
  ↓
Condition Check
  ↓
 ┌───────────────┐
 │               │
Yes             No
 ↓               ↓
Chain 1        Chain 2
 ↓               ↓
Output 1       Output 2
  \             /
   \           /
    Final Output

В LangChain это реализуется с помощью RunnableBranch. Вы указываете список пар, в каждой из которых условие соединяется с операцией выполнения, плюс в конце — стандартная операция выполнения. Условия проверяются по порядку, первое из них, возвращающее значение True, определяет, какая операция выполнения будет применена; если ни одно условие не сработает, обрабатывается стандартная операция.

Пример: маршрутизация по длине текста

В примере текст обрабатывается по-разному в зависимости от его длины:

• Если в тексте более 50 слов, генерируется краткое резюме.
• В противном случае раскрывается суть изложенной идеи.

Результативный рабочий процесс:

Input Text
    ↓
Condition Check
    ↓
Is word count > 50?
    ↓
 ┌───────┴───────┐
Yes              No
 ↓                ↓
Summary          Extend
Chain            Chain
 ↓                ↓
LLM              LLM
 ↓                ↓
Parser           Parser
  \                /
   \              /
    Final Output

Два запроса управляют ветвлениями: summary_prompt запрашивает краткое резюме, а extend_prompt просит модель развить предложенную идею. Небольшая функция is_long_text() считает количество слов, а RunnableBranch использует её результат для выбора последовательности действий. Как и в других случаях, существуют версии для Gemini и OpenAI.

В общем виде двунаправленный условный поток выглядит следующим образом:

User Input
    ↓
Condition Check
    ↓
 ┌───────┴───────┐
Yes              No
 ↓                ↓
Prompt 1        Prompt 2
 ↓                ↓
LLM              LLM
 ↓                ↓
Parser           Parser
  \                /
   └──────┬───────┘
           ↓
      Final Output

Для любого заданного входа выполняется ровно одна ветвь.

Анализ конкретного входа

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

Это предложение содержит 22 слова, что значительно меньше порогового значения в 50 слов, поэтому условие возвращает False, и выбирается ветвь для расширения:

Input Text
    ↓
Condition Check
    ↓
22 > 50?
    ↓
   No
    ↓
Extend Chain
    ↓
Gemini 2.5 Flash
    ↓
StrOutputParser
    ↓
Extended Text

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

Когда условное маршрутизирование полезно

Маршрутизирование оказывается полезным тогда, когда один и тот же тип входных данных требует разного обработания в зависимости от определенного свойства:

• Длина: сокращать длинный текст, расширять краткий.
• Настроение: направлять положительные и отрицательные отзывы в разные цепочки ответов.
• Тип запроса: направлять каждую категорию запросов в специализированную цепочку обработки.

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

Input
  ↓
Condition
  ↓
Choose one branch
  ↓
Execute selected chain
  ↓
Output

Это делает рабочий процесс адаптивным в том смысле, в котором это невозможно при использовании фиксированной последовательности, а также позволяет сократить количество вызовов модели по сравнению с обработкой всех веток и последующим отбрасыванием результатов.

Версия Gemini

Вот полная программа маршрутизации для Gemini:

from langchain_google_genai import GoogleGenerativeAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch
from dotenv import load_dotenv

load_dotenv()

llm = GoogleGenerativeAI(model="gemini-2.5-flash")

parser = StrOutputParser()

# Prompt 1 : Summary
summary_prompt = PromptTemplate(
    input_variables=["text"],
    template="Summarize the following text briefly:\n\n{text}"
)

# Prompt 2 : Extend
extend_prompt = PromptTemplate(
    input_variables=["text"],
    template="Extend the idea in the following long text:\n\n{text}"
)

# Chain 1 : Summary
summary_chain = summary_prompt | llm | parser

# Chain 2 : Extend
extend_chain = extend_prompt | llm | parser

def is_long_text(input: dict) -> bool:
    return len(input["text"].split()) > 50

conditional_chain = RunnableBranch(
    (lambda x: is_long_text(x), summary_chain),
    extend_chain
)

text_input = {
    "text": "Machine learning is a field of artificial intelligence that uses statistical techniques to give computer systems the ability to learn from data."
}

result = conditional_chain.invoke(text_input)

# print("\nOutput:\n")
# print(result)

conditional_chain.get_graph().print_ascii()

В ней определены две небольшие цепочки — summary_chain и extend_chain, а также предикат подсчёта слов; затем RunnableBranch выбирает между этими цепочками на основе данного предиката.

Разбор кода Gemini

Модель снова — Gemini 2.5 Flash:

llm = GoogleGenerativeAI(model="gemini-2.5-flash")

Класс StrOutputParser преобразует каждый ответ в строку:

parser = StrOutputParser()

Запрос на краткое резюме требует предоставления краткого содержания:

summary_prompt = PromptTemplate(
    input_variables=["text"],
    template="Summarize the following text briefly:\n\n{text}"
)

Запрос на расширение используется в другом пути:

extend_prompt = PromptTemplate(
    input_variables=["text"],
    template="Extend the idea in the following long text:\n\n{text}"
)

Внимательно посмотрите на формулировку: здесь говорится о «следующем длинном тексте», однако этот вариант выполняется только для коротких входных данных. Такое несоответствие не нарушает работу системы, но посылает модели вводящую в заблуждение информацию. Замена формулировки на что-то вроде «Раскройте идею, изложенную в следующем тексте», — это небольшое улучшение, которое стоит внести в вашу версию.

Каждый запрос объединяется с моделью и парсером в отдельную цепочку обработки:

summary_chain = summary_prompt | llm | parser

extend_chain = extend_prompt | llm | parser

Это приводит к двум независимым путям обработки:

Summary Prompt → Gemini → Parser

Extend Prompt  → Gemini → Parser

Условный шаг определяет, какой из путей будет использован.

Условие представляет собой обычную функцию на Python:

def is_long_text(input: dict) -> bool:
    return len(input["text"].split()) > 50

Она разделяет значение input[“text”] на части с помощью пробелов и подсчитывает их количество. Её функционал прост:

True  → if word count > 50
False → otherwise

Разделение по пробелам — это приблизительный способ подсчёта количества слов; знаки препинания остаются при словах, а сложные термины с дефисами считаются одним словом. Это подходит для маршрутизации по длине, но если вам действительно важны ограничения модели, следует считать токены, а не слова. Также обратите внимание, что имя параметра input перекрывает встроенную функцию Python; это работает, но более подходящим будет имя вроде data.

После того как все компоненты готовы, RunnableBranch собирает маршрутизатор:

conditional_chain = RunnableBranch(
    (lambda x: is_long_text(x), summary_chain),
    extend_chain
)

Первый аргумент — это кортеж, который интерпретируется как правило:

If is_long_text(x) is True
        ↓
Execute summary_chain

Последний аргумент, который не находится внутри кортежа, является значением по умолчанию:

Otherwise
        ↓
Execute extend_chain

Обёртка в виде лямбды не требуется. RunnableBranch принимает в качестве условия любую функцию, поэтому прямое передача is_long_text также сработает. В итоге маршрутизация выглядит следующим образом:

                    Input
                      ↓
              is_long_text()
                 /        \
              True        False
               ↓            ↓
        summary_chain   extend_chain
               ↓            ↓
             Gemini       Gemini
               ↓            ↓
             Parser       Parser
                \          /
                 \        /
                  Output

Для его запуска требуется словарь входных данных:

result = conditional_chain.invoke(text_input)

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

Что возвращается

При использовании примера входных данных из 22 слов предикат возвращает:

is_long_text() → False

Таким образом, функция extend_chain обрабатывает входные данные. Концептуально:

Input Text
    ↓
Word Count Check
    ↓
False
    ↓
extend_chain
    ↓
Gemini 2.5 Flash
    ↓
StrOutputParser
    ↓
Extended Text

Если передать текст длиной более 50 слов, предикат возвращает:

is_long_text() → True

Это изменяет путь на:

Input Text
    ↓
Word Count Check
    ↓
True
    ↓
summary_chain
    ↓
Gemini 2.5 Flash
    ↓
StrOutputParser
    ↓
Short Summary

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

Тот же маршрутизатор в OpenAI

Версия для OpenAI сохраняет логику маршрутизации и заменяет только модель:

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableBranch
from dotenv import load_dotenv

load_dotenv()
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
# Prompt 1 : Summary
summary_prompt = PromptTemplate(
    input_variables=["text"],
    template="Summarize the following text briefly:\n\n{text}"
)
# Prompt 2 : Extend
extend_prompt = PromptTemplate(
    input_variables=["text"],
    template="Extend the idea in the following long text:\n\n{text}"
)
# Chain 1 : Summary
summary_chain = summary_prompt | llm | parser
# Chain 2 : Extend
extend_chain = extend_prompt | llm | parser
def is_long_text(input: dict) -> bool:
    return len(input["text"].split()) > 50
conditional_chain = RunnableBranch(
    (lambda x: is_long_text(x), summary_chain),
    extend_chain
)
text_input = {
    "text": "Machine learning is a field of artificial intelligence that uses statistical techniques to give computer systems the ability to learn from data."
}
result = conditional_chain.invoke(text_input)
print(result)
conditional_chain.get_graph().print_ascii()

Два потока обработки, один предикат, один объект RunnableBranch — так же, как и в Gemini.

Модель — это gpt-4o-mini:

llm = ChatOpenAI(model="gpt-4o-mini")

Шаблоны запросов остались без изменений: один для краткого изложения, другой — для расширения:

summary_prompt = PromptTemplate(
    input_variables=["text"],
    template="Summarize the following text briefly:\n\n{text}"
)

extend_prompt = PromptTemplate(
    input_variables=["text"],
    template="Extend the idea in the following long text:\n\n{text}"
)

Цепочки формируются точно так же, как и раньше:

summary_chain = summary_prompt | llm | parser

extend_chain = extend_prompt | llm | parser

Теперь оба пути проходят через OpenAI:

Summary Prompt → OpenAI → StrOutputParser

Extend Prompt → OpenAI → StrOutputParser

Предикат — это тот же проверочный фильтр из 50 слов:

def is_long_text(input: dict) -> bool:
    return len(input["text"].split()) > 50

Маршрутизатор создается таким же образом:

conditional_chain = RunnableBranch(
    (lambda x: is_long_text(x), summary_chain),
    extend_chain
)

Результат True отправляет входные данные в функцию summary_chain; все остальное — в extend_chain. Затем вызывается соответствующая цепочка, а результат выводится:

result = conditional_chain.invoke(text_input)
print(result)

Для примерного предложения предикат снова даёт следующий результат:

is_long_text() → False

Поэтому запускается функция расширения, которая возвращает более подробное объяснение входных данных, примерно такое:

Machine learning enables computer systems to learn patterns
from data and improve their performance without being explicitly
programmed for every task. It is widely used in applications
such as recommendation systems, image recognition, and predictive
analytics.

Текст зависит от модели и меняется при каждом запуске.

Печать структуры цепочки

Тот же инструмент для графов используется и с маршрутизаторами:

conditional_chain.get_graph().print_ascii()

Основным элементом отрисованной диаграммы является узел Branch. Это точка принятия решений, где LangChain оценивает условия и передаёт входные данные одной из возможных цепочек.

В чём разница между двумя ветвями

У каждой ветви есть своя задача. Ветвь для подведения итогов:

summary_prompt
      ↓
Gemini
      ↓
StrOutputParser
      ↓
Short Summary

Ветвь для расширения:

extend_prompt
      ↓
Gemini
      ↓
StrOutputParser
      ↓
Extended Text

Предикат выбирает между ними:

                 Input
                   ↓
            is_long_text()
              /         \
           True         False
            ↓             ↓
       Summary Chain   Extend Chain

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

Маршрутизация на основе собственной классификации модели

Условия не ограничиваются простыми критериями, такими как количество слов. Более распространенным и эффективным подходом является использование первой цепочки для классификации входных данных, после чего происходит направление их в соответствии с полученной классификацией. В примере с оценкой настроения поступающие отзывы маркируются одним из двух значений:

positive
negative

а RunnableBranch передает их в цепочку ответов, соответствующую этой маркировке:

Feedback
   ↓
Sentiment Classifier
   ↓
 ┌───────────────┐
 │               │
Positive       Negative
 │               │
 ↓               ↓
Positive       Negative
Response       Response
 │               │
 └───────┬───────┘
         ↓
      Output

Надежным способом создания классификатора является использование модели, возвращающей структурированный результат, проверенный с помощью модели Pydantic — например, поле, ограниченное значениями positive или negative. Затем условия ветвлений проверяют именно это поле, а не анализируют свободный текст, что исключает непредвиденные ситуации, когда модель отвечает «Positive.» с заглавной буквы и точкой. Всегда следует предусмотреть разумную стандартную ветвь, поскольку классификатор рано или поздно может вернуть что-то, чего вы не ожидали.

Это показывает, что условие может зависеть от информации, сгенерированной ранее в цепочке обработки, а не только от сырого входного данных. Для более сложных сценариев маршрутизации в документации LangChain также описывается маршрутизация с использованием пользовательской функции, заключенной в RunnableLambda, которая возвращает цепочку для выполнения; стоит проверить, какой подход рекомендуют документы для вашей версии.

Полный путь данных

Общий шаблон условия выглядит следующим образом:

                       Input
                         ↓
                  Condition Check
                         ↓
                 ┌───────┴───────┐
                 ↓               ↓
              Condition       Condition
               True            False
                 ↓               ↓
             Chain 1          Chain 2
                 ↓               ↓
                LLM             LLM
                 ↓               ↓
              Parser          Parser
                 \               /
                  \             /
                   └─────┬───────┘
                         ↓
                       Output

В примере с длиной текста:

Machine Learning Text
          ↓
   is_long_text()
          ↓
     Word Count > 50?
        /       \
      Yes       No
       ↓         ↓
   Summary     Extend
    Chain       Chain
       ↓         ↓
    Gemini     Gemini
       ↓         ↓
    Parser     Parser
        \       /
         \     /
          Output

Короткие входные данные проходят по одному маршруту:

Word Count ≤ 50
       ↓
Extend Chain

Длинные входные данные проходят по другому маршруту:

Word Count > 50
       ↓
Summary Chain

Если посмотреть в целом, вот четыре варианта рядом друг с другом:

Simple Chain
A → B → C

Sequential Chain
A → B → C → D → E

Parallel Chain
      ┌→ B ─┐
A ────┼→ C ─┼→ Output
      └→ D ─┘

Conditional Chain
       ↓
      A
       ↓
   Condition
    /     \
   B       C

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

Практические рекомендации перед отправкой

Есть несколько часто встречающихся проблем, которые стоит учитывать при переходе от примеров к реальному коду:

• Фиксируйте имена моделей. Конструкторы без аргумента модели наследуют значение по умолчанию от установленной интеграции, и с течением времени такие идентификаторы могут стать устаревшими.
• Сохраняйте парсер. Модели чата возвращают объекты сообщений; StrOutputParser обеспечивает каждому последующему этапу предсказуемый тип str.
• Обращайте внимание на передачу данных. Обычная строка автоматически заполняет промпт только тогда, когда в нем ровно одна переменная. Для более сложных случаев необходимо явно создавать словарь.
• Соответствуйте ключам переменным. При расширении данных затем их объединении параллельные ключи должны совпадать с именами переменных в промпте для объединения.

Чтобы ознакомиться с более полным обзором, в котором эти цепочки рассматриваются на примере работы вместе с механизмами поиска информации, инструментами и агентами, работающими с локальными моделями, посмотрите LangChain 1.x в практике: цепочки, RAG, инструменты и агенты локально.

Основные выводы

• Цепочка — это объявленная конвейерная структура, в которой выход каждого компонента поступает на вход следующего; оператор | формирует её из любых объектов типа Runnable.
• Простые цепочки представляют собой один прямой проход данных; последовательные цепочки объединяют несколько таких проходов, что приводит к увеличению задержки на каждом этапе.
• RunnableParallel выполняет независимые подцепочки одновременно и возвращает словарь, ключи которого соответствуют названиям ветвей; последующие этапы могут использовать эти данные по имени переменной.
• RunnableBranch оценивает условия по порядку и выполняет только первую соответствующую ветвь, при этом при отсутствии совпадений используется значение по умолчанию.
• Условия могут быть простыми проверками входных данных или решениями, основанными на структурированной классификации предыдущей модели.
• Поскольку модель представляет собой всего один компонент, переключение между Gemini, OpenAI или Anthropic обычно влечёт за собой изменение всего одной строки кода, в то время как промпты, парсеры и структура остаются неизменными.

Все четыре подхода основаны на одном и том же принципе — интерфейсе Runnable, который в первую очередь позволяет объединять промпты, модели, парсеры и собственные функции. После того как эти концепции станут знакомыми, следующим логичным шагом будет прямое понимание этого интерфейса.

Связанные материалы