Free lesson · GenAI Data Engineering
Route extraction by complexity to cost-effective models
Use DeepSeek for bulk extraction (140x cheaper than OpenAI o1) and reasoning models for complex documents requiring multi-hop inference.
Course: GenAI Data Pipelines · Chapter 10 · Knowledge Graph Construction with LightRAG
Free to read — no subscription required.
Introduction
When you send every document through a frontier model like GPT-4o, you build the simplest extraction pipeline imaginable — and one of the most expensive. Most corpora are mixed: a long tail of simple inputs (press releases, structured forms, FAQs) sits alongside a smaller share of genuinely hard inputs (legal contracts, clinical notes, multi-party agreements) that need reasoning capability. Teams that route by complexity classify each document up front and dispatch it to the cheapest model that can still extract it correctly; get this wrong and you either burn 5-10× the necessary spend or quietly degrade quality on the documents that matter. By the end of this lesson you will be able to score documents on cheap heuristic signals, map those scores to a tiered model ladder (DeepSeek → GPT-4o-mini → GPT-4o), and emit a per-document cost estimate so downstream budget controls can react.
Key Terminology
- ComplexityLevel — the enum (SIMPLE / MODERATE / COMPLEX) that tags each document so the router knows which model tier to dispatch to; it is the single contract between classifier and router.
- ComplexityClassifier — the fast, pre-extraction component that scores a document on signals such as entity density, sentence length, and technical-term ratio, then bins it into a
ComplexityLevel. It runs in pure Python so routing overhead stays negligible. - ExtractionRouter — the lookup that maps a
ComplexityLevelto a concrete model name and its per-million-token cost, returning both the routing decision and an estimated cost so upstream budget controls can react.
Concepts
Three ideas drive the design. First, routing is a pre-extraction decision: the classifier must be cheap enough that running it on every document does not erase the savings from sending some to cheaper models — which is why the signals here are pure-Python heuristics, not LLM calls. Second, the bins are calibrated, not absolute: the simple_threshold and complex_threshold should be tuned against a labeled slice of your corpus so that the SIMPLE bin actually contains documents the cheap model handles correctly. Third, the router emits cost metadata alongside the model choice so an upstream budget guard can short-circuit, downgrade, or batch when a workload's projected spend exceeds policy. Together these three pieces typically cut extraction cost by 60–80% versus a uniform GPT-4o pipeline without measurable quality loss on the SIMPLE majority.
Code Walkthrough
Document Complexity Classification
The complexity classifier analyzes text characteristics to route documents to appropriate models. Key signals include entity density, sentence complexity, domain specificity, and the presence of implicit relationships that require inference.
ComplexityClassifier scores document chunks on multiple complexity dimensions and routes them to cost-appropriate extraction models.
Code snippet python
1from enum import Enum 2 3class ComplexityLevel(str, Enum): 4 SIMPLE = "simple" 5 MODERATE = "moderate" 6 COMPLEX = "complex" 7 8class ComplexityClassifier: 9 def __init__( 10 self, 11 simple_threshold: float = 0.3, 12 complex_threshold: float = 0.7, 13 ): 14 self.simple_t = simple_threshold 15 self.complex_t = complex_threshold 16 17 def classify( 18 self, 19 text: str, 20 ) -> dict: 21 scores = { 22 "entity_density": ( 23 self._entity_density(text) 24 ), 25 "avg_sentence_length": ( 26 self._avg_sentence_length(text) 27 ), 28 "technical_term_ratio": ( 29 self._technical_ratio(text) 30 ), 31 } 32 composite = sum( 33 scores.values() 34 ) / len(scores) 35 if composite < self.simple_t: 36 level = ComplexityLevel.SIMPLE 37 elif composite > self.complex_t: 38 level = ComplexityLevel.COMPLEX 39 else: 40 level = ComplexityLevel.MODERATE 41 return { 42 "level": level, 43 "composite_score": round( 44 composite, 3 45 ), 46 "scores": scores, 47 } 48 49 def _entity_density( 50 self, text: str 51 ) -> float: 52 words = text.split() 53 capitalized = sum( 54 1 for w in words 55 if w[0].isupper() 56 and len(w) > 1 57 ) 58 return min( 59 capitalized / max(len(words), 1), 60 1.0, 61 ) 62 63 def _avg_sentence_length( 64 self, text: str 65 ) -> float: 66 sentences = text.split(".") 67 avg = sum( 68 len(s.split()) 69 for s in sentences 70 ) / max(len(sentences), 1) 71 return min(avg / 50.0, 1.0) 72 73 def _technical_ratio( 74 self, text: str 75 ) -> float: 76 words = text.lower().split() 77 technical = sum( 78 1 for w in words if len(w) > 12 79 ) 80 return min( 81 technical / max(len(words), 1), 82 1.0, 83 )
- Lines 8-15: The classifier accepts configurable thresholds for simple and complex classification. These thresholds should be tuned on a labeled sample of your document corpus.
- Lines 17-38: The
classifymethod computes three complexity signals and averages them into a composite score. Documents below the simple threshold route to cheap models; documents above the complex threshold route to expensive models. - Lines 40-60: The heuristic signals -- entity density, sentence length, and technical term ratio -- provide a fast approximation of document complexity. These are computed without any API calls, so routing adds negligible overhead.
Model Routing
The router maps complexity levels to specific models, directing simple documents to DeepSeek and complex documents to GPT-4o or Claude.
ExtractionRouter selects the extraction model based on complexity classification and provides cost estimation for each routing decision.
Code snippet python
1class ExtractionRouter: 2 def __init__( 3 self, 4 model_map: dict = None, 5 ): 6 self.model_map = model_map or { 7 ComplexityLevel.SIMPLE: { 8 "model": "deepseek/deepseek-chat", 9 "cost_per_1m_tokens": 0.14, 10 }, 11 ComplexityLevel.MODERATE: { 12 "model": "gpt-4o-mini", 13 "cost_per_1m_tokens": 0.15, 14 }, 15 ComplexityLevel.COMPLEX: { 16 "model": "gpt-4o", 17 "cost_per_1m_tokens": 2.50, 18 }, 19 } 20 21 def route( 22 self, 23 complexity: dict, 24 ) -> dict: 25 level = complexity["level"] 26 config = self.model_map[level] 27 return { 28 "model": config["model"], 29 "complexity_level": level, 30 "estimated_cost_per_1m": ( 31 config["cost_per_1m_tokens"] 32 ), 33 }
- Lines 1-19: The router maps complexity levels to model configurations with associated costs. DeepSeek at $0.14/1M tokens handles simple documents, while GPT-4o at $2.50/1M tokens handles complex documents requiring reasoning capability.
- Lines 21-30: The
routemethod returns the selected model and cost estimate. Including the cost estimate in the routing decision enables upstream budget enforcement to account for per-document cost variation.
Do's and Don'ts
Now that you have worked through the implementation, the practices below separate a durable approach from a fragile one.
Do's
- ✓Tune
simple_thresholdandcomplex_thresholdon a labeled sample of your own corpus before relying on the SIMPLE-bin cost savings — defaults are a starting point, not a calibration. - ✓Emit the per-document
estimated_cost_per_1mfrom the router into your logs or metrics so you can audit actual spend by complexity bin and detect drift when the corpus mix changes. - ✓Keep the classifier signals cheap (string/regex level) so routing overhead stays well under the cost of the cheapest model in the ladder.
Don'ts
- ✗Don't route to a cheaper model on confidence alone — verify on a held-out sample that the SIMPLE-bin model actually matches the COMPLEX-bin model's extraction quality on documents in that bin, otherwise the savings come out of accuracy.
- ✗Don't call an LLM inside the complexity classifier; the whole point of routing is to decide before you pay for an extraction call, so a classifier that itself hits an API defeats the budget argument.
- ✗Don't hard-code model names and prices inside the router — keep
model_mapinjectable so price changes, new providers, or A/B tests don't require code edits.
This lesson is free to read. Its 3 hands-on labs — real code, in a cloud IDE — are part of the GenAI Data Engineering subscription.
From · cancel anytime · Already a subscriber? Sign in →
More free lessons in GenAI Data Pipelines
- Ch 7Track costs in real-time with Langfuse and enforce budgets
- Ch 7Orchestrate pipelines as Argo Workflows with Kafka triggers
- Ch 8Configure AlloyDB with pgvector and ScaNN indexing
- Ch 8Build zero-downtime reindexing for embedding model upgrades
- Ch 9Build semantic caching using Redis LangCache
- Ch 10Route extraction by complexity to cost-effective modelsYou are here
- Ch 12Build agentic RAG with query decomposition and self-verification