Geminiで作る出典付きRAG ― ハルシネーションを止める引用生成と検証パイプラインの設計

# retriever.py
from dataclasses import dataclass
from typing import List
 
@dataclass
class Source:
    id: str          # 例: "doc_2026_legal_001#sec3"
    title: str
    content: str
    url: str
    metadata: dict
 
def retrieve(query: str, top_k: int = 5) -> List[Source]:
    """Vertex AI Search や pgvector などから関連ドキュメントを取得します。
 
    重要: id はドキュメント内の節（section）まで一意に識別できる粒度にしてください。
    """
    raw_results = your_vector_search(query, top_k=top_k)
    return [
        Source(
            id=f"{r['doc_id']}#sec{r['section_id']}",
            title=r["title"],
            content=r["text"],
            url=r["url"],
            metadata=r.get("metadata", {}),
        )
        for r in raw_results
    ]

# generator.py
from google import genai
from google.genai import types
from pydantic import BaseModel, Field
from typing import List
 
class Citation(BaseModel):
    source_id: str = Field(description="参照したコンテキスト内のID")
    quoted_span: str = Field(description="参照箇所からの引用テキスト（最大80文字）")
 
class Claim(BaseModel):
    text: str = Field(description="主張または事実の本体")
    citations: List[Citation] = Field(description="この主張を裏付ける出典のリスト")
 
class GroundedAnswer(BaseModel):
    claims: List[Claim]
    summary: str = Field(description="回答全体の要約（出典に基づくもののみ）")
 
SYSTEM_PROMPT = """\
あなたは情報検索アシスタントです。提供されたコンテキストに含まれる情報のみを使って回答してください。
 
ルール:
- 全ての主張(claim)に最低1つのcitationを付けてください
- citationのsource_idは、提供されたコンテキストのIDから1字も変えずにコピーしてください
- quoted_spanはコンテキストの原文から80文字以内で抜粋してください（要約はしない）
- コンテキストにない情報を推測で補わないでください
- コンテキストに答えがない場合はclaims=[]とし、summaryで「資料からは確認できません」と答えてください
"""
 
def build_context(sources: List[Source]) -> str:
    """source_id を [SRC:id] の形でコンテキストに埋め込みます。"""
    blocks = []
    for s in sources:
        blocks.append(f"[SRC:{s.id}] {s.title}\n{s.content}\n")
    return "\n---\n".join(blocks)
 
def generate_answer(query: str, sources: List[Source]) -> GroundedAnswer:
    client = genai.Client()
    context = build_context(sources)
    response = client.models.generate_content(
        model="gemini-2.5-pro",
        contents=[
            types.Content(
                role="user",
                parts=[types.Part(text=f"質問: {query}\n\nコンテキスト:\n{context}")],
            )
        ],
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_PROMPT,
            response_mime_type="application/json",
            response_schema=GroundedAnswer,
            temperature=0.0,  # 出典生成では創造性は不要
        ),
    )
    return GroundedAnswer.model_validate_json(response.text)

{
  "claims": [
    {
      "text": "本契約は2026年4月1日から効力を発生する。",
      "citations": [
        {
          "source_id": "doc_2026_legal_001#sec3",
          "quoted_span": "第3条 本契約は2026年4月1日から効力を発生し..."
        }
      ]
    }
  ],
  "summary": "本契約の発効日と主要条項に関する確認結果です。"
}

# validator.py
from difflib import SequenceMatcher
 
