osr-enterprise-builder

RouterBuilder(registry, settings=None) is a fluent builder; every with_* returns the builder and build() returns an EnterpriseRouter (thread-safe facade: route, learn, execute, run, health_snapshot). It raises ConfigurationError for an empty registry, no strategies, or duplicate strategy names.

Reference chain

from opensmartroute.enterprise import (RouterBuilder, CacheMiddleware, TenantMiddleware, TimeoutMiddleware,
                                       LoggingTelemetry, MetricsTelemetry, FileAuditSink)
from opensmartroute.observability import MemorySink, MetricsSink
from opensmartroute.security import GuardMiddleware

metrics = MetricsTelemetry()
app = (
    RouterBuilder(registry)
    .with_rules(load_rules("rules.yaml"))            # declarative preferences (RulesStrategy)
    .with_defaults()                                 # capability + similarity + Thompson bandit
    .with_slm()                                      # the routing SLM shipped in the package (or a model file / RouterSLM)
    .with_auto_learning(state_dir=".osr-state")      # IRT + Bradley-Terry + LinUCB + Markov, persisted
    .with_health(latency_slo_ms=3000)                # breakers, rate limits, budgets; HealthPolicy + HealthStrategy
    .with_guard(redact=True, output="redact")        # InputGuard first in the chain; OutputGuard screens every answer
    .with_origin_policy()                            # sensitive tool arguments must come from the user / trusted context
    .with_limits(max_steps=12, max_cost_usd=2.0)     # ResourceLimits per task for runaway agents
    .with_web_grounding(threshold=0.5)               # requests about current facts get screened web hits as a system block
    .with_knowledge("state/knowledge.jsonl")         # knowledge graph: engineered context before, learning after; KnowledgeStrategy
    .with_spread()                                   # weighted spread across same-family deployments (metadata.weight)
    .with_savings(baseline="max")                    # SavingsLedger: what routing saved vs always-the-priciest
    .with_middleware(TenantMiddleware({"acme": {"max_cost_per_1k": 0.005, "data_boundary": "private",
                                                "deny_targets": ["llm-frontier"]}}),   # order = execution order
                     CacheMiddleware(max_size=4096, ttl_s=30),
                     TimeoutMiddleware(budget_ms=250))
    .with_telemetry(LoggingTelemetry(), metrics)     # on_decision / on_outcome / on_error
    .with_tracing(MemorySink(), MetricsSink())       # spans + events of every stage (docs/OBSERVABILITY.md)
    .with_audit(FileAuditSink("audit.jsonl"))        # hash-chained, tamper-evident
    .with_calibration(conformal_alpha=0.1)           # temperature scaling + conformal candidate sets
    .with_retrieval(narrow_above=500, narrow_to=50)  # BM25 + dense RRF narrowing for big catalogues
    .with_llm_judge(LLMJudgeStrategy(judge), escalate_below=0.6)
    .with_objective(Objective(cost=0.3))
    .build()
)

Other options: with_strategy(s, weight), with_policy(policy) (wrapped by HealthPolicy when health is on), with_feedback(FeedbackStore), with_state_store(store, save_every), with_queue_awareness(slo_ms) (in-flight tracking + Erlang-C latency), with_fair_share(weights), with_shadow(candidate_router, mode="shadow"|"ab") (SPRT-judged), with_router_options(**kw), with_components(registry) (decorator SDK). with_guard(guard=None, redact=False, learned=False, output=None) takes output="flag"|"redact"|"block" or an OutputGuard; GuardMiddleware(...) in with_middleware is the same thing placed by hand. with_knowledge(graph_or_path=None, strategy=True, learn=True, save_every=50, budget_tokens=..., max_facts=..., max_web_hits=..., subject_of=...) ingests the catalogue into a knowledge.KnowledgeGraph (in memory, or the JSONL file at the path), places a ContextMiddleware after the web grounding so live hits are in the engineered block, and adds the KnowledgeStrategy; the graph is router.knowledge (to_jsonld / to_ntriples / to_graphml / to_cypher for the graph tooling).

Per-target controls live in the catalogue, not in code: with_health() calls HealthRegistry.apply_limits on every target's metadata - rate_limit: 60 (per minute) or {per_second: 2, burst: 10}, budget: 5 (USD per hour) or {usd: 100, per: day}, breaker: {failure_threshold, window_s, recovery_s, half_open_max_calls}; a malformed declaration raises ConfigurationError at build time. SpreadStrategy reads metadata.weight across targets that share a family (the same model behind several regions or keys).

