端到端系统设计 - 技术学习笔记

一句话总结

群聊审核不是"调用一个模型"这么简单--它是一个多层决策系统,需要平衡延迟, 准确率, 成本和用户体验.
本篇把前面所有知识点串成一个完整的生产级架构.

前置回顾

前两篇搞定了推理服务(第 16 篇)和 Go 集成(第 17 篇).
现在退后一步,看全局:一条群消息从进入到被放行或拦截,整条链路怎么设计?

系统全景

                          群聊消息流入
                              │
                              ▼
┌─────────────────────────────────────────────────────────────────┐
│  Go 审核服务 (Moderation Service)                                │
│                                                                  │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │ Layer 1: 规则引擎 (同步, <1ms)                              │ │
│  │  - 关键词黑名单                                             │ │
│  │  - 正则表达式                                               │ │
│  │  - 频率限制(同一用户 10s 内发 20 条 → 刷屏)                 │ │
│  │  - 用户/群白名单                                            │ │
│  └─────────────────────┬──────────────────────────────────────┘ │
│                        │ 未命中规则                               │
│                        ▼                                         │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │ Layer 2: 小模型快判 (同步, <30ms)                           │ │
│  │  - 情感分类 (positive/negative/neutral)                     │ │
│  │  - 意图识别 (spam/ad/normal/question)                       │ │
│  │  - 毒性检测 (toxic/safe)                                    │ │
│  └─────────────────────┬──────────────────────────────────────┘ │
│                        │ 模型置信度 < 阈值(灰色地带)             │
│                        ▼                                         │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │ Layer 3: LLM 深度判断 (异步, 200ms-2s)                      │ │
│  │  - 上下文理解(结合前后 5 条消息)                             │ │
│  │  - 讽刺/反语识别                                            │ │
│  │  - 变体对抗(谐音, 拆字, 火星文)                             │ │
│  └─────────────────────┬──────────────────────────────────────┘ │
│                        │ LLM 也不确定                             │
│                        ▼                                         │
│  ┌────────────────────────────────────────────────────────────┐ │
│  │ Layer 4: 人工复审队列                                       │ │
│  └────────────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────────┘

每一层都是过滤器. 越上层越快越便宜,越下层越准越贵. 设计目标:让 90% 的消息在前两层就完成判断.

Layer 1: 规则引擎

规则引擎是第一道防线,命中即决策,不走模型:

type RuleEngine struct {
    keywords    *ahocorasick.Matcher  // AC 自动机,多模式串匹配
    patterns    []*regexp.Regexp      // 正则规则
    whitelist   map[string]bool       // 白名单用户/群
    rateLimit   *RateLimiter          // 频率控制
}

type RuleResult struct {
    Hit        bool
    Action     Action   // Block / Pass / Escalate
    Reason     string   // "keyword:赌博" / "rate_limit:20msg/10s"
    Confidence float32  // 规则命中 = 1.0
}

func (e *RuleEngine) Check(ctx context.Context, msg *Message) RuleResult {
    // 白名单直接放行
    if e.whitelist[msg.SenderID] || e.whitelist[msg.GroupID] {
        return RuleResult{Hit: true, Action: Pass}
    }

    // 频率检查
    if e.rateLimit.IsExceeded(msg.SenderID) {
        return RuleResult{Hit: true, Action: Block, Reason: "rate_limit"}
    }

    // 关键词匹配
    if matches := e.keywords.FindAll(msg.Text); len(matches) > 0 {
        return RuleResult{Hit: true, Action: Block, Reason: fmt.Sprintf("keyword:%s", matches[0])}
    }

    // 正则匹配
    for _, p := range e.patterns {
        if p.MatchString(msg.Text) {
            return RuleResult{Hit: true, Action: Block, Reason: fmt.Sprintf("pattern:%s", p.String())}
        }
    }

    return RuleResult{Hit: false}
}

规则引擎的局限

规则引擎只能处理已知模式. 对于"这个群主真是个人才呢"(阴阳怪气), "来 W 我主页有好康的"(变体广告),规则无能为力. 这些需要模型理解语义.

Layer 2: 小模型快判

小模型(DistilBERT 级别)负责处理规则引擎漏过的消息. 目标: p99 延迟 < 30ms.

type ModelLayer struct {
    sentimentClient  *inference.Client  // 情感分类
    intentClient     *inference.Client  // 意图识别
    toxicityClient   *inference.Client  // 毒性检测
    thresholds       ModelThresholds
}

type ModelThresholds struct {
    ToxicityBlock     float32 // > 0.9 直接拦截
    ToxicityEscalate  float32 // 0.6 ~ 0.9 上报 LLM
    SpamBlock         float32 // > 0.85 直接拦截
    NegativeEscalate  float32 // 强负面情绪 + 特定意图 → 上报
}

