AI Engineering Academy · Lekcja

Tworzenie łańcuchów za pomocą LCEL

Tworzenie łańcuchów za pomocą LCEL to bezpłatna lekcja AI Engineering Academy na CoddyKit. To lekcja 2 z 4. Możesz przeczytać całą lekcję poniżej za darmo — a potem ćwiczyć ją interaktywnie w przeglądarce z wbudowanym edytorem kodu i tutorem AI dostępnym 24/7. To część ścieżki edukacyjnej AI Engineering Academy, a Twój postęp synchronizuje się między webem a aplikacją CoddyKit. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Czym jest LCEL?

LCEL (LangChain Expression Language) to deklaratywny sposób łączenia komponentów LangChain w potoki. Zamiast tworzyć rozwlekłe klasy Chain, łączy się obiekty Runnable za pomocą operatora |. Łańcuchy LCEL automatycznie obsługują strumieniowanie, operacje asynchroniczne, przetwarzanie wsadowe i mechanizmy awaryjne — wszystkie te możliwości są dostępne bez dodatkowej pracy, wystarczy użyć składni potoku.

Pierwszy łańcuch LCEL

Najprostszy łańcuch LCEL łączy PromptTemplate, ChatModel i OutputParser. Każdy komponent jest obiektem Runnable, a operator potoku łączy je ze sobą. Po wywołaniu chain.invoke() dane wejściowe przechodzą kolejno przez każdy krok, a wynik każdego kroku staje się danymi wejściowymi następnego.

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template(
    'Write a product description for: {product}'
)
model = ChatOpenAI(model='gpt-4o-mini', temperature=0.7)
parser = StrOutputParser()

chain = prompt | model | parser

result = chain.invoke({'product': 'noise-cancelling headphones'})
print(result)

Sprawdzanie elementów wewnętrznych łańcucha

Łańcuchy LCEL w przejrzysty sposób udostępniają swoją strukturę. chain.steps wyświetla komponenty w kolejności ich wykonywania. Za pomocą chain.input_schema i chain.output_schema można sprawdzić typy danych przepływających do łańcucha i z niego wychodzących. Wywołanie invoke() przyjmuje również parametr config, który umożliwia przekazanie nazw uruchomień, tagów i callbacków do pojedynczego wywołania bez zmieniania definicji łańcucha.

# Inspect chain structure
print('Steps:', chain.steps)
print('Input schema:', chain.input_schema.schema())
print('Output schema:', chain.output_schema.schema())

# Named run for observability
result = chain.invoke(
    {'product': 'coffee maker'},
    config={'run_name': 'product-desc-run', 'tags': ['production']}
)

Synchroniczne a asynchroniczne wywołanie

Łańcuchy LCEL udostępniają synchroniczne i asynchroniczne wersje wszystkich metod. invoke() blokuje wątek wywołujący. ainvoke() jest odpowiednikiem asynchronicznym i należy go używać w handlerach FastAPI oraz innych kontekstach asynchronicznych, aby nie blokować pętli zdarzeń. Wersja asynchroniczna ma identyczne działanie — różnica w interfejsie sprowadza się do użycia await.

import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

chain = ChatPromptTemplate.from_template('Explain {topic}') | ChatOpenAI() | StrOutputParser()

async def main():
    # Async invocation
    result = await chain.ainvoke({'topic': 'quantum computing'})
    print(result)

asyncio.run(main())

Strumieniowanie tokenów przez łańcuch

Łańcuchy LCEL przekazują strumień przez wszystkie swoje elementy. Wywołaj chain.stream(), aby odbierać tokeny w miarę ich generowania przez model. StrOutputParser obsługuje strumieniowanie i przekazuje pojedyncze fragmenty tekstu, zamiast czekać na pełną odpowiedź. Dzięki temu użytkownicy otrzymują natychmiastową informację zwrotną, a nie długą pauzę zakończoną kompletną odpowiedzią.

chain = (
    ChatPromptTemplate.from_template('Tell me about {topic} in detail.')
    | ChatOpenAI(model='gpt-4o-mini')
    | StrOutputParser()
)

# Print each token as it arrives
for chunk in chain.stream({'topic': 'neural networks'}):
    print(chunk, end='', flush=True)
print()  # newline at end

# Async streaming
async def stream_async():
    async for chunk in chain.astream({'topic': 'transformers'}):
        print(chunk, end='', flush=True)

Przetwarzanie wsadowe wielu danych wejściowych

