agent-model-router

agent-model-router

English | 中文

A zero-dependency Python library for explainable LLM routing and recoverable task scheduling.

Latest release: 1.1.2 · Python: 3.10+ · License: MIT · Tests: 297 passed, 1 optional live-provider smoke skipped

• PyPI: https://pypi.org/project/agent-model-router/
• Source: https://github.com/Odd-C/agent-model-router
• Changelog: CHANGELOG.md

What it does

agent-model-router separates three concerns that are often mixed together in multi-model applications:

caller classifies the task
        ↓
router filters candidates and explains the best choice
        ↓
integration calls the provider and reports result/latency/quota
        ↓
optional scheduler persists, claims, retries, recovers, or falls back

The package provides:

• Utility routing across quality, cost, latency, failure risk, quota pressure, and deadline pressure.
• Hard constraints before scoring for cost, quota, cooldown, health, latency, quality tier, capability, and deadline feasibility.
• Natural-language policy compilation using deterministic Chinese/English rules—no model call or NLP dependency.
• Model profiles keyed by id@provider, with JSON overrides.
• Quota and cooldown state, plus an optional sliding-window ProviderHealth profile.
• Persistent tasks with JSON and SQLite backends.
• SQLite CAS claims, leases, heartbeats, stale-worker recovery, and owner-safe completion for multiple scheduler processes.
• Executable failure handling: abort, cooldown retry, exponential retry, and real executor-provided fallback.
• Two stdlib HTTP services: an OpenAI-compatible proxy and a lightweight task dashboard/API.

What it does not do

Accuracy depends on keeping these boundaries explicit:

• It is not an LLM provider SDK. Your integration still owns credentials, upstream requests, streaming, and provider-specific transport.
• It does not infer task_type with an LLM. Pass task_type explicitly or classify it in your access layer.
• ProviderHealth is passive storage and scoring, not active probing. Your integration must call record_result() and pass a ProviderHealth instance into routing.
• Fallback is real only when your executor implements prepare_fallback(task, error) -> bool. Without that hook, the scheduler fails closed.
• Cross-process exactly-once claiming requires the SQLite backend. The JSON backend is intended for single-process use.
• The bundled policy entries and MockExecutor are examples, not production provider configuration.
• Traffic mirroring, active health probes, and per-provider concurrency gates are not implemented in 1.1.2.

Install

python -m pip install agent-model-router==1.1.2

The runtime package has no third-party dependencies.

Quick start

1. Route a task with Utility scoring

from agent_model_router import HardConstraints, list_models, route_with_utility

result = route_with_utility(
    {"task_type": "coding", "priority": "high", "deadline": None},
    list_models(),
    constraints=HardConstraints(cost_max="free"),
)

print(result["model"], result["provider"])
print(result["score"])
print(result["breakdown"])
print(result["why"])

route_with_utility() first removes candidates that violate hard constraints, then scores the remaining candidates. A multi-candidate result includes the following layers:

{
    "model": "...",
    "provider": "...",
    "reason": "...",
    "score": 0.0,
    "breakdown": {
        "raw": {...},
        "normalized": {...},
        "weights": {...},
        "weighted": {...},
    },
    "why": "...",
}

With one remaining candidate, scoring uses absolute feature values: breakdown["normalized"] is None and the nested raw layer is not added because there is no relative normalization reference.

2. Compile a natural-language preference

from agent_model_router import list_models, route_with_intent

result = route_with_intent(
    {"task_type": "coding", "priority": "normal", "deadline": None},
    list_models(),
    "use a free model and prioritize quality",
)

The Policy Compiler uses deterministic rules to translate common cost, latency, quality, and capability phrases into HardConstraints and weights. It does not call an LLM.

3. Wire passive health data into routing

from pathlib import Path

from agent_model_router import HardConstraints, ProviderHealth, route_with_utility

candidates = [
    {
        "id": "healthy-model",
        "provider": "provider-a",
        "tier": "S",
        "cost": "paid",
        "role": "stable",
        "scenarios": ["coding"],
    },
    {
        "id": "degraded-model",
        "provider": "provider-b",
        "tier": "S",
        "cost": "paid",
        "role": "stable",
        "scenarios": ["coding"],
    },
]

health = ProviderHealth(Path("./router-state"))
health.record_result("healthy-model", "provider-a", status=200, latency_ms=420)
health.record_result("degraded-model", "provider-b", status=503, latency_ms=900)

result = route_with_utility(
    {"task_type": "coding", "priority": "normal", "deadline": None},
    candidates,
    health=health,
    constraints=HardConstraints(max_failure_risk=0.5, max_latency_ms=2000),
)
assert result["model"] == "healthy-model"

If health= is omitted, routing uses documented default priors. Merely creating model-health.json does not automatically connect it to every caller.

4. Persist and execute a task

import time
from pathlib import Path

from agent_model_router import MockExecutor, TaskScheduler, TaskStore

state_dir = Path("./router-state")
store = TaskStore(state_dir, backend="sqlite")
scheduler = TaskScheduler(
    store,
    MockExecutor(result={"ok": True}),
    worker_id="worker-a",
)

now = time.time()
task = scheduler.submit(
    "coding",
    {"request": "example"},
    priority="high",
    deadline=now + 600,  # absolute Unix timestamp
)
scheduler.tick(now=now + 1)