type ModelResult struct {
    Action     Action
    Labels     map[string]string   // "sentiment":"negative", "intent":"spam"
    Scores     map[string]float32  // "toxicity":0.92
    NeedLLM    bool                // 是否需要 LLM 二次判断
}

func (m *ModelLayer) Check(ctx context.Context, msg *Message) (ModelResult, error) {
    // 并发调用三个模型
    g, gCtx := errgroup.WithContext(ctx)

    var toxicity, sentiment, intent *inference.PredictResult

    g.Go(func() error {
        r, err := m.toxicityClient.PredictText(gCtx, msg.Text)
        toxicity = &r
        return err
    })
    g.Go(func() error {
        r, err := m.sentimentClient.PredictText(gCtx, msg.Text)
        sentiment = &r
        return err
    })
    g.Go(func() error {
        r, err := m.intentClient.PredictText(gCtx, msg.Text)
        intent = &r
        return err
    })

    if err := g.Wait(); err != nil {
        return ModelResult{}, err
    }

    return m.decide(toxicity, sentiment, intent), nil
}

决策逻辑

func (m *ModelLayer) decide(toxicity, sentiment, intent *inference.PredictResult) ModelResult {
    result := ModelResult{
        Labels: map[string]string{
            "toxicity":  labelName(toxicity.Label),
            "sentiment": labelName(sentiment.Label),
            "intent":    labelName(intent.Label),
        },
        Scores: map[string]float32{
            "toxicity_score": toxicity.Scores[1], // toxic 类的概率
            "spam_score":     intent.Scores[1],   // spam 类的概率
        },
    }

    // 高置信度毒性 → 直接拦截
    if toxicity.Scores[1] > m.thresholds.ToxicityBlock {
        result.Action = Block
        return result
    }

    // 高置信度垃圾信息 → 直接拦截
    if intent.Scores[1] > m.thresholds.SpamBlock {
        result.Action = Block
        return result
    }

    // 灰色地带 → 交给 LLM
    if toxicity.Scores[1] > m.thresholds.ToxicityEscalate {
        result.Action = Escalate
        result.NeedLLM = true
        return result
    }

    result.Action = Pass
    return result
}

多模型并发 = 体检的不同科室. 去医院体检不是一个医生看全科,而是内科, 外科, 眼科并行检查,最后汇总报告. 同理,情感, 意图, 毒性三个"科室"同时看一条消息,再由决策逻辑汇总判断.

Layer 3: LLM 深度判断

当小模型不确定时(置信度在灰色地带),调用 LLM 做更深入的语义分析:

type LLMLayer struct {
    client     *openai.Client
    model      string
    maxTokens  int
}

func (l *LLMLayer) Check(ctx context.Context, msg *Message, history []Message) (LLMResult, error) {
    // 构造 prompt,包含上下文
    prompt := l.buildPrompt(msg, history)

    resp, err := l.client.CreateChatCompletion(ctx, openai.ChatCompletionRequest{
        Model:      l.model,
        MaxTokens:  l.maxTokens,
        Messages: []openai.ChatCompletionMessage{
            {Role: "system", Content: moderationSystemPrompt},
            {Role: "user", Content: prompt},
        },
        ResponseFormat: &openai.ChatCompletionResponseFormat{
            Type: openai.ChatCompletionResponseFormatTypeJSONObject,
        },
    })
    if err != nil {
        return LLMResult{}, err
    }

    return parseLLMResponse(resp.Choices[0].Message.Content)
}

const moderationSystemPrompt = `你是群聊内容审核助手. 分析给定消息是否违规.
输出 JSON: {"action":"block"|"pass"|"review", "reason":"...", "confidence":0.0-1.0}
违规类型: spam(广告垃圾), toxic(辱骂攻击), nsfw(色情), scam(诈骗), harassment(骚扰)
注意识别: 谐音替换, 拆字, 反语讽刺, 上下文暗示.
如果无法判断,action 设为 review.`

LLM 调用的成本控制

LLM 调用成本比小模型高 100-1000 倍. 必须严格控制进入 Layer 3 的流量:

只有小模型不确定的消息才走 LLM(预计 < 5% 流量)
设置每分钟 LLM 调用量上限
LLM 超时(>2s)直接走人工复审,不等待

Layer 4: 人工复审

所有层都无法确定的消息进入人工队列:

type ReviewQueue struct {
    store  ReviewStore  // Redis sorted set 或消息队列
}

type ReviewItem struct {
    MessageID   string
    GroupID     string
    SenderID    string
    Text        string
    ModelScores map[string]float32
    LLMResult   *LLMResult  // 可能为空(LLM 超时)
    Priority    float64     // 越高越紧急
    CreatedAt   time.Time
}

