Core Examples

Production-ready code examples for StrataRouter Core.

Basic Routing

Simple Router

from stratarouter import Router, Route

# Create router
router = Router(dimension=384, threshold=0.5)

# Define routes
billing = Route("billing")
billing.description = "Billing and payment questions"
billing.keywords = ["invoice", "payment", "charge", "refund"]

support = Route("support")
support.description = "Technical support"
support.keywords = ["bug", "error", "crash", "login"]

router.add_route(billing)
router.add_route(support)

# Build index (with real embeddings)
embeddings = get_embeddings([billing.description, support.description])
router.build_index(embeddings)

# Route queries
query_embedding = get_embedding("Where is my invoice?")
result = router.route("Where is my invoice?", query_embedding)

print(f"Route: {result.route_id}")  # "billing"
print(f"Confidence: {result.scores.confidence:.2f}")  # 0.89

With High-Level Python API

from stratarouter import Router, Route

# Create router with encoder
router = Router(encoder="sentence-transformers/all-MiniLM-L6-v2")

# Add routes
router.add(Route(
    id="billing",
    description="Billing questions",
    keywords=["invoice", "payment"]
))

router.add(Route(
    id="support",
    description="Technical support",
    keywords=["bug", "error"]
))

# Build index (automatic encoding)
router.build_index()

# Route (automatic encoding)
result = router.route("I need my invoice")
print(result.route_id)  # "billing"

Customer Support Routing

from stratarouter import Router, Route

# Create router
router = Router(encoder="all-MiniLM-L6-v2", threshold=0.6)

# Define support categories
categories = {
    "billing": {
        "description": "Billing, invoices, payments, refunds, pricing",
        "keywords": ["invoice", "payment", "charge", "refund", "billing", "price"],
        "examples": [
            "Where's my invoice?",
            "I need a refund",
            "Update my payment method",
            "How much does it cost?"
        ]
    },
    "technical": {
        "description": "Technical issues, bugs, errors, crashes",
        "keywords": ["bug", "error", "crash", "broken", "not working"],
        "examples": [
            "App keeps crashing",
            "Getting an error message",
            "Feature not working",
            "System is down"
        ]
    },
    "account": {
        "description": "Account management, login, password, profile",
        "keywords": ["login", "password", "account", "profile", "signup"],
        "examples": [
            "Can't login",
            "Reset my password",
            "Update my email",
            "Close my account"
        ]
    },
    "general": {
        "description": "General inquiries, information, how-to questions",
        "keywords": ["how", "what", "where", "when", "info"],
        "examples": [
            "How do I use this?",
            "What features do you have?",
            "Where is the documentation?"
        ]
    }
}

# Add routes
for cat_id, cat_info in categories.items():
    route = Route(
        id=cat_id,
        description=cat_info["description"],
        keywords=cat_info["keywords"],
        examples=cat_info["examples"]
    )
    router.add(route)

# Build index
router.build_index()

# Route customer queries
queries = [
    "My credit card was charged twice",
    "The app won't open on my phone",
    "I forgot my password",
    "Do you have a mobile app?"
]

for query in queries:
    result = router.route(query)
    print(f"{query:50s} → {result.route_id:12s} ({result.confidence:.2f})")

Output:

My credit card was charged twice                  → billing      (0.92)
The app won't open on my phone                    → technical    (0.88)
I forgot my password                              → account      (0.95)
Do you have a mobile app?                         → general      (0.78)

Intent Classification

from stratarouter import Router, Route

# Create intent classifier
classifier = Router(encoder="all-MiniLM-L6-v2")

# Define intents
intents = [
    Route(
        id="book_flight",
        description="Book a flight, make a reservation",
        keywords=["book", "flight", "ticket", "reservation", "fly"],
        examples=[
            "I want to book a flight to Paris",
            "Book me a ticket to NYC",
            "Make a flight reservation"
        ]
    ),
    Route(
        id="cancel_booking",
        description="Cancel a booking, get a refund",
        keywords=["cancel", "refund", "change", "modify"],
        examples=[
            "Cancel my flight",
            "I need to cancel my reservation",
            "Get a refund for my booking"
        ]
    ),
    Route(
        id="check_status",
        description="Check flight status, track booking",
        keywords=["status", "where", "track", "when", "arrival"],
        examples=[
            "What's my flight status?",
            "Where is my booking?",
            "When does my flight arrive?"
        ]
    )
]

