🧭 语义Agent路由

import numpy as np
from sentence_transformers import SentenceTransformer

model = SentenceTransformer('all-MiniLM-L6-v2')

# 用户请求向量化
user_request = "帮我分析这份财报的盈利能力"
request_embedding = model.encode(user_request)

# Agent能力描述向量化
agent_descriptions = [
    "我是财务分析专家，擅长财报分析、盈利能力评估",
    "我是代码专家，擅长Python、JavaScript、调试",
    "我是通用助手，可以处理各种日常任务"
]
agent_embeddings = model.encode(agent_descriptions)

# 计算相似度
similarities = np.dot(agent_embeddings, request_embedding)
best_agent_idx = np.argmax(similarities)
print(f"最佳Agent: {['FinAgent', 'CodeAgent', 'General'][best_agent_idx]}")

class SemanticRouter:
    def __init__(self):
        self.agents = {
            "fin_agent": {
                "embedding": model.encode("财务分析"),
                "capabilities": ["财报", "KPI", "审计"],
                "cost_per_token": 0.01,
                "avg_latency": 2000,
                "availability": 0.99
            },
            "code_agent": {...}
        }
    
    def route(self, request):
        req_embed = model.encode(request)
        
        candidates = []
        for name, info in self.agents.items():
            # 语义相似度
            sem_score = cosine_similarity(req_embed, info["embedding"])
            
            # 成本评分（越低越好）
            cost_score = 1 / (1 + info["cost_per_token"])
            
            # 延迟评分（越低越好）
            latency_score = 1 / (1 + info["avg_latency"] / 1000)
            
            # 可用性评分
            avail_score = info["availability"]
            
            # 综合评分
            final_score = (sem_score * 0.5 + 
                          cost_score * 0.2 + 
                          latency_score * 0.2 + 
                          avail_score * 0.1)
            
            candidates.append((name, final_score))
        
        return max(candidates, key=lambda x: x[1])[0]

# 动态路由考虑负载
def dynamic_route(request, agent_pool):
    # 获取各Agent当前负载
    loads = {agent: get_current_load(agent) for agent in agent_pool}
    
    # 获取语义匹配分数
    sem_scores = {agent: semantic_match(request, agent) 
                  for agent in agent_pool}
    
    # 综合评分 = 语义分数 / (负载 + 1)
    final_scores = {
        agent: sem_scores[agent] / (loads[agent] + 1)
        for agent in agent_pool
    }
    
    return max(final_scores, key=final_scores.get)

# 结果：语义匹配度高且负载低的Agent被选中

# OpenClaw Skills 语义路由配置
{
  "skills": [
    {"name": "code-analysis", "embedding": "代码分析、调试、重构"},
    {"name": "financial-analysis", "embedding": "财报、KPI、投资分析"},
    {"name": "content-writing", "embedding": "写作、文案、博客"}
  ],
  "router": {
    "type": "semantic",
    "fallback": "general-assistant",
    "min_confidence": 0.7
  }
}

# 用户请求："帮我优化这段Python代码的性能"
# → 语义路由 → 匹配 "code-analysis" skill
# → 调用对应Agent处理

# OpenClaw 模型路由：根据任务复杂度选择模型
model_routing_rules = {
    "simple_qa": "gpt-3.5-turbo",      # 简单问答用便宜模型
    "code_generation": "gpt-4o",         # 代码生成用强模型
    "complex_reasoning": "claude-opus",  # 复杂推理用最强模型
    "multimodal": "gpt-4-vision"         # 多模态用视觉模型
}

# 语义路由决定使用哪个模型
def route_model(request):
    intent = semantic_analyzer.analyze(request)
    return model_routing_rules.get(intent.type, "gpt-4o")

# OpenClaw sessions_spawn 智能路由
async def handle_complex_task(task_description):
    # 语义分析任务
    analysis = await semantic_router.analyze(task_description)
    
    # 根据分析结果spawn不同agent
    if analysis.domain == "code":
        return await sessions_spawn("code-agent", task_description)
    elif analysis.domain == "finance":
        return await sessions_spawn("fin-agent", task_description)
    elif analysis.domain == "research":
        # 复杂研究任务：启动多个Agent协作
        agents = ["search-agent", "summary-agent", "fact-checker"]
        return await asyncio.gather(*[
            sessions_spawn(agent, task_description) 
            for agent in agents
        ])
    else:
        return await sessions_spawn("general-agent", task_description)

# 先规则过滤，再语义精细路由
def hybrid_route(request):
    # 第一层：快速规则过滤
    if request.startswith("/api/"):
        return "api-agent"
    if "紧急" in request or "urgent" in request:
        return "priority-agent"
    
    # 第二层：语义路由
    return semantic_router.route(request)

# 设置置信度阈值，低置信度时请求人工
def route_with_confidence(request):
    scores = semantic_router.score_all(request)
    best_score = max(scores.values())
    
    if best_score < 0.6:  # 置信度太低
        return "human-review"
    elif best_score < 0.8:  # 中等置信度
        return "general-agent"  # 用通用Agent兜底
    else:  # 高置信度
        return max(scores, key=scores.get)

# 根据路由结果反馈优化
def route_and_learn(request):
    agent = semantic_router.route(request)
    
    # 执行任务
    result = agent.execute(request)
    
    # 记录路由效果
    if result.success:
        # 正反馈：增强这个路由路径
        semantic_router.reinforce(request, agent.name)
    else:
        # 负反馈：弱化这个路由路径
        semantic_router.penalize(request, agent.name)
    
    return result