func (q *ReviewQueue) Enqueue(ctx context.Context, item ReviewItem) error {
    // 优先级 = 毒性分数 × 群活跃度. 大群的可疑消息优先处理
    item.Priority = float64(item.ModelScores["toxicity_score"]) * groupActivityWeight(item.GroupID)
    return q.store.Add(ctx, item)
}

主调度器:串联四层

type Moderator struct {
    rules  *RuleEngine
    models *ModelLayer
    llm    *LLMLayer
    review *ReviewQueue
    metrics *Metrics
}

func (m *Moderator) CheckMessage(ctx context.Context, msg *Message) (Decision, error) {
    start := time.Now()
    defer func() {
        m.metrics.RecordLatency("total", time.Since(start))
    }()

    // Layer 1: 规则引擎
    ruleResult := m.rules.Check(ctx, msg)
    if ruleResult.Hit {
        m.metrics.RecordDecision("rules", ruleResult.Action)
        return Decision{Action: ruleResult.Action, Source: "rules", Reason: ruleResult.Reason}, nil
    }

    // Layer 2: 小模型
    modelResult, err := m.models.Check(ctx, msg)
    if err != nil {
        // 模型不可用时降级到规则引擎结果(Pass)
        m.metrics.RecordError("model_unavailable")
        return Decision{Action: Pass, Source: "degraded"}, nil
    }

    if modelResult.Action == Block {
        m.metrics.RecordDecision("model", Block)
        return Decision{Action: Block, Source: "model", Scores: modelResult.Scores}, nil
    }

    if !modelResult.NeedLLM {
        m.metrics.RecordDecision("model", Pass)
        return Decision{Action: Pass, Source: "model"}, nil
    }

    // Layer 3: LLM (异步,先放行消息,后台判断)
    go m.asyncLLMCheck(context.Background(), msg, modelResult)

    // 消息先放行,LLM 结果出来后如果是违规再撤回
    return Decision{Action: Pass, Source: "pending_llm"}, nil
}

func (m *Moderator) asyncLLMCheck(ctx context.Context, msg *Message, modelResult ModelResult) {
    ctx, cancel := context.WithTimeout(ctx, 3*time.Second)
    defer cancel()

    history, _ := m.fetchMessageHistory(ctx, msg.GroupID, 5)
    llmResult, err := m.llm.Check(ctx, msg, history)
    if err != nil {
        // LLM 失败,进人工队列
        m.review.Enqueue(ctx, ReviewItem{
            MessageID:   msg.ID,
            Text:        msg.Text,
            ModelScores: modelResult.Scores,
        })
        return
    }

    if llmResult.Action == "block" {
        m.recallMessage(ctx, msg) // 撤回已发送的消息
        m.metrics.RecordDecision("llm", Block)
    }
}

监控指标

核心指标

指标	含义	告警阈值
`moderation_latency_p99`	审核延迟 p99	> 50ms(同步部分)
`moderation_accuracy`	模型准确率(通过人工标注验证)	< 95%
`false_positive_rate`	误封率(正常消息被拦截)	> 1%
`false_negative_rate`	漏放率(违规消息被放行)	> 5%
`model_qps`	推理服务 QPS	接近容量上限 80%
`llm_escalation_rate`	进入 LLM 层的比例	> 10%
`review_queue_depth`	人工复审队列深度	> 1000

Prometheus 指标埋点

type Metrics struct {
    latency    *prometheus.HistogramVec
    decisions  *prometheus.CounterVec
    errors     *prometheus.CounterVec
    queueDepth prometheus.Gauge
}

func NewMetrics(reg prometheus.Registerer) *Metrics {
    m := &Metrics{
        latency: prometheus.NewHistogramVec(prometheus.HistogramOpts{
            Name:    "moderation_latency_seconds",
            Buckets: []float64{0.001, 0.005, 0.01, 0.03, 0.05, 0.1, 0.5},
        }, []string{"layer"}),
        decisions: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "moderation_decisions_total",
        }, []string{"source", "action"}),
        errors: prometheus.NewCounterVec(prometheus.CounterOpts{
            Name: "moderation_errors_total",
        }, []string{"type"}),
        queueDepth: prometheus.NewGauge(prometheus.GaugeOpts{
            Name: "moderation_review_queue_depth",
        }),
    }
    reg.MustRegister(m.latency, m.decisions, m.errors, m.queueDepth)
    return m
}

func (m *Metrics) RecordLatency(layer string, d time.Duration) {
    m.latency.WithLabelValues(layer).Observe(d.Seconds())
}

func (m *Metrics) RecordDecision(source string, action Action) {
    m.decisions.WithLabelValues(source, action.String()).Inc()
}

A/B 测试:模型版本对比

