416 lines
19 KiB
Python
416 lines
19 KiB
Python
"""主路由器：两级路由（大领域组 → 组内路由模型 → 专业执行器）专家系统编排。
|
||
|
||
两级体系（对齐用户架构决策）：
|
||
第一级：用户通过接口指定大领域组（domain_group: tech/professional/lifestyle/general），
|
||
或系统自动检测（8 领域分类 → 映射到组）
|
||
第二级：组内路由模型（RuleClassifier(domains=组内领域) + 组内知识/模板）识别具体
|
||
领域、子领域、拆解子任务 → 组内专业小模型/规则执行器
|
||
组内路由模型只认识本组领域：体积与匹配开销约为统一路由模型的 1/4，
|
||
且未来 L2 模型层可每组一个更小的路由模型，按需加载不常驻。
|
||
|
||
链路：缓存 → 组路由(分类/子领域/拆解) → 黑板+前向链 → DAG 执行 → 合并
|
||
→ Judge 校验 → （不达标）最后处理者升级 → 缓存/指标
|
||
|
||
L0 模式（默认）：规则分类 + 规则拆解 + 规则执行器 + 规则 Judge —— 零模型参数、零 API。
|
||
L2 模式（可选）：execution.expert_backend = hf/api 时，子任务改由专家池小模型执行
|
||
（≤8B，按需加载），其余流程不变。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import uuid
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
from .cache import RouterCache
|
||
from .classifier import BaseClassifier, RuleClassifier, build_classifier
|
||
from .config import load_config
|
||
from .executors import NodeExecutor, build_node_executor
|
||
from .experts import Expert, build_expert_pool
|
||
from .fallback import FallbackProvider, build_fallback
|
||
from .inference import InferenceEngine
|
||
from .judge import BaseJudge, build_judge
|
||
from .knowledge import KnowledgeBase
|
||
from .memory import TaskGraph, TaskNode, WorkingMemory
|
||
from .models import Classification, ExpertResponse, RouterResult, now_ms
|
||
from .planner import Planner
|
||
from .stats import Stats
|
||
from .trace import TraceStore
|
||
|
||
|
||
class Router:
|
||
def __init__(
|
||
self,
|
||
classifier: BaseClassifier,
|
||
experts: Dict[str, Expert],
|
||
judge: BaseJudge,
|
||
fallback: FallbackProvider,
|
||
cache: Optional[RouterCache] = None,
|
||
stats: Optional[Stats] = None,
|
||
config: Optional[Dict[str, Any]] = None,
|
||
kb: Optional[KnowledgeBase] = None,
|
||
planner: Optional[Planner] = None,
|
||
):
|
||
self.classifier = classifier
|
||
self.experts = experts
|
||
self.judge = judge
|
||
self.fallback = fallback
|
||
self.cache = cache or RouterCache()
|
||
self.stats = stats or Stats()
|
||
cfg = config or {}
|
||
rcfg = cfg.get("router", {})
|
||
self.low_confidence_threshold = rcfg.get("low_confidence_threshold", 0.60)
|
||
self.judge_fallback_threshold = rcfg.get("judge_fallback_threshold", 0.70)
|
||
self.cache_enabled = cfg.get("cache", {}).get("enabled", True)
|
||
# ---- 专家系统内核 ----
|
||
self.kb = kb or KnowledgeBase()
|
||
self.planner = planner or Planner(self.kb)
|
||
self.inference = InferenceEngine(self.kb)
|
||
ecfg = cfg.get("execution", {})
|
||
self.expert_backend = ecfg.get("expert_backend", "rule") # rule | hf | api
|
||
# 子任务执行后端（T1 抽象：NodeExecutor 工厂，新增后端无需改 Router）
|
||
self.node_executor: NodeExecutor = build_node_executor(
|
||
self.expert_backend, kb=self.kb, experts=experts)
|
||
# 推理链轨迹存储（T3：可解释性产品化）
|
||
self.trace_store = TraceStore()
|
||
# ---- 两级路由：大领域分组 + 组内路由模型（更小更专） ----
|
||
self.domain_groups: Dict[str, List[str]] = cfg.get("domain_groups", {}) or {}
|
||
if not self.domain_groups:
|
||
# 兜底：未配置时按单组（全部领域）处理，行为退化为一级路由
|
||
self.domain_groups = {"all": list(self.experts.keys())}
|
||
self._group_of_domain: Dict[str, str] = {}
|
||
for g, domains in self.domain_groups.items():
|
||
for d in domains:
|
||
self._group_of_domain[d] = g
|
||
# 组内路由模型：每组一个轻量分类器（只认识组内领域）
|
||
self._group_classifiers: Dict[str, RuleClassifier] = {
|
||
g: RuleClassifier(domains=domains)
|
||
for g, domains in self.domain_groups.items()
|
||
}
|
||
|
||
# ---------------------------------------------------------------
|
||
async def route(self, query: str, domain_group: Optional[str] = None) -> RouterResult:
|
||
"""两级路由入口。
|
||
|
||
domain_group 指定时：跳过 8 领域统一分类器，直接用组内路由模型
|
||
（RuleClassifier(domains=组内领域)）识别组内领域 —— 更小更专。
|
||
未指定时：统一分类器识别领域 → 自动映射到大领域组（向后兼容）。
|
||
"""
|
||
start = now_ms()
|
||
route: list = []
|
||
request_id = uuid.uuid4().hex[:12]
|
||
|
||
# ---- Step 1: 缓存 ----
|
||
if self.cache_enabled:
|
||
hit = self.cache.get(query)
|
||
if hit is not None:
|
||
level, cached = hit
|
||
latency = now_ms() - start
|
||
result = RouterResult(
|
||
query=query,
|
||
response=cached.get("response", ""),
|
||
domain=cached.get("domain", "general"),
|
||
difficulty=cached.get("difficulty", "medium"),
|
||
confidence=cached.get("confidence", 0.0),
|
||
upgraded=False,
|
||
quality_score=cached.get("quality_score", 0.0),
|
||
model_used=cached.get("model_used", ""),
|
||
route=["cache:" + level],
|
||
latency_ms=latency,
|
||
cache_hit=True,
|
||
cache_level=level,
|
||
cost_est=0.0,
|
||
subdomain=cached.get("subdomain"),
|
||
subdomain2=cached.get("subdomain2"),
|
||
domain_group=cached.get("domain_group"),
|
||
request_id=request_id,
|
||
)
|
||
self._store_trace(
|
||
request_id=request_id, query=query, group=cached.get("domain_group"),
|
||
domain=result.domain, difficulty=result.difficulty,
|
||
confidence=result.confidence, subdomain=result.subdomain,
|
||
subdomain2=result.subdomain2, route=route, quality=result.quality_score,
|
||
upgraded=False, model=result.model_used, latency=latency,
|
||
cache_hit=True, cache_level=level,
|
||
)
|
||
self.stats.record(latency, result.domain, result.difficulty, False, True, level, 0.0, result.model_used)
|
||
return result
|
||
route.append("cache:miss")
|
||
|
||
# ---- Step 2: 组路由（两级第一级）→ 组内分类（两级第二级） ----
|
||
classifier = self.classifier
|
||
group = domain_group
|
||
if group is not None:
|
||
# 用户指定大领域：校验 + 使用组内路由模型
|
||
if group not in self.domain_groups:
|
||
raise ValueError(
|
||
f"未知大领域组: {group}（可用: {sorted(self.domain_groups)}）"
|
||
)
|
||
classifier = self._group_classifiers[group]
|
||
route.append(f"group:{group}@explicit")
|
||
classification = classifier.classify(query)
|
||
if group is None:
|
||
# 自动检测：8 领域分类 → 映射大领域组
|
||
group = self._group_of_domain.get(classification.domain, "general")
|
||
route.append(f"group:{group}@auto")
|
||
route.append(f"classify:{classification.domain}@{classification.confidence:.2f}/{classification.difficulty}")
|
||
subdomain, subdomain2 = self._detect_subdomain(query, classification.domain)
|
||
if subdomain:
|
||
route.append(f"subdomain:{subdomain}")
|
||
if subdomain2:
|
||
route.append(f"subdomain2:{subdomain2}")
|
||
|
||
# ---- Step 3: 低置信度 -> 直接走最后处理者 ----
|
||
if classifier.should_fallback(classification, self.low_confidence_threshold):
|
||
route.append("direct_fallback")
|
||
fb = await self._call_fallback(query)
|
||
latency = now_ms() - start
|
||
result = self._finalize(query, classification, fb, quality_score=0.0,
|
||
upgraded=True, route=route, latency_ms=latency,
|
||
model_used=fb.model_used, cost_est=fb.cost_est,
|
||
subdomain=subdomain, subdomain2=subdomain2,
|
||
domain_group=group)
|
||
result.request_id = request_id
|
||
self._store_trace(
|
||
request_id=request_id, query=query, group=group,
|
||
domain=result.domain, difficulty=result.difficulty,
|
||
confidence=result.confidence, subdomain=subdomain,
|
||
subdomain2=subdomain2, route=route, quality=0.0,
|
||
upgraded=True, model=fb.model_used, latency=latency,
|
||
)
|
||
self._record(result, latency)
|
||
return result
|
||
|
||
# ---- Step 4: Planner 任务拆解（DAG） ----
|
||
graph = self.planner.plan(query, classification)
|
||
route.extend(self.planner.explain_plan(graph))
|
||
|
||
# ---- Step 5: 黑板初始化 + 前向链（规则轨迹） ----
|
||
memory = WorkingMemory()
|
||
self.inference.initialize(
|
||
query, classification.domain, classification.difficulty,
|
||
classification.confidence, memory,
|
||
)
|
||
fired = self.inference.run(query, classification.domain, memory)
|
||
if fired:
|
||
route.append(f"rules:{','.join(fired[:5])}")
|
||
|
||
# ---- Step 6: DAG 顺序执行（拓扑序） ----
|
||
order = graph.topo_order()
|
||
last_model = f"rule:{classification.domain}"
|
||
for node in order:
|
||
last_model = await self._execute_node(graph, node, classification, memory, route) or last_model
|
||
|
||
# ---- Step 7: 黑板合并（节点输出 + 推理机规则产出） ----
|
||
response = memory.merge([n.id for n in order])
|
||
# 追加推理机规则产出的部分解（带 output 的知识规则，如 git/docker/常识条目）
|
||
node_ids = {n.id for n in order}
|
||
extra_sections = [sid for sid in memory.sections if sid not in node_ids]
|
||
extras = [memory.section(s) for s in extra_sections if memory.section(s)]
|
||
if extras:
|
||
extra_text = "\n\n".join(extras)
|
||
response = (response + "\n\n" + extra_text) if response.strip() else extra_text
|
||
if not response.strip():
|
||
response = "（规则执行器）未能生成有效回答：任务均未产出内容。"
|
||
route.append("merge:empty")
|
||
|
||
# ---- Step 8: Judge 校验 ----
|
||
try:
|
||
evaluation = await self.judge.evaluate(query, response, classification.domain)
|
||
except Exception:
|
||
evaluation = None
|
||
route.append("judge_error")
|
||
quality_score = evaluation.overall_score if evaluation else 0.0
|
||
route.append(f"judge:{quality_score:.2f}")
|
||
|
||
upgraded = False
|
||
if evaluation is not None and evaluation.needs_fallback:
|
||
route.append("upgrade")
|
||
fb = await self._call_fallback(query)
|
||
response = fb.text
|
||
last_model = fb.model_used
|
||
upgraded = True
|
||
|
||
latency = now_ms() - start
|
||
result = self._finalize(query, classification, ExpertResponse(
|
||
text=response, model_used=last_model, latency_ms=latency,
|
||
tokens=max(8, int(len(response) / 2.2)), cost_est=0.0,
|
||
), quality_score=quality_score, upgraded=upgraded, route=route,
|
||
latency_ms=latency, model_used=last_model, cost_est=0.0,
|
||
subdomain=subdomain, subdomain2=subdomain2, domain_group=group)
|
||
result.request_id = request_id
|
||
self._store_trace(
|
||
request_id=request_id, query=query, group=group,
|
||
domain=result.domain, difficulty=result.difficulty,
|
||
confidence=result.confidence, subdomain=subdomain,
|
||
subdomain2=subdomain2, route=route, quality=quality_score,
|
||
upgraded=upgraded, model=last_model, latency=latency,
|
||
)
|
||
self._record(result, latency)
|
||
|
||
# 未升级的结果写缓存
|
||
if self.cache_enabled and not upgraded and result.response:
|
||
self.cache.put(query, result.to_dict())
|
||
|
||
return result
|
||
|
||
# ---------------------------------------------------------------
|
||
def _store_trace(self, request_id: str, query: str, group: Optional[str],
|
||
domain: str, difficulty: str, confidence: float,
|
||
subdomain: Optional[str], subdomain2: Optional[str],
|
||
route: list, quality: float, upgraded: bool,
|
||
model: str, latency: float,
|
||
cache_hit: bool = False, cache_level: Optional[str] = None) -> None:
|
||
"""记录完整推理链到轨迹存储（T3：可解释性产品化）。"""
|
||
self.trace_store.put(request_id, {
|
||
"request_id": request_id,
|
||
"query": query,
|
||
"domain_group": group,
|
||
"domain": domain,
|
||
"difficulty": difficulty,
|
||
"confidence": round(confidence, 4),
|
||
"subdomain": subdomain,
|
||
"subdomain2": subdomain2,
|
||
"route": list(route),
|
||
"quality_score": round(quality, 4),
|
||
"upgraded": upgraded,
|
||
"model_used": model,
|
||
"latency_ms": round(latency, 2),
|
||
"cache_hit": cache_hit,
|
||
"cache_level": cache_level,
|
||
})
|
||
|
||
# ---------------------------------------------------------------
|
||
def _detect_subdomain(self, query: str, domain: str) -> tuple:
|
||
"""子领域识别：返回 (二级 subdomain, 三级 subdomain2)。
|
||
|
||
二级取领域内最高优先级带 subdomain 的命中规则；
|
||
三级取最高优先级带 subdomain2 的命中规则（可与二级来自不同规则）。
|
||
"""
|
||
hits = self.kb.match(query, domain=domain)
|
||
sub = None
|
||
sub2 = None
|
||
for h in hits:
|
||
if sub is None and h.subdomain:
|
||
sub = h.subdomain
|
||
if sub2 is None and h.subdomain2:
|
||
sub2 = h.subdomain2
|
||
if sub is not None and sub2 is not None:
|
||
break
|
||
return sub, sub2
|
||
|
||
# ---------------------------------------------------------------
|
||
async def _execute_node(self, graph: TaskGraph, node: TaskNode,
|
||
classification: Classification, memory: WorkingMemory,
|
||
route: list) -> Optional[str]:
|
||
"""执行一个子任务节点；返回使用的 model_used（失败返回 None）。"""
|
||
# 依赖检查：依赖失败/跳过 → 本节点跳过
|
||
for dep_id in node.deps:
|
||
dep = graph.get(dep_id)
|
||
if dep is not None and dep.status in ("failed", "skipped"):
|
||
node.status = "skipped"
|
||
route.append(f"{node.id}:{node.kind}:skip")
|
||
return None
|
||
node.status = "running"
|
||
try:
|
||
# NodeExecutor 后端执行（rule 零参数 / model 专家池 ≤8B）
|
||
resp = await self.node_executor.execute(
|
||
node, classification.domain, classification.difficulty, memory)
|
||
node.output = resp.text
|
||
node.status = "done"
|
||
memory.write_section(node.id, resp.text)
|
||
route.append(f"{node.id}:{node.kind}")
|
||
return resp.model_used
|
||
except Exception as e:
|
||
node.status = "failed"
|
||
node.error = str(e)
|
||
self.stats.record_error()
|
||
route.append(f"{node.id}:{node.kind}:error:{type(e).__name__}")
|
||
return None
|
||
|
||
# ---------------------------------------------------------------
|
||
async def _call_fallback(self, query: str) -> ExpertResponse:
|
||
try:
|
||
return await self.fallback.generate(query)
|
||
except Exception as e:
|
||
# 回退也失败：返回错误占位响应
|
||
return ExpertResponse(
|
||
text=f"[系统错误] 专家与最后处理者均失败：{type(e).__name__}: {e}",
|
||
model_used=f"error:{self.fallback.name}",
|
||
cost_est=0.0,
|
||
)
|
||
|
||
@staticmethod
|
||
def _finalize(query: str, classification: Classification, resp: ExpertResponse,
|
||
quality_score: float, upgraded: bool, route: list,
|
||
latency_ms: float, model_used: str, cost_est: float,
|
||
error: Optional[str] = None,
|
||
subdomain: Optional[str] = None,
|
||
subdomain2: Optional[str] = None,
|
||
domain_group: Optional[str] = None) -> RouterResult:
|
||
return RouterResult(
|
||
query=query,
|
||
response=resp.text,
|
||
domain=classification.domain,
|
||
difficulty=classification.difficulty,
|
||
confidence=classification.confidence,
|
||
upgraded=upgraded,
|
||
quality_score=quality_score,
|
||
model_used=model_used,
|
||
route=route,
|
||
latency_ms=latency_ms,
|
||
cache_hit=False,
|
||
cost_est=cost_est,
|
||
error=error,
|
||
subdomain=subdomain,
|
||
subdomain2=subdomain2,
|
||
domain_group=domain_group,
|
||
)
|
||
|
||
def _record(self, result: RouterResult, latency_ms: float):
|
||
self.stats.record(
|
||
latency_ms,
|
||
result.domain,
|
||
result.difficulty,
|
||
result.upgraded,
|
||
result.cache_hit,
|
||
result.cache_level,
|
||
result.cost_est,
|
||
result.model_used,
|
||
)
|
||
|
||
# ---------------------------------------------------------------
|
||
def health(self) -> Dict[str, Any]:
|
||
return {
|
||
"status": "ok",
|
||
"domains": list(self.experts.keys()),
|
||
"domain_groups": self.domain_groups,
|
||
"classifier": type(self.classifier).__name__,
|
||
"judge": type(self.judge).__name__,
|
||
"fallback": type(self.fallback).__name__,
|
||
"planner": type(self.planner).__name__,
|
||
"execution_mode": self.expert_backend,
|
||
"rules": self.kb.rules_count(),
|
||
}
|
||
|
||
|
||
def build_router(config_path: Optional[str] = None) -> Router:
|
||
"""从配置构建完整 Router（默认 L0 专家系统模式：零参数可跑）。"""
|
||
config = load_config(config_path)
|
||
kb = KnowledgeBase()
|
||
classifier = build_classifier(config.get("classifier", {}))
|
||
experts = build_expert_pool(config.get("experts", {}), config.get("domains", []))
|
||
judge = build_judge(config.get("judge", {}),
|
||
config.get("router", {}).get("judge_fallback_threshold", 0.70),
|
||
kb=kb)
|
||
fallback = build_fallback(config.get("fallback", {}))
|
||
cache_cfg = config.get("cache", {})
|
||
cache = RouterCache(
|
||
semantic_enabled=cache_cfg.get("semantic_enabled", True),
|
||
similarity_threshold=cache_cfg.get("similarity_threshold", 0.88),
|
||
promote_frequency=cache_cfg.get("promote_frequency", 5),
|
||
)
|
||
ecfg = config.get("execution", {})
|
||
planner = Planner(kb, max_depth=ecfg.get("max_plan_depth", 3))
|
||
stats = Stats()
|
||
return Router(classifier, experts, judge, fallback, cache, stats, config,
|
||
kb=kb, planner=planner)
|