Складання конвеєрів LangChain з використанням LCEL: лінійні, паралельні та розгалужені.
Навчіться підключати запити, моделі та парсери до лінійних, багатоступеневих, паралельних та умовних конвеєрів LangChain за допомогою оператора pipe, RunnableParallel та RunnableBranch.
Джерела: Chains in LangChain: Simple, Sequential, Parallel, and Conditional — Part 7
Справжні функції LLM майже ніколи не обмежуються одним викликом моделі. Ви форматуєте запит, викликаєте модель, обробляєте отриманий результат, використовуєте його у наступному запиті, іноді виконуєте кілька запитів одночасно та час від часу обираєте інший шлях залежно від вхідних даних. Ручне під’єднання кожного етапу швидко перетворюється на складний для читання та ще складніший для змін код. У цьому посібнику розглядаються чотири шаблони композиції в LangChain, створені за допомогою мови виразів LangChain (LCEL): прямі пайплайни, багатоетапні пайплайни, розгалуження за допомогою RunnableParallel та маршрутизація за допомогою RunnableBranch. Кожен шаблон показаний двічі — один раз на Google Gemini та один раз на OpenAI, щоб ви могли точно бачити, які частини змінюються при зміні постачальника та які залишаються незмінними.
Що насправді є ланцюгом
Ланцюг — це послідовність компонентів LangChain, у якій кожен етап отримує те, що створив попередній етап. Типовими елементами є шаблони запитів, моделі для чату або завершення тексту та парсери результатів, проте можуть брати участь будь-які компоненти, які реалізують інтерфейс Runnable від LangChain. Найменша корисна послідовність виглядає так:
Prompt → LLM → Output Parser → Output
Кожен етап виконує одну функцію. Шаблон запиту бере ваші вхідні змінні та перетворює їх на інструкції для моделі. Модель читає ці інструкції та генерує відповідь. Парсер результатів перетворює цю відповідь у формат, необхідний вашому додатку — наприклад, у звичайний рядок чи перевірений об’єкт. Результатом роботи парсера є кінцевий результат усієї послідовності.
Цінність абстракції полягає у тому, що вам більше не потрібно самостійно передавати дані між цими етапами. Ви оголошуєте порядок лише один раз, і LangChain бере на себе завдання викликати кожен компонент із правильними даними. Концептуально ви складаєте невеликі блоки разом у більшу одиницю:
Input
↓
Component 1
↓
Component 2
↓
Component 3
↓
Output
Це мало має значення для скрипту з двома кроками, але стає життєво важливим, коли додаток розростається до десятка кроків. Оголошена конвеєрна схема зберігає потік даних у одному місці, замість того щоб розкидати його по допоміжних функціях. Ментальна модель полягає у повторюваному ритмі вхідних даних, їх обробки, результату та подальшої обробки цього результату, поки не буде отримано кінцеве значення.
Чому композиція краща за ручне підключення
Коли кожне проміжне значення передається явно, код стає галасливим та крихким. Ланцюги вирішують три практичні проблеми.
Потоки даних без коду-з’єднувача
Основна обіцянка полягає у тому, що результат одного кроку автоматично стає аргументом наступного. Візьмемо знайому послідовність знову:
Prompt → LLM → Output Parser → Output
Якщо писати вручну, потрібно буде сформатувати запит, викликати модель із сформатованим значенням, витягнути текст з об’єкта відповіді та передати його парсеру, зберігаючи змінну для кожного етапу. З LCEL цей же процес скорочується до одного виразу, створеного за допомогою оператора «трубка»:
chain = prompt | llm | parser
Оператор | працює тому, що запити, моделі та парсери всі реалізують один і той самий інтерфейс Runnable, тож кожен з них знає, як приймати результат попереднього компонента. Кількість тимчасових змінних зменшується, що значно скорочує ризик передачі неправильних значень, а канал обробки даних працює зліва направо у тому самому порядку, у якому виконується.
Багатокрокова логіка залишається зрозумілою
Функції обробки продукції часто вимагають кількох викликів моделі з різними запитами. Поширеним підходом є спочатку створення чогось на основі певної теми, а потім подальша обробка цього проміжного результату за допомогою другого запиту:
Topic
↓
Prompt 1
↓
LLM
↓
Output Parser
↓
Prompt 2
↓
LLM
↓
Output Parser
↓
Final Output
Якщо керувати цим процесом вручну, він розростається з додаванням нових кроків. Якщо представити цей процес у вигляді ланцюга, всі етапи залишаються в одній декларації, а правило про те, що кожен етап передає дані наступному, забезпечується фреймворком, а не самодисципліною.
Компоненти можна замінювати
Оскільки кожен етап є окремим компонентом, його можна замінити, не торкаючись інших. Припустимо, що потік обробки наразі працює з Gemini:
chain = prompt | gemini | parser
Перехід на OpenAI означає заміну лише компонента моделі:
chain = prompt | openai | parser
Запит та парсер залишаються без змін. Це корисно, коли змінюються вимоги, коли команда порівнює постачальників за вартістю чи якістю, або коли потрібна дешевша модель під час розробки та більш потужна — у продакшені.
Переваги в одному погляді
- Компактніший код: результати автоматично передаються до наступного кроку.
- Структуровані багатокрокові процеси: кілька запитів, моделей та парсерів знаходяться в одному оголошеному конвеєрі.
- Простіша еволюція: компоненти можна повторно використовувати чи замінювати, включаючи постачальника моделей, без необхідності перепроектування процесу.
Чотири шаблони композиції
Допомагає групувати пайплайни 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()
Його завдання — повернути відповідь моделі у вигляді рядка. У разі моделі типу completion, як-от 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
За допомогою вхідних даних у форматі крикету кожне проміжне значення стає видимим — від словника до остаточного запиту, від первинної відповіді моделі до кінцевої рядка:
{'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
Що повертається
Кінцеве значення — це п’ятипунктний огляд технології Generative AI, наприклад:
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
Запити, парсер та структура ланцюга є спільними. Єдиною частиною, яка змінюється, є створення моделі:
# 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
Оскільки підланцюги є незалежними, жоден з них не бачить результатів роботи іншого. Механізм досить простий:
- Кожен підланцюг отримує однаковий вхід та генерує власний результат.
- Усі підланцюги запускаються одночасно.
- Паралельний етап чекає, поки всі підланцюги не завершать роботу.
- Їхні результати об’єднуються в одне загальне значення.
Це загальне значення є словником. Кожен ключ — це назва, яку ви дали підланцюгу, а кожен значення — це результат роботи цього підланцюга:
{
"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
})
Ключі позначають підланцюги; значеннями є самі ці підланцюги. Як скорочення, коли звичайний словник Runnables з’являється всередині виразу з трубкою, 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. Ви надаєте йому список пар, у кожній з яких умова поєднується з об’єктом Runnable, а також є стандартний об’єкт Runnable наприкінці. Умови перевіряються у порядку, і перша з них, яка повертає True, визначає, який об’єкт Runnable буде виконаний; якщо жодна умова не підходить, обробкою вхідних даних займається стандартний об’єкт.
Приклад: маршрутизація за довжиною тексту
У цьому прикладі текст обробляється по-різному залежно від його довжини:
- Якщо у тексті більше 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_core.runnables, а не старі шляхи з langchain.schema.get_graph().print_ascii() (з встановленим grandalf) — це простий спосіб переконатися, що складна ланцюгова структура налаштована за задумом.Щоб дізнатися більше про ці ланцюги у поєднанні з функціями пошуку, інструментами та агентами, які працюють на локальних моделях, перегляньте LangChain 1.x на практиці: ланцюги, RAG, інструменти та агенти локально.
Основні висновки
- Ланцюг — це оголошена конвеєрна система, у якій вихід кожного компонента подається наступному; оператор
|створює його з будь-яких об’єктів типу Runnable. - Прості ланцюги передбачають один прямий прохід даних; послідовні ланцюги поєднують кілька проходів, що збільшує затримку на кожному етапі.
RunnableParallelвиконує незалежні підланцюги одночасно та повертає словник, ключі якого відповідають назвам гілок; наступні етапи можуть використовувати цей словник за назвою змінної.RunnableBranchоцінює умови в порядку та виконує лише першу підходящу гілку, у разі необхідності переходячи до значення за замовчуванням.- Умови можуть бути простими перевірками вхідних даних або рішеннями, заснованими на структурованій класифікації попередньої моделі.
- Оскільки модель є лише одним компонентом, зміна між Gemini, OpenAI чи Anthropic зазвичай передбачає зміну лише одного рядка, тоді як запити, парсери та структура залишаються незмінними.
Усі чотири патерни ґрунтуються на одній основі — інтерфейсі Runnable, який саме дозволяє поєднувати запити, моделі, парсери та власні функції. Після того, як ці патерни стануть знайомими, наступним логічним кроком є безпосереднє розуміння цього інтерфейсу.