For a router that already exists (the CLI's, a Router you built by hand): enterprise.harden(router, guard=True, redact_pii=True, output_guard="redact", health=True, audit="audit.jsonl", origin_policy=True, limits=True, web_grounding=True, knowledge=True, state_store="redis://...") returns the same EnterpriseRouter osr serve --guard --redact-pii --output-guard redact --health --audit ... --origin-policy --limits --web-grounding --state-store ... builds (enterprise.stores.open_state_store understands redis://, postgresql://, mysql://, sqlite:/// and file:///).

Optional learners and controls added with with_strategy(...) (each is a Strategy; those with a model are persisted by with_auto_learning / with_state_store like the built-in ones): learning.HistoryTargetStrategy() (multi-turn: history-target joint embeddings), learning.UserAdaptiveStrategy() (per-user shrinkage on context["user_id"]), strategies.EdgeCloudStrategy() (metadata.tier edge/cloud vs latency SLO and data boundary), strategies.TokenBudgetStrategy() (elastic <id>@<budget> siblings), strategies.AuctionStrategy() (bias-corrected second-price bids), strategies.HiddenStateStrategy(state_fn, targets, dim), discovery.SchemaAwareStrategy() (tool input_schema coverage). After the answer: signals.UncertaintyGate as the Cascade quality gate, strategies.ProtocolPolicy to pick single / cascade / aggregate / debate / handoff per request, strategies.MixtureOfAgents for top-k aggregation, strategies.SelfEscalation + wrap_stream for mid-stream escalation and learning.HandoffPolicy for permanent hand-off of failing agent tasks. Energy and carbon: math.EnergyModel / hardware_profile() fill cost.wh_per_1k_tokens / gco2_per_1k_tokens that Objective(energy=, carbon=) weighs.

Middleware contract

class Middleware(ABC):
    def __call__(self, request: RouteRequest, next_: RouteFn) -> RouteDecision: ...

Raise SecurityError/ValidationError to reject; mutate request.constraints to enforce tenant defaults (that is what TenantMiddleware does); wrap next_(request) for caching/timing.

Telemetry contract

Telemetry.on_decision(request, decision), on_outcome(outcome), on_error(request, error). Never log raw text - LoggingTelemetry logs a SHA-256 prefix + length. MetricsTelemetry.snapshot() gives decisions_by_target, route_latency_ms{p50,p95,p99}, policy_rejections_total, mean_confidence. opensmartroute.adapters.OpenTelemetryTelemetry needs the otel extra.

Tracing contract

Telemetry sees decisions; the tracer sees how they were made. with_tracing(*sinks_or_tracer, sample_rate=) (no arguments: Tracer.from_settings() / OSR_OBSERVABILITY_*) records a request span per call with nested route / plan / execute spans and events route.signals, route.policy, route.rank (per-strategy ms), cache.hit|miss, tenant.rejected, route.slow, guard.*, health.breaker, shadow.*, fairshare.*, learn.outcome. Sinks: MemorySink (trace(request_id)), MetricsSink (prometheus()), LoggingSink, FileSink, adapters.optional.OpenTelemetrySink. Inside your own middleware emit with current_tracer().event("name", **attrs) - never pass text, only ids and numbers. See docs/OBSERVABILITY.md.

Persisting learner state

with_auto_learning(state_dir) writes JSON files; for shared infrastructure use enterprise.stores: RedisStateStore, SQLStateStore, plus wrappers EncryptedStateStore (needs crypto extra), VersionedStateStore(Migration...), BatchedStateStore, NamespacedStateStore. Pass with with_state_store(store, save_every=10). Multi-replica: one writer, readers reload; state is quarantined on load failure rather than trusted blindly.

Closing the loop

d = app.route(RouteRequest("...", constraints=RequestConstraints(tenant="acme")))
app.learn(Outcome(d.request_id, d.target.id, success=True, quality=0.9, cost_usd=0.002, latency_ms=1200,
                  domains=d.trace.signals.domains, complexity=d.trace.signals.complexity))
res = app.run("...")          # route -> execute plan -> learn, with the full middleware chain
res = app.run("...", fallbacks=True)  # and, like `osr serve`: a failing or empty answer is a failed outcome and the
                                     # next ranked target takes the request (execution.execute_with_fallback)

app.health_snapshot() exposes breaker state per target.health_key (family-aware); app.shadow returns the ShadowMiddleware for A/B statistics.

Checklist before shipping

• Middleware order: guard -> tenant -> cache -> timeout.
• TenantMiddleware(require=True) unless anonymous traffic is expected.
• Set latency_slo_ms, Objective.quality_floor, and per-tenant max_cost_per_1k.
• Persist state (state_dir or a StateStore) and mount it in the container (OSR_STATE).
• Export metrics.snapshot() and keep the audit sink on durable storage.