Gemini Deep Research のレポートを鵜呑みにしない — MCP連携時に引用元を機械検証してから取り込む受け入れゲート

from urllib.parse import urlparse
 
class TrustResolver:
    def __init__(self, allowed_doc_ids: set[str], allowed_domains: set[str], media_exists):
        self.allowed_doc_ids = allowed_doc_ids
        self.allowed_domains = allowed_domains
        self.media_exists = media_exists  # media_id -> bool（実在解決の関数）
 
    def resolve(self, c: Citation) -> tuple[bool, str]:
        if c.kind == "file_search":
            if not c.doc_id or c.doc_id not in self.allowed_doc_ids:
                return False, "doc_id_not_allowed"
            if c.media_id and not self.media_exists(c.media_id):
                return False, "media_unresolvable"
            return True, "ok"
        if c.kind in ("web", "mcp"):
            if not c.url:
                return False, "missing_url"
            host = (urlparse(c.url).hostname or "").lower()
            if not any(host == d or host.endswith("." + d) for d in self.allowed_domains):
                return False, "domain_not_allowed"
            return True, "ok"
        return False, "unknown_source_type"

def evaluate(claims: list[Claim], resolver: TrustResolver) -> dict:
    reasons: dict[str, int] = {}
    grounded = 0
    checkable = 0
    for cl in claims:
        if not cl.citations:
            continue  # 引用不要の文は分母に入れない
        checkable += 1
        ok_here = False
        for c in cl.citations:
            ok, why = resolver.resolve(c)
            if ok:
                ok_here = True
            else:
                reasons[why] = reasons.get(why, 0) + 1
        if ok_here:
            grounded += 1
    coverage = grounded / checkable if checkable else 0.0
    return {
        "coverage": round(coverage, 3),
        "grounded": grounded,
        "checkable": checkable,
        "reject_reasons": reasons,
    }

def gate(report_id: str, claims, resolver, threshold: float = 0.85) -> dict:
    res = evaluate(claims, resolver)
    res["report_id"] = report_id
    res["decision"] = "accept" if res["coverage"] >= threshold else "quarantine"
    return res

# Before：引用の中身を見ずに取り込む
report = fetch_report(op)
ingest(report.text)   # もっともらしさだけで通過してしまう

# After：受け入れゲートを通してから取り込む
report = fetch_report(op)
claims = normalize_report(report.grounding)
decision = gate(report.id, claims, resolver, threshold=0.85)
 
if decision["decision"] == "accept":
    ingest(report.text)
else:
    quarantine(report.id, report.text, decision)   # 人のレビューへ
log_decision(decision)

import json, sqlite3, datetime
 
def log_decision(d: dict, db="gate_log.sqlite"):
    con = sqlite3.connect(db)
    con.execute("""CREATE TABLE IF NOT EXISTS gate_log(
        ts TEXT, report_id TEXT, decision TEXT, coverage REAL, reasons TEXT)""")
    con.execute("INSERT INTO gate_log VALUES(?,?,?,?,?)", (
        datetime.datetime.now().isoformat(timespec="seconds"),
        d["report_id"], d["decision"], d["coverage"], json.dumps(d["reject_reasons"]),
    ))
    con.commit(); con.close()
 
def nightly_review(db="gate_log.sqlite"):
    con = sqlite3.connect(db)
    rows = con.execute("""SELECT decision, COUNT(*), AVG(coverage)
        FROM gate_log WHERE ts >= datetime('now','-1 day') GROUP BY decision""").fetchall()
    con.close()
    return rows