Belajar LangChain - LCEL: Rantai & Komposisi

#Pendahuluan

Di episode 4, kalian menyempurnakan seni menyusun prompt dan menangani streaming modern. Episode 5 adalah episode yang paling menentukan dari seluruh fase dasar: LCEL. Semua yang kalian pelajari sejauh ini — primitif, chat models, prompt — akan dirangkai di sini menjadi pipeline yang benar-benar berguna.

Konsep inti LCEL sudah kalian kenal dari episode 2: setiap komponen adalah Runnable, dan runnable disambung dengan operator pipa |. Sekarang kita bongkar tuntas kelima runnable utama, seluruh lifecycle eksekusi, lalu cara aplikasi bertahan dari error.

#Runnable dan RunnableSequence

RunnableSequence adalah tulang punggung LCEL — rantai yang mengeksekusi langkah demi langkah. Operator pipa | adalah gula sintaksis untuk membangunnya:

Rantai sekuensial paling umum di LangChain

from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Kamu adalah penerjemah Inggris-Indonesia."),
    ("human", "Terjemahkan: {teks}"),
])
 
model = ChatOpenAI(model="gpt-4o-mini")
 
rantai = prompt | model | StrOutputParser()
 
print(rantai.invoke({"teks": "Hello, how are you today?"}))

Alur data: input dict masuk ke prompt, hasilnya diteruskan ke model, lalu jawaban AI diubah menjadi string polos oleh StrOutputParser. Ekspresi ini juga bisa ditulis eksplisit sebagai RunnableSequence — hasilnya sama, tapi bentuk pipa jauh lebih terbaca. Setiap elemen dalam rantai adalah runnable, dan rantai itu sendiri juga runnable, sehingga ia bisa disambung ke rantai lain.

#RunnableParallel dan RunnablePassthrough

Tidak semua alur lurus. RunnableParallel menjalankan beberapa cabang sekaligus atas input yang sama, lalu mengumpulkan hasilnya dalam satu dict:

Cabang paralel atas satu input

from langchain_core.runnables import RunnableParallel, RunnablePassthrough
 
penerjemah = prompt | model | StrOutputParser()
 
cabang = RunnableParallel(
    terjemahan=penerjemah,
    asli=RunnablePassthrough(),
)
 
hasil = cabang.invoke({"teks": "Good morning!"})
print(hasil["terjemahan"])
print(hasil["asli"])

Di sini RunnablePassthrough memainkan peran penting: ia meneruskan input tanpa diubah. Berguna untuk menyisipkan data asli ke dalam hasil akhir, atau menambahkan konteks di tengah alur. Kedua cabang di atas berjalan secara paralel — jika penerjemahan lambat, bagian passthrough tetap selesai seketika.

#RunnableLambda dan RunnableMap

Rantai tidak selalu hanya berisi komponen LangChain. RunnableLambda membungkus fungsi Python biasa menjadi runnable, sehingga fungsi apa pun — sanitasi, validasi, panggilan ke layanan lain — bisa ikut dalam pipeline:

Menyisipkan fungsi Python dengan RunnableLambda

from langchain_core.runnables import RunnableLambda
 
def bersihkan_teks(teks: str) -> str:
    return teks.strip().lower()
 
def sapaan(nama: str) -> str:
    return f"Halo {nama}, selamat datang di pipeline!"
 
langkah = RunnableLambda(bersihkan_teks) | RunnableLambda(sapaan)
print(langkah.invoke("  BUDI  "))

Setiap fungsi menerima output dari langkah sebelumnya dan meneruskannya ke langkah berikutnya. Satu-satunya aturan: fungsi harus callable dengan satu argumen utama (kecuali jika didekorasi khusus untuk konfigurasi tambahan).

Selain RunnableLambda, ada RunnableMap — versi dict dari paralelisme: kalian mendefinisikan nama kunci dan fungsinya, lalu hasil setiap fungsi dikumpulkan ke kunci yang sama:

RunnableMap menjalankan fungsi dan mengumpulkan hasil

from langchain_core.runnables import RunnableMap
 
analisa = RunnableMap({
    "panjang": lambda teks: len(teks),
    "kata": lambda teks: teks.split(),
})
 
print(analisa.invoke("LCEL itu menarik"))

Hasilnya dict dengan dua kunci: panjang berisi jumlah karakter, kata berisi daftar kata. Pola ini sangat cocok untuk pre-processing yang mengubah satu input menjadi banyak nilai yang nanti dipakai rantai.

#Lifecycle: invoke, batch, stream

