Skip to content

Python API

Complete Python API reference for StrataRouter Core.

Router Class

Constructor

from stratarouter import Router

router = Router(
    encoder="sentence-transformers/all-MiniLM-L6-v2",
    mode="local",  # or "cloud"
    dimension=384,
    threshold=0.5,
    **kwargs
)

Parameters: - encoder (str | BaseEncoder): Embedding model - String: Model name from HuggingFace or OpenAI - Object: Custom encoder instance - mode (str): Deployment mode - "local": Run locally (default) - "cloud": Use StrataRouter Cloud - dimension (int): Embedding dimension (default: 384) - threshold (float): Confidence threshold (default: 0.5) - api_key (str): API key for cloud mode

Example:

# Local with HuggingFace
router = Router(
    encoder="sentence-transformers/all-MiniLM-L6-v2",
    threshold=0.7
)

# Cloud mode
router = Router(
    mode="cloud",
    api_key="sr-xxx"
)

Methods

add()

Add a route to the router.

def add(route: Route) -> None

Parameters: - route (Route): Route definition

Raises: - ValueError: If route ID is empty or duplicate - ValueError: If route has no examples or description

Example:

from stratarouter import Route

route = Route(
    id="billing",
    description="Billing questions",
    keywords=["invoice", "payment"],
    examples=["Where's my invoice?"]
)

router.add(route)

build_index()

Build the routing index from route embeddings.

def build_index() -> None

Raises: - ValueError: If no routes added - RuntimeError: If encoding fails - RuntimeError: If index build fails

Example:

router.add(route1)
router.add(route2)
router.build_index()  # Automatically encodes routes

route()

Route a query to the best matching route.

def route(text: str, top_k: int = 1) -> RouteResult

Parameters: - text (str): Query text - top_k (int): Number of top routes to consider (default: 1)

Returns: - RouteResult: Routing result with route_id, confidence, scores

Raises: - ValueError: If text is empty - RuntimeError: If index not built - RuntimeError: If encoding fails

Example:

result = router.route("I need my invoice")

print(result.route_id)      # "billing"
print(result.confidence)    # 0.89
print(result.scores)        # {"semantic": 0.85, ...}
print(result.latency_ms)    # 2.3

save()

Save router state to file.

def save(path: str) -> None

Parameters: - path (str): File path to save to

Raises: - RuntimeError: If save fails

Example:

router.save("my_router.json")

load()

Load router from file (class method).

@classmethod
def load(path: str, **kwargs) -> Router

Parameters: - path (str): File path to load from - **kwargs: Override saved config

Returns: - Router: Loaded router instance

Raises: - FileNotFoundError: If file doesn't exist - ValueError: If file is invalid

Example:

router = Router.load(
    "my_router.json",
    encoder="sentence-transformers/all-MiniLM-L6-v2"
)

Properties

routes

@property
def routes() -> Dict[str, Route]

Get all routes.

Example:

for route_id, route in router.routes.items():
    print(f"{route_id}: {route.description}")

is_index_built

@property
def is_index_built() -> bool

Check if index is built.

Example:

if not router.is_index_built:
    router.build_index()

Route Class

Constructor

from stratarouter import Route

route = Route(
    id="billing",
    description="Billing questions",
    examples=["Where's my invoice?"],
    keywords=["invoice", "payment"],
    patterns=["show my invoice"],
    threshold=0.7,
    tags=["customer", "finance"]
)

Parameters: - id (str): Unique route identifier (required) - description (str): Human-readable description - examples (List[str]): Example queries - keywords (List[str]): Important keywords - patterns (List[str]): Exact patterns to match - threshold (float): Route-specific threshold - tags (List[str]): Tags for organization

Validation

Routes are validated on creation:

# Valid
route = Route(id="test", description="Test route")

# Invalid - will raise ValueError
route = Route(id="")  # Empty ID
route = Route(id="test")  # No description or examples

Builder Pattern

route = (Route(id="billing")
    .with_description("Billing questions")
    .with_keywords(["invoice", "payment"])
    .with_examples(["Where's my invoice?"])
    .with_threshold(0.7))

RouteResult Class

Result from routing operation.

Attributes

