# 安裝
pip install semantic-router

# 如果要完全本地化（不依賴API）
pip install "semantic-router[local]"

from semantic_router import Route
from semantic_router.encoders import OpenAIEncoder
from semantic_router.routers import SemanticRouter
import os

# 設(shè)置API密鑰
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 1. 定義路由規(guī)則 - 簡(jiǎn)單直觀
order_route = Route(
    name="order",
    utterances=[
        "我的訂單在哪里",
        "訂單什么時(shí)候發(fā)貨",
        "查詢訂單狀態(tài)",
        "物流信息",
        "快遞到哪了",
        "為什么還沒(méi)發(fā)貨",
        "訂單號(hào)12345的狀態(tài)",
        "配送需要多久",
        "可以改地址嗎",
        "取消訂單"
    ]
)

payment_route = Route(
    name="payment",
    utterances=[
        "支付失敗了",
        "可以用信用卡嗎",
        "支持哪些支付方式",
        "退款要多久",
        "發(fā)票怎么開(kāi)",
        "可以分期嗎",
        "支付安全嗎",
        "怎么綁定銀行卡"
    ]
)

product_route = Route(
    name="product",
    utterances=[
        "這個(gè)產(chǎn)品有什么功能",
        "產(chǎn)品參數(shù)",
        "有哪些顏色",
        "保修期多長(zhǎng)",
        "使用說(shuō)明",
        "產(chǎn)品對(duì)比",
        "推薦產(chǎn)品",
        "價(jià)格多少"
    ]
)

complaint_route = Route(
    name="complaint",
    utterances=[
        "我要投訴",
        "太差勁了",
        "服務(wù)態(tài)度不好",
        "產(chǎn)品有問(wèn)題",
        "要求賠償",
        "找你們經(jīng)理",
        "這是欺詐"
    ]
)

# 2. 創(chuàng)建路由器
routes = [order_route, payment_route, product_route, complaint_route]
encoder = OpenAIEncoder()  # 也可以用 CohereEncoder() 或 HuggingFaceEncoder()
router = SemanticRouter(encoder=encoder, routes=routes)

# 3. 使用路由器
def handle_query(user_input: str):
    """處理用戶查詢"""
    decision = router(user_input)
    
    if decision.name == "order":
        return handle_order(user_input)
    elif decision.name == "payment":
        return handle_payment(user_input)
    elif decision.name == "product":
        return handle_product(user_input)
    elif decision.name == "complaint":
        return escalate_to_human(user_input)
    else:
        return handle_general(user_input)

# 測(cè)試
print(router("我昨天買(mǎi)的東西怎么還沒(méi)發(fā)貨"))  # -> Route(name='order')
print(router("可以用花唄支付嗎"))           # -> Route(name='payment')
print(router("這個(gè)手機(jī)防水嗎"))             # -> Route(name='product')

from semantic_router.encoders import HuggingFaceEncoder
from semantic_router.llms import LlamaCppLLM

# 使用本地嵌入模型
encoder = HuggingFaceEncoder(
    model_name="sentence-transformers/all-MiniLM-L6-v2"
)

# 使用本地LLM（可選，用于動(dòng)態(tài)路由）
llm = LlamaCppLLM(
    model_path="./models/llama-2-7b-chat.gguf",
    n_ctx=2048,
    n_gpu_layers=32# 如果有GPU
)

# 創(chuàng)建完全本地化的路由器
router = SemanticRouter(
    encoder=encoder,
    routes=routes,
    llm=llm  # 可選：當(dāng)無(wú)法匹配時(shí)使用LLM判斷
)

from semantic_router import Route

# 創(chuàng)建動(dòng)態(tài)路由 - 可以執(zhí)行函數(shù)
def get_order_status(order_id: str):
    # 查詢訂單系統(tǒng)
    returnf"訂單 {order_id} 正在配送中"

def process_payment(amount: float, method: str):
    # 處理支付
    returnf"正在處理 {amount} 元的 {method} 支付"

# 動(dòng)態(tài)路由可以提取參數(shù)并調(diào)用函數(shù)
dynamic_order_route = Route(
    name="order_status",
    utterances=[
        "訂單[ORDER_ID]的狀態(tài)",
        "查詢訂單[ORDER_ID]",
        "[ORDER_ID]到哪了"
    ],
    function=get_order_status,
    function_schema={
        "type": "object",
        "properties": {
            "order_id": {"type": "string"}
        }
    }
)

from semantic_router.index import QdrantIndex
from semantic_router.routers import SemanticRouter

# 使用Qdrant向量數(shù)據(jù)庫(kù)存儲(chǔ)路由
qdrant_index = QdrantIndex(
    url="http://localhost:6333",  # Qdrant服務(wù)地址
    collection_name="routes"
)

