6. Pipeline Deklaratif dengan LCEL

Di bab-bab sebelumnya, kita telah menulis kode imperatif untuk mengorkestrasi interaksi LLM: membuat prompt, memanggil model, mem-parse respons. Ini berfungsi, tetapi seiring aplikasi AI menjadi lebih kompleks, pendekatan ini menjadi verbose dan lebih sulit dipelihara. Kamu akan berakhir dengan pemanggilan fungsi yang bersarang dalam, penanganan error manual di setiap langkah, dan kesulitan memahami alur data secara keseluruhan.

LangChain Expression Language (LCEL) menyelesaikan ini dengan memungkinkan kamu mendeklarasikan apa yang ingin terjadi, bukan bagaimana membuatnya terjadi. Alih-alih menulis kode prosedural yang memanggil fungsi secara berurutan, kamu menyusun komponen menggunakan operator pipe sederhana (|) yang terbaca seperti pipeline Unix. Hasilnya adalah kode yang lebih bersih, lebih mudah dipelihara yang dengan jelas mengekspresikan alur data melalui sistem AI kamu.

Bab ini memperkenalkan LCEL untuk membangun alur kerja linear - urutan operasi di mana data mengalir dari awal hingga akhir tanpa percabangan atau loop. Kita akan membahas kapan menggunakan LCEL, cara menyusun pipeline, dan cara mengeksekusinya baik secara sinkron maupun dengan output streaming.

6.1) Mengapa LCEL?

Masalah yang Diselesaikan LCEL

Mari mulai dengan contoh konkret. Misalkan kamu membangun asisten dukungan pelanggan yang perlu:

Mengambil pertanyaan pengguna
Memformatnya menjadi prompt dengan instruksi sistem dan pesan pengguna
Mengirimkannya ke LLM
Mem-parse respons untuk mengekstrak hanya konten teks

Berikut cara kamu mungkin menulis ini secara imperatif (tanpa LCEL):

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
def answer_question(question: str) -> str:
    # Langkah 1: Format prompt dengan pertanyaan
    messages = prompt.invoke({"question": question})
    
    # Langkah 2: Panggil LLM
    response = llm.invoke(messages)
    
    # Langkah 3: Parse output untuk mengekstrak konten teks
    result = parser.invoke(response)
    
    return result
 
# Gunakan
result = answer_question("How do I reset my password?")
print(result)

Output:

Untuk mereset password Anda, silakan ikuti langkah-langkah berikut:
1. Buka halaman login
2. Klik "Lupa Password"
3. Masukkan alamat email Anda
4. Periksa email Anda untuk link reset
5. Ikuti link tersebut dan buat password baru

Ini berfungsi, tetapi perhatikan masalahnya:

Verbosity: Setiap langkah memerlukan penugasan variabel eksplisit dan pemanggilan fungsi. Logika sebenarnya (format → invoke → parse) terkubur dalam boilerplate.

Struktur Kaku: Jika kamu ingin menambahkan langkah (seperti memvalidasi pertanyaan atau mencatat respons), kamu perlu menyisipkan kode di tengah fungsi, meningkatkan kompleksitas.

Tidak Ada Streaming Bawaan: Untuk streaming token saat tiba, kamu perlu menulis ulang seluruh fungsi untuk menggunakan llm.stream() dan menangani iterasi async secara manual.

Alur Data Tidak Jelas: Membaca kode, tidak langsung jelas bahwa ini adalah pipeline sederhana. Kamu harus melacak melalui penugasan variabel untuk memahami alurnya.

Sekarang mari lihat versi LCEL:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# Definisikan komponen (sama seperti sebelumnya)
llm = ChatOpenAI(model="gpt-4o-mini")
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful customer support assistant."),
    ("user", "{question}")
])
 
parser = StrOutputParser()
 
# Susun mereka menjadi chain menggunakan operator pipe
chain = prompt | llm | parser
 
# Gunakan
result = chain.invoke({"question": "How do I reset my password?"})
print(result)

Output:

Untuk mereset password Anda, silakan ikuti langkah-langkah berikut:
1. Buka halaman login
2. Klik "Lupa Password"
3. Masukkan alamat email Anda
4. Periksa email Anda untuk link reset
5. Ikuti link tersebut dan buat password baru

Outputnya identik, tetapi kodenya sangat berbeda:

Deklaratif: chain = prompt | llm | parser mengekspresikan seluruh alur dalam satu baris. Baca dari kiri-ke-kanan: prompt → LLM → parser.

Dapat Disusun: Setiap komponen (prompt, llm, parser) independen dan dapat digunakan kembali. Kamu dapat menukar komponen tanpa menulis ulang pipeline.

