ai-stopping-hallucinations

防止 AI 编造内容

引导用户让 AI 的回答有事实依据。核心原则：绝不要轻信未经核实的大模型输出，始终对照来源进行验证。

何时不应使用防幻觉模式

• 创意任务（头脑风暴、小说、构思）——幻觉是功能，不是缺陷
• 简单提取，且答案直接摘自输入内容——引用会增加开销，却没有好处
• 没有来源文档，也无法添加——自洽性检查有所帮助，但基于检索的方案效果要好得多。可以先考虑 /ai-searching-docs 来添加知识库

步骤 1：了解事实依据情况

询问用户：

1. 是否有来源文档？（知识库、文档、数据库）→ 使用基于检索的回答
2. 是否属于常识？（没有文档，只有模型自身的知识）→ 使用自洽性检查
3. 幻觉的后果有多严重？（令人困扰还是会造成危险）→ 这决定了检查应当有多严格

步骤 2：强制引用来源

要求 AI 为每项声明引用来源。使用 dspy.Refine 重试生成，直到引用质量达到阈值。

import dspy
import re

lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
dspy.configure(lm=lm)

class CitedAnswer(dspy.Signature):
    """Answer the question using the provided sources. Cite every claim with [1], [2], etc."""
    context: list[str] = dspy.InputField(desc="Numbered source documents")
    question: str = dspy.InputField()
    answer: str = dspy.OutputField(desc="Answer with inline citations like [1], [2]")

def citation_reward(args: dict, pred: dspy.Prediction) -> float:
    answer = pred.answer
    context = args["context"]

    # Check citation ratio — at least half of sentences must have citations
    sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', answer) if s.strip()]
    citations_found = [bool(re.search(r"\[\d+\]", s)) for s in sentences]
    ratio = sum(citations_found) / max(len(sentences), 1)

    # Check all cited numbers exist in context
    cited_nums = set(int(n) for n in re.findall(r"\[(\d+)\]", answer))
    valid_nums = set(range(1, len(context) + 1))
    all_valid = cited_nums.issubset(valid_nums)

    if not all_valid:
        return 0.0
    return ratio  # 0.0 to 1.0

answerer = dspy.ChainOfThought(CitedAnswer)
citation_enforcer = dspy.Refine(answerer, N=3, reward_fn=citation_reward, threshold=0.5)

# Usage
result = citation_enforcer(context=context, question=question)

dspy.Refine 最多重试 N 次，并将奖励分数作为反馈传回，让模型在每次尝试中改进引用。

步骤 3：验证忠实度

生成答案后，再调用一次大模型，检查答案是否确实有来源支持。将回答模块包装在 dspy.Refine 中，以便在答案不忠实时重试。

class CheckFaithfulness(dspy.Signature):
    """Check if every claim in the answer is supported by the context."""
    context: list[str] = dspy.InputField(desc="Source documents")
    answer: str = dspy.InputField(desc="Generated answer to verify")
    is_faithful: bool = dspy.OutputField(desc="Is every claim supported by the context?")
    unsupported_claims: list[str] = dspy.OutputField(desc="Claims not found in context")

class FaithfulAnswerer(dspy.Module):
    def __init__(self):
        self.answer = dspy.ChainOfThought(CitedAnswer)
        self.verify = dspy.Predict(CheckFaithfulness)

    def forward(self, context, question):
        result = self.answer(context=context, question=question)
        check = self.verify(context=context, answer=result.answer)
        # Expose faithfulness on the prediction so reward_fn can read it
        return dspy.Prediction(
            answer=result.answer,
            is_faithful=check.is_faithful,
            unsupported_claims=check.unsupported_claims,
        )

def faithfulness_reward(args: dict, pred: dspy.Prediction) -> float:
    return 1.0 if pred.is_faithful else 0.0

faithful_responder = dspy.Refine(
    FaithfulAnswerer(), N=3, reward_fn=faithfulness_reward, threshold=1.0
)

# Usage
result = faithful_responder(context=context, question=question)

步骤 4：自检模式

生成答案后，让模型根据来源验证自己的声明。使用提供部分忠实度分数的奖励函数，适用于希望获得当前最佳答案，而非直接阻断答案的情况。

class SelfCheckedAnswerer(dspy.Module):
    def __init__(self):
        self.answer = dspy.ChainOfThought("context, question -> answer")
        self.check = dspy.ChainOfThought(CheckFaithfulness)

    def forward(self, context, question):
        result = self.answer(context=context, question=question)
        verification = self.check(context=context, answer=result.answer)
        return dspy.Prediction(
            answer=result.answer,
            is_verified=verification.is_faithful,
            unsupported=verification.unsupported_claims,
        )