# 創(chuàng)建支持向量數(shù)據(jù)庫(kù)的路由器
router = SemanticRouter(
    encoder=encoder,
    routes=routes,
    index=qdrant_index,  # 使用向量數(shù)據(jù)庫(kù)索引
    auto_sync="local"     # 自動(dòng)同步到本地
)

# 現(xiàn)在可以處理成千上萬(wàn)個(gè)路由規(guī)則

pip install routellm

from routellm.controller import Controller
import os

# 設(shè)置API密鑰
os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["ANYSCALE_API_KEY"] = "your-anyscale-key"# 用于Mixtral

# 創(chuàng)建智能路由控制器
client = Controller(
    routers=["mf"],  # 使用matrix factorization路由器
    strong_model="gpt-4",  # 強(qiáng)模型（貴但效果好）
    weak_model="anyscale/mistralai/Mixtral-8x7B-Instruct-v0.1",  # 弱模型（便宜）
)

# 使用示例
def smart_query(question: str, importance: str = "normal"):
    """
    智能查詢，根據(jù)重要性選擇路由閾值
    importance: "low", "normal", "high"
    """
    
    # 根據(jù)重要性設(shè)置閾值
    thresholds = {
        "low": 0.3,     # 更多使用便宜模型
        "normal": 0.5,  # 平衡
        "high": 0.8     # 更多使用貴模型
    }
    
    threshold = thresholds.get(importance, 0.5)
    
    response = client.chat.completions.create(
        model=f"router-mf-{threshold}",
        messages=[
            {"role": "user", "content": question}
        ],
        temperature=0.7
    )
    
    return response.choices[0].message.content

# 測(cè)試不同復(fù)雜度的問(wèn)題
simple_q = "今天星期幾？"# 簡(jiǎn)單問(wèn)題 -> 路由到Mixtral
complex_q = "解釋量子計(jì)算的原理，并給出實(shí)際應(yīng)用案例"# 復(fù)雜問(wèn)題 -> 路由到GPT-4

print(smart_query(simple_q, "low"))     # 用便宜模型
print(smart_query(complex_q, "high"))   # 用貴模型

# config.yaml
model_providers:
  - provider: openai
    api_key: ${OPENAI_API_KEY}
    models:
      - gpt-4
      - gpt-3.5-turbo

  - provider: anyscale
    api_key: ${ANYSCALE_API_KEY}
    models:
      - mistralai/Mixtral-8x7B-Instruct-v0.1

routers:
  mf:
    checkpoint_path: "routellm/mf_gpt4_augmented"
    strong_model: "gpt-4"
    weak_model: "anyscale/mistralai/Mixtral-8x7B-Instruct-v0.1"

python -m routellm.openai_server --routers mf --config config.yaml --port 8000

import openai

openai.api_base = "http://localhost:8000/v1"
openai.api_key = "PLACEHOLDER"

response = openai.ChatCompletion.create(
    model="router-mf-0.5",  # 使用路由器
    messages=[{"role": "user", "content": "你好"}]
)

from llama_index.core import VectorStoreIndex, SummaryIndex, Document
from llama_index.core.tools import QueryEngineTool
from llama_index.core.query_engine.router_query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector

# 創(chuàng)建不同類型的索引
documents = [
    Document(text="公司2024年?duì)I收100億..."),
    Document(text="產(chǎn)品使用說(shuō)明..."),
    Document(text="技術(shù)文檔...")
]

# 向量索引 - 用于語(yǔ)義搜索
vector_index = VectorStoreIndex.from_documents(
    documents[:2]
)

# 摘要索引 - 用于總結(jié)
summary_index = SummaryIndex.from_documents(
    documents[2:]
)

# 創(chuàng)建查詢工具
vector_tool = QueryEngineTool.from_defaults(
    query_engine=vector_index.as_query_engine(),
    description="用于查找具體信息和事實(shí)"
)

summary_tool = QueryEngineTool.from_defaults(
    query_engine=summary_index.as_query_engine(),
    description="用于獲取總結(jié)和概覽"
)

# 創(chuàng)建路由查詢引擎
router_query_engine = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=[vector_tool, summary_tool]
)

# 使用路由器
response = router_query_engine.query("公司去年的營(yíng)收是多少？")  # -> 路由到vector_tool
response = router_query_engine.query("總結(jié)一下技術(shù)文檔")      # -> 路由到summary_tool

from llama_index.core.query_engine import SimpleMultiModalQueryEngine
from llama_index.core.indices import MultiModalVectorStoreIndex

