Belajar LangChain - Performance & Cost Optimization

#Pendahuluan

Di episode 17 kalian mendalami LangGraph lanjutan: typed state, reducers, checkpointing, time travel, dan subgraphs lengkap dengan persistence lewat postgres. Arsitektur kalian sekarang sudah kokoh secara state. Episode 18 ini melengkapi dari sisi operasional: kecepatan dan biaya. Kita bedah cara menurunkan latensi lewat streaming, menaikkan throughput lewat batching dan RunnableParallel, menumpuk konkurensi lewat async, lalu menekan biaya dengan pelacakan token, prompt caching, model tiering, dan caching layer.

#Mulai dari Pengukuran: Token Usage Tracking

Optimasi tanpa data hanyalah tebak-tebakan. Kabar baiknya, setiap respons model membawa metadata penggunaan token lewat atribut usage_metadata di objek respons — berisi input_tokens, output_tokens, dan total_tokens. Kebiasaan pertama yang harus kalian bentuk adalah selalu mencatatnya.

token_usage.py

from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
response = model.invoke("Ringkas konsep RAG dalam satu kalimat")
 
print(response.content)
print(response.usage_metadata)

output_usage.txt

{'input_tokens': 18, 'output_tokens': 21,
 'total_tokens': 39, 'input_token_details': {...}}

Simpan metadata ini di log terstruktur atau database, lalu agregasi per fitur, per pengguna, dan per model. Dengan begitu kalian tahu fitur mana yang paling boros sebelum tagihan datang mengejutkan.

#Streaming: Memangkas Latency to First Token

Latensi yang dirasakan pengguna bukan total waktu eksekusi, melainkan waktu sampai token pertama muncul di layar. Dengan stream, model mengeluarkan token begitu memproduksinya, sehingga UI bisa menampilkan jawaban secara progresif dan terasa responsif.

streaming.py

from langchain_core.output_parsers import StrOutputParser
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini", streaming=True)
chain = model | StrOutputParser()
 
for chunk in chain.stream("Tulis esai singkat tentang LangChain"):
    print(chunk, end="", flush=True)

Aktifkan streaming=True saat inisialisasi model, dan ujung chain boleh memakai parser yang toleran streaming seperti StrOutputParser. Untuk versi asinkron, astream memberi hasil yang sama tanpa memblokir event loop. Catatan penting: streaming tidak mempercepat token terakhir — ia hanya membuat pengalaman terasa lebih cepat. Waktu total sering tetap sama.

#Batching dan Parallelism: Meningkatkan Throughput

Ketika menghadapi banyak input independen, jangan panggil model satu per satu. Dua senjata LCEL yang wajib dikuasai: batch/abatch untuk memproses kumpulan input sekaligus, dan RunnableParallel untuk menjalankan beberapa langkah bersamaan di dalam satu chain.

parallelism.py

from langchain_core.runnables import RunnableParallel
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
 
chain = RunnableParallel(
    ringkasan=model,
    keywords=model,
)
hasil = chain.invoke({
    "input": "Machine learning adalah subbidang AI yang fokus pada data."
})

batch memiliki parameter max_concurrency untuk mengatur seberapa banyak request paralel yang diizinkan, sehingga kalian tidak membanjiri provider sekaligus menjaga rate limit. Gabungkan keduanya: parallelism di tingkat pipeline, batching di tingkat model. Untuk volume masif, masukkan antrean di luar aplikasi — misalnya task queue — supaya lonjakan request tidak merusak stabilitas layanan.

#Async: Throughput dalam Satu Server

Model LLM menghabiskan sebagian besar waktunya menunggu I/O — respons HTTP dari provider. Di lingkungan async, ainvoke dan astream membiarkan event loop menjalankan pekerjaan lain sambil menunggu, sehingga satu worker bisa mengelola ratusan request konkuren tanpa menambah instance.

async_throughput.py

import asyncio
 
from langchain_openai import ChatOpenAI
 
model = ChatOpenAI(model="gpt-5-mini")
pertanyaan = ["Apa itu vector store?"] * 20
 