def partial_faithfulness_reward(args: dict, pred: dspy.Prediction) -> float:
    """Partial credit - 1.0 if faithful, 0.5 if partially faithful, 0.0 if not."""
    if pred.is_verified:
        return 1.0
    # Give partial credit if there are few unsupported claims
    unsupported = pred.unsupported or []
    if len(unsupported) <= 1:
        return 0.5
    return 0.0

self_checked = dspy.Refine(
    SelfCheckedAnswerer(), N=3, reward_fn=partial_faithfulness_reward, threshold=0.5
)

步骤 5：交叉检查模式

多次独立生成答案，然后选择与自身最一致的答案。dspy.BestOfN 会采样 N 个候选答案，并根据奖励函数选出得分最高的一个。

class CompareAnswers(dspy.Signature):
    """Check if two independently generated answers agree on the facts."""
    answer_a: str = dspy.InputField()
    answer_b: str = dspy.InputField()
    agree: bool = dspy.OutputField(desc="Do they agree on all factual claims?")
    discrepancy: str = dspy.OutputField(desc="What they disagree on, if anything")

class GroundedAnswerer(dspy.Module):
    def __init__(self):
        self.answer = dspy.ChainOfThought("context, question -> answer")
        self.verify = dspy.Predict(CheckFaithfulness)

    def forward(self, context, question):
        result = self.answer(context=context, question=question)
        check = self.verify(context=context, answer=result.answer)
        return dspy.Prediction(
            answer=result.answer,
            is_faithful=check.is_faithful,
            unsupported_claims=check.unsupported_claims,
        )

def faithfulness_reward(args: dict, pred: dspy.Prediction) -> float:
    return 1.0 if pred.is_faithful else 0.0

# BestOfN samples N candidates and returns the one with the highest reward
cross_checked = dspy.BestOfN(
    GroundedAnswerer(), N=3, reward_fn=faithfulness_reward, threshold=0.5
)

# Usage — returns the most faithful of the 3 sampled answers
result = cross_checked(context=context, question=question)

最适合幻觉代价高的关键场景输出。该模式会调用大模型 N 次，但会选出最忠实的结果，而不是在失败时重试。

步骤 6：置信度阈值

将低置信度输出标记出来供人工审核，而不是展示给用户。

class ConfidenceGated(dspy.Signature):
    """Answer the question and rate your confidence."""
    context: list[str] = dspy.InputField()
    question: str = dspy.InputField()
    answer: str = dspy.OutputField()
    confidence: float = dspy.OutputField(desc="0.0 to 1.0, how confident are you?")
    reasoning: str = dspy.OutputField(desc="Why this confidence level?")

class GatedResponder(dspy.Module):
    def __init__(self, threshold=0.7):
        self.respond = dspy.ChainOfThought(ConfidenceGated)
        self.threshold = threshold

    def forward(self, context, question):
        result = self.respond(context=context, question=question)

        if result.confidence < self.threshold:
            return dspy.Prediction(
                answer=result.answer,
                needs_review=True,
                confidence=result.confidence,
                reason=result.reasoning,
            )

        return dspy.Prediction(
            answer=result.answer,
            needs_review=False,
            confidence=result.confidence,
        )

步骤 7：加载用于验证的来源数据

防幻觉模式需要来源文档。以下是加载常见格式的方法：

从转录文件加载

import json, re

def load_vtt(path):
    """Extract text from a VTT transcript, stripping timestamps and cues."""
    text = open(path).read()
    lines = [line.strip() for line in text.split("\n")
             if line.strip() and not line.startswith("WEBVTT")
             and not re.match(r"\d{2}:\d{2}", line)
             and not line.strip().isdigit()]
    return " ".join(lines)

def load_livekit_transcript(path):
    """Extract text from a LiveKit transcript JSON export."""
    data = json.load(open(path))
    segments = data.get("segments", data.get("results", []))
    return " ".join(seg.get("text", "") for seg in segments)

def load_recall_transcript(transcript_data):
    """Extract text from a Recall.ai transcript response."""
    return " ".join(
        entry["words"] for entry in transcript_data if entry.get("words")
    )

从 Langfuse 跟踪记录加载

from langfuse import Langfuse

