基于可验证生成式AI的电商推荐幻觉拦截系统（DLOS）：设计、实现与评估

技术支持：拓世网络技术开发部

摘要

 

随着大语言模型在电商推荐系统中的广泛应用，模型生成虚假、错误或不合逻辑内容（即“幻觉”）的问题日益突出，严重影响用户体验与平台信任度。本文提出并实现了一套完整的可验证生成式AI逻辑控制系统——DLOS（Verifiable Generative AI Logic Operating System），聚焦于电商推荐场景下的幻觉检测与拦截。该系统集成了用户意图建模（TSPR）、LLM生成模块、多维度验证器（WebCheck、LogicCheck、TSPR一致性检查）以及决策引擎，形成从用户查询到安全输出的完整闭环。本文详细描述了系统的架构设计、各模块的实现细节、前端展示方案、评估指标以及商业化路径。通过在模拟电商场景中的实验验证，DLOS能够实现超过60%的幻觉拦截率，显著减少错误输出，提升推荐系统的可信度和转化稳定性。本文为生成式AI的安全应用提供了一个可落地、可验证、可商业化的完整解决方案。

 

关键词：大语言模型；幻觉检测；电商推荐；可验证AI；决策系统

 

---

 

1. 引言

 

1.1 背景与问题

 

大语言模型（Large Language Models, LLMs）的出现极大地推动了自然语言处理领域的发展。在电商推荐系统中，LLM被广泛用于生成个性化的商品推荐文案、回答用户咨询、提供购买建议等场景。然而，LLM的本质决定了它可能生成与事实不符、逻辑错误或过度承诺的内容——这种现象被称为“幻觉”（Hallucination）。

 

例如，当用户询问“我需要一款适合敏感牙齿的廉价电动牙刷”时，LLM可能错误地生成“这款牙刷已获FDA批准并可完全治愈牙龈疾病”。这种虚假陈述不仅误导消费者，还可能使平台面临法律风险和声誉损失。

 

1.2 现有方案的局限性

 

目前针对LLM幻觉问题的解决方案主要分为三类：

 

1. 提示工程：通过精心设计的提示词引导模型生成更可靠的内容，但无法从根本上消除幻觉。

2. 检索增强生成（RAG）：从外部知识库检索相关信息辅助生成，但受限于知识库的覆盖面和时效性。

3. 事后验证：生成后对内容进行事实核查，但现有验证系统往往只关注单一维度（如事实性），缺乏对逻辑一致性和用户意图匹配度的综合评估。

 

1.3 本文贡献

 

本文提出并实现了一套完整的可验证生成式AI逻辑控制系统（DLOS），主要贡献包括：

 

1. 设计了覆盖事实验证、逻辑验证和用户意图一致性检查的多维度验证架构；

2. 实现了完整的闭环系统，包含TSPR意图建模、LLM生成、多模块验证、决策引擎和前端展示；

3. 通过具体电商场景的案例验证了系统的有效性；

4. 提出了清晰的商业化路径和评估指标体系。

 

---

 

2. 系统架构设计

 

2.1 整体架构

 

DLOS采用流水线架构，从用户输入到最终输出经过五个核心阶段：

 

```

用户查询（User Query）

    ↓

TSPR意图建模（TSPR Intent Modeling）

    ↓

LLM生成（LLM Generation）

    ↓

DLOS验证器（DLOS Validator）

    ├── WebCheck（事实验证）

    ├── LogicCheck（逻辑验证）

    └── TSPR一致性检查（TSPR Consistency Check）

    ↓

决策引擎（Decision Engine）

    ├── PASS → 输出

    ├── REWRITE → 重写

    └── BLOCK → 拦截

    ↓

前端展示（Frontend Display）

```

 

2.2 各模块功能定义

 

2.2.1 TSPR意图建模模块

 

TSPR（Temporal-Semantic-Personalized-Ranking）模块负责从用户查询中提取四个维度的信息：

 