class MultiModalRouter:
    """多模態(tài)路由器"""
    
    def __init__(self):
        # 文本查詢引擎
        self.text_engine = self._create_text_engine()
        
        # 圖像查詢引擎
        self.image_engine = self._create_image_engine()
        
        # 混合查詢引擎
        self.hybrid_engine = self._create_hybrid_engine()
    
    def route(self, query: str, has_image: bool = False):
        """根據(jù)查詢類型路由"""
        
        # 檢測(cè)查詢意圖
        if any(word in query.lower() for word in ["圖片", "圖像", "照片", "看"]):
            return self.image_engine.query(query)
        elif has_image:
            return self.hybrid_engine.query(query)
        else:
            return self.text_engine.query(query)
    
    def _create_text_engine(self):
        # 創(chuàng)建文本查詢引擎
        pass
    
    def _create_image_engine(self):
        # 創(chuàng)建圖像查詢引擎
        pass
    
    def _create_hybrid_engine(self):
        # 創(chuàng)建混合查詢引擎
        pass

from llama_index.core.tools import FunctionTool
from llama_index.core.agent import ReActAgent

# 定義工具函數(shù)
def search_product(product_name: str) -> str:
    """搜索產(chǎn)品信息"""
    # 實(shí)現(xiàn)產(chǎn)品搜索邏輯
    returnf"找到產(chǎn)品: {product_name}"

def calculate_price(quantity: int, unit_price: float) -> float:
    """計(jì)算總價(jià)"""
    return quantity * unit_price

def check_inventory(product_id: str) -> int:
    """檢查庫(kù)存"""
    # 實(shí)現(xiàn)庫(kù)存查詢
    return100# 示例返回值

# 創(chuàng)建工具
tools = [
    FunctionTool.from_defaults(
        fn=search_product,
        description="搜索產(chǎn)品信息"
    ),
    FunctionTool.from_defaults(
        fn=calculate_price,
        description="計(jì)算價(jià)格"
    ),
    FunctionTool.from_defaults(
        fn=check_inventory,
        description="檢查庫(kù)存數(shù)量"
    )
]

# 創(chuàng)建智能代理（自動(dòng)路由到合適的工具）
agent = ReActAgent.from_tools(
    tools,
    verbose=True
)

# 使用代理
response = agent.chat("iPhone 15 Pro有貨嗎？")  # 自動(dòng)路由到check_inventory
response = agent.chat("10個(gè)單價(jià)99元的商品總價(jià)是多少？")  # 自動(dòng)路由到calculate_price

你的主要需求是什么？
│
├─ 需要極快的響應(yīng)速度？(<50ms)
│  └─ 選擇 Semantic Router
│
├─ 需要控制API成本？
│  └─ 選擇 RouteLLM
│
├─ 需要復(fù)雜的查詢能力？
│  ├─ 需要多模態(tài)？
│  │  └─ 選擇 LlamaIndex
│  └─ 只需要文本？
│     └─ 選擇 Semantic Router + LlamaIndex 組合
│
└─ 需要簡(jiǎn)單快速上線？
   └─ 選擇 Semantic Router

class HybridRouter:
    """混合路由器 - 結(jié)合多個(gè)開(kāi)源方案的優(yōu)勢(shì)"""
    
    def __init__(self):
        # 第一層：Semantic Router處理常見(jiàn)查詢（最快）
        self.semantic_router = self._init_semantic_router()
        
        # 第二層：RouteLLM處理復(fù)雜查詢（成本優(yōu)化）
        self.cost_router = self._init_routellm()
        
        # 第三層：LlamaIndex處理專業(yè)查詢（功能強(qiáng)大）
        self.index_router = self._init_llamaindex()
    
    asyncdef route(self, query: str, context: dict = None):
        """三層路由策略"""
        
        # 1. 嘗試快速語(yǔ)義路由
        semantic_result = self.semantic_router(query)
        if semantic_result.confidence > 0.8:
            return self.handle_semantic_result(semantic_result)
        
        # 2. 評(píng)估查詢復(fù)雜度，決定是否需要貴模型
        if self.is_complex_query(query):
            returnawait self.cost_router.route_complex(query)
        
        # 3. 對(duì)于需要檢索的查詢，使用LlamaIndex
        if self.needs_retrieval(query):
            return self.index_router.query(query)
        
        # 默認(rèn)處理
        return self.default_handler(query)

import time
from prometheus_client import Counter, Histogram, Gauge

# Prometheus指標(biāo)
route_counter = Counter('routing_total', 'Total routing requests', ['router', 'route'])
route_latency = Histogram('routing_latency_seconds', 'Routing latency', ['router'])
route_errors = Counter('routing_errors_total', 'Routing errors', ['router', 'error_type'])

def monitor_routing(router_name: str):
    """路由監(jiān)控裝飾器"""
    def decorator(func):
        def wrapper(*args, **kwargs):
            start = time.time()
            try:
                result = func(*args, **kwargs)
                route_counter.labels(router=router_name, route=result.get('route', 'unknown')).inc()
                return result
            except Exception as e:
                route_errors.labels(router=router_name, error_type=type(e).__name__).inc()
                raise
            finally:
                route_latency.labels(router=router_name).observe(time.time() - start)
        return wrapper
    return decorator

# 使用監(jiān)控
@monitor_routing("semantic_router")
def semantic_route(query):
    return router(query)