Learn LangChain - Performance & Cost Optimization

#Introduction

In episode 17 you went deep into advanced LangGraph: typed states, reducers, checkpointing, time travel, and subgraphs complete with Postgres persistence. Your architecture is now solid on the state side. Episode 18 rounds it out from the operational side: speed and cost. We'll break down how to lower latency with streaming, raise throughput with batching and RunnableParallel, stack concurrency with async, then cut costs with token tracking, prompt caching, model tiering, and caching layers.

#Start with Measurement: Token Usage Tracking

Optimization without data is just guesswork. The good news: every model response carries token usage metadata via the usage_metadata attribute on the response object — containing input_tokens, output_tokens, and total_tokens. The first habit you must form is always recording it.

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': {...}}

Store this metadata in structured logs or a database, then aggregate per feature, per user, and per model. That way you'll know which feature is the most expensive before the bill arrives as a surprise.

#Streaming: Cutting Latency to First Token

The latency users feel isn't total execution time — it's the time until the first token appears on screen. With stream, the model emits tokens as soon as it produces them, so the UI can display the answer progressively and feel responsive.

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)

Enable streaming=True at model initialization, and the end of the chain may use a streaming-tolerant parser like StrOutputParser. For the async version, astream gives the same result without blocking the event loop. Important note: streaming doesn't make the last token arrive faster — it only makes the experience feel faster. Total time is often the same.

#Batching and Parallelism: Increasing Throughput

When facing many independent inputs, don't call the model one by one. Two LCEL weapons you must master: batch/abatch for processing a set of inputs at once, and RunnableParallel for running several steps simultaneously within one 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 has a max_concurrency parameter to control how many parallel requests are allowed, so you don't flood the provider while staying within rate limits. Combine both: parallelism at the pipeline level, batching at the model level. For massive volume, put a queue outside the application — for example a task queue — so request spikes don't break service stability.

#Async: Throughput in a Single Server

LLM models spend most of their time waiting on I/O — HTTP responses from the provider. In an async environment, ainvoke and astream let the event loop run other work while waiting, so one worker can manage hundreds of concurrent requests without adding instances.

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())

This pattern works because gather schedules all the calls at once and waits for them all to finish. An async FastAPI server can use astream to send tokens to the client while still serving other requests on the same event loop — the ideal combination for production chat applications.

#Prompt Caching: Pay Once for the Same Input

Most LLM cost comes from repeated input tokens: the system prompt, long instructions, and the same RAG context answered over and over. Prompt caching lets the provider recognize identical prefixes and bill far less for subsequent calls.

prompt_caching.py

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

For caching to work, the system prefix must be exactly identical across requests. Don't insert dynamic data at the start of the prompt — put changing parts like the user question at the very end. Starting with recent versions, several providers apply caching automatically; your job is just to keep the prompt pattern stable and deterministic.

#Model Tiering and Caching Layers

Model tiering means using cheap models for easy tasks and expensive models for hard ones. The simplest example: a router decides whether a question needs deep reasoning.

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

The last layer is a caching layer: store answers for questions that have already been answered so subsequent calls don't need the model at all. Start with a simple in-memory cache, then move up to Redis for multi-instance scale.

simple_cache.py

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

For long-term memory needs, ecosystems like mem0 provide a memory layer that stores facts and automatically pulls relevant history into the prompt — more expensive than a simple cache, but it adds personalized context. Install its dependency with pip install mem0ai and adapt it to your architecture.

#Conclusion

Episode 18 closed out the performance and cost side: streaming for fast-perceived latency, batch and RunnableParallel for throughput, async for high concurrency, usage_metadata tracking as the measurement baseline, prompt caching for repeated inputs, model tiering for task routing, and caching layers to cut unnecessary calls.

Key takeaways:

• Measure first: record usage_metadata on every request before optimizing anything.
• Streaming improves the user experience; batching and parallelism improve throughput.
• Async lets a single server serve hundreds of concurrent requests with few instances.
• Prompt caching cuts repeated-input costs, as long as the prompt prefix is stable.
• Model tiering and caching layers reduce costs without sacrificing answer quality.

In episode 19 we install the monitoring tool: LangSmith for tracing, datasets, and automated evaluation — so every optimization you make today can be measured for quality, not just speed. See you there!