维度 含义 在电商场景中的示例

时间维度（Temporal） 用户的时间约束和购买紧迫性 “立即需要”、“一周内送达”

语义维度（Semantic） 用户需求的语义特征 商品类别、属性偏好、价格区间

个性化维度（Personalized） 用户的历史偏好和行为模式 品牌忠诚度、价格敏感度、品质要求

排序维度（Ranking） 用户对各属性的优先级排序 “价格最重要”、“效果 > 品牌 > 价格”

 

2.2.2 LLM生成模块

 

该模块接收TSPR的输出和原始用户查询，调用大语言模型生成推荐文案。在本文的实现中，我们使用模拟LLM进行演示，实际部署时可替换为任何商用或开源LLM（GPT-4、Claude、LLaMA等）。

 

2.2.3 DLOS验证器

 

验证器是系统的核心，包含三个子模块：

 

WebCheck（事实验证）：

 

· 功能：验证LLM输出中的事实性断言是否真实

· 方法：提取断言 → 构建查询 → 检索权威知识源（如产品官网、FDA数据库、权威评测） → 验证匹配度

· 输出：事实置信度分数（FCS），范围为[0,1]

 

LogicCheck（逻辑验证）：

 

· 功能：检查LLM输出的逻辑一致性和合理性

· 方法：检测过度承诺（如“完全治愈”）、矛盾陈述（如“最便宜但也是最贵的”）、不合理因果关系

· 输出：逻辑合理性分数（RCS），范围为[0,1]

 

TSPR一致性检查：

 

· 功能：验证LLM输出是否与TSPR提取的用户意图一致

· 方法：计算生成文案与TSPR各维度的语义相似度，综合评估匹配程度

· 输出：语义对齐分数（SAS），范围为[0,1]

 

2.2.4 决策引擎

 

决策引擎根据三个验证模块的输出分数，综合判断最终决策：

 

决策类型 触发条件 处理方式

PASS FCS > 0.7 AND RCS > 0.7 AND SAS > 0.6 直接输出原始生成内容

REWRITE (FCS > 0.4 AND FCS ≤ 0.7) OR (RCS > 0.4 AND RCS ≤ 0.7) 触发重写机制，修正问题部分

BLOCK FCS ≤ 0.4 OR RCS ≤ 0.4 OR SAS ≤ 0.4 完全拦截，返回安全兜底响应

 

同时系统计算综合可信度评分（HRI - Holistic Reliability Index）：

 

```

HRI = 0.4 × FCS + 0.3 × RCS + 0.3 × SAS

```

 

2.3 数据流设计

 

系统内部数据流包含以下关键数据结构：

 

```python

# 用户查询结构

UserQuery = {

    "raw_text": str,

    "timestamp": datetime,

    "user_id": Optional[str]

}

 

# TSPR输出结构

TSPROutput = {

    "temporal": {"urgency": float, "deadline": Optional[str]},

    "semantic": {"category": str, "attributes": dict, "price_range": tuple},

    "personalized": {"brand_preferences": list, "price_sensitivity": float},

    "ranking": List[tuple] # [(attribute, priority_score)]

}

 

# LLM输出结构

LLMOutput = {

    "raw_text": str,

    "generation_metadata": {"model": str, "temperature": float}

}

 

# 验证结果结构

ValidationResult = {

    "webcheck": {"passed": bool, "fcs": float, "evidence": List[dict]},

    "logiccheck": {"passed": bool, "rcs": float, "issues": List[str]},

    "tspr_consistency": {"passed": bool, "sas": float, "mismatches": List[str]},

    "hri": float

}

 

# 最终输出结构

FinalOutput = {

    "decision": str, # "PASS", "REWRITE", "BLOCK"

    "original_content": Optional[str],

    "final_content": str,

    "validation_result": ValidationResult,

    "timestamp": datetime

}

```

 

---

 

3. 核心模块实现

 

3.1 TSPR意图建模实现

 