def load_langfuse_generations(trace_id):
    """Load LM generations from a Langfuse trace for verification."""
    langfuse = Langfuse()
    trace = langfuse.get_trace(trace_id)
    generations = []
    for obs in trace.observations:
        if obs.type == "GENERATION" and obs.output:
            generations.append({
                "input": obs.input,
                "output": obs.output,
                "model": obs.model,
            })
    return generations

将来源文档拆分为带编号的段落

此处的大多数模式都需要 context: list[str]，即带编号的来源段落。将长文档拆分为多个块，让引用更有意义：

def chunk_document(text, max_chars=500):
    """Split a document into numbered passages for citation."""
    paragraphs = [p.strip() for p in text.split("\n\n") if p.strip()]
    chunks = []
    current = ""
    for para in paragraphs:
        if len(current) + len(para) > max_chars and current:
            chunks.append(current.strip())
            current = para
        else:
            current = current + "\n\n" + para if current else para
    if current:
        chunks.append(current.strip())
    return [f"[{i+1}] {chunk}" for i, chunk in enumerate(chunks)]

# Use with any source
transcript_text = load_vtt("meeting.vtt")
context = chunk_document(transcript_text)
result = citation_enforcer(context=context, question="What was decided about the timeline?")

步骤 8：评估防幻觉质量

你需要使用指标来判断验证是否有效。关键问题是：系统能否发现幻觉并生成忠实的答案？

忠实度指标

def faithfulness_metric(example, prediction, trace=None):
    """Score: does the answer stick to the sources?"""
    verifier = dspy.Predict(CheckFaithfulness)
    check = verifier(context=example.context, answer=prediction.answer)

    # Binary: is it faithful?
    if not check.is_faithful:
        return 0.0

    # Bonus: does it actually answer the question?
    relevance = dspy.Predict("question, answer -> is_relevant: bool")
    rel = relevance(question=example.question, answer=prediction.answer)
    return 1.0 if rel.is_relevant else 0.5

evaluator = dspy.Evaluate(devset=devset, metric=faithfulness_metric, num_threads=4)
score = evaluator(my_grounded_qa)

引用覆盖率指标

def citation_metric(example, prediction, trace=None):
    """Score citation quality: coverage + validity."""
    answer = prediction.answer
    sentences = [s.strip() for s in re.split(r'(?<=[.!?])\s+', answer) if s.strip()]
    cited = [bool(re.search(r"\[\d+\]", s)) for s in sentences]
    coverage = sum(cited) / max(len(sentences), 1)

    # Check all cited sources exist
    cited_nums = set(int(n) for n in re.findall(r"\[(\d+)\]", answer))
    valid_nums = set(range(1, len(example.context) + 1))
    all_valid = cited_nums.issubset(valid_nums)

    if not all_valid:
        return 0.0
    return coverage  # 0.0 to 1.0

优化验证流程

# Create training data: questions with source context and gold answers
trainset = [
    dspy.Example(
        context=["[1] The meeting is on March 5.", "[2] Budget is $50k."],
        question="When is the meeting?",
        answer="The meeting is on March 5 [1]."
    ).with_inputs("context", "question"),
    # ... more examples
]

# Optimize the citation enforcer
optimizer = dspy.BootstrapFewShot(metric=faithfulness_metric, max_bootstrapped_demos=4)
optimized = optimizer.compile(FaithfulAnswerer(), trainset=trainset)
optimized.save("optimized_faithful_answerer.json")

# Load later
enforcer = FaithfulAnswerer()
enforcer.load("optimized_faithful_answerer.json")

使用低成本大模型进行验证

验证步骤不需要昂贵的模型——使用较小的模型根据来源检查声明，效果良好，还能降低成本：

class CostEfficientVerifier(dspy.Module):
    def __init__(self):
        self.answer = dspy.ChainOfThought(CitedAnswer)
        self.verify = dspy.Predict(CheckFaithfulness)

        # Use a cheaper model for the verification step
        cheap_lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-haiku-4-5-20251001", etc.
        self.verify.set_lm(cheap_lm)

    def forward(self, context, question):
        result = self.answer(context=context, question=question)
        check = self.verify(context=context, answer=result.answer)
        return dspy.Prediction(
            answer=result.answer,
            is_faithful=check.is_faithful,
            unsupported_claims=check.unsupported_claims,
        )

def faithfulness_reward(args: dict, pred: dspy.Prediction) -> float:
    return 1.0 if pred.is_faithful else 0.0

cost_efficient = dspy.Refine(
    CostEfficientVerifier(), N=3, reward_fn=faithfulness_reward, threshold=1.0
)