chain.batch() przyjmuje listę słowników danych wejściowych i przetwarza je z konfigurowalnym poziomem współbieżności. Wewnętrznie LangChain wykonuje wywołania w równoległych wątkach, dzięki czemu przetwarzanie wsadowe jest znacznie szybsze niż wywoływanie invoke() w pętli. Parametr max_concurrency ogranicza liczbę równoległych wywołań API, pomagając nie przekroczyć limitów częstotliwości. Wyniki są zwracane w tej samej kolejności co dane wejściowe.

products = [
    {'product': 'laptop'},
    {'product': 'tablet'},
    {'product': 'smartwatch'},
    {'product': 'earbuds'},
]

# Process all 4 concurrently
results = chain.batch(products, config={'max_concurrency': 4})

for product, description in zip(products, results):
    print(f'{product["product"]}: {description[:80]}...')

RunnablePassthrough i RunnableLambda

RunnablePassthrough przekazuje dane wejściowe bez zmian — jest przydatny do przekazywania oryginalnego kontekstu razem z przetworzonymi danymi. RunnableLambda opakowuje dowolną funkcję języka Python jako obiekt Runnable, umożliwiając wstawienie dowolnej logiki do łańcucha LCEL. Razem pozwalają modyfikować dane między krokami łańcucha bez pisania pełnej niestandardowej klasy Runnable.

from langchain_core.runnables import RunnablePassthrough, RunnableLambda

def word_count(text: str) -> str:
    count = len(text.split())
    return f'{text}\n\n[Word count: {count}]'

chain = (
    ChatPromptTemplate.from_template('Write about {topic}')
    | ChatOpenAI(model='gpt-4o-mini')
    | StrOutputParser()
    | RunnableLambda(word_count)  # Add word count to output
)

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

Przekazywanie dodatkowego kontekstu za pomocą RunnablePassthrough

Typowym wzorcem w RAG jest przekazanie do końcowego promptu zarówno pobranych dokumentów, jak i pytania użytkownika. Użyj RunnablePassthrough.assign(), aby dodać obliczone pola do słownika danych wejściowych bez utraty jego oryginalnych kluczy. Wzbogaca to obiekt kontekstu przepływający przez łańcuch, dzięki czemu szablon promptu ma dostęp do wszystkich danych.

from langchain_core.runnables import RunnablePassthrough

# Fake retriever for illustration
def retrieve_docs(question: str):
    return ['Doc 1 content', 'Doc 2 content']

retriever = RunnableLambda(retrieve_docs)

rag_chain = (
    RunnablePassthrough.assign(context=retriever)  # adds 'context' key
    | ChatPromptTemplate.from_template(
        'Context: {context}\n\nAnswer: {question}'
    )
    | ChatOpenAI()
    | StrOutputParser()
)

result = rag_chain.invoke({'question': 'What is machine learning?'})

Dodawanie mechanizmów awaryjnych do łańcuchów

Łańcuchy LCEL obsługują mechanizmy awaryjne — alternatywne łańcuchy uruchamiane, gdy główny łańcuch zgłosi wyjątek. Wywołaj chain.with_fallbacks([backup_chain]), aby zarejestrować jeden lub więcej łańcuchów awaryjnych. LangChain najpierw uruchamia główny łańcuch, przechwytuje ewentualny wyjątek, a następnie wypróbowuje kolejne łańcuchy awaryjne w podanej kolejności. Jest to standardowy sposób łagodnego obsługiwania awarii dostawcy lub błędów modelu.

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

primary_chain = (
    ChatPromptTemplate.from_template('Answer: {question}')
    | ChatOpenAI(model='gpt-4o')
    | StrOutputParser()
)

fallback_chain = (
    ChatPromptTemplate.from_template('Answer: {question}')
    | ChatAnthropic(model='claude-3-haiku-20240307')
    | StrOutputParser()
)

robust_chain = primary_chain.with_fallbacks([fallback_chain])
# If OpenAI fails, automatically tries Anthropic

Debugowanie za pomocą intermediate_steps

Debugowanie łańcuchów LCEL jest łatwiejsze dzięki konfiguracji invoke() i opcji verbose=True. Można również użyć .with_config({'verbose': True}) na dowolnym łańcuchu, aby logować dane wejściowe i wyjściowe każdego kroku. Do debugowania strukturalnego można globalnie opakować łańcuch za pomocą set_verbose(True) lub użyć śledzenia LangSmith, aby rejestrować każdą wartość pośrednią w wizualnym eksploratorze śladów.

from langchain.globals import set_verbose, set_debug

# Verbose: logs LLM inputs/outputs
set_verbose(True)

