基于大模型的分布式事务异常检测与自动回滚决策：从指标异常到智能止损

一、分布式事务的"黑箱"困境：异常检测的滞后与回滚的犹豫

分布式事务跨越多个参与节点，任何一方的超时、宕机或网络分区都可能导致事务悬挂、数据不一致。传统监控依赖固定阈值告警——事务执行时间超过 5 秒就报警，但不同业务的事务耗时差异巨大，固定阈值要么误报成灾，要么漏报致命。更棘手的是回滚决策：超时后是立即回滚，还是再等一等？盲目回滚可能打断本可完成的事务，延迟回滚则让锁持有时间拉长，拖垮整个系统。

大模型在分布式事务异常检测中的价值在于：它能理解多维指标的语义关联，将事务执行时间、锁等待时长、网络延迟、CPU 负载等信号综合判断，而非简单阈值比对。同时，基于历史事务的执行模式，模型可以预测事务最终成功或失败的概率，为回滚决策提供量化依据。

二、异常检测与回滚决策的架构

flowchart TD
    A[分布式事务执行] --> B[指标采集层]
    B --> C[事务指标: 执行时间/锁等待/重试次数]
    B --> D[系统指标: CPU/内存/网络延迟]
    B --> E[日志指标: 错误日志/超时日志]
    C & D & E --> F[特征拼接与向量化]
    F --> G[大模型异常判断]
    G --> H{异常置信度}
    H -->|高置信异常| I[生成回滚决策]
    H -->|低置信/正常| J[继续等待执行]
    I --> K[回滚执行器]
    K --> L[记录决策日志与反馈]
    J --> L

三、核心代码实现

3.1 事务指标采集与特征工程

from dataclasses import dataclass, field
from typing import List, Optional
import numpy as np
from datetime import datetime, timedelta

@dataclass
class TransactionMetrics:
    """分布式事务运行时指标"""
    tx_id: str
    start_time: datetime
    participants: List[str]

    # 事务级指标
    elapsed_ms: float              # 已执行时间
    lock_wait_ms: float            # 锁等待累计时间
    retry_count: int               # 重试次数
    participant_count: int         # 参与者数量

    # 系统级指标
    avg_cpu_usage: float           # 参与节点平均 CPU
    avg_network_latency_ms: float  # 参与节点间平均网络延迟
    error_log_rate: float          # 近 1 分钟错误日志频率

    # 历史统计指标
    p99_duration_ms: float         # 同类事务 P99 耗时
    historical_success_rate: float # 同类事务历史成功率

class TransactionFeatureBuilder:
    """事务特征构建器：将原始指标转换为模型输入"""

    def build_features(self, metrics: TransactionMetrics) -> dict:
        # 归一化：已执行时间与同类 P99 的比值
        duration_ratio = metrics.elapsed_ms / max(metrics.p99_duration_ms, 1)

        # 锁等待占比：锁等待时间占已执行时间的比例
        lock_ratio = metrics.lock_wait_ms / max(metrics.elapsed_ms, 1)

        # 重试密度：重试次数与参与者数量的比值
        retry_density = metrics.retry_count / max(metrics.participant_count, 1)

        # 综合异常评分：多个指标的加权组合
        anomaly_score = (
            duration_ratio * 0.3 +
            lock_ratio * 0.2 +
            retry_density * 0.2 +
            metrics.avg_cpu_usage * 0.1 +
            (metrics.avg_network_latency_ms / 100) * 0.1 +
            (1 - metrics.historical_success_rate) * 0.1
        )

        return {
            "duration_ratio": round(duration_ratio, 3),
            "lock_ratio": round(lock_ratio, 3),
            "retry_density": round(retry_density, 3),
            "cpu_usage": round(metrics.avg_cpu_usage, 3),
            "network_latency": round(metrics.avg_network_latency_ms, 1),
            "error_log_rate": round(metrics.error_log_rate, 3),
            "historical_success_rate": round(metrics.historical_success_rate, 3),
            "anomaly_score": round(anomaly_score, 3),
        }

3.2 基于大模型的异常判断与回滚决策

import json
from typing import Tuple

class TransactionAnomalyDetector:
    """分布式事务异常检测器：基于大模型判断事务是否异常"""

    def __init__(self, llm_client):
        self.llm = llm_client
        self.feature_builder = TransactionFeatureBuilder()

    def analyze(self, metrics: TransactionMetrics) -> Tuple[bool, float, str]:
        """
        分析事务是否异常，返回 (是否异常, 置信度, 决策理由)
        """
        features = self.feature_builder.build_features(metrics)

        prompt = f"""你是一个分布式事务监控专家。根据以下事务运行时指标，判断该事务是否异常，并给出回滚建议。

事务指标：
- 已执行时间与 P99 比值: {features['duration_ratio']}
- 锁等待占比: {features['lock_ratio']}
- 重试密度: {features['retry_density']}
- 参与节点 CPU: {features['cpu_usage']}
- 网络延迟: {features['network_latency']}ms
- 错误日志频率: {features['error_log_rate']}
- 历史成功率: {features['historical_success_rate']}
- 综合异常评分: {features['anomaly_score']}

请以 JSON 格式输出：
{{
  "is_anomaly": true/false,
  "confidence": 0.0-1.0,
  "should_rollback": true/false,
  "reason": "判断理由"
}}"""

        response = self.llm.chat(prompt)
        result = json.loads(response)

        return (
            result["is_anomaly"],
            result["confidence"],
            result.get("reason", "")
        )

