Gemini RAGの選び方 2026 — シンプル・Advanced・Agenticの3パターンを実装で比べた

import google.generativeai as genai
import numpy as np
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
# ドキュメントを事前にチャンク化・埋め込み済みと仮定
# chunks: [{"text": "...", "embedding": [...]}]
 
def embed_text(text: str, task_type: str = "RETRIEVAL_DOCUMENT") -> list[float]:
    result = genai.embed_content(
        model="models/text-embedding-004",
        content=text,
        task_type=task_type,
    )
    return result["embedding"]
 
def cosine_similarity(a: list[float], b: list[float]) -> float:
    a, b = np.array(a), np.array(b)
    return float(np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)))
 
def simple_rag(query: str, chunks: list[dict], top_k: int = 3) -> str:
    # クエリをREトリーバルQUERYタスクで埋め込む（DOCUMENTと区別する）
    query_embedding = embed_text(query, task_type="RETRIEVAL_QUERY")
 
    # 類似度でソートして上位k件を取得
    scored = sorted(
        chunks,
        key=lambda c: cosine_similarity(query_embedding, c["embedding"]),
        reverse=True,
    )[:top_k]
    context = "\n\n---\n\n".join(c["text"] for c in scored)
 
    # Geminiで回答生成
    model = genai.GenerativeModel("gemini-2.5-flash")
    prompt = f"""以下の情報のみをもとに質問に答えてください。情報に含まれていない内容は「分かりません」と答えてください。
 
【参照情報】
{context}
 
【質問】
{query}"""
 
    response = model.generate_content(prompt)
    return response.text

def generate_hypothetical_document(query: str) -> str:
    """クエリに答える仮想ドキュメントを生成（HyDE）"""
    model = genai.GenerativeModel("gemini-2.5-flash")
    prompt = f"""次の質問に答える文書の一部を、実際の文書であるかのように200字程度で書いてください。
事実かどうかは問いません。検索精度向上のための仮想文書です。
 
質問: {query}
 
文書の一部:"""
    response = model.generate_content(prompt)
    return response.text
 
def advanced_rag_with_hyde(query: str, chunks: list[dict], top_k: int = 5) -> str:
    # HyDEで仮想ドキュメントを生成してから埋め込む
    hypothetical_doc = generate_hypothetical_document(query)
    search_embedding = embed_text(hypothetical_doc, task_type="RETRIEVAL_QUERY")
 
    # 上位k件を広めに取得
    scored = sorted(
        chunks,
        key=lambda c: cosine_similarity(search_embedding, c["embedding"]),
        reverse=True,
    )[:top_k]
 
    # 上位5件のうち本当に関連性が高いものだけに絞る（簡易リランク）
    # 実際はCross-Encoderなどを使うが、ここではGeminiで再スコアリング
    reranked = rerank_with_gemini(query, [c["text"] for c in scored], top_n=3)
    context = "\n\n---\n\n".join(reranked)
 
    model = genai.GenerativeModel("gemini-2.5-pro")
    prompt = f"""以下の情報のみをもとに、正確に質問に答えてください。
 
【参照情報】
{context}
 
【質問】
{query}"""
 
    response = model.generate_content(prompt)
    return response.text
 
def rerank_with_gemini(query: str, passages: list[str], top_n: int = 3) -> list[str]:
    """Geminiでパッセージを再スコアリング（簡易リランク）"""
    model = genai.GenerativeModel("gemini-2.5-flash")
    numbered = "\n\n".join(f"[{i+1}] {p}" for i, p in enumerate(passages))
    prompt = f"""質問に最も関連性の高いパッセージを{top_n}件選び、番号をカンマ区切りで返してください。他の文字は含めないでください。
 
質問: {query}
 
パッセージ:
{numbered}
 
最も関連性が高い{top_n}件の番号:"""
 
    response = model.generate_content(prompt)
    indices_str = response.text.strip()
    try:
        indices = [int(x.strip()) - 1 for x in indices_str.split(",")]
        return [passages[i] for i in indices if 0 <= i < len(passages)]
    except Exception:
        return passages[:top_n]  # パースに失敗した場合は上位からそのまま返す

import json
 
# ツール定義：Geminiに「検索できる」ことを伝える
search_tool = {
    "function_declarations": [
        {
            "name": "search_documents",
            "description": "ドキュメントベースを検索して関連情報を返す。複雑な質問では複数回呼び出してよい。",
            "parameters": {
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "検索クエリ。具体的なほど精度が上がる。",
                    },
                    "top_k": {
                        "type": "integer",
                        "description": "取得件数（1〜10の整数）",
                    },
                },
                "required": ["query"],
            },
        }
    ]
}
 
def execute_search(query: str, chunks: list[dict], top_k: int = 3) -> str:
    """実際のベクトル検索を実行"""
    query_embedding = embed_text(query, task_type="RETRIEVAL_QUERY")
    scored = sorted(
        chunks,
        key=lambda c: cosine_similarity(query_embedding, c["embedding"]),
        reverse=True,
    )[:top_k]
    return "\n\n".join(c["text"] for c in scored)
 
def agentic_rag(question: str, chunks: list[dict], max_iterations: int = 5) -> str:
    model = genai.GenerativeModel(
        "gemini-2.5-pro",
        tools=[search_tool],  # type: ignore
    )
    messages = [{"role": "user", "parts": [question]}]
 
    for _ in range(max_iterations):
        response = model.generate_content(messages)
        candidate = response.candidates[0]
 
        # Function Callがあれば実行する
        function_calls = [
            p.function_call
            for p in candidate.content.parts
            if hasattr(p, "function_call") and p.function_call.name
        ]
 
        if not function_calls:
            # ツール呼び出しなし → 最終回答
            return candidate.content.parts[0].text
 
        # モデルの発話を履歴に追加
        messages.append({"role": "model", "parts": candidate.content.parts})
 
        # 各Function Callを実行して結果を返す
        tool_results = []
        for fc in function_calls:
            args = dict(fc.args)
            result = execute_search(
                query=args["query"],
                chunks=chunks,
                top_k=int(args.get("top_k", 3)),
            )
            tool_results.append(
                {
                    "function_response": {
                        "name": fc.name,
                        "response": {"result": result},
                    }
                }
            )
 
        messages.append({"role": "user", "parts": tool_results})
 
    return "回答の生成に失敗しました（反復制限に達しました）"