# Debug: logs everything including prompts, parsers, and runnables
set_debug(True)

result = chain.invoke({'product': 'smart watch'})

# Or per-chain verbose config
chain_debug = chain.with_config({'verbose': True})
result = chain_debug.invoke({'product': 'laptop'})

Łańcuchy konfigurowalne w czasie działania

Łańcuchy LCEL obsługują konfigurację w czasie działania za pomocą ConfigurableField. Można udostępnić parametry modelu — temperaturę, nazwę modelu i max_tokens — jako opcje konfigurowalne, bez wpisywania ich na stałe w kodzie. W czasie wywołania należy przekazać konfigurację Configurable, aby zastąpić wartości domyślne. Dzięki temu jeden łańcuch może obsługiwać wielu użytkowników z różnymi ustawieniami lub służyć do testów A/B konfiguracji modeli.

from langchain_core.runnables import ConfigurableField

llm = ChatOpenAI(model='gpt-4o-mini').configurable_fields(
    temperature=ConfigurableField(
        id='temperature',
        name='LLM Temperature',
        description='Controls output creativity'
    )
)

chain = ChatPromptTemplate.from_template('Write about {topic}') | llm | StrOutputParser()

# Low temperature for factual output
factual = chain.invoke(
    {'topic': 'history of Rome'},
    config={'configurable': {'temperature': 0.0}}
)

Szybki sprawdzian

Sprawdź swoją wiedzę na temat tworzenia łańcuchów za pomocą LCEL.

Podsumowanie lekcji

W tej lekcji poznał Pan/Pani: kompozycję potoków LCEL, która łączy obiekty Runnable za pomocą | w przejrzyste, czytelne potoki; RunnablePassthrough i RunnableLambda, które umożliwiają dodawanie kroków modyfikowania danych bez opuszczania paradygmatu LCEL; oraz mechanizmy awaryjne, dzięki którym łańcuchy są odporne na błędy, ponieważ automatycznie wypróbowują zapasowe łańcuchy. W następnej części omówimy rozgałęzianie i równoległe łańcuchy na potrzeby bardziej złożonego routingu.

Często zadawane pytania

Czy lekcja „Tworzenie łańcuchów za pomocą LCEL” jest bezpłatna?

Tak — pełny tekst „Tworzenie łańcuchów za pomocą LCEL” jest dostępny za darmo tutaj w sieci. Aby ćwiczyć ją interaktywnie (wbudowany edytor kodu i tutor AI dostępny 24/7) i odblokować resztę kursu AI Engineering Academy, przejdź na CoddyKit PRO. Kurs AI Engineering Academy zawiera 4 lekcji w sumie.

Co nauczysz się w „Tworzenie łańcuchów za pomocą LCEL”?

Uczestnicy użyją operatora potoku do połączenia szablonu promptu, LLM-a i parsera danych wyjściowych w sekwencję Runnable, wywołają ją synchronicznie i asynchronicznie oraz przeanalizują dane pośredn… Ćwiczysz AI Engineering Academy z praktycznym kodem, który uruchamiasz bezpośrednio w przeglądarce, a tutor AI dostępny 24/7 odpowiada na Twoje pytania podczas pracy nad lekcją.

Czy potrzebuję doświadczenia, aby zacząć AI Engineering Academy?

Nie wymagamy żadnego doświadczenia. AI Engineering Academy w CoddyKit jest strukturyzowany dla początkujących i zaawansowanych użytkowników, więc możesz zacząć tutaj lub od początku i uczyć się w swoim tempie. To lekcja 2 z 4.

Ile czasu zajmuje lekcja „Tworzenie łańcuchów za pomocą LCEL”?

Większość lekcji CoddyKit trwa około 5–10 minut. Każda lekcja to mały, interaktywny krok, dzięki czemu robisz systematyczne postępy i zawsze wracasz dokładnie do tego samego miejsca — na webie i w aplikacji.

Czy mogę pisać i uruchamiać kod w tej lekcji AI Engineering Academy?

Tak. Każda lekcja AI Engineering Academy zawiera wbudowany edytor kodu, więc piszesz i uruchamiasz prawdziwy kod bezpośrednio w przeglądarce i od razu otrzymujesz sprzężenie zwrotne od AI — bez konfiguracji na komputerze.

Wszystkie lekcje w tym kursie

Architektura LangChain i podstawowe abstrakcje
Tworzenie łańcuchów za pomocą LCEL
Rozgałęzianie i równoległe łańcuchy
Strumieniowanie danych wyjściowych w LangChain

← Powrót do AI Engineering Academy