Tworzenie pipeline’ów LangChain przy użyciu LCEL: liniowe, równoległe i z rozgałęzieniami
Naucz się łączenia promptów, modeli oraz parserów w liniowe, wieloetapowe, równoległe i warunkowe pipeline’y LangChain za pomocą operatora pipe, RunnableParallel oraz RunnableBranch.
Źródła: Chains in LangChain: Simple, Sequential, Parallel, and Conditional — Part 7
Cechy prawdziwych LLM prawie nigdy nie kończą się na jednej wywołaniu modelu. Sformatowujesz prompt, wywołujesz model, przetwarzasz otrzymany wynik, wprowadzasz go do kolejnego promptu, czasami uruchamiasz kilka zapytań równocześnie, a czasem wybierasz inną ścieżkę w zależności od danych wejściowych. Ręczne łączenie każdego etapu szybko przeradza się w kod „klejący”, który jest trudny do odczytania i jeszcze trudniejszy do modyfikacji. Ten przewodnik przedstawia cztery wzorce kompozycji w LangChain, stworzone przy użyciu języka wyrażeń LangChain (LCEL): proste pipeline’y, pipeline’y wieloetapowe, rozgałęzienia za pomocą RunnableParallel oraz kierowanie przepływem danych za pomocą RunnableBranch. Każdy wzorzec jest pokazany dwa razy – raz z Google Gemini, a raz z OpenAI – abyś mógł dokładnie zobaczyć, które elementy zmieniają się przy zmianie dostawcy, a które pozostają bez zmian.
Czym właściwie jest łańcuch
Łańcuch to szyna przetwarzania składników LangChain, w której każdy krok otrzymuje to, co wygenerował poprzedni krok. Typowymi elementami budulcowymi są szablony promptów, modele do rozmowy lub uzupełniania tekstu oraz parserzy wyników, ale może w nim uczestniczyć cokolwiek, co implementuje interfejs Runnable od LangChain. Najprostsza użyteczna szyna przetwarzania wygląda w ten sposób:
Prompt → LLM → Output Parser → Output
Każdy etap ma jedno zadanie. Szablon promptu bierze Twoje zmienne wejściowe i przekształca je w instrukcje dla modelu. Model czyta te instrukcje i generuje odpowiedź. Parser wyników przekształca tę odpowiedź na formę wymaganą przez Twoją aplikację, na przykład zwykłą ciąg znaków lub zweryfikowany obiekt. To, co wychodzi z parsera, jest wynikiem całej szyny przetwarzania.
Zaletą abstrakcji jest to, że nie musisz już sam przekazywać danych pomiędzy tymi etapami. Deklarujesz kolejność raz, a LangChain zajmuje się wywoływaniem każdego komponentu z odpowiednimi danymi wejściowymi. Koncepcyjnie łączysz małe elementy w jedną większą całość:
Input
↓
Component 1
↓
Component 2
↓
Component 3
↓
Output
To ma niewielkie znaczenie w przypadku skryptu składającego się z dwóch kroków, ale staje się kluczowe, gdy aplikacja obejmuje już tuzin kroków. Deklarowany pipeline umożliwia widzenie przepływu danych w jednym miejscu, zamiast rozproszenia ich w różnych funkcjach pomocniczych. Model mentalny opiera się na powtarzającym się rytmie: dane wejściowe, przetwarzanie, wynik, a następnie ponowne przetwarzanie tego wyniku, aż ostateczna wartość zostanie uzyskana.
Dlaczego kompozycja jest lepsza od ręcznego łączenia
Gdy każda wartość pośrednia jest przekazywana wyraźnie, kod staje się chaotyczny i kruchy. Łańcuchy rozwiązują trzy praktyczne problemy.
Przepływ danych bez kodu łączącego
Główne obietnice to to, że wynik jednego kroku automatycznie staje się argumentem następnego. Weźmy ponownie tę dobrze znaną sekwencję:
Prompt → LLM → Output Parser → Output
Gdybyś pisał ręcznie, musiałbyś sformatować zapytanie, wywołać model z uformowaną wartością, wydobyć tekst z obiektu odpowiedzi i przekazać go parserowi, przechowując zmienną dla każdego etapu. Z LCEL ten sam proces sprowadza się do jednego wyrażenia utworzonego za pomocą operatora pipe:
chain = prompt | llm | parser
Operator | działa, ponieważ zapytania, modele i parserzy implementują tę samą interfejs Runnable, dzięki czemu każdy z nich wie, jak przyjąć wynik poprzedniej komponenty. Mniejsza liczba zmiennych tymczasowych oznacza mniej miejsc, w których może dojść do przekazania niewłaściwej wartości, a pipeline jest czytany od lewej do prawej w tej samej kolejności, w jakiej jest wykonywany.
Logika wieloetapowa pozostaje czytelna
Cechy produkcji często wymagają kilku wywołań modelu z różnymi promptami. Powszechnym podejściem jest najpierw wygenerowanie czegoś na podstawie określonego tematu, a następnie przetworzenie tego tymczasowego wyniku za pomocą drugiego promptu:
Topic
↓
Prompt 1
↓
LLM
↓
Output Parser
↓
Prompt 2
↓
LLM
↓
Output Parser
↓
Final Output
Gdy taki proces jest zarządzany ręcznie, rozrasta się w miarę dodawania kolejnych kroków. Wyrażony jako łańcuch, poszczególne etapy znajdują się w jednej deklaracji, a zasada, że każdy etap dostarcza dane do następnego, jest egzekwowana przez framework, a nie przez zasady dyscypliny.
Komponenty można wymieniać
Ponieważ każdy etap jest niezależną komponentą, można zastąpić jeden z nich bez wpływu na pozostałe. Załóżmy, że obecnie pipeline działa na Gemini:
chain = prompt | gemini | parser
Przejście na OpenAI oznacza zmianę tylko komponentu modelu:
chain = prompt | openai | parser
Zapytania i parserzy pozostają bez zmian. Jest to przydatne, gdy wymagania ulegają zmianie, gdy zespół porównuje dostawców pod kątem kosztu lub jakości, lub gdy chce się mieć tańszy model w fazie rozwoju, a bardziej wydajny w produkcji.
Korzyści na pierwszy rzut oka
- Mniej kodu łączącego: wyniki są automatycznie przekazywane do następnego kroku.
- Zorganizowane wieloetapowe procesy pracy: kilka zapytań, modeli i parserów znajduje się w jednym deklarowanym pipeline’u.
- Latwiejsza ewolucja: komponenty można ponownie wykorzystać lub zastąpić, w tym dostawcę modeli, bez konieczności przeprojektowania całego procesu.
Cztery wzorce kompozycji
Pomaga grupowanie pipeline’ów LangChain w cztery kategorie. Ściśle mówiąc, nie są to odrębne klasy we współczesnym LangChain; to wzory budowane z tych samych prymitywów typu Runnable. Nazwy te nadal stanowią przydatny słownictwo:
- Prosta łańcuchowa struktura: pojedyncza linia od promptu do parsera.
- Sekwencyjna łańcuchowa struktura: kilka takich linii połączonych ze sobą.
- Paralelna łańcuchowa struktura: kilka niezależnych pod-łańcuchów działających jednocześnie na tym samym wejściu.
- Warunkowa łańcuchowa struktura: punkt decyzyjny, który wybiera, który pod-łańcuch ma obsłużyć dane wejściowe.
Razem obejmują one wszystko, od pojedynczego promptu po procesy pracy, które rozgałęziają się i podejmują decyzje. Każda z nich jest omówiona poniżej przy użyciu praktycznego przykładu z dwoma dostawcami.
Proste łańcuchowe struktury: jedna prosta ścieżka
Najprostszy łańcuch połączeń łączy swoje komponenty w linii prostej. Wyjście każdego komponentu stanowi wejście dla następnego, a w trakcie nie ma żadnych odgałęzień ani decyzji:
Input
↓
Prompt Template
↓
LLM
↓
Output Parser
↓
Output
Przykład działania służy jako narzędzie pomocnicze do nauki: na podstawie określonego tematu model powinien przedstawić kilka łatwo zrozumiałych faktów, bez zagłębiania się w skomplikowane szczegóły techniczne. Łańcuch realizacji tego zadania jest jak najkrótszy:
prompt → llm → parser
Mówiąc prościej, wartość topic trafia do szablonu promptu, sformatowany prompt jest wysyłany do modelu, a odpowiedź modelu przechodzi przez parser wyjściowy przed tym, jak zostanie zwrócona. Parser jest technicznie opcjonalny, ale jego użycie zapewnia przewidywalny typ zwracanego wyniku.
W praktyce przykład używa tematu kricket i prosi o pięć ciekawych faktów. Dane przepływają w następujący sposób:
{'topic': 'cricket'}
↓
PromptTemplate
↓
LLM
↓
StrOutputParser
↓
Five interesting facts about cricket
Wersja Gemini
Oto pełny program wykorzystujący Gemini od Google poprzez integrację 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()
Buduje model GoogleGenerativeAI na bazie gemini-2.5-flash, promptu jednowymiarowego oraz StrOutputParser, a następnie łączy je za pomocą sekwencji prompt | llm | parser. Ostatni wiersz wyświetla diagram ASCII przedstawiający tę łańcuchową strukturę, który zostanie omówiony później. Należy zauważyć, że wersja ta ma skomentowany wiersz print(result), więc skrypt pokazuje jedynie diagram, chyba że zostanie ponownie włączony.
Przebieg kodu Gemini
Importy pozwalają na zaimportowanie czterech elementów niezbędnych skryptowi:
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 to otoczka wokół modelu Gemini. PromptTemplate definiuje prompt z polem {topic}, które ma zostać wypełnione podczas wykonywania. StrOutputParser przekształca wynik modelu na zwykłą ciąg znaków w Pythonie. load_dotenv odczytuje lokalny plik .env i umieszcza jego zawartość w zmiennych środowiskowych.
Zawołanie load_dotenv sprawia, że klucz API oraz wszelkie inne ustawienia z pliku .env stają się dostępne dla procesu przed utworzeniem jakiegokolwiek klienta:
load_dotenv()
Integracja z Gemini odczytuje swój klucz ze środowiska, dlatego to wywołanie musi zostać wykonane przed utworzeniem instancji modelu. Zwykłą praktyką jest przechowywanie kluczy w pliku .env (poza systemem kontroli wersji).
Następnie tworzona jest sama instancja modelu, przypisana do wersji gemini-2.5-flash:
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
Ścieżka do promptu określa zarówno tekst, jak i zmienne, których oczekuje:
prompt = PromptTemplate(
template='Generate 5 interesting facts about {topic}',
input_variables=['topic']
)
Podczas wywołania miejsce zastępcze {topic} jest wypełniane z słownika danych wejściowych. Przekazanie takich danych:
{'topic': 'cricket'}
powoduje powstanie następującego konkretnego promptu:
Generate 5 interesting facts about cricket
Parser jest tworzony bez żadnych argumentów:
parser = StrOutputParser()
Jego zadaniem jest zwrócenie odpowiedzi modelu w postaci ciągu znaków. W przypadku modelu typu completion, takiego jak GoogleGenerativeAI, który już zwraca tekst, jego działanie jest niemal identyczne z bezpośrednim przekazaniem danych. W przypadku modelu do rozmów, który zwraca obiekt wiadomości, wydobywa on treść tekstową. W obu przypadkach kolejny użytkownik otrzymuje zwykły typ str.
Następnie przychodzi linia, która definiuje cały proces przetwarzania:
chain = prompt | llm | parser
Operator rury łączy te trzy komponenty w taki sposób, że sformatowany prompt jest przekazywany modelowi, a odpowiedź modelu – parserowi:
PromptTemplate → Gemini 2.5 Flash → StrOutputParser
Na koniec łańcuch jest uruchamiany za pomocą słownika zawierającego temat:
result = chain.invoke({'topic':'cricket'})
To pojedyncze wywołanie uruchamia każdy etap po kolei, bez użycia żadnych zmiennych pośrednich w kodzie:
cricket
↓
PromptTemplate
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Final Result
Co jest zwracane
Wartość zwracana to ciąg znaków zawierający pięć faktów o krykieta, coś w tym rodzaju:
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.
Ponieważ tekst jest generowany, sformułowania i konkretne fakty różnią się przy każdym uruchomieniu, a przykładowy wynik zawiera nawet błąd pisowni ("Zeakand"). Traktuj przykładowe wyniki jako ilustracje struktury, a nie jako stałe rezultaty. Jeśli potrzebujesz deterministycznego formatowania, to zadanie należy do bardziej rygorystycznego parsera lub ustrukturyzowanego wyjścia, a nie do oczekiwania, że model będzie się powtarzał.
Ta sama ścieżka przetwarzania w OpenAI
Teraz identyczny narzędzie do pomocy w nauce, działające na OpenAI. Najważniejsze jest to, jak mało zmian nastąpiło:
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()
Modelem jest teraz ChatOpenAI z gpt-4o-mini. Prompty, parser oraz proces kompozycji są dokładnie takie same jak wcześniej, a ta wersja wyświetla zarówno wynik, jak i graf.
Jedyną nowością jest obiekt modelu:
llm = ChatOpenAI(model="gpt-4o-mini")
ChatOpenAI z biblioteki langchain_openai zastępuje GoogleGenerativeAI. Jest to model do rozmów, więc zwraca obiekt wiadomości zamiast surowej ciągu znaków, co właśnie uzasadnia konieczność zachowania parsera w łańcuchu przetwarzania.
Śablon promptu pozostał niezmieniony:
prompt = PromptTemplate(
template='Generate 5 interesting facts about {topic}',
input_variables=['topic']
)
Podobnie jest z parserem:
parser = StrOutputParser()
I tak samo z procesem kompozycji:
chain = prompt | llm | parser
Struktura jest taka sama – trzyetapowa linia z OpenAI pośrodku:
PromptTemplate → OpenAI → StrOutputParser
Wywołanie łańcucha i wydrukowanie wyniku wyglądają w ten sposób:
result = chain.invoke({'topic':'cricket'})
print(result)
Wykonanie z użyciem OpenAI zwraca również ciąg pięciu faktów o krykiecie, na przykład:
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.
Również dokładne sformułowanie będzie się różnić w zależności od wykonań i modeli.
Wydrukowanie struktury łańcucha
Oba skrypty kończą się wywołaniem, które przedstawia pipeline jako diagram tekstowy:
chain.get_graph().print_ascii()
get_graph() zwraca reprezentację graficzną obiektu Runnable, a print_ascii() rysuje ją w terminalu. Dla łańcucha Gemini otrzymujemy coś takiego:
PromptInput
*
*
PromptTemplate
*
*
GoogleGenerativeAI
*
*
StrOutputParser
*
*
StrOutputParserOutput
Węzły PromptInput i StrOutputParserOutput stanowią punkty wejścia i wyjścia pochodzące ze schematu; trzy węzły pomiędzy nimi to twoje komponenty. Jest to szybka weryfikacja, czy łańcuch jest połączony tak, jak sądzisz. W większości środowisk funkcja print_ascii() zależy od opcjonalnego pakietu grandalf, więc zainstaluj go, jeśli wywołanie powoduje błąd importu.
Tylko model się zmienił
Umieść oba pipeline obok siebie – różnica polega na jednym węźle. W przypadku Gemini:
Input
↓
PromptTemplate
↓
GoogleGenerativeAI
↓
StrOutputParser
↓
Output
W przypadku OpenAI:
Input
↓
PromptTemplate
↓
ChatOpenAI
↓
StrOutputParser
↓
Output
W kodzie cała różnica sprowadza się do tego, jaki obiekt modelu tworzysz:
# Google Gemini
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
# OpenAI
llm = ChatOpenAI(model="gpt-4o-mini")
Wszystko wokół może pozostać bez zmian. To jest praktyczna zaleta budowania na standaryzowanych komponentach: wybór dostawcy staje się decyzją konfiguracyjną, a nie koniecznością przepisywania kodu. W praktyce wciąż istnieją niewielkie różnice, na które warto zwrócić uwagę, takie jak modele do czatowania w porównaniu z modelami do uzupełniania tekstu, nazwy parametrów oraz sposób, w jaki każdy dostawca obsługuje instrukcje systemowe, ale sam proces jest przenośny.
Pełna ścieżka danych
Ogólnie rzecz biorąc, prosta łańcuchowa struktura wygląda tak:
Input
↓
PromptTemplate
↓
LLM
↓
StrOutputParser
↓
Output
Przy użyciu danych z krykieta każda wartość pośrednia staje się widoczna – od słownika po wygenerowany prompt, surową odpowiedź modelu aż po ostateczną ścieżkę tekstową:
{'topic': 'cricket'}
↓
PromptTemplate
↓
"Generate 5 interesting facts about cricket"
↓
LLM
↓
Raw LLM Response
↓
StrOutputParser
↓
Final String Output
To jest cała idea. Każdy komponent wykonuje jedną czynność, a jego wynik staje się wejściem dla następnego komponentu, w ramach jednego prostego procesu.
Łańcuchy sekwencyjne: przekazywanie wyników jednego wywołania modelu do następnego
Łańcuch sekwencyjny rozszerza tę samą ideę na kilka etapów. Prosty łańcuch zazwyczaj zawiera jeden takiego ciąg:
Prompt → LLM → Output Parser
Łańcuch sekwencyjny może zawierać kilka szablonów zapytań, kilka wywołań modelu oraz kilku parserów, wszystkie połączone ze sobą. Zasada działania pozostaje niezmienna: to, co wytwarza bieżący etap, jest tym, co przyjmuje następny etap. Nadal nie ma rozgałęzień; etapy zawsze są wykonywane w tym samym, ustalonym porządku. Wersja dwuetapowa wygląda tak:
Input
↓
Prompt Template 1
↓
LLM 1
↓
Output Parser
↓
Prompt Template 2
↓
LLM 2
↓
Output Parser
↓
Output
Przykład: szczegółowy raport, a następnie streszczenie
Zadanie przykładowe składa się z dwóch faz. Najpierw generuje się szczegółowy raport na dany temat. Następnie raport ten jest skracany do pięciopunktowego streszczenia. Druga faza nigdy nie ma bezpośredniego dostępu do oryginalnego tematu; pracuje wyłącznie na podstawie raportu stworzonego w pierwszej fazie:
Input
↓
Explanation Prompt
↓
LLM
↓
StrOutputParser
↓
Detailed Report
↓
Summary Prompt
↓
LLM
↓
StrOutputParser
↓
5-Point Summary
Dwa szablony promptów pełnią tę funkcję. explaination_prompt (błąd ortograficzny znajduje się w kodzie, więc zachowaj go przy kopiowaniu fragmentów) prosi o szczegółowy raport. summary_prompt bierze ten raport jako dane wejściowe i prosi o pięć punktów streszczających.
wersja Gemini
Oto pełny program dla 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()
Jedna instancja GoogleGenerativeAI obsługuje obie fazy, a cały proces to jedno wyrażenie składające się z sześciu komponentów: dwóch promptów, modelu dwa razy oraz parsera dwa razy.
Przebieg kodu Gemini
Importy odpowiadają prostemu łańcuchowi:
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 do obsługi modelu, PromptTemplate dla dwóch ponownie używalnych promptów, StrOutputParser do przekształcania każdej odpowiedzi w ciąg znaków oraz load_dotenv do konfiguracji.
Zmienne środowiskowe są ładowane najpierw, dokładnie tak jak wcześniej:
load_dotenv()
Model jest tworzony tylko raz:
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
Ponowne użycie jednego obiektu modelu w obu etapach jest w pełni dozwolone. Wrappers modeli w LangChain nie przechowują stanu w odniesieniu do poszczególnych wywołań, więc ta sama instancja może być referowana dowolną liczbę razy w łańcuchu.
Pierwsze polecenie przyjmuje topic i prosi o szczegółowy raport:
explaination_prompt = PromptTemplate(
template='Generate a detailed report on {topic}',
input_variables=['topic']
)
Biorąc pod uwagę ten wpis:
{"topic": "Generative AI"}
wyświetla się to jako:
Generate a detailed report on Generative AI
To w drugim poleceniu następuje ciekawe przekazanie danych:
summary_prompt = PromptTemplate(
template='Generate a 5 pointer summary from the following text \n {text}',
input_variables=['text']
)
Oczekuje zmiennej o nazwie text, ale w słowniku wejściowym nic nie ma tak nazwane. Wartość pochodzi z pierwszego etapu: to raport wygenerowany przez pierwsze wywołanie modelu wypełnia tę pozycję.
To działa dzięki parserowi znajdującemu się pomiędzy poszczególnymi etapami:
parser = StrOutputParser()
Po pierwszym wywołaniu modelu parser wytwarza zwykłą ciąg znaków. Gdy PromptTemplate z dokładnie jedną zmienną wejściową otrzymuje prosty ciąg znaków zamiast słownika, LCEL przypisuje ten ciąg do tej pojedynczej zmiennej. Dlatego raport trafia do {text} bez żadnego wyraźnego przypisania. Gdyby drugi prompt miał dwie zmienne, trzeba byłoby samodzielnie utworzyć słownik, na przykład za pomocą małego słownika Runnables lub RunnablePassthrough.assign.
Cały proces składa się z jednej linii:
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
Czytanie jej od lewej do prawej pokazuje kolejność wykonywania:
Explanation Prompt
↓
Gemini
↓
StrOutputParser
↓
Summary Prompt
↓
Gemini
↓
StrOutputParser
↓
Final Output
Każdy komponent otrzymuje wynik poprzedniego komponentu. Nic w twoim kodzie nie dotyka raportu pośredniego.
Jego uruchomienie wymaga jednego tematu:
result = chain.invoke({"topic": "Generative AI"})
Za tym wywołaniem obie fazy są wykonywane jeden po drugim:
"Generative AI"
↓
Explanation Prompt
↓
Gemini
↓
StrOutputParser
↓
Detailed Report
↓
Summary Prompt
↓
Gemini
↓
StrOutputParser
↓
Final 5-Point Summary
Co wraca
Ostateczna wartość to pięciopunktowy podsumowanie sztucznej inteligencji generatywnej, na przykład:
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.
Należy zauważyć, że to podsumowanie nie jest tworzone bezpośrednio na podstawie tematu. Gemini najpierw tworzy pełny raport, a tylko ten raport jest wysyłany do promptu służącego do tworzenia podsumowania. Taki dwuetapowy podejście zazwyczaj zapewnia drugiemu wywołaniu bogatszy materiał do skompresowania, ale kosztem dodatkowego wywołania modelu oraz związanej z tym opóźnienia i liczby tokenów. Jak zawsze, konkretne sformułowania różnią się w zależności od każdego wykonywania.
Ta sama ścieżka w OpenAI
Wariant OpenAI zachowuje niezmieniony przepływ pracy:
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()
Jest to ta sama dwuetapowa struktura: prompt do generowania raportu, model, parser, prompt do stworzenia podsumowania, model, parser. Skrypt importuje również funkcję load_dotenv dwa razy, co nie stanowi problemu, ale można to poprawić.
Tym razem model jest tworzony bez żadnych argumentów:
llm = ChatOpenAI()
Przy braku argumentu model, funkcja ChatOpenAI korzysta z domyślnego modelu biblioteki, który zmieniał się w różnych wersjach langchain_openai. W rzeczywistych projektach należy wyraźnie określić nazwę modelu, aby aktualizacja nie zmieniała w tajemnicy zachowania lub kosztów.
Prompt do generowania raportu jest identyczny jak w wersji Gemini:
explaination_prompt = PromptTemplate(
template="Generate a detailed report on {topic}",
input_variables=["topic"]
)
Tak samo jest z promptem do stworzenia podsumowania, który ponownie oczekuje raportu w formacie {text}:
summary_prompt = PromptTemplate(
template="Generate a 5 pointer summary from the following text \n {text}",
input_variables=["text"],
)
Tutaj właśnie przydaje się sekwencyjny wzorzec: drugi prompt jest automatycznie wypełniany na podstawie wyników pierwszego etapu.
Parser jest taki sam – StrOutputParser:
parser = StrOutputParser()
A składanie odbywa się znak po znaku w ten sam sposób:
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
Jako strumień, jest to następujące:
Prompt 1
↓
OpenAI
↓
StrOutputParser
↓
Prompt 2
↓
OpenAI
↓
StrOutputParser
↓
Final Output
Tylko węzeł modelu różni się od wersji Gemini.
Wywołanie przekazuje temat ponownie. Należy zauważyć, że pokazany fragment brakuje zamkniętej nawiasu; pełny program zawiera poprawną linię:
result = chain.invoke({"topic": "Generative AI"}
Następnie łańcuch uruchamia obie fazy samodzielnie.
Co wraca
Rezultatem jest ponownie pięciopunktowy podsumowanie wywiedzione z raportu pośredniego:
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.
Detale będą się różnić pomiędzy poszczególnymi uruchomieniami.
Drukowanie struktury łańcucha
Pomocnik do grafów działa również przy dłuższych łańcuchach:
chain.get_graph().print_ascii()
Dla sekwencyjnego łańcucha Gemini diagram pokazuje obie fazy ułożone pionowo:
PromptInput
*
*
PromptTemplate
*
*
GoogleGenerativeAI
*
*
StrOutputParser
*
*
PromptTemplate
*
*
GoogleGenerativeAI
*
*
StrOutputParser
*
*
StrOutputParserOutput
Węzeł modelu pojawia się dwa razy, ponieważ ten sam obiekt modelu jest używany w dwóch miejscach w łańcuchu przetwarzania, a graf odzwierciedla te pozycje, a nie unikalne obiekty.
Porównanie dwóch implementacji
Oba warianty mają tę samą sekwencyjną strukturę; jedyną rzeczywistą różnicą jest dostawca. łańcuch Gemini:
Input
↓
Explanation Prompt
↓
GoogleGenerativeAI
↓
StrOutputParser
↓
Summary Prompt
↓
GoogleGenerativeAI
↓
StrOutputParser
↓
Output
Łańcuch OpenAI:
Input
↓
Explanation Prompt
↓
ChatOpenAI
↓
StrOutputParser
↓
Summary Prompt
↓
ChatOpenAI
↓
StrOutputParser
↓
Output
Prompty, parser oraz układ łańcucha są wspólne. Jedyną zmieniającą się częścią jest konstrukcja modelu:
# Google Gemini
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
# OpenAI
llm = ChatOpenAI()
Sam złożony łańcuch pozostaje identyczny w różnych dostawcach:
chain = explaination_prompt | llm | parser | summary_prompt | llm | parser
Abstrakcyjnie mówiąc, sekwencyjny łańcuch wygląda w ten sposób:
Prompt 1
↓
LLM 1
↓
Parser
↓
Prompt 2
↓
LLM 2
↓
Parser
↓
Output
Różnica w porównaniu z prostym łańcuchem polega na tym, że istnieje wiele etapów, a każdy kolejny etap zależy od poprzedniego.
Pełna ścieżka danych
W ogólnej formie łańcuch sekwencyjny przekazuje dane w ten sposób:
Input
↓
PromptTemplate 1
↓
LLM 1
↓
StrOutputParser
↓
Detailed Report
↓
PromptTemplate 2
↓
LLM 2
↓
StrOutputParser
↓
Final Output
Śledzony przy użyciu danych wejściowych generatywnej sztucznej inteligencji, włączając renderowane prompty:
{"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
Pierwsze wezwanie modelu stanowi krok pośredni, a nie odpowiedź. Jego wynik staje się surowcem dla kolejnej operacji. Można dalej łączyć etapy, gdy proces wymaga ich więcej:
Prompt
↓
LLM
↓
Parser
↓
Prompt
↓
LLM
↓
Parser
↓
Prompt
↓
LLM
↓
Output
Niezależnie od długości łańcucha, kolejność jest stała i każdy etap czeka na poprzedni. Oznacza to również, że opóźnienia się kumulują: trzy kolejne wezwania modelu kosztują mniej więcej tyle, ile suma ich czasów odpowiedzi. Warto również dodać mechanizm ochronny przed niekontrolowanym wyjściem pośrednim, ponieważ bardzo długi raport trafia bezpośrednio do następnego promptu i zwiększa jego liczbę tokenów.
Łańcuchy równoległe: rozwijanie się na podstawie tego samego wejścia
Łańcuch równoległy uruchamia kilka pod-łańcuchów obok siebie. Podczas gdy łańcuch sekwencyjny czeka, aż każdy etap się zakończy, zanim rozpocznie następny, łańcuch równoległy uruchamia niezależne pod-łańcuchy jednocześnie:
┌──→ Sub-Chain 1 ──→ Output 1
│
Input ──────────────┼──→ Sub-Chain 2 ──→ Output 2
│
└──→ Sub-Chain 3 ──→ Output 3
↓
Combined Output
Ponieważ pod-łańcuchy są niezależne, żaden z nich nie widzi wyniku innego. Zasada działania jest prosta:
- Każdy pod-łańcuch otrzymuje ten sam dane wejściowe i wytwarza swój własny wynik.
- Krok równoległy czeka, aż wszystkie pod-łańcuchy się zakończą.
- Ich wyniki są łączone w jeden wartość kombinowaną.
Ta wartość kombinowana to słownik. Każdy klucz to nazwa nadana pod-łańcuchowi, a każda wartość to wynik tego pod-łańcucha:
{
"facts": "...",
"definition": "...",
"use_case": "..."
}
Przykład: trzy perspektywy jednego tematu
Gdy mamy temat taki jak Sztuczna inteligencja, celem jest stworzenie trzech elementów na podstawie tego samego wejścia: pięciu ciekawych faktów, zwięzłej definicji oraz przykładu zastosowania w praktyce. Żaden z tych elementów nie zależy od pozostałych, więc wykonywanie ich po kolei tylko marnowałoby czas. Zamiast tego są one wykonywane równolegle:
Input
↓
RunnableParallel
/ | \
/ | \
↓ ↓ ↓
Facts Definition Use Case
↓ ↓ ↓
LLM LLM LLM
↓ ↓ ↓
Parser Parser Parser
\ | /
\ | /
↓ ↓ ↓
Combined Result
Pracę wykonują trzy podprocesy: facts odpowiada za pięć faktów, definition za krótką definicję, a use_case za praktyczne zastosowanie. Są one łączone za pomocą RunnableParallel. Pierwsza implementacja wykorzystuje Gemini we wszystkich trzech przypadkach; druga łączy OpenAI i Anthropic oraz dodaje krok łączenia wyników.
Dlaczego wykonywać podprocesy równolegle
Istnieją dwa powody, by stosować ten wzorzec.
Pierwszym czynnikiem jest szybkość. Wywołania modeli są uzależnione od czasu przesyłania danych i generowania wyników, więc niezależne wywołania znacznie korzystają z jednoczesnego ich wykonywania. Wyobraźmy sobie trzy podprocesy, z których każdy zajmuje około dwóch sekund. Jeśli zostaną uruchomione jeden po drugim, ich łączny czas wyniesie więcej:
Chain 1 → 2 seconds
Chain 2 → 2 seconds
Chain 3 → 2 seconds
Total ≈ 6 seconds
Jeśli zostaną uruchomione jednocześnie, łączny czas będzie mniej więcej równy czasowi trwania naj wolniejszego z nich:
Chain 1 ─┐
Chain 2 ─┼→ execute together → ≈ 2 seconds
Chain 3 ─┘
Rzeczywista oszczędność zależy od ograniczeń prędkości dostawcy oraz od tego, jak równomiernie kończą się poszczególne wywołania, ale w przypadku niezależnych zadań poprawa jest zazwyczaj znaczna.
Drugi powód ma charakter strukturalny: zadania te dzielą ten sam wprowadzony materiał, ale nie potrzebują wyników pozostałych zadań. Na przykład z jednego dokumentu można chcieć stworzyć streszczenie, listę kluczowych tematów oraz zestaw pytań do nauki. Każde z tych zadań może pracować bezpośrednio z dokumentem:
Document
↓
┌───────┼───────┐
↓ ↓ ↓
Summary Topics Questions
Streszczenie nie musi być powiązane z wyodrębnianiem tematu ani generatorem pytań. Gdy zadania są ze sobą powiązane w ten sposób, przymuszanie ich do sekwencji powoduje opóźnienia i łączy kroki, które nie mają rzeczywistej zależności. Gdy faktycznie od siebie zależą, paralelizm jest niewłaściwym narzędziem, a sekwencyjna łańcuchowa struktura stanowi dokładny opis sytuacji.
wersja Gemini
Oto trójstronna struktura działania w 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()
Trzy szablony promptów, jeden model oraz jeden parser są połączone w trzy małe łańcuchy, które RunnableParallel uruchamia jednocześnie na danych wejściowych związanych z Sztuczną Inteligencją. Kilka linii z poleceniem print jest skomentowanych; ich aktywacja pokazuje typ wyniku oraz poszczególne pola.
Przebieg kodu Gemini
Importy dodają jeden nowy komponent:
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
Oprócz otoczki modelu, PromptTemplate i StrOutputParser, skrypt importuje RunnableParallel z modułu langchain_core.runnables, który odpowiada za jednoczesne uruchamianie podłańcuchów.
Tworzony jest jeden model Gemini 2.5 Flash:
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
Wszystkie trzy podłańcuchy go używają. Dzielenie się jednym obiektem modelu pomiędzy równoległymi gałęziami jest bezpieczne; każde wezwanie jest niezależne.
Dzielony jest również jeden parser:
parser = StrOutputParser()
Pierwsze polecenie prosi o pięć krótkich faktów:
fact_prompt = PromptTemplate(
input_variables=["topic"],
template="Give five interesting short facts about: {topic}"
)
Drugie prosi o zwięzłą definicję:
definition_prompt = PromptTemplate(
input_variables=["topic"],
template="Give a concise definition of: {topic}"
)
Trzecie prosi o przykład zastosowania w praktyce:
usecase_prompt = PromptTemplate(
input_variables=["topic"],
template="What is a real-world use case of: {topic}?"
)
Wszystkie trzy polecenia używają tego samego nazwy zmiennej:
{topic}
To wspólne nazwanie umożliwia użycie jednego słownika wejściowego do zaopatrywania wszystkich gałęzi. Same zadania pozostają całkowicie oddzielone.
A teraz kluczowy element – łańcuch równoległy:
parallel_chain = RunnableParallel({
"facts": fact_prompt | llm | parser,
"definition": definition_prompt | llm | parser,
"use_case": usecase_prompt | llm | parser
})
RunnableParallel przyjmuje słownik, którego wartościami są obiekty typu Runnable. Klucze tego słownika stają się kluczami wyników:
facts
definition
use_case
Każda wartość to osobny, kompletny mały łańcuch:
fact_prompt → LLM → parser
definition_prompt → LLM → parser
usecase_prompt → LLM → parser
Ponieważ żadna z nich nie polega na innej, wszystkie mogą być wykonywane jednocześnie. Przy użyciu synchronicznego invoke LangChain uruchamia gałęzie w puli wątków; przy użyciu ainvoke działają one jako równoległe zadania asynchroniczne.
Łańcuch jest wywoływany raz:
result = parallel_chain.invoke({"topic": "Artificial Intelligence"})
z jednym wejściem:
{"topic": "Artificial Intelligence"}
Ten słownik jest przekazywany każdej gałęzi. Koncepcyjnie:
Artificial Intelligence
↓
RunnableParallel
/ | \
↓ ↓ ↓
Facts Definition Use Case
↓ ↓ ↓
LLM LLM LLM
↓ ↓ ↓
Parser Parser Parser
\ | /
\ | /
Combined Result
Co wraca
Wynikiem jest słownik z jednym wpisem na każdą pod-łańcuchową strukturę:
{
"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."
}
Tekst w każdym polu będzie się różnił przy kolejnych uruchomieniach, ale jego struktura pozostaje stała, i to właśnie od niej zależy działanie twojego kodu:
result["facts"]
result["definition"]
result["use_case"]
Każdy klucz zwraca wynik odpowiadającej mu gałęzi, dzięki czemu kod poniżej może wybrać dokładnie ten fragment, który mu potrzebny.
Mieszanie dostawców i łączenie wyników
Druga implementacja pokazuje dwa dodatkowe rozwiązania: różne gałęzie mogą korzystać z różnych dostawców, a krok równoległy może sam stanowić jedną z faz większego sekwencyjnego procesu. W tym przypadku jedna gałąź używa OpenAI, druga – Anthropic, a ostateczne wezwanie do OpenAI łączy oba wyniki w jeden dokument:
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()
Wpis to tekst o maszynach wektorów wspierających. Jeden z gałęzi tworzy na jego podstawie notatki do nauki, druga – test składający się z pięciu pytań, a proces łączenia łączy te notatki i test w jeden dokument.
Dwa szczegóły w tym opisie wymagają uwagi. Został tu importowany RunnableParallel z langchain.schema.runnable, co jest starszą ścieżką z czasów przed podziałem na langchain_core; w nowszych wersjach dostępna jest ona pod adresem langchain_core.runnables, jak w przykładzie z Gemini, a starsza ścieżka może nie istnieć w nowszych wersjach. Ponadto użyty został identyfikator modelu claude-3-7-sonnet-20250219, który jest przestarzały; dostawcy z czasem usuwają stare wersje modeli, dlatego przed ponownym użyciem należy sprawdzić aktualną listę modeli Anthropic.
Stworzone zostają dwa obiekty modelu:
model1 = ChatOpenAI()
model2 = ChatAnthropic(model_name='claude-3-7-sonnet-20250219')
ChatOpenAI zapisuje notatki, a następnie przeprowadza łączenie. ChatAnthropic tworzy testy. W RunnableParallel nie ma żadnego wymogu, aby każda gałąź używała tego samego modelu, co jest przydatne, gdy jeden dostawca jest lepszy lub tańszy do wykonania określonego zadania.
Zapytanie dotyczące notatek przekształca tekst wejściowy w krótkie, proste notatki:
prompt1 = PromptTemplate(
template='Generate short and simple notes from the following text \n {text}',
input_variables=['text']
)
Zapytanie dotyczące testów generuje pięć krótkich par pytań i odpowiedzi na podstawie tego samego tekstu:
prompt2 = PromptTemplate(
template='Generate 5 short question answers from the following text \n {text}',
input_variables=['text']
)
Żadne z zadań nie wymaga wyników drugiego, więc stanowią one naturalne gałęzie równoległe.
Trzecie zapytanie jest inne. Jest wykonywane po rozgałęzieniach i wymaga obu wyników:
prompt3 = PromptTemplate(
template='Merge the provided notes and quiz into a single document \n notes -> {notes} and quiz -> {quiz}',
input_variables=['notes', 'quiz']
)
Deklaruje dwie zmienne wejściowe:
notes
quiz
Te nazwy nie są arbitralne. Muszą odpowiadać kluczom kroku równoległego, ponieważ słownik wygenerowany przez rozgałęzienie jest dokładnie tym, co otrzymuje to polecenie.
Krok równoległy definiuje dwa gałęzie:
parallel_chain = RunnableParallel({
'notes': prompt1 | model1 | parser,
'quiz': prompt2 | model2 | parser
})
Jego struktura to podział dwukierunkowy, który łączy się w słownik:
Input Text
↓
RunnableParallel
/ \
↓ ↓
Notes Chain Quiz Chain
↓ ↓
OpenAI Anthropic
↓ ↓
Parser Parser
\ /
↓ ↓
Dictionary
Koncepcyjnie wartość pośrednia wygląda w ten sposób:
{
"notes": "...",
"quiz": "..."
}
Łańcuch łączenia to zwykły, prosty łańcuch:
merge_chain = prompt3 | model1 | parser
Wykorzystuje słownik z kroku równoległego, wypełnia {notes} i {quiz}, a następnie prosi OpenAI o stworzenie jednego dokumentu.
Ostatecznie oba etapy łączy się za pomocą kolejnego rurykowania:
chain = parallel_chain | merge_chain
Rezultatem jest hybrydowy przepływ pracy: równoległy na początku, sekwencyjny na końcu:
Input Text
↓
RunnableParallel
/ \
↓ ↓
Notes Quiz
↓ ↓
OpenAI Anthropic
↓ ↓
Parser Parser
\ /
↓ ↓
Combined Dictionary
↓
Merge Prompt
↓
OpenAI
↓
Parser
↓
Final Output
Kluczową cechą zachowania jest to, że obie gałęzie działają niezależnie, ale etap łączenia nie może rozpocząć się, dopóki obie nie zakończą swojej pracy. Dlatego naj wolniejsza gałąź określa tempo całego procesu rozgałęziania. Jeśli jedna z gałęzi zawiedzie, krok równoległy zostanie przerwany i łączenie nigdy nie nastąpi, więc w takich przypadkach należy otoczyć niestabilne dostawcy mechanizmami ponownych prób lub rozwiązaniami awaryjnymi.
Celowy proces jest uruchamiany za pomocą następującego tekstu:
result = chain.invoke({'text': text})
Gałęzie z adnotacjami i quizem przetwarzają tekst jednocześnie. Gdy obie zakończą pracę, ich wyniki trafiają do łańcucha łączenia, a jego rezultat w postaci ciągu znaków jest tym, co zwraca funkcja invoke.
Drukowanie struktury łańcucha
Pomocnik do tworzenia grafów wizualizuje również strukturę rozgałęzienia:
parallel_chain.get_graph().print_ascii()
W przypadku łańcucha Gemini składającego się z trzech gałęzi diagram pokazuje moment rozdzielenia i połączenia:
Parallel<...>Input
/ | \
/ | \
PromptTemplate PromptTemplate PromptTemplate
↓ ↓ ↓
LLM LLM LLM
↓ ↓ ↓
StrOutputParser StrOutputParser StrOutputParser
\ | /
\ | /
Parallel<...>Output
Węzły Parallel<...>Input i Parallel<...>Output wskazują miejsce, gdzie dane wejściowe są rozdzielane oraz gdzie wyniki są ponownie gromadzone w słowniku.
Porównanie obu implementacji
Oba warianty ilustrują tę samą ideę: niezależne podłańcuchy działające jednocześnie. Wersja Gemini to czysta struktura rozgałęzienia z trzema gałęziami w ramach jednego modelu:
Input
↓
RunnableParallel
├── Facts → Gemini → Parser
├── Definition → Gemini → Parser
└── Use Case → Gemini → Parser
↓
Dictionary Output
Wersja z mieszanymi dostawcami dzieli się na dwie gałęzie korzystające z różnych dostawców, a następnie dodaje etap łączenia:
Input
↓
RunnableParallel
├── Notes → OpenAI → Parser
└── Quiz → Anthropic → Parser
↓
Dictionary Output
↓
Merge Chain
↓
OpenAI
↓
Output
U podstaw obu leży ta sama struktura:
RunnableParallel({
"chain_1": chain_1,
"chain_2": chain_2,
"chain_3": chain_3
})
Klucze określają podłańcuchy; wartości to same te podłańcuchy. Jako uproszczenie, gdy zwykły słownik obiektów typu Runnable pojawia się wewnątrz wyrażenia typu pipe, LCEL automatycznie przekształca go na RunnableParallel, więc często zobaczysz taki słownik bez wyraźnej deklaracji klasy.
Pełna ścieżka danych
W ogólnej formie, równoległy łańcuch rozdzielają jeden wejście i gromadzą słownik:
Input
↓
RunnableParallel
/ | \
↓ ↓ ↓
Sub-chain Sub-chain Sub-chain
↓ ↓ ↓
Output Output Output
\ | /
\ | /
↓ ↓ ↓
Dictionary
Przykład sztucznej inteligencji na Gemini:
{"topic": "Artificial Intelligence"}
↓
RunnableParallel
/ | \
↓ ↓ ↓
Facts Definition Use Case
↓ ↓ ↓
Gemini Gemini Gemini
↓ ↓ ↓
Parser Parser Parser
\ | /
\ | /
Dictionary
Ostateczna wartość ma po jednym kluczu na każdą gałąź:
{
"facts": "...",
"definition": "...",
"use_case": "..."
}
To jest istotna różnica w porównaniu z łańcuchem sekwencyjnym. W łańcuchu sekwencyjnym każdy etap wykorzystuje wynik poprzedniego etapu:
A → B → C → D
W łańcuchu równoległym niezależne etapy działają jednocześnie na tym samym wejściu, a ich wyniki są łączone:
┌→ A ─┐
Input ─┼→ B ─┼→ Combined Output
└→ C ─┘
Wykorzystuj paralelizm do zadań, które dzielą się danymi wejściowymi, ale nie wynikami, a zachowaj sekwencyjną kompozycję dla zadań, które rzeczywiście opierają się na wcześniejszych krokach. Mieszanie tych dwóch podejść, jak to robi przykład z notatkami i quizem, jest powszechne i w pełni naturalne.
Łańcuchy warunkowe: kierowanie danymi wejściowymi do odpowiedniego podłańcucha
Łańcuch warunkowy wybiera następny krok w czasie wykonania. Łańcuch sekwencyjny zawsze wykonywa każdy etap w ustalonej kolejności, natomiast łańcuch paralelny wykonywa wszystkie gałęzie jednocześnie. Łańcuch warunkowy ocenia warunek dotyczący danych wejściowych i uruchamia tylko tę gałąź, która się z nim zgadza:
Input
↓
Condition Check
↓
┌───────────────┐
│ │
Yes No
↓ ↓
Chain 1 Chain 2
↓ ↓
Output 1 Output 2
\ /
\ /
Final Output
W LangChain to jest reprezentowane za pomocą RunnableBranch. Podaje się mu listę par, w których każda łączy warunek z obiektem Runnable, plus domyślny obiekt Runnable na końcu. Warunki są sprawdzane w kolejności, a pierwszy, który zwróci True, decyduje o tym, który obiekt Runnable ma zostać uruchomiony; jeśli żaden warunek się nie spełnia, obsłużony zostaje domyślny obiekt.
Przykład: kierowanie według długości tekstu
Przykład traktuje tekst inaczej w zależności od jego długości:
- Jeśli tekst zawiera więcej niż 50 słów, generuje się krótkie streszczenie.
- w przeciwnym razie rozwija się pomysł zawarty w tekście.
Otrzymany proces pracy:
Input Text
↓
Condition Check
↓
Is word count > 50?
↓
┌───────┴───────┐
Yes No
↓ ↓
Summary Extend
Chain Chain
↓ ↓
LLM LLM
↓ ↓
Parser Parser
\ /
\ /
Final Output
Dwa prompty sterują gałęziami: summary_prompt prosi o krótkie podsumowanie, a extend_prompt każe modelowi rozwijać tę ideę. Mała funkcja is_long_text() liczy słowa, a RunnableBranch wykorzystuje jej wynik do wyboru odpowiedniej ścieżki. Podobnie jak w innych przypadkach, istnieje wersja dla Gemini oraz wersja dla OpenAI.
W uogólnionej formie dwukierunkowy pipeline warunkowy wygląda w ten sposób:
User Input
↓
Condition Check
↓
┌───────┴───────┐
Yes No
↓ ↓
Prompt 1 Prompt 2
↓ ↓
LLM LLM
↓ ↓
Parser Parser
\ /
└──────┬───────┘
↓
Final Output
Dla danego wejścia uruchamia się dokładnie jedna gałąź.
Analiza konkretnego wejścia
Przykładowe wejście to jednozdaniowa definicja uczenia maszynowego, opisująca je jako gałąź sztucznej inteligencji, w której metody statystyczne umożliwiają komputerom uczenie się na podstawie danych.
To zdanie ma 22 słowa, co jest znacznie poniżej progu 50 słów, więc warunek zwraca False, w wyniku czego wybierana jest ścieżka rozszerzania:
Input Text
↓
Condition Check
↓
22 > 50?
↓
No
↓
Extend Chain
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Extended Text
Gdy podstawi się fragment dłuższy niż 50 słów, warunek zmienia się na True, co skutkuje przesłaniem danych do łańcucha podsumowawczego.
Kiedy pomocne jest warunkowe routowanie
Routowanie jest przydatne wtedy, gdy ten sam rodzaj danych wymaga różnego potraktowania w zależności od określonej cechy:
- Długość: skracać długie teksty, rozwijać krótkie teksty.
- Nastroje: przesyłać pozytywne i negatywne opinie do różnych łańcuchów odpowiedzi.
- Rodzaj żądania: kierować każdą kategorię żądania do specjalistycznego łańcucha.
Ważna jest kwestia, że nie trzeba płacić za gałęzie, których się nie potrzebuje. Uruchamia się tylko wybrana ścieżka:
Input
↓
Condition
↓
Choose one branch
↓
Execute selected chain
↓
Output
Dzięki temu proces pracy staje się adaptacyjny w sposób niemożliwy dla stałej sekwencji, a także oszczędza wywołania modelu w porównaniu z uruchamianiem każdej gałęzi i odrzucaniem wyników.
Wersja Gemini
Oto pełny program routingu w 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()
Definiuje dwa krótkie łańcuchy, summary_chain i extend_chain, oraz predykat liczący słowa, a następnie pozwala RunnableBranch wybrać jeden z łańcuchów na podstawie tego predykatu.
Przebieg kodu Gemini
Modelem jest ponownie Gemini 2.5 Flash:
llm = GoogleGenerativeAI(model="gemini-2.5-flash")
StrOutputParser przekształca każdą odpowiedź w ciąg znaków:
parser = StrOutputParser()
Pytanie dotyczące streszczenia prosi o krótkie podsumowanie:
summary_prompt = PromptTemplate(
input_variables=["text"],
template="Summarize the following text briefly:\n\n{text}"
)
Pytanie typu „extend” jest używane na drugiej ścieżce:
extend_prompt = PromptTemplate(
input_variables=["text"],
template="Extend the idea in the following long text:\n\n{text}"
)
Przyjrzyj się uważnie sformułowaniu: mowa jest o „poniższym długim tekście”, a jednak ta gałąź działa tylko przy krótkich danych wejściowych. Ta niezgodność niczego nie psuje, ale dostarcza modelowi mylącego sygnału. Przekształcenie tego na coś w rodzaju „Rozwiń ideę zawartą w poniższym tekście” to niewielka poprawka, którą warto wprowadzić we własnej wersji.
Każde polecenie jest łączone z modelem i parserem w osobną łańcuchową strukturę:
summary_chain = summary_prompt | llm | parser
extend_chain = extend_prompt | llm | parser
To daje dwa niezależne ścieżki:
Summary Prompt → Gemini → Parser
Extend Prompt → Gemini → Parser
Krok warunkowy decyduje, która z nich zostanie uruchomiona.
Warunek to zwykła funkcja w Pythonie:
def is_long_text(input: dict) -> bool:
return len(input["text"].split()) > 50
Dzieli wartość input[„text”] na części za pomocą przestrzeni międzyznakowej i liczy je. Jego funkcja jest prosta:
True → if word count > 50
False → otherwise
Rozdzielanie tekstu na podstawie przerw między znakami to przybliżona metoda liczenia słów; znaki interpunkcyjne pozostają przy słowach, a wyrażenia połączone myślnikiem są liczone jako jedno. To wystarcza do routingu według długości, ale jeśli naprawdę zależy ci na ograniczeniach modelu, licz tokeny zamiast słów. Zwróć również uwagę, że nazwa parametru input przykrywa wbudowaną funkcję w Pythonie; działa to, ale nazwa taka jak data jest bardziej przejrzysta.
Gdy wszystko jest już przygotowane, RunnableBranch składa router:
conditional_chain = RunnableBranch(
(lambda x: is_long_text(x), summary_chain),
extend_chain
)
Pierwszy argument to tuple, który stanowi regułę:
If is_long_text(x) is True
↓
Execute summary_chain
Ostatni argument, który nie jest umieszczony w tuple, stanowi wartość domyślną:
Otherwise
↓
Execute extend_chain
Opakowanie lambda nie jest konieczne. RunnableBranch akceptuje każdą funkcję wywołalną jako warunek, więc bezpośrednie przekazanie is_long_text również działa równie dobrze. W rezultacie routing wygląda w ten sposób:
Input
↓
is_long_text()
/ \
True False
↓ ↓
summary_chain extend_chain
↓ ↓
Gemini Gemini
↓ ↓
Parser Parser
\ /
\ /
Output
Aby je uruchomić, potrzebny jest słownik wejściowy:
result = conditional_chain.invoke(text_input)
Najpierw sprawdza się warunek, a tylko wybrana gałąź wykonuje wywołanie modelu.
Co jest zwracane
Z przykładowym wejściem składającym się z 22 słów predykat zwraca:
is_long_text() → False
zatem extend_chain obsługuje dane wejściowe. Koncepcyjnie:
Input Text
↓
Word Count Check
↓
False
↓
extend_chain
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Extended Text
Jeśli podać tekst dłuższy niż 50 słów, predykat zwraca:
is_long_text() → True
co zmienia ścieżkę na:
Input Text
↓
Word Count Check
↓
True
↓
summary_chain
↓
Gemini 2.5 Flash
↓
StrOutputParser
↓
Short Summary
Jeden pipeline, dwa możliwe szlaki, wybrane przez same dane wejściowe.
Ten sam router w OpenAI
Wersja OpenAI zachowuje logikę routingu i zmienia tylko model:
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()
Dwie łańcuchy, jeden predykat, jedna RunnableBranch – tak samo jak w przypadku Gemini.
Modelem jest gpt-4o-mini:
llm = ChatOpenAI(model="gpt-4o-mini")
Dwa szablony promptów pozostają niezmienione – jeden służy do streszczania, a drugi do rozszerzania:
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}"
)
Łańcuchy są budowane dokładnie tak samo jak wcześniej:
summary_chain = summary_prompt | llm | parser
extend_chain = extend_prompt | llm | parser
Teraz obie ścieżki przechodzą przez OpenAI:
Summary Prompt → OpenAI → StrOutputParser
Extend Prompt → OpenAI → StrOutputParser
Warunek to ta sama kontrola składająca się z 50 słów:
def is_long_text(input: dict) -> bool:
return len(input["text"].split()) > 50
Router jest budowany w ten sam sposób:
conditional_chain = RunnableBranch(
(lambda x: is_long_text(x), summary_chain),
extend_chain
)
Wynik True wysyła dane do summary_chain; wszystko inne trafia do extend_chain.
Łańcuch jest uruchamiany, a wynik wyświetlany:
result = conditional_chain.invoke(text_input)
print(result)
Dla przykładowego zdania warunek ponownie daje wynik:
is_long_text() → False
dlatego łańcuch rozszerzający jest uruchamiany i zwraca rozszerzone wyjaśnienie danych, podobne do tego:
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.
Dokładny tekst zależy od modelu i będzie się zmieniać przy kolejnych uruchomieniach.
Drukowanie struktury łańcucha
Ten sam narzędzie do tworzenia grafów działa również w przypadku routerów:
conditional_chain.get_graph().print_ascii()
Wyróżniającym elementem wygenerowanego diagramu jest węzeł Branch. Jest to punkt decyzyjny, w którym LangChain ocenia warunki i przekazuje dane do jednego z możliwych łańcuchów.
Jak różnią się te dwa gałęzie
Każda z gałęzi ma swoje zadanie. Gałąź podsumowująca:
summary_prompt
↓
Gemini
↓
StrOutputParser
↓
Short Summary
Gałąź rozszerzająca:
extend_prompt
↓
Gemini
↓
StrOutputParser
↓
Extended Text
Warunek decyduje, która z nich zostanie wybrana:
Input
↓
is_long_text()
/ \
True False
↓ ↓
Summary Chain Extend Chain
To jest kluczowa różnica w porównaniu z łańcuchem równoległym: obie gałęzie nie są wykonywane jednocześnie. Wykonuje się tylko tę ścieżkę, której warunek jest spełniony, co sprawia, że koszt i opóźnienie są proporcjonalne do rzeczywiście potrzebnej pracy.
Routerowanie na podstawie własnej klasyfikacji modelu
Warunki nie ograniczają się tylko do prostych kryteriów, takich jak liczba słów. Powszechnym i skuteczniejszym podejściem jest pozwolenie pierwszemu łańcuchowi na sklasyfikowanie danych wejściowych, a następnie kierowanie ich zgodnie z tą klasyfikacją. W przykładzie dotyczącym nastroju, otrzymane opinie są oznaczane jedną z dwóch wartości:
positive
negative
a RunnableBranch przekazuje je do łańcucha odpowiedzi odpowiadającego tej etykiecie:
Feedback
↓
Sentiment Classifier
↓
┌───────────────┐
│ │
Positive Negative
│ │
↓ ↓
Positive Negative
Response Response
│ │
└───────┬───────┘
↓
Output
Skutecznym sposobem na stworzenie klasyfikatora jest sprawienie, by model zwracał ustrukturyzowany wynik zweryfikowany przez model Pydantic – na przykład pole ograniczone do wartości positive lub negative. Warunki w poszczególnych gałęziach sprawdzają wtedy to pole zamiast analizować tekst w formie swobodnej, co zapobiega nieoczekiwaniom, gdy model odpowiada „Positive.” z wielką literą i kropką. Zawsze należy zachować rozsądnie dobraną domyślną gałąź, ponieważ klasyfikator w końcu może zwrócić coś, czego się nie spodziewaliśmy.
To pokazuje, że warunek może zależeć od informacji wygenerowanych przez wcześniejszy łańcuch, a nie tylko od surowych danych wejściowych. W przypadku bardziej złożonego routingu aktualna dokumentacja LangChain opisuje również routowanie za pomocą funkcji niestandardowej umieszczonej w RunnableLambda, która zwraca łańcuch do uruchomienia; warto sprawdzić, jakie podejście zalecają dokumenty dla danej wersji.
Pełna ścieżka danych
Ogólny wzorzec warunkowy jest następujący:
Input
↓
Condition Check
↓
┌───────┴───────┐
↓ ↓
Condition Condition
True False
↓ ↓
Chain 1 Chain 2
↓ ↓
LLM LLM
↓ ↓
Parser Parser
\ /
\ /
└─────┬───────┘
↓
Output
W przykładzie dotyczącym długości tekstu:
Machine Learning Text
↓
is_long_text()
↓
Word Count > 50?
/ \
Yes No
↓ ↓
Summary Extend
Chain Chain
↓ ↓
Gemini Gemini
↓ ↓
Parser Parser
\ /
\ /
Output
Krótkie dane wejściowe kierowane są jedną ścieżką:
Word Count ≤ 50
↓
Extend Chain
Długie dane wejściowe kierowane są drugą ścieżką:
Word Count > 50
↓
Summary Chain
Patrząc szerzej, oto te cztery kształty obok siebie:
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
Łańcuch warunkowy daje procesowi LLM możliwość podejmowania decyzji, uruchamiając różną logikę w zależności od danych wejściowych lub wartości obliczonej wcześniej w łańcuchu przetwarzania.
Wskazówki praktyczne przed wysyłką
Podczas przechodzenia od przykładów do rzeczywistego kodu warto pamiętać o kilku często występujących problemach:
- Zadaj nazwy modeli. Konstruktory bez argumentu modelu dziedziczą domyślne wartości używane przez zainstalowaną integrację, a identyfikatory o określonej dacie są z czasem usuwane.
- Zachowaj parser. Modele do czatowania zwracają obiekty wiadomości;
StrOutputParserzapewnia każdemu kolejnemu kroku przewidywalną wartość typustr. - Zwracaj uwagę na przekazywanie danych. Pusta strona tekstowa wypełnia automatycznie prompt tylko wtedy, gdy prompt zawiera dokładnie jedną zmienną. W innych przypadkach należy wyraźnie utworzyć słownik.
- Dopasuj klucze do zmiennych. Podczas rozgałęziania, po którym następuje łączenie, klucze muszą odpowiadać nazwom zmiennych w promptzie używanym do łączenia.
langchain_core.runnables, a nie ze starszych ścieżek z langchain.schema.get_graph().print_ascii() (przy zainstalowanym grandalf) to tani sposób na potwierdzenie, że złożona łańcuchowa struktura jest połączona zgodnie z zamysłem.Aby zapoznać się z bardziej szczegółowym przeglądem, w którym te łańcuchy są przedstawione obok mechanizmów wyszukiwania, narzędzi oraz agentów działających na lokalnych modelach, zobacz LangChain 1.x w praktyce: łańcuchy, RAG, narzędzia i agenci lokalnie.
Główne wnioski
- Łańcuch to zadeklarowana ścieżka przetwarzania, w której wyjście każdego komponentu trafia do następnego; operator
|buduje go z dowolnych obiektów typu Runnable. - Proste łańcuchy obejmują jeden prosty etap przetwarzania; łańcuchy sekwencyjne składają się z kilku etapów, co powoduje wzrost opóźnienia przy każdym kolejnym kroku.
RunnableParalleluruchamia niezależne pod-łańcuchy równocześnie i zwraca słownik oparty na nazwach gałęzi, który późniejsze etapy mogą wykorzystywać podając nazwę zmiennej.RunnableBranchocenia warunki w kolejności i uruchamia tylko pierwszą pasującą gałąź, przechodząc w razie potrzeby do wartości domyślnej.- Warunki mogą być prostymi sprawdzeniami danych wejściowych lub decyzjami opartymi na uporządkowanej klasyfikacji wcześniejszego modelu.
- Ponieważ model składa się z jednego komponentu, zmiana między Gemini, OpenAI lub Anthropic zazwyczaj dotyczy tylko jednej linii kodu, podczas gdy prompty, parserzy i struktura pozostają bez zmian.
Wszystkie cztery wzorce opierają się na tej samej podstawie – interfejsie Runnable, który jest tym, co pozwala na łączenie promptów, modeli, parserów oraz własnych funkcji. Zrozumienie tego interfejsu bezpośrednio jest naturalnym następnym krokiem, gdy te struktury staną się już znajome.