步骤 9：批量验证

当你需要一次验证许多响应时（例如，审核转录问答系统）：

import json

def verify_batch(qa_pairs, context, output_path="verification_results.json"):
    """Verify a batch of question-answer pairs against source context."""
    verifier = dspy.Predict(CheckFaithfulness)
    results = []

    for qa in qa_pairs:
        check = verifier(context=context, answer=qa["answer"])
        results.append({
            "question": qa["question"],
            "answer": qa["answer"],
            "is_faithful": check.is_faithful,
            "unsupported_claims": check.unsupported_claims,
        })

    # Summary
    faithful_count = sum(1 for r in results if r["is_faithful"])
    print(f"Faithful: {faithful_count}/{len(results)} "
          f"({faithful_count/len(results):.0%})")

    with open(output_path, "w") as f:
        json.dump(results, f, indent=2)

    return results

Refine 的工作原理

使用 dspy.Refine 时：

1. 封装的模块会生成一个候选预测
2. 奖励函数会对该预测进行评分（0.0 到 1.0）
3. 如果分数达到阈值，Refine 会立即返回该预测
4. 如果没有达到阈值，则会使用分数作为反馈信号，最多重试 N 次
5. N 次尝试后，返回迄今为止得分最高的候选结果

dspy.BestOfN 与此类似，但会始终采样全部 N 个候选结果，并返回最佳结果——当你希望采样过程一致，而不是成功后提前退出时，这种方式很有用。

出色的奖励函数能让 Refine 表现更好——与可衡量属性（引用比例、忠实度检查）挂钩的具体评分，效果优于模糊的二元通过／失败判断。

选择合适的模式

模式	成本	延迟	最适用场景
引用强制（Refine）	1-3 次 LM 调用	低至中	有编号来源时
忠实度验证（Refine）	2-6 次 LM 调用	中	RAG 系统、文档问答
自检（Refine）	2-6 次 LM 调用	中	通用事实核查
交叉检查（BestOfN）	N 次 LM 调用	高	高风险、关键输出
置信度门控	1 次 LM 调用	低	人类参与的系统
廉价验证器（Refine）	1 次昂贵调用 + 1-3 次廉价调用	低至中	对成本敏感的生产环境

注意事项

• 按 . 分割句子会在缩写和小数处出错。 简单的 .split(".") 会错误处理 "Dr. Smith"、"$50.00" 和 URL，导致未引用句子数量虚高，并产生不正确的奖励分数。应使用 re.split(r'(?<=[.!?])\s+', text) 来分割句子。
• 引用正则表达式无法匹配组合引用。 模型写出 [1, 2] 或 [1-3] 时，标准的 \[(\d+)\] 模式只能匹配 [1] 格式。应扩展模式以处理范围和逗号分隔的列表，或在签名说明中统一引用格式。
• 忠实度验证器默认使用同一个昂贵的 LM。 验证属于分类任务，较便宜的模型也能很好地处理。务必在验证器预测器上调用 .set_lm()，以使用较小的模型。这样通常可将验证成本降低 5-10 倍，且准确率损失极小。
• dspy.Retrieve 需要配置检索器。 dspy.Retrieve 需要通过 dspy.configure(rm=...) 配置检索模型。未配置时，调用会在运行时失败。请配置检索器，或直接将上下文作为函数参数传入。
• 奖励函数必须妥善处理缺失字段。 如果模块抛出异常，或返回的预测缺少预期字段，奖励函数就会报错。添加 getattr(pred, "is_faithful", False) 这类防御式访问，避免 Refine 的重试循环崩溃。
• threshold=1.0 的 Refine 必须取得满分才能提前返回。 对于部分得分奖励函数（0.0 到 1.0），请将阈值设为低于 1.0，或改用 BestOfN；否则 Refine 总会执行完全部 N 次尝试。

交叉参考

安装任意技能：npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• 检索增强生成（RAG）设置——参见 /ai-searching-docs
• 通用输出验证（格式、安全性、质量）——参见 /ai-checking-outputs
• 强制执行业务规则和内容政策——参见 /ai-following-rules
• 使用奖励函数进行迭代优化——参见 /dspy-refine
• 采样并选择最佳输出——参见 /dspy-best-of-n
• 检索模型配置与搜索——参见 /dspy-retrieval
• 如果你还没有 /ai-do，请安装它——它会将任意 AI 问题路由到合适的技能，是最高效的工作方式：npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

更多资源

• 完整的示例请参见 examples.md
• API 签名和参数表请参见 reference.md