Gemini API で『自己批判するエージェント』を実装する — Reflection × Critic × Refiner で本番品質を継続的に上げる実装

# Critic-Refiner: Generator/Refiner = gemini-3-pro, Critic = gemini-2.5-flash
import os
from google import genai
from google.genai import types
from pydantic import BaseModel
from typing import List
 
client = genai.Client(api_key=os.environ["YOUR_GEMINI_API_KEY"])
GEN_MODEL = "gemini-3-pro"
CRITIC_MODEL = "gemini-2.5-flash"
 
class Issue(BaseModel):
    category: str
    snippet: str
    suggestion: str
    severity: int  # 1(軽微) 〜 5(致命)
 
class CriticReview(BaseModel):
    issues: List[Issue]
    overall_score: int  # 0〜100
 
def critic(draft: str, constraints: List[str]) -> CriticReview:
    """軽量モデルで初稿をレビュー。severity 判断基準を Few-shot で明示する。"""
    severity_rubric = (
        "severity 5 (致命): 事実誤認、禁止語混入、フォーマット崩壊で機械処理不能\n"
        "severity 4 (重大): 指示違反、必須項目欠落\n"
        "severity 3 (中):   表現が不自然、敬体・常体の混在\n"
        "severity 2 (軽微): より良い表現の提案、句読点の好み\n"
        "severity 1 (微小): 完全に好みの問題"
    )
    prompt = (
        "次の本文を厳格にレビューしてください。\n"
        "本当に問題のある箇所だけ列挙し、好みの違いは無視してください。\n"
        f"severity の判断基準:\n{severity_rubric}\n\n"
        f"制約:\n" + "\n".join(f"- {c}" for c in constraints) +
        f"\n\n本文:\n{draft}"
    )
    try:
        resp = client.models.generate_content(
            model=CRITIC_MODEL,
            contents=prompt,
            config=types.GenerateContentConfig(
                temperature=0.0,
                response_mime_type="application/json",
                response_schema=CriticReview,
            ),
        )
        return CriticReview.model_validate_json(resp.text)
    except Exception as e:
        print(f"[critic] fallback: {e}")
        return CriticReview(issues=[], overall_score=100)
 
def refiner(draft: str, review: CriticReview, severity_threshold: int = 3) -> str:
    """severity >= threshold の指摘のみ修正対象にする。過剰修正を防ぐ。"""
    target_issues = [i for i in review.issues if i.severity >= severity_threshold]
    if not target_issues:
        return draft  # 修正対象なし
    issues_text = "\n".join(
        f"- [sev{i.severity}/{i.category}] {i.snippet} → {i.suggestion}"
        for i in target_issues
    )
    prompt = (
        "次の本文の、指摘されている箇所のみを最小限修正してください。\n"
        "指摘以外の部分は一切変更しないでください。\n\n"
        f"修正点:\n{issues_text}\n\n元の本文:\n{draft}"
    )
    resp = client.models.generate_content(
        model=GEN_MODEL,
        contents=prompt,
        config=types.GenerateContentConfig(temperature=0.2),
    )
    return resp.text
 
def critic_refiner_pipeline(task: str, constraints: List[str]) -> dict:
    """Generator → Critic → Refiner の3段パイプライン。"""
    gen_resp = client.models.generate_content(
        model=GEN_MODEL,
        contents=(
            f"要件: {task}\n\n制約:\n" + "\n".join(f"- {c}" for c in constraints)
        ),
        config=types.GenerateContentConfig(temperature=0.4),
    )
    draft = gen_resp.text
    review = critic(draft, constraints)
    final = refiner(draft, review, severity_threshold=3)
    return {"draft": draft, "review": review.model_dump(), "final": final}
 
# 期待出力: draft / review / final の3つを含む辞書。
# - severity 3 未満の指摘は無視されるため、final は draft とほぼ同じ場合もある

import time
from dataclasses import dataclass
from typing import List
 
@dataclass
class LoopConfig:
    max_iterations: int = 2          # 反復は最大2回(初稿 + 1回修正)まで
    max_total_tokens: int = 50_000   # 1リクエスト合計トークン上限
    timeout_seconds: float = 30.0    # 全体タイムアウト
    severity_threshold: int = 3      # 修正対象の severity
 
@dataclass
class LoopResult:
    final_text: str
    iterations: int
    total_input_tokens: int
    total_output_tokens: int
    stopped_reason: str  # "no_issues" / "max_iterations" / "budget" / "timeout"
 
def reflect_with_budget(
    task: str,
    constraints: List[str],
    config: LoopConfig = LoopConfig(),
) -> LoopResult:
    """予算ガード付き Reflection ループ。3つの上限のいずれかに達したら停止する。"""
    started = time.monotonic()
    in_tokens = 0
    out_tokens = 0
 
    # 初稿
    gen_resp = client.models.generate_content(
        model=GEN_MODEL,
        contents=(
            f"要件: {task}\n制約:\n" + "\n".join(f"- {c}" for c in constraints)
        ),
        config=types.GenerateContentConfig(temperature=0.4),
    )
    current = gen_resp.text
    in_tokens += gen_resp.usage_metadata.prompt_token_count
    out_tokens += gen_resp.usage_metadata.candidates_token_count
 
    for iteration in range(1, config.max_iterations + 1):
        # ガード: タイムアウト
        if time.monotonic() - started > config.timeout_seconds:
            return LoopResult(current, iteration, in_tokens, out_tokens, "timeout")
        # ガード: トークン予算
        if in_tokens + out_tokens > config.max_total_tokens:
            return LoopResult(current, iteration, in_tokens, out_tokens, "budget")
 
        review = critic(current, constraints)
        target_issues = [i for i in review.issues if i.severity >= config.severity_threshold]
 
        # ガード: 直すべき問題がない
        if not target_issues:
            return LoopResult(current, iteration, in_tokens, out_tokens, "no_issues")
 
        current = refiner(current, review, config.severity_threshold)
 
    return LoopResult(current, config.max_iterations, in_tokens, out_tokens, "max_iterations")
 
# 期待出力: LoopResult。stopped_reason に必ずいずれかの停止理由が入る
# 例: LoopResult(final_text='...', iterations=2, total_input_tokens=1234,
#                total_output_tokens=567, stopped_reason='max_iterations')

# ❌ 危険: 全体を書き直す指示
prompt = f"レビュー結果を踏まえて全体を書き直してください。\n\nレビュー:\n{review_text}\n\n本文:\n{draft}"
 
# ✅ 安全: 指摘箇所のみを最小修正する指示
prompt = (
    "指摘されている箇所のみを最小限修正してください。\n"
    "指摘以外の部分は一切変更しないでください。\n\n"
    f"修正点:\n{issues_text}\n\n元の本文:\n{draft}"
)