def verify_citations(answer: GroundedAnswer, sources: List[Source]):
    """各citationの妥当性を3段階で検証します。"""
    source_by_id = {s.id: s for s in sources}
    issues = []
 
    for i, claim in enumerate(answer.claims):
        for j, c in enumerate(claim.citations):
            # 検証1: source_id が実在するか
            if c.source_id not in source_by_id:
                issues.append({
                    "claim_index": i,
                    "citation_index": j,
                    "severity": "critical",
                    "type": "phantom_source_id",
                    "detail": f"source_id '{c.source_id}' はコンテキストに存在しません",
                })
                continue
 
            src = source_by_id[c.source_id]
 
            # 検証2: quoted_span が原文に含まれるか
            if c.quoted_span and c.quoted_span not in src.content:
                # 完全一致しない場合、ウィンドウをずらしながら類似度を測る
                best = 0.0
                step = max(1, len(c.quoted_span) // 4)
                for k in range(0, max(1, len(src.content) - len(c.quoted_span)), step):
                    window = src.content[k:k + len(c.quoted_span)]
                    ratio = SequenceMatcher(None, c.quoted_span, window).ratio()
                    if ratio > best:
                        best = ratio
                        if best >= 0.95:
                            break
                if best < 0.8:
                    issues.append({
                        "claim_index": i,
                        "citation_index": j,
                        "severity": "warning",
                        "type": "quote_drift",
                        "detail": f"引用テキストが原文と一致しません（類似度: {best:.2f}）",
                    })
 
            # 検証3: claim のキーワードが source の内容に含まれるか
            keywords = [w for w in claim.text.split() if len(w) > 1][:5]
            if keywords and not any(k in src.content for k in keywords):
                issues.append({
                    "claim_index": i,
                    "citation_index": j,
                    "severity": "warning",
                    "type": "weak_grounding",
                    "detail": "主張のキーワードが出典内容に見当たりません",
                })
 
    return issues

// CitedAnswer.tsx
import React from "react";
 
type Citation = { source_id: string; quoted_span: string };
type Claim = { text: string; citations: Citation[] };
type Source = { id: string; title: string; url: string };
 
export function CitedAnswer({
  claims,
  sources,
}: {
  claims: Claim[];
  sources: Source[];
}) {
  const sourceMap = new Map(sources.map((s) => [s.id, s]));
  const refIndex = new Map(sources.map((s, i) => [s.id, i + 1]));
 
  return (
    <article className="cited-answer">
      {claims.map((claim, i) => (
        <p key={i}>
          {claim.text}
          {dedupeBySource(claim.citations).map((c, j) => {
            const src = sourceMap.get(c.source_id);
            if (!src) return null;
            return (
              <sup key={j} className="citation-marker">
                <a
                  href={src.url}
                  title={c.quoted_span}
                  target="_blank"
                  rel="noopener noreferrer"
                >
                  [{refIndex.get(c.source_id)}]
                </a>
              </sup>
            );
          })}
        </p>
      ))}
 
      <footer className="citations-footer">
        <h3>参考文献</h3>
        <ol>
          {sources.map((s) => (
            <li key={s.id} id={`ref-${s.id}`}>
              <a href={s.url} target="_blank" rel="noopener noreferrer">
                {s.title}
              </a>
            </li>
          ))}
        </ol>
      </footer>
    </article>
  );
}
 
function dedupeBySource(citations: Citation[]): Citation[] {
  const seen = new Set<string>();
  return citations.filter((c) => {
    if (seen.has(c.source_id)) return false;
    seen.add(c.source_id);
    return true;
  });
}

def generate_with_validation(
    query: str,
    sources: List[Source],
    max_retry: int = 2,
) -> tuple[GroundedAnswer, list]:
    last_issues = []
    for attempt in range(max_retry + 1):
        answer = generate_answer(query, sources)
        issues = verify_citations(answer, sources)
        critical = [i for i in issues if i["severity"] == "critical"]
        if not critical:
            return answer, issues
        last_issues = issues
        # critical issue をプロンプトに追記して再試行する
        # （実装は次の落とし穴2と合わせて検討）
    raise ValueError(f"出典検証に失敗しました: {last_issues}")

def grounding_score(answer: GroundedAnswer, sources: List[Source]) -> float:
    """0.0(全くgroundedしていない) 〜 1.0(完璧にgrounded)"""
    if not answer.claims:
        return 1.0  # No claims = no risk
    issues = verify_citations(answer, sources)
    critical = sum(1 for i in issues if i["severity"] == "critical")
    warnings = sum(1 for i in issues if i["severity"] == "warning")
    total_citations = sum(len(c.citations) for c in answer.claims)
    if total_citations == 0:
        return 0.0
    penalty = (critical + warnings * 0.5) / total_citations
    return max(0.0, 1.0 - penalty)

# eval_harness.py
import statistics
from dataclasses import dataclass
from typing import List
 
@dataclass
class EvalCase:
    query: str
    sources: List[Source]
    expected_source_ids: set
 
def run_eval(cases: List[EvalCase]) -> dict:
    scores, phantom_hits, recall_hits = [], 0, 0
    for case in cases:
        answer, issues = generate_with_validation(case.query, case.sources)
        scores.append(grounding_score(answer, case.sources))
        # 捏造ID（critical）が1件でもあればカウント
        if any(i["type"] == "phantom_source_id" for i in issues):
            phantom_hits += 1
        # 期待した出典を最低1つ引けているか（出典リコール）
        cited = {c.source_id for cl in answer.claims for c in cl.citations}
        if case.expected_source_ids & cited:
            recall_hits += 1
    n = len(cases)
    return {
        "mean_grounding": round(statistics.mean(scores), 3),
        "phantom_rate": round(phantom_hits / n, 3),
        "source_recall": round(recall_hits / n, 3),
    }