上线新模型时,不能直接全量替换. 用 A/B 测试验证效果:

type ABRouter struct {
    modelA    string // "sentiment_model/1" (当前生产版本)
    modelB    string // "sentiment_model/2" (新版本)
    bPercent  int    // B 版本流量百分比(如 10%)
}

func (r *ABRouter) Route(msg *Message) string {
    // 按 group_id hash 分流,保证同一个群的消息走同一个版本
    hash := fnv32(msg.GroupID)
    if int(hash%100) < r.bPercent {
        return r.modelB
    }
    return r.modelA
}

func fnv32(s string) uint32 {
    h := fnv.New32a()
    h.Write([]byte(s))
    return h.Sum32()
}

A/B 测试观察周期内需要对比的指标:

┌────────────────────────────────────────────────────┐
│  A/B 测试 Dashboard                                │
│                                                     │
│  版本 A (v1, 90% 流量):                            │
│    准确率: 96.2%  |  误封率: 0.8%  |  p99: 22ms    │
│                                                     │
│  版本 B (v2, 10% 流量):                            │
│    准确率: 97.1%  |  误封率: 0.5%  |  p99: 25ms    │
│                                                     │
│  结论: B 准确率提升, 误封率下降, 延迟略增可接受     │
│  建议: 扩大 B 流量到 50%, 继续观察 3 天             │
└────────────────────────────────────────────────────┘

优雅降级

当推理服务不可用时,系统不能完全停摆:

type DegradationStrategy struct {
    circuitBreaker *CircuitBreaker
    fallbackRules  *RuleEngine  // 加强版规则引擎(比正常模式更严格的关键词列表)
}

func (d *DegradationStrategy) OnModelUnavailable(ctx context.Context, msg *Message) Decision {
    // 熔断状态下的降级策略
    switch d.level() {
    case DegradeLevel1:
        // 模型超时但服务未完全宕 → 只用规则引擎,灰色消息放行
        result := d.fallbackRules.Check(ctx, msg)
        if result.Hit {
            return Decision{Action: Block, Source: "degraded_rules"}
        }
        return Decision{Action: Pass, Source: "degraded_pass"}

    case DegradeLevel2:
        // 模型服务完全不可用 → 严格模式,可疑消息全部进人工队列
        result := d.fallbackRules.Check(ctx, msg)
        if result.Hit {
            return Decision{Action: Block, Source: "degraded_rules"}
        }
        return Decision{Action: Escalate, Source: "degraded_review"}

    case DegradeLevel3:
        // 极端情况:人工队列也满了 → 只拦截规则明确命中的,其余全放行
        result := d.fallbackRules.Check(ctx, msg)
        if result.Hit {
            return Decision{Action: Block, Source: "emergency_rules"}
        }
        return Decision{Action: Pass, Source: "emergency_pass"}
    }
    return Decision{Action: Pass}
}

降级不能沉默

进入降级模式时必须触发告警. 否则模型服务挂了几个小时,违规消息一直在放行,没人知道.
降级告警 + 定时恢复探测,是生产必备.

生产注意事项

先放行后撤回的风险

LLM 异步审核意味着违规消息会短暂展示(200ms-2s). 对于极端违规内容(如暴力恐怖),这不可接受. 解决方案:对高风险关键词(哪怕模型不确定)先拦截再审核,宁可误伤不可漏放.

上下文窗口

单条消息看起来无害,结合上下文才能判断(如"+1"本身无害,但如果前一条是赌博广告,"+1"就是参与). LLM 层需要传入前后 N 条消息. 但上下文太长会增加延迟和成本,推荐 5-10 条.

快速回顾

四层架构: 规则(快) → 小模型(准) → LLM(深) → 人工(兜底), 逐层过滤
90% 在前两层解决: 规则引擎 + 小模型覆盖绝大多数场景
异步 LLM: 灰色消息先放行后撤回,平衡用户体验和审核准确度
A/B 测试: 按群 hash 分流,同群同版本,观察指标对比后全量
优雅降级: 模型挂了不等于系统挂了,规则引擎兜底 + 告警
监控为王: 延迟, 准确率, 误封率, 漏放率, 队列深度,缺一不可

动手练习

实现主调度器: 完成 Moderator.CheckMessage 的完整逻辑,包含四层调用
并发模型调用: 用 errgroup 并发调用三个模型,测量相比串行调用的延迟提升
熔断器: 实现一个简单的熔断器,连续 5 次推理失败后进入降级模式,30 秒后自动恢复探测
A/B 路由: 实现按 group_id hash 分流逻辑,验证分布均匀性
指标埋点: 接入 Prometheus,用 Grafana 展示审核延迟分布和各层决策比例
降级演练: 手动停止推理服务,验证系统自动切换到降级模式并触发告警