for intent in intents:
    classifier.add(intent)

classifier.build_index()

# Classify queries
test_queries = [
    "I want to fly to London next week",
    "I need to cancel my trip",
    "Is my flight on time?"
]

for query in test_queries:
    result = classifier.route(query)
    print(f"'{query}' → {result.route_id}")

Multi-Language Routing

from stratarouter import Router, Route

# Create multi-language router
router = Router(encoder="sentence-transformers/paraphrase-multilingual-mpnet-base-v2")

# Define language routes
languages = [
    Route(
        id="en",
        description="English language queries",
        keywords=["hello", "thanks", "please", "help"],
        examples=[
            "Hello, how are you?",
            "Thank you very much",
            "Can you help me?"
        ]
    ),
    Route(
        id="es",
        description="Spanish language queries",
        keywords=["hola", "gracias", "por favor", "ayuda"],
        examples=[
            "Hola, ¿cómo estás?",
            "Muchas gracias",
            "¿Puedes ayudarme?"
        ]
    ),
    Route(
        id="fr",
        description="French language queries",
        keywords=["bonjour", "merci", "s'il vous plaît", "aide"],
        examples=[
            "Bonjour, comment allez-vous?",
            "Merci beaucoup",
            "Pouvez-vous m'aider?"
        ]
    )
]

for lang in languages:
    router.add(lang)

router.build_index()

# Route multi-language queries
queries = [
    "Hello, I need help",
    "Hola, necesito ayuda",
    "Bonjour, j'ai besoin d'aide"
]

for query in queries:
    result = router.route(query)
    print(f"{query:40s} → {result.route_id} ({result.confidence:.2f})")

Confidence Thresholding

from stratarouter import Router, Route

router = Router(encoder="all-MiniLM-L6-v2", threshold=0.7)

# Add routes
router.add(Route(id="billing", keywords=["invoice"]))
router.add(Route(id="support", keywords=["bug"]))
router.build_index()

# Route with confidence handling
query = "I have a problem"  # Ambiguous query

result = router.route(query)

if result.confidence > 0.8:
    # High confidence - route directly
    print(f"High confidence: Routing to {result.route_id}")
    handle_route(result.route_id, query)

elif result.confidence > 0.5:
    # Medium confidence - ask for clarification
    print(f"Medium confidence: Ask user to clarify")
    print(f"Did you mean: {result.route_id}?")

else:
    # Low confidence - escalate to human
    print("Low confidence: Escalate to human agent")
    escalate_to_human(query)

Batch Routing

from stratarouter import Router, Route
import time

router = Router(encoder="all-MiniLM-L6-v2")

# Add routes
for i in range(10):
    router.add(Route(id=f"route_{i}", description=f"Route {i}"))

router.build_index()

# Batch route queries
queries = [f"Query {i}" for i in range(100)]

start = time.time()
results = []

for query in queries:
    result = router.route(query)
    results.append(result)

elapsed = time.time() - start

print(f"Routed {len(queries)} queries in {elapsed:.3f}s")
print(f"Average: {elapsed/len(queries)*1000:.2f}ms per query")
print(f"Throughput: {len(queries)/elapsed:.0f} req/s")

# Analyze results
avg_confidence = sum(r.confidence for r in results) / len(results)
print(f"Average confidence: {avg_confidence:.2f}")

Route with Metadata

from stratarouter import Router, Route

router = Router(encoder="all-MiniLM-L6-v2")

# Add routes with metadata
billing_route = Route(
    id="billing",
    description="Billing questions",
    keywords=["invoice"],
    metadata={
        "priority": "high",
        "handler": "billing_team@company.com",
        "sla_minutes": 15
    }
)

support_route = Route(
    id="support",
    description="Technical support",
    keywords=["bug"],
    metadata={
        "priority": "medium",
        "handler": "support_team@company.com",
        "sla_minutes": 60
    }
)

router.add(billing_route)
router.add(support_route)
router.build_index()

# Route and use metadata
result = router.route("I need my invoice")
route = router.routes[result.route_id]

