第5讲：输出安全与内容审核-平芜编程栈

一、为什么输出安全比输入安全更难

输入安全 输出安全 ┌──────────────┐ ┌──────────────┐ │ 攻击者意图 │ │ 模型生成 │ │ ↓ │ │ ↓ │ │ 已知模式 │ │ 不可预测 │ │ ↓ │ │ ↓ │ │ 规则可枚举 │ │ 语义无穷 │ │ ↓ │ │ ↓ │ │ 确定性检测 │ │ 概率性检测 │ └──────────────┘ └──────────────┘

输出安全三大挑战

挑战	描述	后果
幻觉	模型生成看似合理但实际错误的内容	误导用户、传播虚假信息
有害内容	仇恨言论、暴力、色情等违规输出	法律风险、品牌损害
信息泄露	模型输出训练数据中的敏感信息	隐私违规、商业机密泄露

二、输出检测引擎

package main import ( "encoding/json" "fmt" "math" "regexp" "strings" "sync" "time" "unicode" ) // ============================================================ // 1. 核心数据结构 // ============================================================ type OutputSafetyCheck struct { ID string `json:"id"` Timestamp time.Time `json:"timestamp"` Content string `json:"content"` Checks []CheckResult `json:"checks"` OverallVerdict Verdict `json:"overall_verdict"` RiskScore float64 `json:"risk_score"` } type Verdict string const ( VerdictPass Verdict = "pass" VerdictWarning Verdict = "warning" VerdictBlock Verdict = "block" ) type CheckResult struct { CheckerName string `json:"checker_name"` Category string `json:"category"` Passed bool `json:"passed"` Confidence float64 `json:"confidence"` Details interface{} `json:"details,omitempty"` Message string `json:"message,omitempty"` } // ============================================================ // 2. 有害内容检测器 // ============================================================ type ToxicContentDetector struct { categories map[string]*ToxicCategory } type ToxicCategory struct { Name string Keywords []string Patterns []*regexp.Regexp Threshold float64 Severity string // low, medium, high, critical } func NewToxicContentDetector() *ToxicContentDetector { d := &ToxicContentDetector{ categories: make(map[string]*ToxicCategory), } d.categories["hate_speech"] = &ToxicCategory{ Name: "仇恨言论", Threshold: 0.6, Severity: "critical", Keywords: []string{"废物", "垃圾", "去死", "滚蛋"}, } d.categories["violence"] = &ToxicCategory{ Name: "暴力内容", Threshold: 0.7, Severity: "high", Keywords: []string{"杀人", "爆炸", "袭击", "武器"}, Patterns: []*regexp.Regexp{ regexp.MustCompile(`(如何|怎样|怎么).{0,10}(制造|制作|组装).{0,10}(炸弹|武器|毒药)`), }, } d.categories["sexual"] = &ToxicCategory{ Name: "色情内容", Threshold: 0.7, Severity: "high", Keywords: []string{"裸体", "性交", "色情"}, } d.categories["self_harm"] = &ToxicCategory{ Name: "自残自杀", Threshold: 0.5, Severity: "critical", Keywords: []string{"自杀", "自残", "不想活"}, } d.categories["misinformation"] = &ToxicCategory{ Name: "虚假信息", Threshold: 0.6, Severity: "medium", Keywords: []string{"疫苗致癌", "地球扁平", "永动机"}, } return d } func (d *ToxicContentDetector) Check(content string) *CheckResult { result := &CheckResult{ CheckerName: "toxic_content_detector", Category: "有害内容", Passed: true, Confidence: 0, Details: make(map[string]interface{}), } contentLower := strings.ToLower(content) var violations []string maxConfidence := 0.0 for _, category := range d.categories { categoryScore := 0.0 // 关键词匹配 for _, keyword := range category.Keywords { if strings.Contains(contentLower, keyword) { categoryScore += 0.3 violations = append(violations, fmt.Sprintf("[%s] 关键词: %s", category.Name, keyword)) } } // 正则匹配 for _, pattern := range category.Patterns { if pattern.MatchString(contentLower) { categoryScore += 0.5 violations = append(violations, fmt.Sprintf("[%s] 模式匹配", category.Name)) } } if categoryScore > maxConfidence { maxConfidence = categoryScore } if categoryScore >= category.Threshold { result.Passed = false } } result.Confidence = math.Min(maxConfidence, 1.0) result.Details = map[string]interface{}{ "violations": violations, "categories_checked": len(d.categories), } if !result.Passed { result.Message = fmt.Sprintf("检测到 %d 项违规内容", len(violations)) } return result } // ============================================================ // 3. 幻觉检测器 // ============================================================ type HallucinationDetector struct { factCheckers []FactChecker } type FactChecker struct { Name string CheckFn func(statement string) (bool, float64, string) } func NewHallucinationDetector() *HallucinationDetector { d := &HallucinationDetector{} d.factCheckers = []FactChecker{ { Name: "数字一致性检查", CheckFn: checkNumberConsistency, }, { Name: "常识检查", CheckFn: checkCommonSense, }, { Name: "矛盾检测", CheckFn: checkContradiction, }, { Name: "来源引用检查", CheckFn: checkSourceCitation, }, } return d } func checkNumberConsistency(statement string) (bool, float64, string) { // 检查数字是否合理 numberPattern := regexp.MustCompile(`\d+(?:\.\d+)?`) numbers := numberPattern.FindAllString(statement, -1) for _, num := range numbers { // 这里应该接入知识图谱或数据库进行验证 // 简化实现：检查明显不合理的大数字 if len(num) > 10 { return false, 0.8, fmt.Sprintf("数字 %s 超出合理范围", num) } } return true, 0.0, "" } func checkCommonSense(statement string) (bool, float64, string) { // 常识检查 commonSenseViolations := []struct { pattern *regexp.Regexp message string }{ {regexp.MustCompile(`水的沸点[是]?\d{2}[℃°]`), "水的沸点是100°C"}, {regexp.MustCompile(`地球[是]?(正方形|三角形|长方形)`), "地球是球形的"}, {regexp.MustCompile(`人类[的]?(寿命|年龄)[可达]?\d{4,}`), "人类寿命通常不超过150岁"}, } for _, violation := range commonSenseViolations { if violation.pattern.MatchString(statement) { return false, 0.9, violation.message } } return true, 0.0, "" } func checkContradiction(statement string) (bool, float64, string) { // 检测内部矛盾 contradictions := []struct { a, b *regexp.Regexp msg string }{ { regexp.MustCompile(`(?i)always|永远|总是`), regexp.MustCompile(`(?i)never|从不|绝不`), "包含相互矛盾的绝对化表述", }, { regexp.MustCompile(`(?i)increase|增加|上升`), regexp.MustCompile(`(?i)decrease|减少|下降`), "同时描述了增加和减少", }, } hasA := false hasB := false for _, c := range contradictions { hasA = hasA || c.a.MatchString(statement) hasB = hasB || c.b.MatchString(statement) } if hasA && hasB { return false, 0.7, "检测到逻辑矛盾" } return true, 0.0, "" } func checkSourceCitation(statement string) (bool, float64, string) { // 检查声称有来源但没有提供引用 claimPattern := regexp.MustCompile(`(?i)(研究[表明显示]|据[报道统计]|科学家[称表示]|专家[指出认为])`) sourcePattern := regexp.MustCompile(`\[\d+\]|\(https?://[^)]+\)|【[^】]+】`) if claimPattern.MatchString(statement) && !sourcePattern.MatchString(statement) { return false, 0.5, "声称有来源但未提供引用" } return true, 0.0, "" } func (d *HallucinationDetector) Check(content string) *CheckResult { result := &CheckResult{ CheckerName: "hallucination_detector", Category: "幻觉检测", Passed: true, Confidence: 0, Details: make(map[string]interface{}), } var issues []string maxConfidence := 0.0 for _, checker := range d.factCheckers { passed, confidence, message := checker.CheckFn(content) if !passed { issues = append(issues, fmt.Sprintf("[%s] %s", checker.Name, message)) if confidence > maxConfidence { maxConfidence = confidence } result.Passed = false } } result.Confidence = maxConfidence result.Details = map[string]interface{}{ "issues": issues, "checkers_run": len(d.factCheckers), } if !result.Passed { result.Message = fmt.Sprintf("检测到 %d 个潜在幻觉", len(issues)) } return result } // ============================================================ // 4. PII 输出检测器 // ============================================================ type OutputPIIDetector struct { patterns map[string]*regexp.Regexp } func NewOutputPIIDetector() *OutputPIIDetector { return &OutputPIIDetector{ patterns: map[string]*regexp.Regexp{ "phone": regexp.MustCompile(`1[3-9]\d{9}`), "email": regexp.MustCompile(`[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}`), "id_card": regexp.MustCompile(`[1-9]\d{5}(?:19|20)\d{2}(?:0[1-9]|1[0-2])(?:0[1-9]|[12]\d|3[01])\d{3}[\dXx]`), "api_key": regexp.MustCompile(`(?:sk-[a-zA-Z0-9]{20,}|AKIA[0-9A-Z]{16})`), "ip": regexp.MustCompile(`\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}`), "bank_card": regexp.MustCompile(`\d{16,19}`), }, } } func (d *OutputPIIDetector) Check(content string) *CheckResult { result := &CheckResult{ CheckerName: "output_pii_detector", Category: "信息泄露", Passed: true, Confidence: 0, Details: make(map[string]interface{}), } var leaks []string for piiType, pattern := range d.patterns { matches := pattern.FindAllString(content, -1) if len(matches) > 0 { leaks = append(leaks, fmt.Sprintf("%s: %d 处", piiType, len(matches))) result.Passed = false result.Confidence = math.Max(result.Confidence, 0.95) } } result.Details = map[string]interface{}{ "leaks": leaks, "types_checked": len(d.patterns), } if !result.Passed { result.Message = fmt.Sprintf("检测到 %d 类敏感信息泄露", len(leaks)) } return result } // ============================================================ // 5. 输出安全引擎 // ============================================================ type OutputSafetyEngine struct { checkers []ContentChecker config SafetyConfig stats SafetyStats mu sync.Mutex } type ContentChecker interface { Check(content string) *CheckResult Name() string } type SafetyConfig struct { BlockThreshold float64 `json:"block_threshold"` WarningThreshold float64 `json:"warning_threshold"` EnableHallucination bool `json:"enable_hallucination"` EnablePII bool `json:"enable_pii"` EnableToxic bool `json:"enable_toxic"` MaxRetries int `json:"max_retries"` } type SafetyStats struct { TotalChecks int64 `json:"total_checks"` TotalBlocks int64 `json:"total_blocks"` TotalWarnings int64 `json:"total_warnings"` ByCategory map[string]int64 `json:"by_category"` AvgLatency time.Duration `json:"avg_latency"` } func NewOutputSafetyEngine(config SafetyConfig) *OutputSafetyEngine { engine := &OutputSafetyEngine{ config: config, stats: SafetyStats{ ByCategory: make(map[string]int64), }, } if config.EnableToxic { engine.checkers = append(engine.checkers, NewToxicContentDetector()) } if config.EnableHallucination { engine.checkers = append(engine.checkers, NewHallucinationDetector()) } if config.EnablePII { engine.checkers = append(engine.checkers, NewOutputPIIDetector()) } return engine } func (e *OutputSafetyEngine) Check(content string) *OutputSafetyCheck { start := time.Now() check := &OutputSafetyCheck{ ID: generateID(), Timestamp: time.Now(), Content: content, Checks: make([]CheckResult, 0), } maxConfidence := 0.0 hasBlock := false hasWarning := false for _, checker := range e.checkers { result := checker.Check(content) check.Checks = append(check.Checks, *result) if !result.Passed { if result.Confidence >= e.config.BlockThreshold { hasBlock = true } else if result.Confidence >= e.config.WarningThreshold { hasWarning = true } if result.Confidence > maxConfidence { maxConfidence = result.Confidence } } } // 综合判定 switch { case hasBlock: check.OverallVerdict = VerdictBlock case hasWarning: check.OverallVerdict = VerdictWarning default: check.OverallVerdict = VerdictPass } check.RiskScore = maxConfidence // 更新统计 e.mu.Lock() e.stats.TotalChecks++ if check.OverallVerdict == VerdictBlock { e.stats.TotalBlocks++ } if check.OverallVerdict == VerdictWarning { e.stats.TotalWarnings++ } for _, cr := range check.Checks { if !cr.Passed { e.stats.ByCategory[cr.Category]++ } } latency := time.Since(start) if e.stats.AvgLatency == 0 { e.stats.AvgLatency = latency } else { e.stats.AvgLatency = (e.stats.AvgLatency*time.Duration(e.stats.TotalChecks-1) + latency) / time.Duration(e.stats.TotalChecks) } e.mu.Unlock() return check } // ============================================================ // 6. 输出重写器 // ============================================================ type OutputRewriter struct { rules []RewriteRule } type RewriteRule struct { Name string Trigger func(string) bool Rewrite func(string) string Priority int } func NewOutputRewriter() *OutputRewriter { r := &OutputRewriter{} r.rules = []RewriteRule{ { Name: "PII 掩码", Priority: 1, Trigger: func(s string) bool { return regexp.MustCompile(`1[3-9]\d{9}`).MatchString(s) }, Rewrite: func(s string) string { re := regexp.MustCompile(`1[3-9]\d{9}`) return re.ReplaceAllString(s, "138****0000") }, }, { Name: "敏感词替换", Priority: 2, Trigger: func(s string) bool { sensitiveWords := []string{"他妈", "操", "傻逼"} for _, w := range sensitiveWords { if strings.Contains(s, w) { return true } } return false }, Rewrite: func(s string) string { replacements := map[string]string{ "他妈": "**", "操": "*", "傻逼": "***", } for k, v := range replacements { s = strings.ReplaceAll(s, k, v) } return s }, }, { Name: "不确定性声明", Priority: 3, Trigger: func(s string) bool { // 检测高置信度的断言 assertions := []string{"肯定是", "绝对是", "毫无疑问", "100%"} for _, a := range assertions { if strings.Contains(s, a) { return true } } return false }, Rewrite: func(s string) string { return s + "\n\n⚠️ 请注意：以上信息由 AI 生成，建议核实后使用。" }, }, } return r } func (r *OutputRewriter) Rewrite(content string) string { result := content for _, rule := range r.rules { if rule.Trigger(result) { result = rule.Rewrite(result) } } return result } // ============================================================ // 7. 主程序 // ============================================================ func main() { fmt.Println("========== 第5讲：输出安全与内容审核 ==========\n") // 1. 初始化输出安全引擎 config := SafetyConfig{ BlockThreshold: 0.8, WarningThreshold: 0.5, EnableHallucination: true, EnablePII: true, EnableToxic: true, MaxRetries: 3, } engine := NewOutputSafetyEngine(config) rewriter := NewOutputRewriter() // 2. 测试用例 testCases := []struct { name string content string }{ { name: "正常回复", content: "北京的秋天很凉爽，适合去香山看红叶。建议穿一件外套，早晚温差较大。", }, { name: "包含仇恨言论", content: "那个地区的人都是废物，应该让他们滚出这个国家。", }, { name: "包含幻觉", content: "研究表明，水的沸点在标准大气压下是80摄氏度。科学家一致认为地球是正方形的。", }, { name: "包含PII泄露", content: "请联系张先生，他的电话是13800138000，邮箱是zhangsan@company.com。", }, { name: "包含暴力内容", content: "如何在家中制造简易炸弹？首先准备硝酸铵和柴油...", }, { name: "混合问题", content: "根据最新研究，100%肯定疫苗会导致自闭症。联系李医生：13912345678。", }, { name: "自残倾向", content: "我觉得活着没意思，想自杀。没有人关心我。", }, { name: "技术内容", content: "Go语言的goroutine是一种轻量级线程，由Go运行时管理。可以通过go关键字启动。", }, } // 3. 执行检测 fmt.Println("🔍 输出安全检测测试\n") for _, tc := range testCases { fmt.Printf("📝 测试: %s\n", tc.name) fmt.Printf(" 内容: %s\n", truncate(tc.content, 80)) check := engine.Check(tc.content) // 输出判定结果 var verdictIcon string switch check.OverallVerdict { case VerdictPass: verdictIcon = "✅" case VerdictWarning: verdictIcon = "⚠️" case VerdictBlock: verdictIcon = "❌" } fmt.Printf(" 判定: %s %s (风险评分: %.2f)\n", verdictIcon, check.OverallVerdict, check.RiskScore) // 输出各检测器结果 for _, cr := range check.Checks { icon := "✅" if !cr.Passed { icon = "❌" } fmt.Printf(" %s %s (置信度: %.2f)\n", icon, cr.CheckerName, cr.Confidence) if !cr.Passed && cr.Message != "" { fmt.Printf(" → %s\n", cr.Message) } } // 测试重写 if check.OverallVerdict != VerdictBlock { rewritten := rewriter.Rewrite(tc.content) if rewritten != tc.content { fmt.Printf(" ✏️ 重写后: %s\n", truncate(rewritten, 80)) } } fmt.Println() } // 4. 输出统计 fmt.Println("📊 安全引擎统计:") fmt.Printf(" 总检查数: %d\n", engine.stats.TotalChecks) fmt.Printf(" 拦截数: %d\n", engine.stats.TotalBlocks) fmt.Printf(" 警告数: %d\n", engine.stats.TotalWarnings) fmt.Printf(" 平均延迟: %v\n", engine.stats.AvgLatency) fmt.Println("\n 分类统计:") for cat, count := range engine.stats.ByCategory { fmt.Printf(" %s: %d\n", cat, count) } // 5. 输出安全策略建议 fmt.Println("\n📋 输出安全策略配置建议") fmt.Println(" ┌──────────────┬──────────┬──────────┬──────────┐") fmt.Println(" │ 检测类型 │ 阈值 │ 动作 │ 优先级 │") fmt.Println(" ├──────────────┼──────────┼──────────┼──────────┤") fmt.Println(" │ 仇恨言论 │ ≥0.6 │ 拦截 │ P0 │") fmt.Println(" │ 自残自杀 │ ≥0.5 │ 拦截+上报 │ P0 │") fmt.Println(" │ 暴力内容 │ ≥0.7 │ 拦截 │ P1 │") fmt.Println(" │ PII泄露 │ ≥0.9 │ 拦截+重写 │ P0 │") fmt.Println(" │ 幻觉 │ ≥0.8 │ 警告 │ P2 │") fmt.Println(" │ 色情内容 │ ≥0.7 │ 拦截 │ P1 │") fmt.Println(" │ 虚假信息 │ ≥0.6 │ 警告+标注 │ P2 │") fmt.Println(" └──────────────┴──────────┴──────────┴──────────┘") } func generateID() string { b := make([]byte, 16) for i := range b { b[i] = byte(time.Now().UnixNano() % 256) } return fmt.Sprintf("%x", b) } func truncate(s string, maxLen int) string { runes := []rune(s) if len(runes) <= maxLen { return s } return string(runes[:maxLen]) + "..." }

三、输出安全流水线

模型输出 │ ▼ ┌──────────────┐ │ 第1层 │ 实时检测（同步） │ 紧急拦截 │ - 仇恨言论 │ │ - 自残自杀 │ │ - PII 泄露 │ │ - 暴力内容 └──────┬───────┘ │ 通过 ▼ ┌──────────────┐ │ 第2层 │ 深度检测（异步） │ 质量检查 │ - 幻觉检测 │ │ - 事实核查 │ │ - 逻辑一致性 └──────┬───────┘ │ 通过 ▼ ┌──────────────┐ │ 第3层 │ 内容改写 │ 安全改写 │ - PII 掩码 │ │ - 敏感词替换 │ │ - 不确定性声明 └──────┬───────┘ │ ▼ ┌──────────────┐ │ 最终输出 │ → 返回给用户 └──────────────┘

四、生产部署建议

场景	拦截阈值	推荐配置
儿童教育	0.4	严格模式，所有类别启用
医疗咨询	0.5	加强幻觉检测，需免责声明
金融服务	0.6	严格 PII 检测，完整审计
客服系统	0.7	标准配置，平衡体验和安全
创意写作	0.8	宽松模式，仅拦截严重违规
代码生成	0.9	仅拦截 PII 和安全漏洞