Skip to content

Python SDK Reference

Complete Python API reference for StrataRouter Core and Runtime.

Installation

pip install stratarouter

Core API

Router

Main routing engine for semantic routing decisions.

Class: Router

from stratarouter import Router, RouterConfig

router = Router(config: RouterConfig = None, **kwargs)

Parameters: - config (RouterConfig, optional): Router configuration - **kwargs: Configuration parameters (alternative to config object) - dimension (int): Embedding dimension (default: 384) - threshold (float): Confidence threshold (default: 0.5) - top_k (int): Number of candidates (default: 5) - enable_calibration (bool): Enable confidence calibration (default: True)

Example: 

# Method 1: Using config object
config = RouterConfig(dimension=384, threshold=0.7)
router = Router(config)

# Method 2: Direct parameters
router = Router(dimension=384, threshold=0.7)

Method: add_route

Add a route to the router.

router.add_route(route: Route) -> None

Parameters: - route (Route): Route definition to add

Raises: - ValueError: If route validation fails - DuplicateRouteError: If route ID already exists

Example: 

route = Route(
    id="billing",
    description="Billing and payment questions",
    examples=["Where's my invoice?", "Update payment method"],
    keywords=["invoice", "payment", "billing"]
)
router.add_route(route)

Method: build_index

Build the routing index from embeddings.

router.build_index(embeddings: List[List[float]]) -> None

Parameters: - embeddings (List[List[float]]): List of embedding vectors, one per route

Raises: - ValueError: If embedding count doesn't match route count - DimensionMismatch: If embedding dimensions don't match config

Example: 

# Get embeddings for all routes
embeddings = [
    get_embedding(route.description + " " + " ".join(route.examples))
    for route in router.routes
]

# Build index
router.build_index(embeddings)

Method: route

Route a query to the best matching route.

router.route(
    query: str,
    embedding: List[float],
    k: int = 1
) -> RouteResult

Parameters: - query (str): Query text - embedding (List[float]): Query embedding vector - k (int, optional): Number of alternatives to return (default: 1)

Returns: - RouteResult: Routing result with scores and latency

Raises: - IndexNotBuilt: If index not built before routing - ValueError: If query is empty or embedding is invalid - DimensionMismatch: If embedding dimension doesn't match

Example: 

query = "I need my invoice"
embedding = get_embedding(query)

result = router.route(query, embedding)

print(f"Route: {result.route_id}")
print(f"Confidence: {result.scores.confidence:.2f}")
print(f"Latency: {result.latency_ms:.1f}ms")

# With alternatives
result = router.route(query, embedding, k=3)
for alt_id, alt_conf in result.alternatives:
    print(f"Alternative: {alt_id} ({alt_conf:.2f})")

Property: route_count

Get the number of routes in the router.

count = router.route_count

Returns: - int: Number of routes

Property: is_index_built

Check if the index has been built.

built = router.is_index_built

Returns: - bool: True if index is built

Route

Route definition with examples, keywords, and patterns.

Class: Route

from stratarouter import Route

route = Route(
    id: str,
    description: str = "",
    examples: List[str] = None,
    keywords: List[str] = None,
    patterns: List[str] = None,
    metadata: Dict[str, str] = None,
    threshold: float = None,
    tags: List[str] = None
)

Parameters: - id (str): Unique route identifier - description (str, optional): Human-readable description - examples (List[str], optional): Example queries - keywords (List[str], optional): Important keywords - patterns (List[str], optional): Regex patterns to match - metadata (Dict[str, str], optional): Additional metadata - threshold (float, optional): Route-specific confidence threshold - tags (List[str], optional): Tags for categorization

Example: 

route = Route(
    id="billing",
    description="Billing and payment questions",
    examples=[
        "Where's my invoice?",
        "How do I update my payment method?",
        "I want a refund"
    ],
    keywords=["invoice", "payment", "refund", "billing", "charge"],
    patterns=[
        r"(?i)invoice",
        r"(?i)payment.*fail",
        r"(?i)charge.*card"
    ],
    metadata={
        "category": "finance",
        "priority": "high"
    },
    threshold=0.75,
    tags=["customer-service", "finance"]
)

Method: validate

Validate route configuration.

route.validate() -> None