Semua runnable berbagi tiga mode eksekusi yang sama, dan rantai yang kalian bangun tidak terkecuali — ini yang membuat LCEL begitu konsisten:

Satu rantai, tiga mode eksekusi

jawaban_tunggal = rantai.invoke({"teks": "Hello!"})
 
banyak_jawaban = rantai.batch([
    {"teks": "Hello!"},
    {"teks": "Thank you!"},
    {"teks": "Goodbye!"},
])
 
for potongan in rantai.stream({"teks": "Hello!"}):
    print(potongan, end="", flush=True)

Ketiganya memiliki pasangan async — ainvoke, abatch, astream — yang dipakai saat aplikasi berjalan di server. Pola penting: karena rantai itu sendiri runnable, kalian bisa menyusun rantai berisi rantai lain (komposisi bersarang), dan setiap lapisan tetap mendukung ketiga mode ini.

#Event Streaming

Selain mode eksekusi, LCEL mengekspos event stream — catatan detail per langkah dalam rantai. Saat sebuah rantai dijalankan dengan astream_events, kalian melihat setiap event: saat prompt mulai disusun, saat model mulai mengalir, saat parser selesai, lengkap dengan waktu dan urutannya. Ini fondasi dari observability dan kemajuan antarmuka pengguna:

Mengamati event per langkah rantai

async def amati():
    async for event in rantai.astream_events(
        {"teks": "Hello!"}, version="v3"
    ):
        if event["event"] == "on_chat_model_stream":
            print("Model mengalir:", event["data"]["chunk"].content, end="")
 
asyncio.run(amati())

Dengan pola ini kalian bisa menampilkan indikator "model sedang mengetik" di antarmuka, atau mencatat event untuk debugging. Di episode 19 kita akan melihat versi penuhnya lewat LangSmith, yang menangkap event-stream ini secara otomatis.

#Error Handling dan Retry

Aplikasi produksi harus tahan terhadap kegagalan — dan ini bagian yang paling sering terlewat. Dua mekanisme utama di LCEL: retry untuk kegagalan yang bisa diulang (misalnya rate limit sesaat), dan fallback untuk kegagalan yang perlu dialihkan ke cadangan.

Retry otomatis dengan backoff diaktifkan lewat with_retry:

Mengaktifkan retry dengan aturan tersendiri

from langchain_core.runnables import Runnable
 
rantai_tangguh = rantai.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
)

Fallback memakai with_fallbacks — jika model utama gagal, rantai otomatis mencoba alternatif:

Fallback ke model cadangan

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
 
rantai_utama = prompt | ChatOpenAI(model="gpt-4o") | StrOutputParser()
rantai_cadangan = prompt | ChatAnthropic(model="claude-3-5-sonnet-latest") | StrOutputParser()
 
rantai_aman = rantai_utama.with_fallbacks([rantai_cadangan])

Di luar itu, error tetap bisa ditangkap biasa dengan try/except — misalnya untuk validasi input sebelum masuk rantai, atau untuk mencatat kegagalan ke sistem monitoring. Kombinasi retry, fallback, dan penanganan manual membuat aplikasi kalian jauh lebih siap menghadapi penyedia model yang kadang tidak stabil.

#Penutup

Episode 5 menuntaskan dasar LCEL: membangun RunnableSequence dengan operator pipa, memparalelkan cabang dengan RunnableParallel, meneruskan data dengan RunnablePassthrough, membungkus fungsi Python dengan RunnableLambda dan RunnableMap, menjalankan lifecycle invoke/batch/stream dengan varian async, memantau event streaming, serta mengamankan aplikasi dengan retry dan fallback.

Inti yang harus dibawa pulang:

• prompt | model | parser membangun RunnableSequence; rantai itu sendiri tetap runnable.
• RunnableParallel menjalankan cabang bersamaan; RunnablePassthrough meneruskan input tanpa ubah.
• RunnableLambda menyisipkan fungsi Python; RunnableMap mengumpulkan hasil fungsi ke dict.
• Setiap runnable mendukung invoke/batch/stream dan varian async-nya secara konsisten.
• Lindungi aplikasi dengan with_retry untuk percobaan ulang dan with_fallbacks untuk cadangan model.

Di episode 6 berikutnya kita akan belajar mengendalikan bentuk output model: output parsers & structured output — StrOutputParser, JsonOutputParser, PydanticOutputParser, with_structured_output berbasis JSON schema, streaming parsing, hingga hasil terstruktur bertipe Pydantic. Sampai jumpa!