This langchain routing by cost tutorial walks through building a dynamic model selector that keeps your LLM spend predictable without fragmenting your integration code. We’ll point LangChain at a single OpenAI-compatible endpoint exposed by n4n.ai, then layer a lightweight router on top that picks a model per request based on estimated token cost. You end up with one client, one auth path, and a clear knob for spend control.
Step 1: Point LangChain at a single OpenAI-compatible endpoint
Install the OpenAI integration for LangChain. You do not need provider-specific SDKs or separate retry clients.
pip install langchain-openai python-dotenv tiktokenLoad your key and set the base URL. The gateway speaks the OpenAI chat completions protocol, so ChatOpenAI works unchanged.
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
llm = ChatOpenAI(
api_key=os.environ["N4N_API_KEY"],
base_url="https://api.n4n.ai/v1",
model="openai/gpt-4o-mini",
temperature=0,
)One endpoint now fronts 240+ models. You switch models by changing the model string, not the client class. That matters because most cost-routing logic dies in boilerplate when you maintain three provider clients. Keep the transport dumb and push routing decisions into your own code.
Step 2: Define cost tiers and estimate spend
A useful langchain routing by cost tutorial should not pretend you can predict exact cents without pricing tables that change weekly. Use relative tiers: cheap, standard, premium. Map them to models available through the gateway.
COST_TIERS = {
"cheap": "anthropic/claude-3-haiku",
"standard": "openai/gpt-4o-mini",
"premium": "openai/gpt-4o",
}Publicly listed prices show Haiku and GPT-4o mini are roughly an order of magnitude cheaper than full GPT-4o for similar token counts. That spread is enough to route on.
Estimate tokens before the call. For a quick heuristic, use tiktoken on the prompt text. If you batch many calls, cache the encoder.
import tiktoken
def estimate_tokens(text: str) -> int:
enc = tiktoken.get_encoding("o200k_base")
return len(enc.encode(text))Set a hard token ceiling per tier. These are not price limits; they are size gates that correlate with cost.
MAX_TOKENS_CHEAP = 2000
MAX_TOKENS_STANDARD = 8000If your prompt exceeds the standard ceiling, you are shipping enough context that the premium model’s higher quality may pay for itself. Below that, drop to cheaper tiers aggressively.
Step 3: Build the router Runnable
LangChain’s expression language makes this clean. Wrap selection in a function that returns a configured ChatOpenAI instance, then invoke. Reconstructing clients per call is wasteful, so cache them.
from langchain_openai import ChatOpenAI
def _make_llm(model: str) -> ChatOpenAI:
return ChatOpenAI(
api_key=os.environ["N4N_API_KEY"],
base_url="https://api.n4n.ai/v1",
model=model,
temperature=0,
)
LLM_CACHE = {tier: _make_llm(model) for tier, model in COST_TIERS.items()}
def select_cached(prompt: str) -> ChatOpenAI:
tok = estimate_tokens(prompt)
if tok <= MAX_TOKENS_CHEAP:
return LLM_CACHE["cheap"]
elif tok <= MAX_TOKENS_STANDARD:
return LLM_CACHE["standard"]
return LLM_CACHE["premium"]This pattern keeps the langchain routing by cost tutorial grounded: you still get a single Runnable interface, but the underlying model flips based on input size. For async pipelines, use ainvoke on the cached clients; they are thread-safe.
from langchain_core.runnables import RunnableLambda
routed = RunnableLambda(lambda inp: select_cached(inp["prompt"]).invoke(inp["prompt"]))If you need structured output, bind a schema to the cached client before returning it. The router stays identical.
Step 4: Forward routing directives and cache hints
The gateway honors client routing directives. If you want to force a specific routing preference, pass headers through default_headers. This is useful when you want the gateway to prefer the cheapest available provider for a given model family.
llm = ChatOpenAI(
api_key=os.environ["N4N_API_KEY"],
base_url="https://api.n4n.ai/v1",
model="openai/gpt-4o-mini",
default_headers={"x-n4n-route": "prefer-cheapest", "x-n4n-cache": "read-write"},
)Cache-control hints are forwarded to the upstream provider when supported. Use read-write for repetitive system prompts and read-only for one-off user content. Combined with tier selection, cache hits compound your savings.
Step 5: Let the gateway handle degradation
Writing exponential backoff for every provider is busywork. n4n.ai performs automatic fallback when a provider is rate-limited or degraded, so a single request to the gateway can survive an upstream outage without your code catching RateLimitError. You still need an outer guard for genuine failures, but you can skip multi-provider orchestration entirely.
from langchain_core.exceptions import LangChainException
def safe_call(prompt: str) -> str:
try:
return select_cached(prompt).invoke(prompt).content
except LangChainException as e:
# gateway already tried fallbacks; log and escalate
raiseDo not wrap this in a manual loop that retries across models unless you have a quality reason. The cost router already picked the cheapest viable tier; a retry storm just burns tokens.
Step 6: Meter usage per token
The gateway returns OpenAI-compatible usage objects. Read them to track spend at the token level.
response = select_cached("Summarize: " + long_text).invoke(...)
print(response.usage)
# {'completion_tokens': 120, 'prompt_tokens': 850, 'total_tokens': 970}Because metering is per-token and unified across models, you can aggregate cost in one pipeline regardless of which tier served the request. Pipe the usage block into your logging backend:
import json, time
def log_usage(tier: str, usage: dict):
with open("usage.jsonl", "a") as f:
f.write(json.dumps({"tier": tier, "ts": time.time(), **usage}) + "\n")Over a week, this file shows you exactly how often the cheap tier handled load. That is the real measure of whether the router works.
Step 7: Compose a reusable cost-routed chain
Wrap the router in a proper chain with a prompt template so callers never touch model names.
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "You are a concise classifier."),
("user", "{input}")
])
def routed_chain(input_text: str):
llm = select_cached(input_text)
chain = prompt | llm
return chain.invoke({"input": input_text})
out = routed_chain("Short text: classify as spam or ham.")
print(out.content)For async services, swap to chain.ainvoke and keep the same cached clients.
How to verify success
Run the script with a mix of short and long inputs. Add logging to confirm the model string changes across tiers:
print(select_cached("hi").model_name) # anthropic/claude-3-haiku
print(select_cached("x" * 4000).model_name) # openai/gpt-4o-miniCheck the usage block on each response to confirm tokens are counted. If you see total_tokens populated and no provider-specific errors, the route worked. Deploy behind a flag and compare your weekly token histogram to the pre-router baseline.
Step 8: Avoid these routing mistakes
- Reconstructing clients per call. Client creation is not free. Cache by tier.
- Estimating tokens with
len(text.split()). Word counts lie for code and CJK text. Use a real tokenizer. - Routing on response quality after the fact. If the cheap tier fails your eval, tighten
MAX_TOKENS_STANDARD, don’t add a silent upgrade path that defeats the purpose. - Ignoring cache headers. A 90% cache hit rate on system prompts beats any model downgrade.
This langchain routing by cost tutorial gave you a single integration point, a tier-based selector, and gateway-level fallback. Ship it, meter it, and let the usage log tell you when to adjust the ceilings.