class RollbackDecisionEngine:
    """回滚决策引擎：综合异常检测结果与业务规则做出最终决策"""

    # 回滚置信度阈值：超过此值才触发回滚
    ROLLBACK_CONFIDENCE_THRESHOLD = 0.75

    # 最大等待时间：即使模型判断正常，超过此时间也强制回滚
    MAX_WAIT_MS = 30000

    def decide(
        self,
        metrics: TransactionMetrics,
        is_anomaly: bool,
        confidence: float,
        reason: str
    ) -> dict:
        """做出回滚决策"""
        decision = {
            "tx_id": metrics.tx_id,
            "action": "continue",
            "confidence": confidence,
            "reason": reason,
        }

        # 规则 1：超过最大等待时间，强制回滚
        if metrics.elapsed_ms > self.MAX_WAIT_MS:
            decision["action"] = "force_rollback"
            decision["reason"] = (
                f"事务已执行 {metrics.elapsed_ms}ms，"
                f"超过最大等待时间 {self.MAX_WAIT_MS}ms"
            )
            return decision

        # 规则 2：模型判断异常且置信度足够高
        if is_anomaly and confidence >= self.ROLLBACK_CONFIDENCE_THRESHOLD:
            decision["action"] = "rollback"
            decision["reason"] = (
                f"模型检测异常(置信度={confidence:.2f}): {reason}"
            )
            return decision

        # 规则 3：模型判断异常但置信度不足，继续等待但标记观察
        if is_anomaly and confidence < self.ROLLBACK_CONFIDENCE_THRESHOLD:
            decision["action"] = "observe"
            decision["reason"] = (
                f"异常信号但置信度不足({confidence:.2f})，继续观察"
            )
            return decision

        return decision

3.3 反馈闭环：决策效果追踪

class DecisionFeedbackTracker:
    """决策反馈追踪器：记录每次决策的实际结果，用于模型迭代"""

    def record(self, tx_id: str, decision: dict, actual_outcome: str):
        """
        actual_outcome: "committed" | "rolled_back" | "timed_out"
        """
        feedback = {
            "tx_id": tx_id,
            "decision_action": decision["action"],
            "decision_confidence": decision["confidence"],
            "actual_outcome": actual_outcome,
            "decision_correct": self._evaluate_correctness(
                decision["action"], actual_outcome
            ),
        }
        # 写入反馈存储，供后续模型训练使用
        self._persist(feedback)

    def _evaluate_correctness(self, action: str, outcome: str) -> bool:
        """评估决策是否正确"""
        if action == "rollback" and outcome == "timed_out":
            return True   # 正确回滚了最终超时的事务
        if action == "continue" and outcome == "committed":
            return True   # 正确等待了最终提交的事务
        if action == "rollback" and outcome == "committed":
            return False  # 误回滚了本可提交的事务
        if action == "continue" and outcome == "timed_out":
            return False  # 应该更早回滚
        return True

四、异常检测与回滚决策的边界分析

大模型的推理延迟。每次异常判断需要调用大模型，延迟在 200ms-2s 之间。对于超时临界点的事务，这个延迟可能错过最佳回滚窗口。建议对高优先级事务预计算异常评分，模型仅做最终确认。

误回滚的代价不对称。回滚一个本可提交的事务（误杀）比延迟回滚一个最终超时的事务（漏杀）代价更高——前者导致业务中断，后者仅增加锁持有时间。决策引擎应偏向保守，提高回滚置信度阈值。

冷启动问题。新上线的事务类型没有历史数据，P99 和成功率无法计算。建议冷启动阶段采用更严格的超时阈值，积累足够样本后再启用模型决策。

适用边界：该方案适合事务类型多样、执行时间分布差异大、固定阈值频繁误报的场景。对于事务类型单一、执行时间稳定的系统，固定阈值已经足够。

五、总结

基于大模型的分布式事务异常检测，通过多维指标的语义关联判断事务健康状态，替代固定阈值告警。回滚决策引擎结合模型判断与业务规则，在置信度足够时触发回滚，避免盲目操作。落地时需关注推理延迟对决策时效的影响、误回滚与漏回滚的代价不对称，以及新事务类型的冷启动策略。建议采用"规则兜底 + 模型增强"的混合模式，在模型不可靠时回退到固定阈值。