Semantic Routing#

This guide covers how to use semantic routing to direct queries to appropriate agents.

Overview#

Semantic routing uses vector similarity to match user queries with predefined routes, enabling:

Zero LLM calls for routing decisions
Sub-millisecond latency using Redis vector search
Dynamic route management without code changes

import os
os.environ.setdefault("OPENAI_API_KEY", "your-api-key-here")

REDIS_URL = os.environ.get("REDIS_URL", "redis://localhost:6379")

Basic Usage#

from redis_openai_agents import SemanticRouter, Route

# Define routes with reference phrases
router = SemanticRouter(
    name="support-router",
    redis_url=REDIS_URL,
    routes=[
        Route(
            name="billing",
            references=[
                "payment issue",
                "invoice problem",
                "refund request",
                "subscription billing",
                "charge on my card"
            ],
            metadata={"agent": "billing_agent", "priority": "high"}
        ),
        Route(
            name="technical",
            references=[
                "bug report",
                "error message",
                "not working",
                "crash",
                "performance issue"
            ],
            metadata={"agent": "tech_agent", "priority": "medium"}
        ),
        Route(
            name="sales",
            references=[
                "pricing information",
                "enterprise plan",
                "demo request",
                "upgrade options",
                "volume discount"
            ],
            metadata={"agent": "sales_agent", "priority": "high"}
        )
    ]
)

print(f"Router '{router.name}' initialized with {len(router.routes)} routes")

Routing Queries#

# Test various queries
test_queries = [
    "I need help with my payment",
    "The app keeps crashing on startup",
    "Can I get a demo of your product?",
    "I want a refund for last month",
    "What are your enterprise pricing options?"
]

for query in test_queries:
    match = router.route(query)
    print(f"Query: '{query}'")
    print(f"  → Route: {match.name}")
    print(f"  → Agent: {match.metadata.get('agent')}")
    print(f"  → Distance: {match.distance:.4f}")
    print()

Distance Thresholds#

Each route can have its own distance threshold for matching precision.

# Router with custom thresholds
precise_router = SemanticRouter(
    name="precise-router",
    redis_url=REDIS_URL,
    routes=[
        Route(
            name="password_reset",
            references=["reset password", "forgot password", "can't log in"],
            distance_threshold=0.2,  # Strict matching
            metadata={"action": "password_reset_flow"}
        ),
        Route(
            name="general_help",
            references=["help", "question", "how do I", "support"],
            distance_threshold=0.4,  # Lenient matching (catch-all)
            metadata={"action": "general_support"}
        )
    ]
)

# Test threshold behavior
queries = [
    "I forgot my password",           # Should match password_reset
    "How do I change my settings?",   # Should match general_help
    "Reset my account password",      # Should match password_reset
]

for query in queries:
    match = precise_router.route(query)
    print(f"'{query}' → {match.name} (distance: {match.distance:.4f})")

Integration with Agents#

from agents import Agent, Runner

# Define specialized agents
billing_agent = Agent(
    name="billing_agent",
    instructions="You handle billing and payment issues. Be helpful with refunds and subscriptions."
)

tech_agent = Agent(
    name="tech_agent",
    instructions="You handle technical issues. Help troubleshoot bugs and errors."
)

sales_agent = Agent(
    name="sales_agent",
    instructions="You handle sales inquiries. Provide pricing and demo information."
)

# Agent registry
agents = {
    "billing_agent": billing_agent,
    "tech_agent": tech_agent,
    "sales_agent": sales_agent
}

async def route_and_run(query: str) -> str:
    """Route query to appropriate agent and run."""
    # Route the query (no LLM call!)
    match = router.route(query)
    agent_name = match.metadata.get("agent")
    
    print(f"Routing to: {agent_name}")
    
    # Get the agent
    agent = agents.get(agent_name)
    if not agent:
        return f"No agent found for route: {match.name}"
    
    # Run the agent
    result = await Runner.run(agent, input=query)
    return result.final_output

# Route and run
response = await route_and_run("I was charged twice for my subscription")
print(f"\nResponse: {response[:200]}...")

Dynamic Route Management#

Routes can be added or removed at runtime.

# Add a new route
router.add_route(
    Route(
        name="feedback",
        references=["feedback", "suggestion", "feature request", "improvement idea"],
        metadata={"agent": "feedback_agent"}
    )
)

print(f"Router now has {len(router.routes)} routes")

# Test new route
match = router.route("I have a suggestion for a new feature")
print(f"Query routed to: {match.name}")

# Remove a route
router.remove_route("feedback")
print(f"Router now has {len(router.routes)} routes")

Multi-Match Routing#

Get multiple potential matches for a query.

# Get top-k matches
query = "I need help with billing and have a technical question"
matches = router.route_many(query, k=3)

print(f"Query: '{query}'")
print("\nTop matches:")
for i, match in enumerate(matches, 1):
    print(f"  {i}. {match.name} (distance: {match.distance:.4f})")

Best Practices#

1. Use Diverse Reference Phrases#

Include various ways users might express the same intent:

Route(
    name="cancel",
    references=[
        "cancel subscription",
        "stop my plan",
        "end membership",
        "don't want to continue",
        "unsubscribe"
    ]
)

2. Have a Fallback Route#

Route(
    name="general",
    references=["help", "question", "other"],
    distance_threshold=0.5,  # Very lenient
    metadata={"agent": "general_agent"}
)

3. Monitor Route Performance#

Track which routes are used and their match distances to refine references.

Cleanup#

# Clean up routers
router.clear()
precise_router.clear()

print("Routers cleaned up!")

Version