TSPR模块的实现采用规则基与轻量级NLP模型相结合的方式。

 

```python

# tspr_engine.py

import re

from typing import Dict, List, Tuple, Optional

from datetime import datetime

 

class TSPREngine:

    """TSPR意图建模引擎"""

    

    def __init__(self):

        # 定义关键模式

        self.patterns = {

            "price_range": r"(\$?\d+)\s*-\s*(\$?\d+)", # 价格区间模式

            "urgency_keywords": ["immediate", "urgent", "asap", "right now", "quick"],

            "category_mapping": {

                "toothbrush": "Oral Care",

                "electric toothbrush": "Oral Care > Electric Toothbrushes",

                "sensitive teeth": "Oral Care > Sensitive Products"

            }

        }

        

        # 预定义属性权重

        self.attribute_weights = {

            "price": 0.4,

            "effectiveness": 0.3,

            "safety": 0.2,

            "brand": 0.1

        }

    

    def extract_temporal(self, query: str) -> Dict:

        """提取时间维度"""

        urgency_score = 0.0

        deadline = None

        

        query_lower = query.lower()

        for keyword in self.patterns["urgency_keywords"]:

            if keyword in query_lower:

                urgency_score += 0.3

        

        # 检测具体时间要求

        date_pattern = r"(\d+)\s*(day|week|month)"

        date_match = re.search(date_pattern, query_lower)

        if date_match:

            deadline = f"{date_match.group(1)} {date_match.group(2)}"

            urgency_score = min(1.0, urgency_score + 0.4)

        

        return {

            "urgency": min(1.0, urgency_score),

            "deadline": deadline

        }

    

    def extract_semantic(self, query: str) -> Dict:

        """提取语义维度"""

        query_lower = query.lower()

        

        # 识别类别

        category = "General"

        for key, value in self.patterns["category_mapping"].items():

            if key in query_lower:

                category = value

                break

        

        # 提取属性

        attributes = {}

        price_match = re.search(self.patterns["price_range"], query_lower)

        if price_match:

            attributes["price_range"] = (price_match.group(1), price_match.group(2))

        

        if "cheap" in query_lower or "budget" in query_lower:

            attributes["price_sensitivity"] = "high"

        elif "premium" in query_lower or "luxury" in query_lower:

            attributes["price_sensitivity"] = "low"

        

        if "sensitive" in query_lower:

            attributes["sensitivity_requirement"] = "high"

        

        # 确定价格区间

        price_range = (0, 100)

        if attributes.get("price_sensitivity") == "high":

            price_range = (0, 50)

        elif attributes.get("price_sensitivity") == "low":

            price_range = (100, 500)

        

        return {

            "category": category,

            "attributes": attributes,

            "price_range": price_range

        }

    

    def extract_personalized(self, query: str, user_id: Optional[str] = None) -> Dict:

        """提取个性化维度（模拟版本，实际应接入用户数据库）"""

        # 模拟用户偏好

        simulated_preferences = {

            "brand_preferences": ["Oral-B", "Philips", "Colgate"],

            "price_sensitivity": 0.8, # 0-1, 越高越敏感

            "quality_preference": 0.6,

            "previous_purchases": ["Oral-B Pro 1000", "Sonicare 4100"]

        }

        

        # 从查询中推断偏好调整

        query_lower = query.lower()

        if "cheap" in query_lower:

            simulated_preferences["price_sensitivity"] = 0.9

        

        return simulated_preferences

    

    def compute_ranking(self, semantic: Dict, personalized: Dict) -> List[Tuple[str, float]]:

        """计算属性排序"""

        ranking = []

        

        # 根据语义和个性化计算各属性的优先级

        if semantic["attributes"].get("price_sensitivity") == "high":

            ranking.append(("price", 0.9))

        else:

            ranking.append(("price", personalized.get("price_sensitivity", 0.5)))

        

        if semantic["attributes"].get("sensitivity_requirement") == "high":

            ranking.append(("effectiveness", 0.8))

            ranking.append(("safety", 0.85))

        else:

            ranking.append(("effectiveness", 0.7))

            ranking.append(("safety", 0.6))

        

        ranking.append(("brand", 0.4))

        

        # 按优先级排序

        ranking.sort(key=lambda x: x[1], reverse=True)

        

        return ranking

    

    def process(self, query: str, user_id: Optional[str] = None) -> Dict:

        """执行完整的TSPR处理流程"""

        temporal = self.extract_temporal(query)

        semantic = self.extract_semantic(query)

        personalized = self.extract_personalized(query, user_id)

        ranking = self.compute_ranking(semantic, personalized)

        

        return {

            "temporal": temporal,

            "semantic": semantic,

            "personalized": personalized,

            "ranking": ranking

        }

```

 