Streaming Bawaan: Beralih antara chain.invoke() dan chain.stream() tanpa mengubah definisi pipeline.

Intent Jelas: Operator | membuat alur data jelas sekilas.

Di mana LCEL cocok: Alur kerja linear

LCEL dirancang untuk alur kerja linear - urutan di mana data mengalir dalam satu arah dari awal hingga akhir tanpa percabangan atau loop.

Pola linear ini mencakup banyak aplikasi AI. Pertimbangkan sistem Q&A dokumen: kamu menerima pertanyaan → mengambil dokumen relevan → memformat prompt → mengirim ke LLM → mem-parse jawaban. Setiap langkah adalah urutan yang jelas di mana output dari satu langkah menjadi input untuk langkah berikutnya.

Tetapi bagaimana jika alur kerja kamu perlu:

Membuat LLM memutuskan tool mana yang akan dipanggil berdasarkan pertanyaan
Memanggil tool, mengamati hasilnya, kemudian memutuskan apa yang harus dilakukan selanjutnya
Mencoba ulang operasi yang gagal dengan pendekatan berbeda

Skenario ini memerlukan loop dan percabangan kondisional - hal-hal yang tidak dapat ditangani LCEL. Inilah mengapa LangGraph ada (kita akan memperkenalkannya di Bab 15).

Berikut perbandingan visual dari dua pendekatan:

LCEL sempurna untuk pipeline langsung di mana setiap langkah memproses output dari langkah sebelumnya. Data mengalir dalam satu arah saja.

LangGraph untuk ketika kamu memerlukan pengambilan keputusan dan loop. Agen dapat bertindak, mengamati hasil, dan berpikir lagi.

Untuk bab ini, kita fokus pada LCEL. Mengapa menguasai pipeline linear terlebih dahulu?

Fondasi: Operator pipe (|) adalah sintaks inti LangChain. Memahami ini membuat segalanya lebih mudah.
Prasyarat untuk LangGraph: Agen LangGraph menggunakan chain LCEL secara ekstensif di dalam node mereka
Pola dunia nyata: Agen kompleks dibangun dengan menggabungkan chain LCEL

Sisa bab ini akan menunjukkan cara membangun pipeline linear ini dengan operator pipe (|).

6.2) Menyusun Pipeline dengan Operator |

Cara Kerja Pipe LCEL

Operator pipe LCEL (|) berfungsi karena interface Runnable.

Apa itu Runnable?

Runnable adalah interface standar LangChain. Ketika komponen mengimplementasikan interface Runnable, ia dapat di-chain dengan komponen lain menggunakan operator |.

Setiap Runnable menyediakan metode-metode ini:

.invoke(input) - Eksekusi sekali dan dapatkan hasil lengkap
.stream(input) - Eksekusi dan terima setiap kata saat LLM menghasilkannya
.batch(inputs) - Eksekusi beberapa kali dengan input berbeda dan dapatkan semua hasil

Untuk bab ini, kita akan fokus pada .invoke() dan .stream() (kita akan membahas .batch() nanti saat diperlukan).

Mengapa | berfungsi?

Karena class Runnable mengimplementasikan operator | menggunakan operator overloading Python. Ketika kamu menulis prompt | llm, ia membuat Runnable baru yang mengeksekusi dua komponen secara berurutan.

Sebagian besar komponen LangChain adalah Runnable

Inilah mengapa kamu dapat me-chain begitu banyak komponen berbeda:

ChatPromptTemplate adalah Runnable
ChatOpenAI adalah Runnable
StrOutputParser adalah Runnable
Bahkan chain kustom yang kamu buat dengan | sendiri adalah Runnable!

Ini berarti kamu dapat membangun pipeline kompleks dengan menggabungkan yang lebih sederhana.

Menghubungkan Komponen: Tipe Input/Output

Saat menghubungkan komponen dengan |, kamu perlu memastikan tipe output dari satu komponen cocok dengan tipe input dari yang berikutnya.

Signature komponen kunci:

Komponen	Tipe Input	Tipe Output
`ChatPromptTemplate`	`dict`	`list[BaseMessage]`
`ChatOpenAI` (LLM)	`list[BaseMessage]`	`AIMessage`
`StrOutputParser`	`AIMessage`	`str`

Contoh alur:

python

chain = prompt | llm | parser

Berikut cara tipe data bertransformasi saat melewati setiap komponen:

prompt: Mengambil dict sebagai input dan mengonversi ke list[BaseMessage]
llm: Mengambil list[BaseMessage] sebagai input dan mengonversi ke AIMessage
parser: Mengambil AIMessage sebagai input dan mengonversi ke str

Mari lihat ini dalam aksi:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
parser = StrOutputParser()
 