@dataclass
class RouteResult:
    route_id: str           # Matched route ID
    confidence: float       # Confidence score (0-1)
    scores: Dict[str, float]  # Score breakdown
    latency_ms: float       # Routing latency
    metadata: Dict[str, Any]  # Additional data

Score Breakdown

result.scores = {
    "semantic": 0.85,   # Dense embedding score
    "keyword": 0.78,    # BM25 keyword score
    "pattern": 0.0,     # Rule matching score
    "total": 0.82,      # Fused score
    "confidence": 0.89  # Calibrated confidence
}

Custom Encoders

BaseEncoder

Base class for custom encoders.

from stratarouter.encoders import BaseEncoder

class CustomEncoder(BaseEncoder):
    @property
    def dimension(self) -> int:
        return 384

    def encode(self, texts: Union[str, List[str]]) -> np.ndarray:
        # Your encoding logic
        if isinstance(texts, str):
            texts = [texts]
        return np.array([self._encode_single(t) for t in texts])

    def _encode_single(self, text: str) -> np.ndarray:
        # Encode single text
        return np.random.rand(self.dimension)

Using Custom Encoder

encoder = CustomEncoder()
router = Router(encoder=encoder)

Built-in Encoders

HuggingFaceEncoder

from stratarouter.encoders import HuggingFaceEncoder

encoder = HuggingFaceEncoder(
    model_name="sentence-transformers/all-MiniLM-L6-v2",
    device="cuda",  # or "cpu"
    batch_size=32
)

router = Router(encoder=encoder)

OpenAIEncoder

from stratarouter.encoders import OpenAIEncoder

encoder = OpenAIEncoder(
    model="text-embedding-3-small",
    api_key="sk-..."
)

router = Router(encoder=encoder)

Error Handling

from stratarouter import Router, Route

try:
    router = Router(encoder="invalid-model")
except ImportError:
    print("Encoder dependencies not installed")

try:
    router.add(Route(id=""))  # Empty ID
except ValueError as e:
    print(f"Invalid route: {e}")

try:
    result = router.route("query")  # Index not built
except RuntimeError as e:
    if "Index not built" in str(e):
        router.build_index()
        result = router.route("query")

Type Hints

from typing import Dict, List, Optional
from stratarouter import Router, Route, RouteResult

def create_router(
    routes: List[Route],
    encoder: Optional[str] = None
) -> Router:
    router = Router(encoder=encoder or "all-MiniLM-L6-v2")
    for route in routes:
        router.add(route)
    router.build_index()
    return router

def route_queries(
    router: Router,
    queries: List[str]
) -> List[RouteResult]:
    return [router.route(q) for q in queries]

Async Support

For async operations (cloud mode):

import asyncio
from stratarouter import Router

async def route_async():
    router = Router(mode="cloud", api_key="sr-xxx")

    # Cloud routing is async
    result = await router.route_async("query")
    return result

result = asyncio.run(route_async())

Configuration

Global Config

from stratarouter import config

# Set default encoder
config.DEFAULT_ENCODER = "all-MiniLM-L6-v2"

# Set default threshold
config.DEFAULT_THRESHOLD = 0.7

# Enable debug logging
config.DEBUG = True

Per-Router Config

router = Router(
    encoder="all-MiniLM-L6-v2",
    threshold=0.7,
    dimension=384,
    cache_embeddings=True,
    batch_size=32
)

Best Practices

1. Reuse Router Instances

# Good - reuse router
router = Router()
for query in queries:
    router.route(query)

# Bad - create new router each time
for query in queries:
    router = Router()  # Expensive!
    router.route(query)

2. Build Index Once

# Good
router.build_index()
for query in queries:
    router.route(query)

# Bad
for query in queries:
    router.build_index()  # Rebuilds every time!
    router.route(query)

3. Use Batch Encoding

# Good - encode all at once
texts = [r.description for r in routes]
embeddings = encoder.encode(texts)

# Bad - encode one by one
for text in texts:
    encoder.encode([text])  # Slower!

4. Handle Errors Gracefully

def safe_route(router: Router, query: str) -> Optional[RouteResult]:
    try:
        return router.route(query)
    except RuntimeError as e:
        logger.error(f"Routing failed: {e}")
        return None

Examples

See Examples for complete code examples.

API Reference (Rust Core)

For the underlying Rust API, see the Rust documentation.