3.2 LLM生成模块实现

 

```python

# llm_engine.py

import random

from typing import Dict, Optional

from datetime import datetime

 

class LLMEngine:

    """LLM生成引擎（支持模拟模式和真实API模式）"""

    

    def __init__(self, mode: str = "simulate", api_key: Optional[str] = None):

        self.mode = mode

        self.api_key = api_key

        

        # 预定义响应模板

        self.responses = {

            "safe": "Based on your request for a cheap electric toothbrush for sensitive teeth, I recommend the Oral-B Pro 500. It features sensitive mode, soft bristles, and costs only $29.99. Users report 40% less gum irritation within 2 weeks.",

            "hallucinated": "This toothbrush is FDA approved and cures gum disease completely. It uses quantum bristle technology that repairs 100% of enamel damage within days.",

            "partially_false": "This toothbrush has been clinically proven to eliminate 100% of plaque and is recommended by 99% of dentists worldwide. It comes with a lifetime warranty that covers everything including battery degradation."

        }

    

    def generate_response(self, user_input: str, use_hallucination: bool = True) -> str:

        """生成推荐响应"""

        if self.mode == "simulate":

            # 模拟模式：根据参数返回不同质量的响应

            if use_hallucination:

                # 根据输入内容选择合适的幻觉响应

                if "cheap" in user_input.lower() and "sensitive" in user_input.lower():

                    return self.responses["hallucinated"]

                else:

                    return random.choice([self.responses["hallucinated"], self.responses["partially_false"]])

            else:

                return self.responses["safe"]

        else:

            # 真实API模式（示例使用OpenAI）

            return self._call_real_llm(user_input)

    

    def _call_real_llm(self, user_input: str) -> str:

        """调用真实LLM API"""

        # 这里需要导入openai库并配置API密钥

        # 以下为示例代码，实际使用时需要取消注释并配置

        """

        import openai

        openai.api_key = self.api_key

        response = openai.ChatCompletion.create(

            model="gpt-3.5-turbo",

            messages=[

                {"role": "system", "content": "You are an e-commerce recommendation assistant."},

                {"role": "user", "content": f"Recommend a product: {user_input}"}

            ],

            temperature=0.7

        )

        return response.choices[0].message.content

        """

        # 临时返回

        return self.responses["safe"]

 

# 保持向后兼容的函数接口

def generate_response(user_input: str, use_hallucination: bool = True) -> str:

    engine = LLMEngine(mode="simulate")

    return engine.generate_response(user_input, use_hallucination)

```

 

3.3 验证器实现

 

验证器是DLOS系统的核心，实现三个独立的验证模块。

 

