LangChain Integration¶
Use StrataRouter inside LangChain chains, agents, and retrievers.
Installation¶
Quick Start¶
The fastest way to add semantic routing to a LangChain application:
from stratarouter import Router, Route
from stratarouter.integrations.langchain import StrataRouterChain
from langchain_openai import ChatOpenAI
# 1. Build your router
router = Router(dimension=1536)
billing = Route("billing")
billing.description = "Billing, payments, invoices, and refunds"
billing.examples = ["Where's my invoice?", "I need a refund", "Payment failed"]
billing.keywords = ["invoice", "payment", "refund", "billing", "charge"]
support = Route("support")
support.description = "Technical issues, bugs, and product help"
support.examples = ["App is crashing", "Can't login", "Feature not working"]
support.keywords = ["bug", "crash", "error", "broken", "login"]
router.add_route(billing)
router.add_route(support)
router.build_index()
# 2. Create the router chain
router_chain = StrataRouterChain(router)
# 3. Route a query
result = router_chain.route("My payment keeps failing")
print(f"Route: {result.route_id}") # "billing"
print(f"Confidence: {result.confidence}") # 0.94
Router as a LangChain Chain¶
Use StrataRouter as a drop-in routing component in any chain:
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
# Billing chain
billing_chain = LLMChain(
llm=llm,
prompt=PromptTemplate.from_template(
"You are a billing specialist. Answer: {query}"
)
)
# Support chain
support_chain = LLMChain(
llm=llm,
prompt=PromptTemplate.from_template(
"You are a technical support engineer. Answer: {query}"
)
)
# Route between chains
def handle(query: str):
route_result = router_chain.route(query)
if route_result.route_id == "billing":
return billing_chain.run(query=query)
elif route_result.route_id == "support":
return support_chain.run(query=query)
print(handle("I haven't received my invoice"))
LangChain Retriever Integration¶
Use StrataRouter to dynamically select the best retriever:
from langchain.schema import BaseRetriever
from langchain_community.vectorstores import FAISS
from langchain_openai import OpenAIEmbeddings
embeddings = OpenAIEmbeddings()
# Specialized vector stores per domain
billing_db = FAISS.from_texts(billing_docs, embeddings)
support_db = FAISS.from_texts(support_docs, embeddings)
# Routing retriever
class RoutingRetriever(BaseRetriever):
def get_relevant_documents(self, query: str):
route = router_chain.route(query)
if route.route_id == "billing":
return billing_db.as_retriever().get_relevant_documents(query)
else:
return support_db.as_retriever().get_relevant_documents(query)
# Use in RAG chain
from langchain.chains import RetrievalQA
rag = RetrievalQA.from_chain_type(
llm=llm,
retriever=RoutingRetriever()
)
answer = rag.run("Where can I download my invoice?")
Agent Routing with LangChain¶
Use StrataRouter to dispatch tasks to specialized agents:
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import Tool
def create_billing_agent():
tools = [Tool(name="lookup_invoice", func=lookup_invoice, description="...")]
return create_openai_tools_agent(llm, tools, billing_prompt)
def create_support_agent():
tools = [Tool(name="check_status", func=check_status, description="...")]
return create_openai_tools_agent(llm, tools, support_prompt)
billing_agent = AgentExecutor(agent=create_billing_agent(), tools=[...])
support_agent = AgentExecutor(agent=create_support_agent(), tools=[...])
def dispatch(query: str):
route = router_chain.route(query)
if route.confidence < 0.5:
return "I'm not sure how to help with that. Could you clarify?"
if route.route_id == "billing":
return billing_agent.run(query)
else:
return support_agent.run(query)
Using Real OpenAI Embeddings¶
StrataRouter works with any embedding dimension. For OpenAI:
from openai import OpenAI
client = OpenAI()
def embed(text: str) -> list[float]:
resp = client.embeddings.create(
model="text-embedding-3-small", # 1536-dim
input=text
)
return resp.data[0].embedding
# Use 1536 dimensions for text-embedding-3-small
router = Router(dimension=1536)
# Build index from route embeddings
route_texts = [
billing.description + " " + " ".join(billing.examples),
support.description + " " + " ".join(support.examples),
]
embeddings = [embed(t) for t in route_texts]
router.build_index(embeddings)
# Route with real embedding
query = "Show me my last 3 invoices"
result = router.route(query, embed(query))
print(result.route_id) # "billing"
Full Production Example¶
import asyncio
from stratarouter import Router, Route
from stratarouter.integrations.langchain import StrataRouterChain
from stratarouter_runtime import CoreRuntimeBridge, RuntimeConfig
from langchain_openai import ChatOpenAI
async def setup():
# Router
router = Router(dimension=1536, threshold=0.6)
for route_data in ROUTES:
r = Route(route_data["id"])
r.description = route_data["description"]
r.examples = route_data["examples"]
r.keywords = route_data["keywords"]
router.add_route(r)
router.build_index()
# Runtime (caching + observability)
bridge = CoreRuntimeBridge(RuntimeConfig(
cache_enabled=True,
batch_enabled=True,
metrics_enabled=True,
))
# LangChain chain
router_chain = StrataRouterChain(router)
return router_chain, bridge
async def handle(query: str, user_id: str):
router_chain, bridge = await setup()
# Route
route = router_chain.route(query)
# Execute via Runtime (with caching)
result = await bridge.execute(route, context={"user_id": user_id})
return result.response
asyncio.run(handle("Can I get a refund?", "user-123"))