Raises: - ValueError: If validation fails

Validation Rules: - ID cannot be empty - Must have either examples or description - Patterns must be valid regex - Threshold must be between 0 and 1

RouteResult

Result of a routing operation.

Class: RouteResult

class RouteResult:
    route_id: str
    scores: RouteScores
    latency_ms: float
    alternatives: List[Tuple[str, float]]  # If k > 1

Attributes: - route_id (str): Matched route ID - scores (RouteScores): Score breakdown - latency_ms (float): Routing latency in milliseconds - alternatives (List[Tuple[str, float]]): Alternative routes with confidence

Example: 

result = router.route(query, embedding, k=3)

# Primary route
print(f"Route: {result.route_id}")
print(f"Confidence: {result.scores.confidence}")

# Score breakdown
print(f"Semantic: {result.scores.semantic}")
print(f"Keyword: {result.scores.keyword}")
print(f"Pattern: {result.scores.pattern}")
print(f"Total: {result.scores.total}")

# Alternatives
for route_id, confidence in result.alternatives:
    print(f"{route_id}: {confidence:.2f}")

# Performance
print(f"Latency: {result.latency_ms:.1f}ms")

RouteScores

Detailed score breakdown.

Class: RouteScores

class RouteScores:
    semantic: float      # Semantic similarity (0-1)
    keyword: float       # Keyword matching (0-1)
    pattern: float       # Pattern matching (0-1)
    total: float         # Fused score (0-1)
    confidence: float    # Calibrated confidence (0-1)

Score Types:

  1. Semantic Score - Cosine similarity on embeddings
  2. Based on dense vector similarity
  3. Captures semantic meaning and context

  4. Weight: 0.64 (default)

  5. Keyword Score - BM25 keyword matching

  6. Intersection over union of keywords
  7. Sparse signal for exact terms and domain vocabulary

  8. Weight: 0.29 (default)

  9. Pattern Score - Regex pattern matching

  10. Rule-based deterministic matching
  11. Handles known phrases with certainty

  12. Weight: 0.07 (default)

  13. Total Score - Weighted fusion

  14. total = 0.64*semantic + 0.29*keyword + 0.07*pattern

  15. Confidence Score - Calibrated probability

  16. Isotonic regression on total score
  17. True probability estimate
  18. Use for thresholding

RouterConfig

Configuration for the router.

Class: RouterConfig

from stratarouter import RouterConfig

config = RouterConfig(
    dimension: int = 384,
    default_threshold: float = 0.5,
    top_k: int = 5,
    enable_calibration: bool = True,
    semantic_weight: float = 0.64,
    keyword_weight: float = 0.29,
    pattern_weight: float = 0.07
)

Parameters: - dimension (int): Embedding dimension (default: 384) - default_threshold (float): Default confidence threshold (default: 0.5) - top_k (int): Number of candidates to consider (default: 5) - enable_calibration (bool): Enable confidence calibration (default: True) - semantic_weight (float): Weight for semantic score (default: 0.64) - keyword_weight (float): Weight for keyword score (default: 0.29) - pattern_weight (float): Weight for pattern score (default: 0.07)

Example: 

config = RouterConfig(
    dimension=1536,       # OpenAI text-embedding-3-small
    default_threshold=0.7,  # Higher threshold for precision
    top_k=10,             # More candidates for accuracy
    enable_calibration=True,
    semantic_weight=0.70,  # Emphasize semantics
    keyword_weight=0.23,   # Moderate keyword signal
    pattern_weight=0.07    # Unchanged rule weight
)

Runtime API

CoreRuntimeBridge

Integration layer between Core routing and Runtime execution.

Class: CoreRuntimeBridge

from stratarouter_runtime import CoreRuntimeBridge, RuntimeConfig

bridge = CoreRuntimeBridge(config: RuntimeConfig)

Parameters: - config (RuntimeConfig): Runtime configuration

Example: 

config = RuntimeConfig(
    cache_enabled=True,
    batch_enabled=True,
    execution_timeout=60
)
bridge = CoreRuntimeBridge(config)

Method: execute (async)

Execute a routing decision.

result = await bridge.execute(
    decision: RouteResult,
    context: Dict[str, Any] = None
) -> ExecutionResult