```python

# validator.py

import re

from typing import Dict, List, Tuple, Optional

from tspr_engine import TSPREngine

 

class Validator:

    """DLOS验证器，包含三个核心验证模块"""

    

    def __init__(self):

        self.tspr_engine = TSPREngine()

        

        # 定义过度承诺关键词模式

        self.overclaim_patterns = [

            r"cure.*completely",

            r"eliminate.*100%",

            r"guarantee.*perfect",

            r"repair.*100%",

            r"never.*fail",

            r"no side effect",

            r"miracle",

            r"magical"

        ]

        

        # 定义权威知识源（模拟）

        self.knowledge_base = {

            "fda_approved_products": ["Oral-B Pro 500", "Sonicare 4100", "Colgate Omron"],

            "dental_facts": {

                "gum_disease_treatment": "Gum disease requires professional dental treatment and cannot be cured by any toothbrush alone.",

                "enamel_repair": "Enamel cannot be naturally regenerated or repaired once lost."

            }

        }

    

    # ==================== WebCheck 事实验证 ====================

    

    def webcheck_verify(self, llm_output: str) -> Dict:

        """执行事实验证"""

        fcs = 1.0 # 初始分数

        evidence = []

        failed_claims = []

        

        # 提取关键断言

        claims = self._extract_claims(llm_output)

        

        for claim in claims:

            claim_result = self._verify_claim(claim)

            evidence.append(claim_result)

            

            if not claim_result["verified"]:

                failed_claims.append(claim)

                fcs -= 0.2 # 每个未经验证的断言扣0.2

        

        # 确保分数在[0,1]范围内

        fcs = max(0.0, min(1.0, fcs))

        

        return {

            "passed": fcs > 0.6,

            "fcs": fcs,

            "evidence": evidence,

            "failed_claims": failed_claims

        }

    

    def _extract_claims(self, text: str) -> List[str]:

        """提取文本中的事实性断言"""

        claims = []

        

        # 使用正则提取断言模式

        claim_patterns = [

            r"([A-Z][^.!?]+(?:is|are|can|will|has|have)[^.!?]+[.!?])",

            r"([^.!?]+(?:approved|certified|proven|shown|demonstrated)[^.!?]+[.!?])"

        ]

        

        for pattern in claim_patterns:

            matches = re.findall(pattern, text, re.IGNORECASE)

            claims.extend(matches)

        

        # 去重

        return list(set(claims))

    

    def _verify_claim(self, claim: str) -> Dict:

        """验证单个断言"""

        claim_lower = claim.lower()

        

        # 检查FDA批准断言

        if "fda approved" in claim_lower:

            # 检查是否有具体产品

            product_match = re.search(r"([A-Z][a-z]+ [A-Z][a-z]+ \d+)", claim)

            if product_match:

                product = product_match.group(1)

                if product in self.knowledge_base["fda_approved_products"]:

                    return {"claim": claim, "verified": True, "source": "fda_database"}

                else:

                    return {"claim": claim, "verified": False, "source": None, "reason": "Product not in FDA database"}

            else:

                return {"claim": claim, "verified": False, "source": None, "reason": "No specific product mentioned"}

        

        # 检查治愈断言

        if "cure" in claim_lower and "gum disease" in claim_lower:

            return {

                "claim": claim, 

                "verified": False, 

                "source": None, 

                "reason": self.knowledge_base["dental_facts"]["gum_disease_treatment"]

            }

        

        # 检查修复断言

        if "repair" in claim_lower and "enamel" in claim_lower:

            return {

                "claim": claim,

                "verified": False,

                "source": None,

                "reason": self.knowledge_base["dental_facts"]["enamel_repair"]

            }

        

        # 默认：无法验证的断言标记为不确定

        return {"claim": claim, "verified": None, "source": None, "reason": "Unable to verify"}

    

    # ==================== LogicCheck 逻辑验证 ====================

    

    def logiccheck_verify(self, llm_output: str) -> Dict:

        """执行逻辑验证"""

        rcs = 1.0

        issues = []

        

        # 检测过度承诺

        overclaim_matches = self._detect_overclaims(llm_output)

        if overclaim_matches:

            rcs -= len(overclaim_matches) * 0.2

            issues.extend([f"Overclaim detected: {match}" for match in overclaim_matches])

        

        # 检测矛盾陈述

        contradictions = self._detect_contradictions(llm_output)

        if contradictions:

            rcs -= len(contradictions) * 0.3

            issues.extend(contradictions)

        

        # 检测不合理因果

        causal_issues = self._detect_faulty_causality(llm_output)

        if causal_issues:

            rcs -= len(causal_issues) * 0.15

            issues.extend(causal_issues)

        

        rcs = max(0.0, min(1.0, rcs))

        

        return {

            "passed": rcs > 0.6,

            "rcs": rcs,

            "issues": issues

        }

    

    def _detect_overclaims(self, text: str) -> List[str]:

        """检测过度承诺"""

        matches = []

        for pattern in self.overclaim_patterns:

            found = re.findall(pattern, text.lower())

            matches.extend(found)

        return matches

    

    def _detect_contradictions(self, text: str) -> List[str]:

        """检测矛盾陈述"""

        contradictions = []

        

        # 价格矛盾

        price_patterns = [

            (r"cheap|budget|inexpensive|low[-\s]?cost", r"premium|luxury|expensive|high[-\s]?end")

        ]

        

        for low_pattern, high_pattern in price_patterns:

            if re.search(low_pattern, text.lower()) and re.search(high_pattern, text.lower()):

                contradictions.append("Price contradiction: product described as both cheap and premium")

        

        # 功效矛盾

        if "cure" in text.lower() and "may help" in text.lower():

            contradictions.append("Efficacy contradiction: claims both cure and may help")

        

        return contradictions

    

    def _detect_faulty_causality(self, text: str) -> List[str]:

        """检测错误的因果推理"""

        issues = []

        

        # 检查绝对因果

        absolute_causal_patterns = [

            r"if you use.*then you will",

            r"using.*guarantees",

            r"leads to.*always"

        ]

        

        for pattern in absolute_causal_patterns:

            if re.search(pattern, text.lower()):

                issues.append("Faulty causality: absolute causal claim without evidence")

        

        return issues

    

    # ==================== TSPR一致性检查 ====================

    

    def tspr_consistency_check(self, llm_output: str, tspr_result: Dict) -> Dict:

        """检查LLM输出与TSPR意图的一致性"""

        sas = 1.0

        mismatches = []

        

        # 检查价格一致性

        price_range = tspr_result["semantic"]["price_range"]

        price_sensitivity = tspr_result["semantic"]["attributes"].get("price_sensitivity")

        

        price_indicators = self._extract_price_indicators(llm_output.lower())

        

        if price_sensitivity == "high" and any(word in price_indicators["premium_indicators"] for word in price_indicators["found"]):

            sas -= 0.3

            mismatches.append("Price mismatch: user requested cheap product but response suggests premium")

        

        if price_sensitivity == "low" and any(word in price_indicators["budget_indicators"] for word in price_indicators["found"]):

            sas -= 0.2

            mismatches.append("Price mismatch: user requested premium product but response suggests budget")

        

        # 检查敏感牙齿需求一致性

        sensitivity_required = tspr_result["semantic"]["attributes"].get("sensitivity_requirement") == "high"

        if sensitivity_required:

            sensitivity_keywords = ["gentle", "soft", "sensitive", "irritation", "gentle on gums"]

            has_sensitivity_content = any(keyword in llm_output.lower() for keyword in sensitivity_keywords)

            

            if not has_sensitivity_content:

                sas -= 0.35

                mismatches.append("Content mismatch: user requested sensitive teeth product but response lacks relevant features")

        

        # 检查类别一致性

        expected_category = tspr_result["semantic"]["category"]

        if "Oral Care" in expected_category:

            category_keywords = ["toothbrush", "brush", "dental", "oral"]

            has_category_content = any(keyword in llm_output.lower() for keyword in category_keywords)

            if not has_category_content:

                sas -= 0.2

                mismatches.append("Category mismatch: response not focused on oral care products")

        

        sas = max(0.0, min(1.0, sas))

        

        return {

            "passed": sas > 0.5,

            "sas": sas,

            "mismatches": mi