saved = store.get(task.task_id)
print(saved.status, saved.result)  # done {'ok': True}

For multiple scheduler processes, use SQLite and give every process a distinct worker_id.

Routing model

Candidate identity

Every candidate is identified by id@provider. The model ID and provider are returned separately so callers can map them to their own selector format.

Model profiles

Profiles describe routing facts, not provider credentials:

{
  "models": [
    {
      "key": "example-large@provider-a",
      "id": "example-large",
      "provider": "provider-a",
      "tier": "S",
      "capability": 0.95,
      "cost": "paid",
      "quota_per_window": null,
      "role": "stable",
      "scenarios": ["coding", "complex"],
      "fallback_chain": ["example-small@provider-b"]
    }
  ]
}

The five built-in public profiles are mechanism samples. Production users should provide a writable state directory and their own model-policy.json.

Utility dimensions

Multiple candidates are min-max normalized within the candidate set. Hard constraints run first.

Task types

The library understands the routing meaning of task types such as coding, complex, daily, simple, image, vision, batch, and maintenance. It does not classify natural-language tasks into those types.

For image and vision, candidates without matching vision/image scenarios or roles are removed rather than rescued by normalization.

Scheduler semantics

State flow

queued ──CAS claim──> running ──success──> done
   │                    │
   │                    └─failure──> failed
   │                                      │
deferred ──due──> queued                  ├─cooldown/backoff──> deferred
                                          └─prepared fallback──> queued

running + expired lease ──> queued or failed
user cancellation ──> cancelled
missed deadline ──> expired

Claim ownership

A successful claim records worker_id, attempt_id, lease_until, and heartbeat_at. Completion is accepted only from the exact owner and attempt. A late result from an expired or replaced attempt is discarded.

max_retries is the global failure ceiling. retry_before_fallback controls how many retry-then-fallback failures are retried before asking the executor to prepare a fallback. If no fallback hook succeeds, the task stays terminally failed even if the global ceiling has room left.

Failure actions

The scheduler never pretends that fallback happened. The executor must actually mutate its opaque payload or routing state and return True from prepare_fallback().

Migration from 1.0.x

Stop old 1.0.x workers before starting 1.1.x schedulers. Legacy running rows without ownership fields are deliberately recovered as worker-lost on the first 1.1.x tick.

Services

Task dashboard/API

python -m agent_model_router.taskserver \
  --host 127.0.0.1 \
  --port 8099 \
  --state-dir ./router-state

The dashboard uses MockExecutor unless embedded with a different executor. It is useful for task/state/API demonstrations; it is not an LLM runtime by itself. Pass --state-dir explicitly in generic deployments; if omitted, taskserver retains its historical Work-PWA-oriented default ~/.hermes/webui.

Stable HTTP endpoints are documented in docs/API.md.

OpenAI-compatible proxy

agent-model-router \
  --config ./model-policy.json \
  --host 127.0.0.1 \
  --port 8765

The policy must include provider connection configuration. Keep keys in environment variables; do not hardcode credentials in profile files.

Integration checklist

Before production use:

1. Set a writable, instance-specific state directory with LLM_ROUTER_STATE_DIR or configure_state_dir().
2. Replace built-in sample profiles with your real models and quotas.
3. Configure provider transport and credentials in the access layer or proxy configuration.
4. Pass a meaningful task_type.
5. Record quota usage and upstream failures so cooldown/quota decisions have data.
6. Record provider status/latency and pass ProviderHealth into routing if health-aware selection is required.
7. Use SQLite for multiple scheduler processes.
8. Implement and test prepare_fallback() if tasks must switch models automatically.
9. Treat recommendations as advice; retain an explicit user/operator override.

State files

State filenames intentionally keep their historical names for backward compatibility:

• model-policy.json
• model-quota.json
• model-cooldown.json
• model-health.json
• preferences.json
• model-tasks.json (JSON task backend)
• model-scheduler.db

The package/import/CLI names are agent-model-router, agent_model_router, and agent-model-router respectively.

Benchmark

The bundled benchmark is a deterministic synthetic strategy comparison, not a claim about real provider quality or latency:

python -m agent_model_router.benchmark --tasks 300 --seed 42

Current 1.1.2 output for that command:

See docs/BENCHMARK.md for methodology, real local concurrency measurements, and the known quota-write bottleneck.

Tests

python -m pytest tests/ -q

Release 1.1.2 baseline:

297 passed, 1 skipped, 59 subtests passed

The skipped test is an opt-in live smoke test. Enable it with:

MODEL_SCHEDULER_SMOKE_BASE_URL=... \
MODEL_SCHEDULER_SMOKE_API_KEY=... \
MODEL_SCHEDULER_SMOKE_MODEL=... \
python -m pytest tests/test_live_smoke.py -v

Compatibility API

The early rule-chain APIs—assess_difficulty, route_model, and recommend_for_session—remain available for existing integrations. New code should prefer route_with_utility() or route_with_intent().

Release history

• 1.1.2: documentation release; rewritten public guide and synchronized API, benchmark, and release notes.
• 1.1.1: release workflow and validation hardening; heartbeat input validation.
• 1.1.0: SQLite claim ownership, leases/heartbeats, stale-worker recovery, owner-safe completion, and executable degradation actions.
• 1.0.0: package/project rename to agent-model-router.

See docs/RELEASES.md and CHANGELOG.md for details.

License

MIT. See LICENSE.