Skip to content

complexity

complexity

Query complexity analyzer — scores queries and suggests token budgets.

Produces a numeric complexity score (0.0–1.0) and a suggested max_tokens budget based on query characteristics such as length, domain signals (code, math, multi-step reasoning), and whether the target model is a thinking model that needs extra headroom.

Classes

ComplexityResult dataclass

ComplexityResult(score: float, tier: str, suggested_max_tokens: int, signals: dict)

Output of the complexity analysis.

ComplexityQueryAnalyzer

Bases: QueryAnalyzer

Query analyzer that produces a complexity-aware RoutingContext.

Drop-in replacement for DefaultQueryAnalyzer — adds complexity_score, has_reasoning, and suggested_max_tokens to the returned RoutingContext.

Functions

score_complexity

score_complexity(query: str) -> ComplexityResult

Score a query's complexity from 0.0 (trivial) to 1.0 (very complex).

The score is a weighted combination of independent signals, each contributing a fraction between 0 and 1. Weights reflect how much each signal correlates with the amount of reasoning and output tokens a model will need.

Source code in src/openjarvis/learning/routing/complexity.py
def score_complexity(query: str) -> ComplexityResult:
    """Score a query's complexity from 0.0 (trivial) to 1.0 (very complex).

    The score is a weighted combination of independent signals, each
    contributing a fraction between 0 and 1.  Weights reflect how much
    each signal correlates with the amount of reasoning and output
    tokens a model will need.
    """
    signals: dict = {}
    score = 0.0

    # --- Length signal (0–0.20) ---
    length = len(query)
    if length < 20:
        length_score = 0.0
    elif length < 100:
        length_score = 0.3
    elif length < 300:
        length_score = 0.6
    elif length < 800:
        length_score = 0.8
    else:
        length_score = 1.0
    signals["length"] = length_score
    score += 0.20 * length_score

    # --- Domain signals (0–0.25) ---
    has_code = bool(_CODE_PATTERNS.search(query))
    has_math = bool(_MATH_PATTERNS.search(query))
    domain_score = 0.0
    if has_code:
        domain_score = max(domain_score, 0.7)
    if has_math:
        domain_score = max(domain_score, 0.8)
    if has_code and has_math:
        domain_score = 1.0
    signals["domain"] = domain_score
    signals["has_code"] = has_code
    signals["has_math"] = has_math
    score += 0.25 * domain_score

    # --- Reasoning signal (0–0.25) ---
    has_reasoning = bool(_REASONING_PATTERNS.search(query))
    has_multi_step = bool(_MULTI_STEP_PATTERNS.search(query))
    reasoning_score = 0.0
    if has_reasoning:
        reasoning_score = 0.6
    if has_multi_step:
        reasoning_score = max(reasoning_score, 0.8)
    if has_reasoning and has_multi_step:
        reasoning_score = 1.0
    signals["reasoning"] = reasoning_score
    signals["has_reasoning"] = has_reasoning
    signals["has_multi_step"] = has_multi_step
    score += 0.25 * reasoning_score

    # --- Question / sub-task count (0–0.15) ---
    n_questions = _count_questions(query)
    n_subtasks = _count_sub_tasks(query)
    multi_part = n_questions + n_subtasks
    if multi_part <= 1:
        multi_score = 0.0
    elif multi_part <= 3:
        multi_score = 0.5
    else:
        multi_score = 1.0
    signals["multi_part"] = multi_score
    signals["n_questions"] = n_questions
    signals["n_subtasks"] = n_subtasks
    score += 0.15 * multi_score

    # --- Creative / generative signal (0–0.15) ---
    has_creative = bool(_CREATIVE_PATTERNS.search(query))
    creative_score = 0.7 if has_creative else 0.0
    signals["creative"] = creative_score
    signals["has_creative"] = has_creative
    score += 0.15 * creative_score

    # Clamp
    score = max(0.0, min(1.0, score))

    # Map to tier
    if score < 0.15:
        tier = "trivial"
    elif score < 0.30:
        tier = "simple"
    elif score < 0.55:
        tier = "moderate"
    elif score < 0.80:
        tier = "complex"
    else:
        tier = "very_complex"

    suggested_max_tokens = _TOKEN_TIERS[tier]

    return ComplexityResult(
        score=round(score, 3),
        tier=tier,
        suggested_max_tokens=suggested_max_tokens,
        signals=signals,
    )

is_thinking_model

is_thinking_model(model_name: str) -> bool

Return True if the model is known to use internal chain-of-thought.

Source code in src/openjarvis/learning/routing/complexity.py
def is_thinking_model(model_name: str) -> bool:
    """Return True if the model is known to use internal chain-of-thought."""
    return bool(_THINKING_MODEL_PATTERNS.search(model_name))

adjust_tokens_for_model

adjust_tokens_for_model(suggested: int, model_name: Optional[str] = None) -> int

Multiply the token budget when the target is a thinking model.

Source code in src/openjarvis/learning/routing/complexity.py
def adjust_tokens_for_model(suggested: int, model_name: Optional[str] = None) -> int:
    """Multiply the token budget when the target is a thinking model."""
    if model_name and is_thinking_model(model_name):
        return suggested * _THINKING_TOKEN_MULTIPLIER
    return suggested