取り込んだのに一度も引かれない資料 — File Search ストアを引用ログで棚卸しする隔離つき剪定設計

from google import genai
from google.genai import types
 
client = genai.Client()  # GEMINI_API_KEY を環境変数から読む
STORE_NAME = "fileSearchStores/faq-assistant-01"
 
 
def ask_with_citations(question: str) -> dict:
    """質問を投げ、回答本文と引用された資料の識別子を返す。"""
    response = client.models.generate_content(
        model="gemini-flash-latest",
        contents=question,
        config=types.GenerateContentConfig(
            tools=[
                types.Tool(
                    file_search=types.FileSearch(
                        file_search_store_names=[STORE_NAME]
                    )
                )
            ],
        ),
    )
 
    cited = set()
    # grounding_metadata / grounding_chunks は存在しないことがあるため必ずガードする
    for cand in (response.candidates or []):
        meta = getattr(cand, "grounding_metadata", None)
        if not meta:
            continue
        for chunk in (getattr(meta, "grounding_chunks", None) or []):
            ctx = getattr(chunk, "retrieved_context", None)
            if not ctx:
                continue
            # File Search はドキュメント名や title を返す。環境差に備え両方見る
            doc_id = getattr(ctx, "document_name", None) or getattr(ctx, "title", None)
            if doc_id:
                cited.add(doc_id)
 
    return {"answer": response.text, "cited_docs": sorted(cited)}
 
 
if __name__ == "__main__":
    result = ask_with_citations("解約後もデータは残りますか？")
    print(result["answer"])
    print("引用:", result["cited_docs"])
    # 引用: ['documents/policy-cancellation-2026', 'documents/data-retention-note']

import json
import time
from pathlib import Path
 
LEDGER = Path("citation_ledger.jsonl")
 
 
def record_citation(question: str, cited_docs: list[str]) -> None:
    """1クエリ分の引用実績を追記する。個人開発ならこれで十分軽い。"""
    entry = {
        "ts": int(time.time()),
        "q_hash": hash(question) & 0xFFFFFFFF,  # 質問の生文は残さずハッシュのみ
        "cited": cited_docs,
    }
    with LEDGER.open("a", encoding="utf-8") as f:
        f.write(json.dumps(entry, ensure_ascii=False) + "\n")
 
 
def coverage_since(days: int = 90) -> dict[str, int]:
    """直近 days 日で、各資料が何回引用されたかを集計する。"""
    cutoff = int(time.time()) - days * 86400
    counts: dict[str, int] = {}
    if not LEDGER.exists():
        return counts
    for line in LEDGER.read_text(encoding="utf-8").splitlines():
        row = json.loads(line)
        if row["ts"] < cutoff:
            continue
        for doc in row["cited"]:
            counts[doc] = counts.get(doc, 0) + 1
    return counts

import datetime as dt
 
# ストア内ドキュメントの一覧を取得する（取り込み日を含む想定）
def list_store_docs() -> list[dict]:
    docs = []
    for d in client.file_search_stores.documents.list(parent=STORE_NAME):
        docs.append({
            "id": d.name,
            "created": d.create_time,  # datetime
        })
    return docs
 
 
def find_prune_candidates(min_age_days: int = 90) -> list[dict]:
    """未引用かつ取り込みから min_age_days 以上経過した資料を候補にする。"""
    counts = coverage_since(days=min_age_days)
    now = dt.datetime.now(dt.timezone.utc)
    candidates = []
    for doc in list_store_docs():
        age_days = (now - doc["created"]).days
        cited = counts.get(doc["id"], 0)
        if cited == 0 and age_days >= min_age_days:
            candidates.append({"id": doc["id"], "age_days": age_days})
    # 古いものから順に並べる
    return sorted(candidates, key=lambda x: -x["age_days"])

QUARANTINE_STORE = "fileSearchStores/quarantine-01"
ALLOWLIST = {
    "documents/policy-cancellation-2026",  # 稀だが決定的。恒久保護
    "documents/legal-tokushoho-notice",
}
 
 
def quarantine(doc_id: str) -> None:
    """本番ストアから外し、隔離ストアへ退避する。元は消さない。"""
    if doc_id in ALLOWLIST:
        return  # 保護対象は絶対に触らない
    # 退避（実装はストア間のコピー）→ 本番から削除
    _copy_document(doc_id, src=STORE_NAME, dst=QUARANTINE_STORE)
    client.file_search_stores.documents.delete(name=doc_id)
 
 
def promote_deletion(grace_days: int = 30) -> list[str]:
    """隔離から grace_days 経過し、その間に影の照合ヒットが無ければ本削除する。"""
    deleted = []
    now = dt.datetime.now(dt.timezone.utc)
    for d in client.file_search_stores.documents.list(parent=QUARANTINE_STORE):
        age = (now - d.create_time).days
        if age >= grace_days and not _shadow_hit(d.name, since_days=grace_days):
            client.file_search_stores.documents.delete(name=d.name)
            deleted.append(d.name)
    return deleted

def _shadow_hit(doc_id: str, since_days: int) -> bool:
    """隔離期間中の実クエリを隔離ストアに流し、doc_id が引かれたか確認する。"""
    cutoff = int(time.time()) - since_days * 86400
    recent_q_hashes = _recent_question_hashes(since=cutoff)
    for q in _sample_questions_for(recent_q_hashes):
        resp = client.models.generate_content(
            model="gemini-flash-latest",
            contents=q,
            config=types.GenerateContentConfig(
                tools=[types.Tool(file_search=types.FileSearch(
                    file_search_store_names=[QUARANTINE_STORE]))],
            ),
        )
        for cand in (resp.candidates or []):
            meta = getattr(cand, "grounding_metadata", None)
            chunks = getattr(meta, "grounding_chunks", None) or [] if meta else []
            for c in chunks:
                ctx = getattr(c, "retrieved_context", None)
                name = getattr(ctx, "document_name", None) if ctx else None
                if name == doc_id:
                    return True  # 隔離中に必要とされた。復帰させるべき
    return False