chain = prompt | llm | parser
 
result = chain.invoke({"question": "What is 2+2?"})
print(result)  # "2+2 sama dengan 4."

Apa yang terjadi di setiap langkah:

Langkah	Input	Komponen	Output
1	`- dict -` `{"question": "What is 2+2?"}`	→ prompt →	`- list[BaseMessage] -` `[SystemMessage(...), HumanMessage(...)]`
2	`- list[BaseMessage] -` `[SystemMessage(...), HumanMessage(...)]`	→ llm →	`- AIMessage -` `AIMessage(content="2+2 sama dengan 4.")`
3	`- AIMessage -` `AIMessage(content="2+2 sama dengan 4.")`	→ parser →	`- str -` `"2+2 sama dengan 4."`

Apa yang terjadi jika tipe tidak cocok?

Jika kamu mencoba menghubungkan komponen yang tidak kompatibel, kamu akan mendapat error:

python

# ERROR: Ini tidak akan berfungsi
chain = llm | prompt  # LLM menghasilkan AIMessage, tetapi prompt memerlukan dict sebagai input

Pesan error akan memberi tahu kamu tipe input apa yang diharapkan komponen berikutnya vs. tipe input apa yang sebenarnya diterima.

6.3) Mengeksekusi Chain: .invoke() dan .stream()

Menjalankan Pipeline

Setelah kamu menyusun chain, kamu mengeksekusinya menggunakan metode .invoke(). Ini adalah cara sinkron untuk menjalankan pipeline - ia menunggu seluruh respons sebelum mengembalikan hasil.

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Invokasi sinkron
result = chain.invoke({"question": "What is 2+2?"})
print(result)

Output:

2+2 sama dengan 4.

Metode .invoke() sangat sederhana: masukkan parameter input yang diharapkan komponen pertama, dan dapatkan kembali output yang dihasilkan oleh komponen terakhir.

Streaming Token dari Chain yang Sama

Meskipun .invoke() sederhana, ia memiliki keterbatasan untuk aplikasi yang menghadap pengguna: pengguna tidak melihat apa pun sampai seluruh respons selesai. Untuk respons panjang (10-20 detik), ini menciptakan pengalaman pengguna yang buruk.

Streaming menampilkan token segera saat dihasilkan, daripada menunggu respons lengkap. Ini adalah efek mengetik yang kamu lihat di ChatGPT.

Mari lihat streaming dalam aksi:

python

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
 
llm = ChatOpenAI(model="gpt-4o-mini")
chain = prompt | llm | StrOutputParser()
 
# Stream chunk saat tiba
for chunk in chain.stream({"question": "Explain what Python is in one sentence"}):
    print(chunk, end="", flush=True)

Output (ditampilkan secara real-time, token demi token):

Python adalah bahasa pemrograman tingkat tinggi yang serbaguna, dikenal karena kesederhanaan dan keterbacaannya, banyak digunakan dalam pengembangan web, data science, dan otomasi.

Perhatikan bahwa definisi chain identik dengan contoh .invoke() di atas. Kita tidak perlu membangunnya kembali - kita hanya memanggil .stream() alih-alih .invoke().

Metode .stream() mengembalikan hasil sebagai chunk saat LLM menghasilkan token. print(chunk, end="", flush=True) menampilkan setiap chunk segera di layar, menciptakan efek mengetik langsung.

Kapan Menggunakan .invoke() vs .stream()

Gunakan .invoke() ketika:

Kamu hanya memerlukan hasil akhir (analisis, terjemahan, klasifikasi)
Responsnya pendek dan waktu tunggu bukan masalah
Kamu memerlukan output lengkap sebelum melanjutkan ke langkah berikutnya

Gunakan .stream() ketika:

Pengguna perlu melihat progres (chat, pembuatan konten)
Responsnya panjang dan waktu tunggu akan terasa
Kamu membangun UI di mana feedback real-time penting

Kedua metode bekerja pada chain yang sama. Definisikan chain sekali, kemudian pilih mode eksekusi berdasarkan kebutuhan kamu.

Di bab ini, kamu mempelajari LCEL - sintaks pipeline deklaratif LangChain:

Membangun pipeline dengan operator pipe: prompt | llm | parser
Eksekusi fleksibel: Gunakan .invoke() untuk hasil lengkap atau .stream() untuk output real-time
Type safety: Cocokkan tipe output dengan tipe input saat me-chain komponen

Chain yang sama berfungsi untuk kedua mode eksekusi - definisikan sekali, gunakan di mana saja.

Selanjutnya: Bab 7 membahas output terstruktur dengan Pydantic, memungkinkan kamu mengekstrak data JSON yang tervalidasi dari respons LLM.