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.
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.
Raises:
- ValueError: If no routes added
- RuntimeError: If encoding fails
- RuntimeError: If index build fails
Example:
route()¶
Route a query to the best matching route.
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.
Parameters:
- path (str): File path to save to
Raises:
- RuntimeError: If save fails
Example:
load()¶
Load router from file (class method).
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:
Properties¶
routes¶
Get all routes.
Example:
is_index_built¶
Check if index is built.
Example:
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¶
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.