混合推理模式：当规则引擎遇上机器学习
单一推理模式无法满足企业级决策的复杂需求。coomia-dip 采用规则层 + ML 层 + 人工审核层的三层混合推理架构，将确定性规则推理、概率性机器学习推断和人工专家判断有机结合。本文深入解析三层架构的设计原理、层间路由机制、置信度融合算法以及降级策略，展示如何在 Reasoning & Decision Layer 中构建一个既精确又智能的决策系统。
Coomia发布于 2025年8月25日19 分钟阅读
分享本文Twitter / X
“系列：S5 智能决策 · 第 3 篇 | 难度：高级 | 阅读时间：20 分钟
混合推理模式：当规则引擎遇上机器学习
#TL;DR
单一推理模式无法满足企业级决策的复杂需求。coomia-dip 采用规则层 + ML 层 + 人工审核层的三层混合推理架构，将确定性规则推理、概率性机器学习推断和人工专家判断有机结合。本文深入解析三层架构的设计原理、层间路由机制、置信度融合算法以及降级策略，展示如何在 Reasoning & Decision Layer 中构建一个既精确又智能的决策系统。
#1. 为什么需要混合推理
#1.1 单一推理的局限
Code
单一推理模式的问题：
规则引擎（确定性） ML 模型（概率性）
+-------------------+ +-------------------+
| 优势： | | 优势： |
| - 可解释 | | - 处理模糊场景 |
| - 可审计 | | - 发现隐含模式 |
| - 响应快 | | - 自适应学习 |
+-------------------+ +-------------------+
| 局限： | | 局限： |
| - 无法处理模糊 | | - 黑箱不可解释 |
| - 规则爆炸 | | - 需要大量数据 |
| - 无法自学习 | | - 合规风险高 |
+-------------------+ +-------------------+
| 场景类型 | 规则引擎适合度 | ML 适合度 | 推荐 |
|---|---|---|---|
| 合规检查（法规明确） | 高 | 低 | 规则 |
| 欺诈检测（模式复杂） | 低 | 高 | ML |
| 信用评估（规则+趋势） | 中 | 中 | 混合 |
| 库存预警（阈值+预测） | 中 | 中 | 混合 |
| 客户流失（多因子） | 低 | 高 | ML + 人工 |
#1.2 三层混合架构总览
Code
三层混合推理架构：
请求入口
|
v
+--------+--------+
| 路由分发器 |
| (Router/Triage) |
+--------+--------+
|
+------------+------------+
| | |
v v v
+--------+ +--------+ +-----------+
| Layer 1| | Layer 2| | Layer 3 |
| 规则层 | | ML 层 | | 人工审核层 |
| (确定) | | (概率) | | (判断) |
+--------+ +--------+ +-----------+
| | |
+------+-----+------+----+
| |
v v
+------------+ +----------+
| 置信度融合 | | 审批队列 |
| (Fusion) | | (Queue) |
+------+-----+ +-----+----+
| |
v v
最终决策 <--------+
#2. 路由分发器设计
#2.1 请求分类与路由
路由分发器根据请求特征决定推理路径：
Python
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import Any
class ReasoningPath(Enum):
"""推理路径类型"""
RULE_ONLY = auto() # 纯规则，不需要 ML
ML_ONLY = auto() # 纯 ML，无规则覆盖
RULE_THEN_ML = auto() # 规则初筛 + ML 精细化
ML_THEN_RULE = auto() # ML 预测 + 规则验证
PARALLEL_FUSION = auto() # 规则与 ML 并行 + 融合
HUMAN_REVIEW = auto() # 直接人工审核
@dataclass
class ReasoningRequest:
"""推理请求"""
request_id: str
domain: str # "fraud", "credit", "inventory" 等
facts: dict[str, Any]
urgency: str = "normal" # "critical", "normal", "low"
confidence_threshold: float = 0.85
@dataclass
class RoutingRule:
"""路由规则"""
domain: str
condition: str # 条件表达式
path: ReasoningPath
priority: int = 0
class ReasoningRouter:
"""推理请求路由分发器"""
def __init__(self):
self._rules: list[RoutingRule] = []
self._domain_config: dict[str, dict] = {}
self._default_path = ReasoningPath.RULE_THEN_ML
def configure_domain(self, domain: str, config: dict) -> None:
"""配置领域路由策略"""
self._domain_config[domain] = config
# 自动注册路由规则
if config.get("rules_complete", False):
self._rules.append(RoutingRule(
domain=domain,
condition="rules_coverage >= 0.95",
path=ReasoningPath.RULE_ONLY,
priority=10,
))
if config.get("ml_model_available", False):
self._rules.append(RoutingRule(
domain=domain,
condition="ml_model_accuracy >= 0.90",
path=ReasoningPath.PARALLEL_FUSION,
priority=5,
))
def route(self, request: ReasoningRequest) -> ReasoningPath:
"""决定推理路径"""
config = self._domain_config.get(request.domain, {})
# 紧急请求：跳过 ML，仅用规则
if request.urgency == "critical":
return ReasoningPath.RULE_ONLY
# 检查规则覆盖率
rules_coverage = config.get("rules_coverage", 0.0)
ml_available = config.get("ml_model_available", False)
if rules_coverage >= 0.95 and not ml_available:
return ReasoningPath.RULE_ONLY
if rules_coverage < 0.3 and ml_available:
return ReasoningPath.ML_ONLY
if ml_available:
return ReasoningPath.PARALLEL_FUSION
return self._default_path
#2.2 路由决策矩阵
Code
路由决策矩阵：
ML 模型可用
Yes No
+------------+ +------------+
规则覆盖 >= 95%| PARALLEL | | RULE_ONLY |
| _FUSION | | |
+------------+ +------------+
规则覆盖 30-95%| RULE_THEN | | RULE_ONLY |
| _ML | | + HUMAN |
+------------+ +------------+
规则覆盖 < 30% | ML_ONLY | | HUMAN |
| | | _REVIEW |
+------------+ +------------+
#3. Layer 1：规则推理层
#3.1 确定性规则执行
规则层处理具有明确条件和结论的业务逻辑：
Python
from dataclasses import dataclass
from typing import Optional
import time
@dataclass
class RuleResult:
"""规则层推理结果"""
rule_ids: list[str] # 触发的规则
conclusion: str # 结论
confidence: float # 确定性规则置信度通常为 1.0
explanation: list[str] # 解释链
elapsed_ms: float
is_conclusive: bool = True # 是否得出最终结论
class RuleReasoningLayer:
"""Layer 1: 确定性规则推理"""
def __init__(self, rete_network, working_memory):
self._rete = rete_network
self._wm = working_memory
self._inconclusive_threshold = 0.7
def reason(self, request: ReasoningRequest) -> RuleResult:
"""执行规则推理"""
start = time.monotonic()
# 将请求事实注入工作记忆
for key, value in request.facts.items():
self._wm.assert_fact(key, value)
# 执行 Rete 前向链
fired_rules = self._rete.run_to_completion(max_cycles=500)
elapsed = (time.monotonic() - start) * 1000
if not fired_rules:
return RuleResult(
rule_ids=[],
conclusion="no_match",
confidence=0.0,
explanation=["无规则匹配当前事实集合"],
elapsed_ms=elapsed,
is_conclusive=False,
)
# 聚合规则结论
conclusion = self._aggregate_conclusions(fired_rules)
coverage = self._compute_coverage(request.facts, fired_rules)
return RuleResult(
rule_ids=[r.rule_id for r in fired_rules],
conclusion=conclusion,
confidence=min(coverage, 1.0),
explanation=self._build_explanation(fired_rules),
elapsed_ms=elapsed,
is_conclusive=coverage >= self._inconclusive_threshold,
)
def _aggregate_conclusions(self, fired_rules) -> str:
"""聚合多条规则的结论"""
# 按优先级取最高的
sorted_rules = sorted(fired_rules, key=lambda r: r.priority, reverse=True)
return sorted_rules[0].conclusion
def _compute_coverage(self, facts: dict, fired_rules) -> float:
"""计算规则对事实的覆盖程度"""
total_facts = len(facts)
covered_facts = set()
for rule in fired_rules:
for cond in rule.conditions:
covered_facts.add(cond.attribute)
return len(covered_facts) / max(total_facts, 1)
def _build_explanation(self, fired_rules) -> list[str]:
"""构建推理解释链"""
explanations = []
for rule in fired_rules:
conditions_text = " AND ".join(
f"{c.attribute} {c.operator} {c.value}"
for c in rule.conditions
)
explanations.append(
f"规则 [{rule.name}]: 因为 {conditions_text} -> {rule.conclusion}"
)
return explanations
#4. Layer 2：ML 推理层
#4.1 模型注册与管理
Python
from dataclasses import dataclass, field
from typing import Protocol
import numpy as np
class MLModel(Protocol):
"""ML 模型接口"""
def predict(self, features: dict[str, Any]) -> tuple[str, float]: ...
def predict_proba(self, features: dict[str, Any]) -> dict[str, float]: ...
@property
def model_id(self) -> str: ...
@property
def version(self) -> str: ...
@dataclass
class ModelMetadata:
"""模型元数据"""
model_id: str
version: str
domain: str
accuracy: float
f1_score: float
training_date: str
feature_names: list[str]
is_active: bool = True
class ModelRegistry:
"""ML 模型注册表"""
def __init__(self):
self._models: dict[str, MLModel] = {}
self._metadata: dict[str, ModelMetadata] = {}
self._domain_index: dict[str, list[str]] = {}
def register(self, model: MLModel, metadata: ModelMetadata) -> None:
key = f"{metadata.model_id}:{metadata.version}"
self._models[key] = model
self._metadata[key] = metadata
self._domain_index.setdefault(metadata.domain, []).append(key)
def get_best_model(self, domain: str) -> tuple[MLModel, ModelMetadata] | None:
"""获取指定领域最佳活跃模型"""
keys = self._domain_index.get(domain, [])
active = [
(self._models[k], self._metadata[k])
for k in keys
if self._metadata[k].is_active
]
if not active:
return None
# 按 F1 分数排序
active.sort(key=lambda x: x[1].f1_score, reverse=True)
return active[0]
#4.2 ML 推理执行
Python
@dataclass
class MLResult:
"""ML 推理结果"""
model_id: str
model_version: str
prediction: str
confidence: float
probabilities: dict[str, float]
feature_importance: dict[str, float]
elapsed_ms: float
class MLReasoningLayer:
"""Layer 2: 机器学习推理"""
def __init__(self, model_registry: ModelRegistry):
self._registry = model_registry
self._feature_extractors: dict[str, callable] = {}
def register_feature_extractor(self, domain: str, extractor: callable):
self._feature_extractors[domain] = extractor
def reason(self, request: ReasoningRequest) -> MLResult | None:
"""执行 ML 推理"""
result = self._registry.get_best_model(request.domain)
if result is None:
return None
model, metadata = result
start = time.monotonic()
# 特征提取
extractor = self._feature_extractors.get(request.domain)
if extractor:
features = extractor(request.facts)
else:
features = request.facts
# 推理
prediction, confidence = model.predict(features)
probabilities = model.predict_proba(features)
# 特征重要性（如果模型支持）
importance = self._compute_importance(model, features)
elapsed = (time.monotonic() - start) * 1000
return MLResult(
model_id=metadata.model_id,
model_version=metadata.version,
prediction=prediction,
confidence=confidence,
probabilities=probabilities,
feature_importance=importance,
elapsed_ms=elapsed,
)
def _compute_importance(self, model: MLModel,
features: dict) -> dict[str, float]:
"""计算特征重要性（SHAP 近似）"""
base_pred, base_conf = model.predict(features)
importance = {}
for key in features:
modified = {**features}
del modified[key]
_, mod_conf = model.predict(modified)
importance[key] = abs(base_conf - mod_conf)
# 归一化
total = sum(importance.values()) or 1.0
return {k: v / total for k, v in importance.items()}
#4.3 ML 模型的沙箱执行
coomia-dip 使用 nsjail 隔离 ML 模型推理，防止恶意模型：
Code
ML 模型沙箱执行流程：
+---------------------+
| ReasoningEngine |
| |
| 1. 提取特征向量 |
| 2. 序列化请求 |
+----------+----------+
|
v
+----------+----------+
| nsjail Sandbox |
| +-----------------+|
| | Model Runtime ||
| | - 只读文件系统 ||
| | - 无网络 ||
| | - CPU/内存限制 ||
| | - 超时 5s ||
| +-----------------+|
+----------+----------+
|
v
+----------+----------+
| 结果反序列化 |
| + 置信度校验 |
+---------------------+
#5. Layer 3：人工审核层
#5.1 审核触发条件
Python
@dataclass
class HumanReviewRequest:
"""人工审核请求"""
request_id: str
original_request: ReasoningRequest
rule_result: RuleResult | None
ml_result: MLResult | None
trigger_reason: str
priority: str = "normal"
deadline_hours: int = 24
class HumanReviewTrigger:
"""判断是否需要人工审核"""
def __init__(self):
self._confidence_threshold = 0.7
self._conflict_threshold = 0.3
self._high_impact_domains = {"credit", "fraud", "compliance"}
def should_review(self, rule_result: RuleResult | None,
ml_result: MLResult | None,
request: ReasoningRequest) -> tuple[bool, str]:
"""
判断是否需要人工审核
Returns: (是否需要, 原因)
"""
# 条件 1：规则和 ML 结果冲突
if rule_result and ml_result:
if rule_result.conclusion != ml_result.prediction:
conflict_severity = abs(
rule_result.confidence - ml_result.confidence
)
if conflict_severity < self._conflict_threshold:
return True, "规则与 ML 结论冲突且置信度接近"
# 条件 2：两层均低置信度
rule_conf = rule_result.confidence if rule_result else 0.0
ml_conf = ml_result.confidence if ml_result else 0.0
max_conf = max(rule_conf, ml_conf)
if max_conf < self._confidence_threshold:
return True, f"最高置信度 {max_conf:.2f} 低于阈值"
# 条件 3：高影响领域
if request.domain in self._high_impact_domains:
if max_conf < 0.95:
return True, f"高影响领域 {request.domain} 需要高置信度"
# 条件 4：规则层不确定
if rule_result and not rule_result.is_conclusive:
return True, "规则层未得出确定性结论"
return False, ""
#5.2 审核队列管理
Python
from collections import deque
from datetime import datetime, timedelta
class ReviewQueue:
"""人工审核队列"""
def __init__(self):
self._queue: deque[HumanReviewRequest] = deque()
self._assigned: dict[str, HumanReviewRequest] = {}
self._completed: dict[str, dict] = {}
def enqueue(self, review: HumanReviewRequest) -> int:
"""入队，返回队列位置"""
self._queue.append(review)
return len(self._queue)
def assign_next(self, reviewer_id: str) -> HumanReviewRequest | None:
"""分配下一个审核任务给审核员"""
if not self._queue:
return None
# 按优先级和截止时间排序
sorted_queue = sorted(
self._queue,
key=lambda r: (
0 if r.priority == "critical" else 1,
r.deadline_hours,
),
)
review = sorted_queue[0]
self._queue.remove(review)
self._assigned[review.request_id] = review
return review
def complete_review(self, request_id: str, decision: str,
reviewer_id: str, notes: str) -> None:
"""完成审核"""
review = self._assigned.pop(request_id, None)
if review is None:
raise KeyError(f"审核 {request_id} 未在分配列表中")
self._completed[request_id] = {
"review": review,
"decision": decision,
"reviewer_id": reviewer_id,
"notes": notes,
"completed_at": datetime.utcnow().isoformat(),
}
@property
def pending_count(self) -> int:
return len(self._queue) + len(self._assigned)
#6. 置信度融合算法
#6.1 融合策略
当规则层和 ML 层同时产生结果时，需要智能融合：
Code
置信度融合流程：
Rule Result ML Result
conclusion: "approve" prediction: "approve"
confidence: 0.92 confidence: 0.87
| |
+----------+---------------+
|
v
+--------+--------+
| Fusion Engine |
| |
| 1. 一致性检查 |
| 2. 加权融合 |
| 3. 校准调整 |
+--------+---------+
|
v
Final: "approve"
confidence: 0.94
#6.2 融合实现
Python
from dataclasses import dataclass
@dataclass
class FusionResult:
"""融合后的推理结果"""
conclusion: str
confidence: float
rule_contribution: float # 规则层贡献度
ml_contribution: float # ML 层贡献度
fusion_method: str
explanation: list[str]
class ConfidenceFusion:
"""置信度融合引擎"""
def __init__(self):
# 领域权重配置
self._domain_weights: dict[str, tuple[float, float]] = {
# (rule_weight, ml_weight)
"compliance": (0.9, 0.1), # 合规偏规则
"fraud": (0.3, 0.7), # 欺诈偏 ML
"credit": (0.5, 0.5), # 信用均衡
"inventory": (0.4, 0.6), # 库存偏 ML
}
self._default_weights = (0.5, 0.5)
def fuse(self, rule_result: RuleResult | None,
ml_result: MLResult | None,
domain: str) -> FusionResult:
"""融合规则和 ML 结果"""
# 只有规则结果
if rule_result and not ml_result:
return FusionResult(
conclusion=rule_result.conclusion,
confidence=rule_result.confidence,
rule_contribution=1.0,
ml_contribution=0.0,
fusion_method="rule_only",
explanation=rule_result.explanation,
)
# 只有 ML 结果
if ml_result and not rule_result:
return FusionResult(
conclusion=ml_result.prediction,
confidence=ml_result.confidence,
rule_contribution=0.0,
ml_contribution=1.0,
fusion_method="ml_only",
explanation=[
f"ML 模型 {ml_result.model_id} 预测: "
f"{ml_result.prediction} (置信度: {ml_result.confidence:.2f})"
],
)
# 双层融合
rule_w, ml_w = self._domain_weights.get(domain, self._default_weights)
if rule_result.conclusion == ml_result.prediction:
# 结论一致：加权提升
return self._fuse_agreement(
rule_result, ml_result, rule_w, ml_w, domain
)
else:
# 结论冲突：加权竞争
return self._fuse_conflict(
rule_result, ml_result, rule_w, ml_w, domain
)
def _fuse_agreement(self, rule: RuleResult, ml: MLResult,
rw: float, mw: float, domain: str) -> FusionResult:
"""双层结论一致时的融合"""
# 加权平均，结论一致时提升置信度
weighted_conf = rule.confidence * rw + ml.confidence * mw
# 一致性加成（Bayesian update 近似）
boosted = 1 - (1 - weighted_conf) * (1 - min(rule.confidence, ml.confidence))
return FusionResult(
conclusion=rule.conclusion,
confidence=min(boosted, 0.99),
rule_contribution=rw,
ml_contribution=mw,
fusion_method="weighted_agreement",
explanation=rule.explanation + [
f"ML 模型同意此结论 (置信度 {ml.confidence:.2f})，融合后提升"
],
)
def _fuse_conflict(self, rule: RuleResult, ml: MLResult,
rw: float, mw: float, domain: str) -> FusionResult:
"""双层结论冲突时的融合"""
rule_score = rule.confidence * rw
ml_score = ml.confidence * mw
if rule_score >= ml_score:
winner = "rule"
conclusion = rule.conclusion
confidence = rule_score / (rule_score + ml_score)
explanation = rule.explanation + [
f"注意: ML 模型给出不同结论 '{ml.prediction}' "
f"(置信度 {ml.confidence:.2f})，但规则层加权分更高"
]
else:
winner = "ml"
conclusion = ml.prediction
confidence = ml_score / (rule_score + ml_score)
explanation = [
f"ML 模型结论 '{ml.prediction}' (置信度 {ml.confidence:.2f}) "
f"加权分高于规则层"
] + [f"[被否决的规则推理]: {e}" for e in rule.explanation]
# 冲突惩罚：降低最终置信度
conflict_penalty = 0.15
confidence = max(confidence - conflict_penalty, 0.1)
return FusionResult(
conclusion=conclusion,
confidence=confidence,
rule_contribution=rw if winner == "rule" else 1 - mw,
ml_contribution=mw if winner == "ml" else 1 - rw,
fusion_method=f"conflict_resolution_{winner}_wins",
explanation=explanation,
)
#7. 混合推理编排器
#7.1 完整编排流程
Python
@dataclass
class HybridReasoningResult:
"""混合推理最终结果"""
request_id: str
conclusion: str
confidence: float
path_taken: ReasoningPath
rule_result: RuleResult | None
ml_result: MLResult | None
fusion_result: FusionResult | None
human_review_needed: bool
review_reason: str
total_elapsed_ms: float
trace: list[str]
class HybridReasoningOrchestrator:
"""混合推理编排器 —— 核心协调组件"""
def __init__(
self,
router: ReasoningRouter,
rule_layer: RuleReasoningLayer,
ml_layer: MLReasoningLayer,
fusion: ConfidenceFusion,
review_trigger: HumanReviewTrigger,
review_queue: ReviewQueue,
):
self._router = router
self._rule_layer = rule_layer
self._ml_layer = ml_layer
self._fusion = fusion
self._review_trigger = review_trigger
self._review_queue = review_queue
async def reason(self, request: ReasoningRequest) -> HybridReasoningResult:
"""执行混合推理"""
start = time.monotonic()
trace = []
# Step 1: 路由
path = self._router.route(request)
trace.append(f"路由决策: {path.name}")
rule_result = None
ml_result = None
fusion_result = None
# Step 2: 按路径执行推理
if path == ReasoningPath.RULE_ONLY:
rule_result = self._rule_layer.reason(request)
trace.append(f"规则层: {rule_result.conclusion} "
f"(置信度 {rule_result.confidence:.2f})")
elif path == ReasoningPath.ML_ONLY:
ml_result = self._ml_layer.reason(request)
trace.append(f"ML 层: {ml_result.prediction} "
f"(置信度 {ml_result.confidence:.2f})")
elif path == ReasoningPath.RULE_THEN_ML:
rule_result = self._rule_layer.reason(request)
trace.append(f"规则层: {rule_result.conclusion}")
if not rule_result.is_conclusive:
ml_result = self._ml_layer.reason(request)
trace.append(f"规则层不确定，进入 ML 层: {ml_result.prediction}")
elif path == ReasoningPath.PARALLEL_FUSION:
# 并行执行（在实际异步环境中）
rule_result = self._rule_layer.reason(request)
ml_result = self._ml_layer.reason(request)
trace.append(f"并行执行: 规则={rule_result.conclusion}, "
f"ML={ml_result.prediction}")
elif path == ReasoningPath.HUMAN_REVIEW:
trace.append("直接进入人工审核")
# Step 3: 融合
if rule_result or ml_result:
fusion_result = self._fusion.fuse(
rule_result, ml_result, request.domain
)
trace.append(f"融合结果: {fusion_result.conclusion} "
f"(置信度 {fusion_result.confidence:.2f})")
# Step 4: 人工审核判断
need_review, review_reason = self._review_trigger.should_review(
rule_result, ml_result, request
)
if path == ReasoningPath.HUMAN_REVIEW:
need_review = True
review_reason = "路由指定人工审核"
if need_review:
self._review_queue.enqueue(HumanReviewRequest(
request_id=request.request_id,
original_request=request,
rule_result=rule_result,
ml_result=ml_result,
trigger_reason=review_reason,
))
trace.append(f"已提交人工审核: {review_reason}")
elapsed = (time.monotonic() - start) * 1000
return HybridReasoningResult(
request_id=request.request_id,
conclusion=fusion_result.conclusion if fusion_result else "pending_review",
confidence=fusion_result.confidence if fusion_result else 0.0,
path_taken=path,
rule_result=rule_result,
ml_result=ml_result,
fusion_result=fusion_result,
human_review_needed=need_review,
review_reason=review_reason,
total_elapsed_ms=elapsed,
trace=trace,
)
#8. 降级与容错策略
#8.1 降级场景
Code
降级策略矩阵：
场景 | 降级动作 | 日志级别
-----------------------|----------------------------|---------
ML 模型服务不可用 | 回退到纯规则推理 | WARN
ML 推理超时 (>5s) | 使用规则结果 + 标记 | WARN
规则引擎异常 | 使用 ML 结果 + 强制人工审核 | ERROR
两层均失败 | 进入人工审核队列 | CRITICAL
人工审核队列积压 >100 | 自动按 ML 结论执行低风险请求 | WARN
#8.2 断路器模式
Python
from datetime import datetime, timedelta
class CircuitBreaker:
"""ML 推理断路器"""
def __init__(self, failure_threshold: int = 5,
recovery_timeout_s: int = 60):
self._failure_count = 0
self._failure_threshold = failure_threshold
self._recovery_timeout = timedelta(seconds=recovery_timeout_s)
self._last_failure_time: datetime | None = None
self._state = "CLOSED" # CLOSED, OPEN, HALF_OPEN
def record_success(self) -> None:
self._failure_count = 0
self._state = "CLOSED"
def record_failure(self) -> None:
self._failure_count += 1
self._last_failure_time = datetime.utcnow()
if self._failure_count >= self._failure_threshold:
self._state = "OPEN"
def allow_request(self) -> bool:
if self._state == "CLOSED":
return True
if self._state == "OPEN":
if (datetime.utcnow() - self._last_failure_time
> self._recovery_timeout):
self._state = "HALF_OPEN"
return True
return False
# HALF_OPEN: 允许一个试探请求
return True
@property
def state(self) -> str:
return self._state
#9. 性能与监控
#9.1 关键指标
| 指标 | 目标值 | 监控方式 |
|---|---|---|
| 规则层 P99 延迟 | < 50ms | Prometheus histogram |
| ML 层 P99 延迟 | < 200ms | Prometheus histogram |
| 融合层延迟 | < 10ms | Prometheus histogram |
| 人工审核比例 | < 15% | 每日报表 |
| 规则-ML 冲突率 | < 10% | 实时告警 |
| 端到端 P99 | < 500ms | 全链路追踪 |
#9.2 gRPC 接口定义
PROTOBUF
// hybrid_reasoning.proto
syntax = "proto3";
package onto.reasoning.v1;
service HybridReasoningService {
rpc Reason(HybridReasoningRequest) returns (HybridReasoningResponse);
rpc GetReasoningTrace(TraceRequest) returns (ReasoningTrace);
rpc GetQueueStatus(QueueStatusRequest) returns (QueueStatusResponse);
}
message HybridReasoningRequest {
string request_id = 1;
string domain = 2;
map<string, string> facts = 3;
string urgency = 4;
double confidence_threshold = 5;
}
message HybridReasoningResponse {
string conclusion = 1;
double confidence = 2;
string path_taken = 3;
bool human_review_needed = 4;
string review_reason = 5;
repeated string trace = 6;
double elapsed_ms = 7;
}
#10. 实战案例：信用风控混合推理
Python
# 场景：信用申请审批
# 1. 配置领域
router = ReasoningRouter()
router.configure_domain("credit", {
"rules_coverage": 0.7,
"ml_model_available": True,
"ml_model_accuracy": 0.92,
})
# 2. 发起推理请求
request = ReasoningRequest(
request_id="CR-2026-001",
domain="credit",
facts={
"applicant_age": 35,
"annual_income": 250000,
"debt_ratio": 0.45,
"credit_score": 680,
"employment_years": 3,
"previous_defaults": 0,
},
confidence_threshold=0.85,
)
# 3. 路由决策
path = router.route(request)
# -> ReasoningPath.PARALLEL_FUSION (因为 rules_coverage=0.7 且 ML 可用)
# 4. 规则层结果
# Rule: credit_score >= 650 AND debt_ratio < 0.5 -> "conditional_approve"
# confidence: 0.85
# 5. ML 层结果
# Model: gradient_boosting_v3
# prediction: "approve"
# confidence: 0.91
# 6. 融合 (credit 领域权重 0.5:0.5)
# 结论一致方向 ("approve" 系列)
# 融合后置信度: 0.94 (一致性加成)
# 最终: "approve", confidence=0.94
# 7. 人工审核：不需要 (confidence 0.94 > 0.85 阈值)
#Key Takeaways
- 三层架构（规则 + ML + 人工审核）覆盖从确定性到模糊性的全部决策场景
- 路由分发器根据领域配置、规则覆盖率和 ML 可用性动态选择推理路径
- 置信度融合使用加权平均 + 一致性加成/冲突惩罚，确保最终决策可靠
- 人工审核触发基于置信度阈值、层间冲突和领域风险等级自动判断
- 断路器模式防止 ML 服务故障拖垮整个推理系统
- 全链路追踪记录每个推理步骤的决策和耗时，支持事后审计
- 通过 gRPC 与 Reasoning & Decision Layer 其他组件集成，实现毫秒级推理响应
#Next Article
下一篇 S5-04 低代码规则：用 YAML 定义复杂业务规则 将展示如何让业务人员通过 YAML 配置文件而非编程代码来定义和管理业务规则。
tags: #hybrid-reasoning #rule-engine #machine-learning #confidence-fusion #human-review #circuit-breaker #coomia-dip