Parameters: - decision (RouteResult): Routing decision from Core - context (Dict[str, Any], optional): Execution context

Returns: - ExecutionResult: Execution result with response and metrics

Example: 

# Route with Core
route_decision = router.route(query, embedding)

# Execute with Runtime
result = await bridge.execute(
    decision=route_decision,
    context={
        "user_id": "user-123",
        "session_id": "sess-456"
    }
)

print(f"Response: {result.response}")
print(f"Cache hit: {result.cache_hit}")
print(f"Latency: {result.latency_ms}ms")
print(f"Cost: ${result.cost_usd}")

RuntimeConfig

Configuration for runtime execution.

Class: RuntimeConfig

from stratarouter_runtime import RuntimeConfig

config = RuntimeConfig(
    # Execution
    execution_timeout: int = 60,
    max_retries: int = 3,
    retry_delay_ms: int = 100,

    # Cache
    cache_enabled: bool = True,
    cache_backend: str = "redis",
    cache_ttl: int = 3600,
    cache_similarity_threshold: float = 0.95,

    # Batch
    batch_enabled: bool = True,
    batch_window_ms: int = 50,
    batch_max_size: int = 32,
    batch_similarity_threshold: float = 0.98,

    # State
    state_backend: str = "postgresql",
    checkpoint_interval: int = 10,

    # Observability
    metrics_enabled: bool = True,
    tracing_enabled: bool = True,
    log_level: str = "info"
)

Example: 

config = RuntimeConfig(
    execution_timeout=120,  # 2 minutes
    cache_enabled=True,
    cache_backend="redis",
    batch_enabled=True,
    metrics_enabled=True
)

ExecutionResult

Result of a runtime execution.

Class: ExecutionResult

class ExecutionResult:
    response: Any
    cache_hit: bool
    latency_ms: float
    cost_usd: float
    provider: str
    metadata: Dict[str, Any]

Attributes: - response (Any): LLM response - cache_hit (bool): Whether cache was hit - latency_ms (float): Total latency - cost_usd (float): Execution cost - provider (str): Provider used - metadata (Dict[str, Any]): Additional metadata

Complete Example

import asyncio
from stratarouter import Router, Route, RouterConfig
from stratarouter_runtime import CoreRuntimeBridge, RuntimeConfig

async def main():
    # Create router
    router_config = RouterConfig(
        dimension=384,
        threshold=0.7,
        enable_calibration=True
    )
    router = Router(router_config)

    # Add routes
    router.add_route(Route(
        id="billing",
        description="Billing and payment questions",
        examples=["Where's my invoice?", "Update payment"],
        keywords=["invoice", "payment", "billing"]
    ))

    router.add_route(Route(
        id="technical",
        description="Technical support",
        examples=["App crashes", "Error message"],
        keywords=["bug", "error", "crash"]
    ))

    # Build index
    embeddings = [
        get_embedding(r.description) 
        for r in router.routes
    ]
    router.build_index(embeddings)

    # Create runtime bridge
    runtime_config = RuntimeConfig(
        cache_enabled=True,
        batch_enabled=True,
        execution_timeout=60
    )
    bridge = CoreRuntimeBridge(runtime_config)

    # Process query
    query = "My payment failed yesterday"
    embedding = get_embedding(query)

    # Route
    route_result = router.route(query, embedding)
    print(f"Routed to: {route_result.route_id}")
    print(f"Confidence: {route_result.scores.confidence:.2f}")

    # Execute
    exec_result = await bridge.execute(
        decision=route_result,
        context={"user_id": "user-123"}
    )

    print(f"Response: {exec_result.response}")
    print(f"Cache hit: {exec_result.cache_hit}")
    print(f"Latency: {exec_result.latency_ms}ms")
    print(f"Cost: ${exec_result.cost_usd:.4f}")

if __name__ == "__main__":
    asyncio.run(main())

Error Handling

See Error Codes Reference for complete error documentation.

from stratarouter import Router, Error

try:
    result = router.route(query, embedding)
except Error.DimensionMismatch as e:
    print(f"Dimension mismatch: {e.expected} != {e.actual}")
except Error.IndexNotBuilt:
    print("Build index first")
    router.build_index(embeddings)
except Error as e:
    print(f"Error: {e}")

Next Steps