print(f"Route: {result.route_id}")
print(f"Priority: {route.metadata['priority']}")
print(f"Handler: {route.metadata['handler']}")
print(f"SLA: {route.metadata['sla_minutes']} minutes")

Pattern-Based Routing

from stratarouter import Router, Route

router = Router(encoder="all-MiniLM-L6-v2")

# Add routes with patterns
help_route = Route(
    id="help",
    description="Help and documentation",
    patterns=["/help", "/docs", "/faq"]
)

agent_route = Route(
    id="human_agent",
    description="Human agent escalation",
    patterns=["talk to agent", "human support", "speak to person"]
)

router.add(help_route)
router.add(agent_route)
router.build_index()

# These will match patterns perfectly (rule score = 1.0)
queries = [
    "/help",  # Exact pattern match
    "I want to talk to agent",  # Pattern match
    "show me /docs"  # Contains pattern
]

for query in queries:
    result = router.route(query)
    print(f"{query:30s} → {result.route_id:15s} (pattern: {result.scores.pattern:.1f})")

Dynamic Route Updates

from stratarouter import Router, Route

# Create router
router = Router(encoder="all-MiniLM-L6-v2")

# Initial routes
router.add(Route(id="billing", keywords=["invoice"]))
router.add(Route(id="support", keywords=["bug"]))
router.build_index()

# Route some queries
result1 = router.route("invoice issue")
print(f"Before: {result1.route_id}")

# Add new route
router.add(Route(
    id="urgent_billing",
    description="Urgent billing issues",
    keywords=["urgent", "invoice", "overdue"]
))

# Rebuild index
router.build_index()

# Route again - may go to new route
result2 = router.route("urgent invoice issue")
print(f"After: {result2.route_id}")

Saving and Loading

from stratarouter import Router, Route

# Create and save
router = Router(encoder="all-MiniLM-L6-v2")
router.add(Route(id="billing", keywords=["invoice"]))
router.add(Route(id="support", keywords=["bug"]))
router.build_index()

router.save("my_router.json")

# Load later
loaded_router = Router.load(
    "my_router.json",
    encoder="all-MiniLM-L6-v2"  # Must provide encoder
)

# Use loaded router
result = loaded_router.route("I need my invoice")
print(result.route_id)

Performance Monitoring

from stratarouter import Router, Route
import time
import statistics

router = Router(encoder="all-MiniLM-L6-v2")

# Add routes
for i in range(100):
    router.add(Route(id=f"route_{i}", description=f"Route {i}"))

router.build_index()

# Measure performance
latencies = []

for i in range(1000):
    query = f"Test query {i}"
    start = time.perf_counter()
    result = router.route(query)
    latency = (time.perf_counter() - start) * 1000
    latencies.append(latency)

# Statistics
print(f"P50: {statistics.median(latencies):.2f}ms")
print(f"P95: {statistics.quantiles(latencies, n=20)[18]:.2f}ms")
print(f"P99: {statistics.quantiles(latencies, n=100)[98]:.2f}ms")
print(f"Mean: {statistics.mean(latencies):.2f}ms")
print(f"Std: {statistics.stdev(latencies):.2f}ms")

Error Handling

from stratarouter import Router, Route

def safe_route(router: Router, query: str):
    """Route with comprehensive error handling"""
    try:
        if not query or not query.strip():
            return {"error": "Empty query"}

        if not router.is_index_built:
            router.build_index()

        result = router.route(query)

        return {
            "success": True,
            "route_id": result.route_id,
            "confidence": result.confidence
        }

    except ValueError as e:
        return {"error": f"Validation error: {e}"}

    except RuntimeError as e:
        if "dimension mismatch" in str(e).lower():
            return {"error": "Encoder dimension doesn't match router"}
        elif "index not built" in str(e).lower():
            return {"error": "Index not built"}
        else:
            return {"error": f"Runtime error: {e}"}

    except Exception as e:
        return {"error": f"Unexpected error: {e}"}

# Use it
router = Router()
router.add(Route(id="test", description="Test"))

result = safe_route(router, "test query")
if "error" in result:
    print(f"Error: {result['error']}")
else:
    print(f"Route: {result['route_id']}")

Next Steps

• API Reference - Complete API documentation
• Performance - Performance tuning
• Integrations - Framework integrations