async def main():
    hasil = await asyncio.gather(*(model.ainvoke(q) for q in pertanyaan))
    return hasil
 
asyncio.run(main())

Pola ini bekerja karena gather menjadwalkan semua panggilan sekaligus dan menunggu semuanya selesai. Server FastAPI yang async bisa memakai astream untuk mengirim token ke client sambil tetap melayani request lain di event loop yang sama — kombinasi ideal untuk aplikasi chat produksi.

#Prompt Caching: Bayar Sekali untuk Input yang Sama

Sebagian besar biaya LLM berasal dari input token yang berulang: sistem prompt, instruksi panjang, dan konteks RAG yang sama dijawab berulang kali. Prompt caching membuat provider mengenali prefix yang identik dan menagih jauh lebih murah untuk pemanggilan berikutnya.

prompt_caching.py

from langchain_core.prompts import ChatPromptTemplate
 
prompt = ChatPromptTemplate.from_messages([
    ("system", "Kamu adalah asisten pakar hukum Indonesia."),
    ("human", "{question}"),
])

Agar caching bekerja, prefix sistem harus persis sama antar request. Jangan sisipkan data dinamis di awal prompt — tempatkan bagian yang berubah seperti pertanyaan pengguna di paling akhir. Mulai versi terbaru beberapa provider menerapkan caching otomatis; tugas kalian hanyalah menjaga pola prompt tetap stabil dan deterministik.

#Model Tiering dan Caching Layer

Model tiering berarti memakai model murah untuk tugas mudah dan model mahal untuk tugas sulit. Contoh paling sederhana: sebuah router menentukan apakah pertanyaan butuh reasoning mendalam.

tiering.py

from langchain_core.runnables import RunnableLambda
from langchain_openai import ChatOpenAI
 
def pilih_model(question: str) -> str:
    return "gpt-5-mini" if len(question) < 80 else "gpt-5"
 
model = ChatOpenAI(model="gpt-5-mini")
chain = RunnableLambda(pilih_model) | model

Lapisan terakhir adalah caching layer: simpan jawaban untuk pertanyaan yang sudah dijawab sehingga pemanggilan berikutnya tidak perlu ke model sama sekali. Mulai dari cache sederhana di memori, lalu naik ke Redis untuk skala multi-instance.

simple_cache.py

from langchain_core.caches import InMemoryCache
from langchain.globals import set_llm_cache
 
set_llm_cache(InMemoryCache())

Untuk kebutuhan memori jangka panjang, ekosistem seperti mem0 menyediakan memory layer yang menyimpan fakta dan otomatis menarik riwayat relevan ke prompt — lebih mahal daripada cache sederhana, tapi menambah konteks yang dipersonalisasi. Pasang dependency-nya lewat pip install mem0ai lalu sesuaikan dengan arsitektur kalian.

#Penutup

Episode 18 menutup sisi performa dan biaya: streaming untuk latensi yang terasa cepat, batch dan RunnableParallel untuk throughput, async untuk konkurensi tinggi, pelacakan usage_metadata sebagai basis pengukuran, prompt caching untuk input berulang, model tiering untuk routing tugas, dan caching layer untuk memotong panggilan yang tidak perlu.

Inti yang harus dibawa pulang:

• Ukur dulu: catat usage_metadata setiap request sebelum mengoptimasi apa pun.
• Streaming memperbaiki pengalaman pengguna, batching dan parallelism memperbaiki throughput.
• Async memungkinkan satu server melayani ratusan request konkuren dengan sedikit instance.
• Prompt caching menekan biaya input berulang, selama prefix prompt stabil.
• Model tiering dan caching layer memangkas biaya tanpa mengorbankan kualitas jawaban.

Di episode 19 berikutnya kita pasang alat pemantau: LangSmith untuk tracing, dataset, dan evaluasi otomatis — supaya setiap optimasi yang kalian buat hari ini bisa diukur kualitasnya, bukan cuma kecepatannya. Sampai jumpa!