ai-stopping-hallucinations

AI にでたらめを言わせない

ユーザーが AI の回答を事実に基づいたものにできるよう案内します。基本原則は、LM の出力をそのまま信用せず、必ず情報源と照合することです。

ハルシネーション対策のパターンを使わない場合

• 創造的なタスク（ブレインストーミング、フィクション、アイデア出し）— ハルシネーションは不具合ではなく、機能です
• 単純な抽出で、回答が入力からの直接的なコピーである場合 — 引用の手間が増える一方で、メリットはありません
• 情報源となるドキュメントがなく、追加もできない場合 — 自己一貫性チェックは役立ちますが、検索で根拠を得る方法のほうがはるかに効果的です。まず /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：忠実性の検証

回答の生成後、2 回目の LM 呼び出しを使って、その回答が実際に情報源によって裏付けられているかを確認します。回答生成器を 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 回の LM 呼び出しを使用しますが、失敗時に再試行するのではなく、最も忠実な結果を選びます。

ステップ 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")

検証に安価な LM を使用する

検証のステップに高価なモデルは必要ありません。情報源と照らし合わせて主張を確認する小型モデルでも十分に機能し、コストを削減できます。

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：一括検証

多数の応答を一度に検証する必要がある場合（例：トランスクリプトの Q&A システムの監査）：

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 システム、ドキュメントに関する Q&A
セルフチェック（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 分の 1 に削減できます。
• dspy.Retrieve には設定済みのリトリーバーが必要です。 dspy.Retrieve では、dspy.configure(rm=...) を使って検索モデルを設定する必要があります。設定しないと、実行時に呼び出しが失敗します。リトリーバーを設定するか、関数のパラメーターとしてコンテキストを直接渡してください。
• 報酬関数では、フィールドが欠落している場合にも適切に対処する必要があります。 モジュールが例外を発生させたり、予測に想定されたフィールドが含まれていなかったりすると、報酬関数でエラーが発生します。Refine の再試行ループがクラッシュしないよう、getattr(pred, "is_faithful", False) のような防御的なアクセスを追加してください。
• 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 を参照してください