If you’re building production LLM applications, you’ve probably hit the tension between cost and capability. You want GPT-4o for complex reasoning but don’t want to pay premium prices for simple classification tasks. This langchain n4n.ai floor suffix routing tutorial shows you how to implement automatic model selection based on price floors, so your application routes requests to the cheapest model that meets your quality threshold.
Prerequisites
Before we start, make sure you have:
- Python 3.10+
- An n4n.ai API key (get one at n4n.ai)
- Basic familiarity with LangChain’s
Runnableinterface
Install the dependencies:
pip install langchain-openai langchain-core python-dotenvCreate a .env file with your credentials:
N4N_API_KEY=your-api-key-here
N4N_BASE_URL=https://api.n4n.ai/v1Understanding the floor price suffix
The :floor suffix is a routing directive you append to a model name. It tells the gateway: “Give me the cheapest available model that costs at least this much per million tokens.” This is useful when you have a minimum quality bar — you don’t want the absolute cheapest model, but you also don’t want to overpay for capability you don’t need.
For example, gpt-4o-mini:floor:0.50 means “route to the cheapest model priced at or above $0.50 per million input tokens.” If gpt-4o-mini is $0.15 and gpt-4o is $2.50, you’ll get gpt-4o. If a new model launches at $0.60, you’ll get that instead.
Step 1: Basic client setup
Let’s start with a minimal LangChain client configured for n4n.ai:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
llm = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="gpt-4o-mini", # we'll override this per-request
temperature=0,
)
response = llm.invoke("What is 2 + 2?")
print(response.content)Run it:
python step1_basic.pyExpected output:
4Nothing fancy yet — just confirming the gateway connection works.
Step 2: Using the floor suffix in model names
Now let’s apply the floor suffix. We’ll create a helper that builds model strings with the directive:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
load_dotenv()
def make_floor_model(base_model: str, floor_price: float) -> str:
"""Construct a model string with :floor suffix."""
return f"{base_model}:floor:{floor_price:.2f}"
# Example: minimum $0.50 per million input tokens
model_name = make_floor_model("gpt-4o-mini", 0.50)
print(f"Routing with model: {model_name}")
llm = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model=model_name,
temperature=0,
)
response = llm.invoke("Explain quantum entanglement in two sentences.")
print(response.content)Run it:
python step2_floor.pyExpected output (model will vary based on current pricing):
Routing with model: gpt-4o-mini:floor:0.50
Quantum entanglement is a phenomenon where pairs of particles become correlated such that the quantum state of each particle cannot be described independently, even when separated by large distances. Measuring one particle instantly determines the state of its entangled partner.The gateway selected the cheapest model meeting your $0.50 floor. Check the response headers if you want to verify which model actually served the request — n4n.ai returns x-model-used in the response metadata.
Step 3: Building a routing chain with fallbacks
Real applications need more than a single model. Let’s build a chain that tries a cheap model first, then escalates if the response quality is insufficient. We’ll use LangChain’s RunnableWithFallbacks:
import os
import json
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda, RunnableWithFallbacks
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
load_dotenv()
# Tier 1: Cheap model for simple tasks
cheap_model = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="gpt-4o-mini",
temperature=0,
max_tokens=500,
)
# Tier 2: Floor-routed model for medium complexity
medium_model = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="gpt-4o-mini:floor:0.50",
temperature=0,
max_tokens=1000,
)
# Tier 3: High-capability model for hard problems
expensive_model = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="gpt-4o",
temperature=0,
max_tokens=2000,
)
# Quality checker — returns True if response looks complete
def check_quality(response: str) -> bool:
"""Heuristic: reject very short or error-like responses."""
if not response or len(response.strip()) < 20:
return False
if "i don't know" in response.lower() or "i cannot" in response.lower():
return False
return True
# Build the fallback chain
def make_chain(model):
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant. Be concise."),
("human", "{input}"),
])
return prompt | model | StrOutputParser()
cheap_chain = make_chain(cheap_model)
medium_chain = make_chain(medium_model)
expensive_chain = make_chain(expensive_model)
# Wrap with fallback logic
def try_chain(chain, input_data):
result = chain.invoke(input_data)
if check_quality(result):
return result
raise ValueError("Quality check failed")
routing_chain = (
RunnableLambda(lambda x: try_chain(cheap_chain, x))
.with_fallbacks([
RunnableLambda(lambda x: try_chain(medium_chain, x)),
RunnableLambda(lambda x: try_chain(expensive_chain, x)),
])
)
# Test with varying complexity
test_inputs = [
"What is 2 + 2?",
"Write a haiku about debugging.",
"Explain the proof of Fermat's Last Theorem in detail.",
]
for inp in test_inputs:
print(f"\n--- Input: {inp} ---")
try:
result = routing_chain.invoke({"input": inp})
print(f"Result: {result[:200]}...")
except Exception as e:
print(f"All tiers failed: {e}")Run it:
python step3_fallback.pyExpected output (models selected will vary):
--- Input: What is 2 + 2? ---
Result: 4
--- Input: Write a haiku about debugging. ---
Result: Bugs hide in the code
Patient search finds the flaw
Logic brings the light
--- Input: Explain the proof of Fermat's Last Theorem in detail. ---
Result: Fermat's Last Theorem states that no three positive integers a, b, and c can satisfy the equation a^n + b^n = c^n for any integer value of n greater than 2. The proof, completed by Andrew Wiles in 1994...The first two queries likely succeeded on the cheap tier. The Fermat query probably escalated to medium or expensive tier because the cheap model’s response was too short or contained “I cannot” language.
Step 4: Dynamic floor pricing based on task classification
Hardcoding floor prices works, but you can do better by classifying the task first and selecting an appropriate floor. Here’s a pattern that uses a lightweight classifier to pick the routing strategy:
import os
import json
from enum import Enum
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda, RunnableBranch
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
load_dotenv()
class TaskComplexity(Enum):
SIMPLE = "simple" # classification, extraction, simple QA
MEDIUM = "medium" # summarization, translation, coding help
COMPLEX = "complex" # reasoning, creative writing, analysis
# Classifier prompt — uses cheapest model
CLASSIFIER_PROMPT = """Classify the task complexity. Respond with ONLY one word: simple, medium, or complex.
Simple: factual lookup, classification, extraction, yes/no questions
Medium: summarization, translation, code generation, explanation
Complex: multi-step reasoning, creative writing, mathematical proof, strategy
Task: {input}
Classification:"""
classifier = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="gpt-4o-mini",
temperature=0,
max_tokens=10,
)
classifier_chain = (
ChatPromptTemplate.from_template(CLASSIFIER_PROMPT)
| classifier
| StrOutputParser()
)
# Model selectors per complexity
def get_model_for_complexity(complexity: str) -> ChatOpenAI:
configs = {
"simple": {"model": "gpt-4o-mini", "max_tokens": 500},
"medium": {"model": "gpt-4o-mini:floor:0.50", "max_tokens": 1500},
"complex": {"model": "gpt-4o:floor:2.00", "max_tokens": 3000},
}
cfg = configs.get(complexity, configs["simple"])
return ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
**cfg,
temperature=0,
)
def route_and_execute(input_data: dict) -> str:
task = input_data["input"]
complexity = classifier_chain.invoke({"input": task}).strip().lower()
print(f" [Classifier] '{task[:50]}...' -> {complexity}")
model = get_model_for_complexity(complexity)
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant."),
("human", "{input}"),
])
chain = prompt | model | StrOutputParser()
return chain.invoke({"input": task})
# Test
test_cases = [
"Is Paris the capital of France?",
"Summarize the plot of Hamlet in three sentences.",
"Design a distributed rate limiter for a microservices architecture. Include code.",
]
for case in test_cases:
print(f"\n--- Task: {case[:60]}... ---")
result = route_and_execute({"input": case})
print(f"Result: {result[:300]}...")Run it:
python step4_dynamic.pyExpected output:
--- Task: Is Paris the capital of France?... ---
[Classifier] 'Is Paris the capital of France?' -> simple
Result: Yes, Paris is the capital city of France.
--- Task: Summarize the plot of Hamlet in three sentences.... ---
[Classifier] 'Summarize the plot of Hamlet in three sentences.' -> medium
Result: Hamlet, Prince of Denmark, seeks revenge against his uncle Claudius...
--- Task: Design a distributed rate limiter for a microservices architecture... ---
[Classifier] 'Design a distributed rate limiter for a microservices architecture...' -> complex
Result: A distributed rate limiter for microservices typically uses...The classifier routes each task to an appropriate price floor. Simple tasks hit the cheapest model. Complex tasks get a $2.00 floor, which typically routes to GPT-4o or equivalent capability.
Step 5: Observability — tracking which model actually ran
The floor suffix means the model name in your code isn’t the model that executes. You need to capture the actual model from response headers for cost tracking and debugging. Here’s how to extract it:
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict
load_dotenv()
class ModelTrackingCallback(BaseCallbackHandler):
"""Captures the actual model used from response headers."""
def __init__(self):
self.last_model = None
self.last_usage = None
def on_llm_end(self, response: Any, **kwargs: Any) -> None:
# LangChain's OpenAI wrapper stores response metadata in generation_info
if hasattr(response, 'generations') and response.generations:
gen = response.generations[0][0]
if hasattr(gen, 'generation_info') and gen.generation_info:
self.last_model = gen.generation_info.get('model_name')
self.last_usage = gen.generation_info.get('usage')
# Usage
callback = ModelTrackingCallback()
llm = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model="gpt-4o-mini:floor:0.50",
temperature=0,
callbacks=[callback],
)
response = llm.invoke("Write a one-sentence definition of recursion.")
print(f"Requested model: gpt-4o-mini:floor:0.50")
print(f"Actual model: {callback.last_model}")
print(f"Usage: {callback.last_usage}")
print(f"Response: {response.content}")Run it:
python step5_observability.pyExpected output:
Requested model: gpt-4o-mini:floor:0.50
Actual model: gpt-4o
Usage: {'prompt_tokens': 18, 'completion_tokens': 22, 'total_tokens': 40}
Response: Recursion is a programming technique where a function calls itself to solve smaller instances of the same problem.The gateway chose gpt-4o because it was the cheapest model above the $0.50 floor at request time. Your callback captures this for logging, cost allocation, or alerting.
Step 6: Production pattern — configurable routing policy
Hardcoding floors in application code creates deployment friction. Let’s externalize the routing policy into a JSON config that you can update without code changes:
import os
import json
from pathlib import Path
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.runnables import RunnableLambda
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
load_dotenv()
# routing_policy.json — deploy this separately from code
POLICY_PATH = Path("routing_policy.json")
DEFAULT_POLICY = {
"tiers": [
{"name": "simple", "model": "gpt-4o-mini", "max_tokens": 500, "floor": None},
{"name": "medium", "model": "gpt-4o-mini", "max_tokens": 1500, "floor": 0.50},
{"name": "complex", "model": "gpt-4o", "max_tokens": 3000, "floor": 2.00},
],
"classifier_model": "gpt-4o-mini",
"quality_threshold_chars": 20,
}
def load_policy() -> dict:
if POLICY_PATH.exists():
with open(POLICY_PATH) as f:
return json.load(f)
return DEFAULT_POLICY
def build_model(tier: dict) -> ChatOpenAI:
model_name = tier["model"]
if tier.get("floor") is not None:
model_name = f"{model_name}:floor:{tier['floor']:.2f}"
return ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model=model_name,
temperature=0,
max_tokens=tier["max_tokens"],
)
class RoutingEngine:
def __init__(self, policy: dict = None):
self.policy = policy or load_policy()
self.tiers = {t["name"]: build_model(t) for t in self.policy["tiers"]}
self.classifier = ChatOpenAI(
api_key=os.getenv("N4N_API_KEY"),
base_url=os.getenv("N4N_BASE_URL"),
model=self.policy["classifier_model"],
temperature=0,
max_tokens=10,
)
def classify(self, task: str) -> str:
prompt = f"""Classify: simple, medium, or complex.
Simple: lookup, classification, yes/no
Medium: summary, translation, code
Complex: reasoning, creative, math proof
Task: {task}
Classification:"""
result = self.classifier.invoke(prompt).content.strip().lower()
return result if result in self.tiers else "simple"
def execute(self, task: str) -> dict:
tier_name = self.classify(task)
model = self.tiers[tier_name]
prompt = ChatPromptTemplate.from_messages([
("system", "You are a helpful assistant."),
("human", "{input}"),
])
chain = prompt | model | StrOutputParser()
response = chain.invoke({"input": task})
return {
"tier": tier_name,
"response": response,
"model_requested": model.model_name,
}
# Create policy file if it doesn't exist
if not POLICY_PATH.exists():
with open(POLICY_PATH, "w") as f:
json.dump(DEFAULT_POLICY, f, indent=2)
print(f"Created {POLICY_PATH} — edit it to adjust routing without code changes")
# Demo
engine = RoutingEngine()
tasks = [
"What's the capital of Japan?",
"Translate 'hello world' to Spanish.",
"Prove that the square root of 2 is irrational.",
]
for task in tasks:
result = engine.execute(task)
print(f"\nTask: {task}")
print(f" Tier: {result['tier']}")
print(f" Requested: {result['model_requested']}")
print(f" Response: {result['response'][:150]}...")Run it:
python step6_policy.pyExpected output:
Created routing_policy.json — edit it to adjust routing without code changes
Task: What's the capital of Japan?
Tier: simple
Requested: gpt-4o-mini
Response: The capital of Japan is Tokyo.
Task: Translate 'hello world' to Spanish.
Tier: medium
Requested: gpt-4o-mini:floor:0.50
Response: "Hello world" in Spanish is "Hola mundo".
Task: Prove that the square root of 2 is irrational.
Tier: complex
Requested: gpt-4o:floor:2.00
Response: Proof by contradiction: Assume √2 is rational, so √2 = a/b where a,b are integers...Now you can tune routing by editing routing_policy.json — raise floors when you need higher quality, lower them to reduce costs, add new tiers for specialized models — all without redeploying application code.
Common pitfalls
Assuming the floor price maps to a specific model. It doesn’t. The gateway selects the cheapest currently available model meeting the floor. Provider pricing changes, new models launch, and regional availability varies. Your code should never assume gpt-4o-mini:floor:0.50 equals GPT-4o.
Setting floors too tight. If you set a $0.51 floor and the cheapest model above that is $5.00, you just increased costs 10x. Check the model catalog periodically or implement a ceiling fallback.
Skipping quality checks. The floor suffix guarantees minimum price, not minimum quality. A $2.00 model might still hallucinate on your specific task. Keep the fallback chain from Step 3 or implement task-specific evaluators.
Ignoring latency implications. Higher-tier models often have higher latency. If your application has strict SLAs, add latency-aware routing alongside price-aware routing.
What’s next
You now have a complete pattern for cost-aware routing in LangChain:
- Floor suffix for minimum-capability guarantees
- Fallback chains for quality escalation
- Dynamic classification for task-appropriate routing
- Observability callbacks for actual model tracking
- Externalized policy for runtime configuration
From here, consider adding:
- Token budget enforcement — track cumulative spend per user/session
- A/B testing framework — compare floor thresholds against quality metrics
- Provider-specific routing — use
:provider:anthropicsuffixes alongside floors for vendor diversity
The floor suffix is a lever, not a solution. Combine it with the patterns above and you’ll build LLM applications that are both cost-effective and reliable.