Gemini API で『人間が確認するAI』を作る — Human-in-the-Loop ワークフローを本番導入する実装ガイド

# pip install google-genai pydantic
import os
from typing import Literal
from google import genai
from google.genai import types
from pydantic import BaseModel
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
class CritiqueResult(BaseModel):
    score: float  # 0.0 - 1.0
    issues: list[str]
    severity: Literal["none", "minor", "major", "critical"]
 
PRO_MODEL = "gemini-2.5-pro"
FLASH_MODEL = "gemini-2.5-flash"
 
def generate_with_logprobs(prompt: str) -> tuple[str, float | None]:
    """メイン生成。avg_logprob を信頼度として返す（取得できなければ None）。"""
    resp = client.models.generate_content(
        model=PRO_MODEL,
        contents=prompt,
        config=types.GenerateContentConfig(
            response_logprobs=True,
            logprobs=5,
            temperature=0.3,
        ),
    )
    text = resp.text
    # avg_logprob は -inf 〜 0。0 に近いほど確信度が高い
    avg_lp = resp.candidates[0].avg_logprobs
    if avg_lp is None:
        # モデルや設定によっては返らない。「確信度ゼロ」と誤読せず、欠測として扱う
        return text, None
    # 確信度を 0..1 に正規化（経験則: -0.3 以上が高確信、-1.5 以下は低確信）
    confidence = max(0.0, min(1.0, (avg_lp + 1.5) / 1.2))
    return text, confidence
 
def self_critique(prompt: str, output: str) -> CritiqueResult:
    """生成済み出力を Flash でセルフレビュー。"""
    review_prompt = f"""次の【質問】に対する【回答】を厳密に評価してください。
事実誤認・論理破綻・指示無視・有害表現があれば issues に列挙し、
深刻度を severity（none/minor/major/critical）で判定してください。
質を 0.0〜1.0 のスコアでも返してください。
 
【質問】
{prompt}
 
【回答】
{output}
"""
    resp = client.models.generate_content(
        model=FLASH_MODEL,
        contents=review_prompt,
        config=types.GenerateContentConfig(
            response_mime_type="application/json",
            response_schema=CritiqueResult,
            temperature=0.0,
        ),
    )
    return CritiqueResult.model_validate_json(resp.text)
 
def external_verify(output: str) -> float:
    """形式・ポリシー・禁止語などの機械チェック。"""
    score = 1.0
    if len(output) < 20 or len(output) > 4000:
        score -= 0.3
    forbidden = ["保証します", "100%確実", "絶対に"]  # 業務ドメインに合わせて調整
    if any(w in output for w in forbidden):
        score -= 0.4
    return max(0.0, score)
 
def hitl_decision(prompt: str) -> dict:
    """3 指標を合算して『自動採用 / レビュー要 / 再生成』を決定する。"""
    output, lp_conf = generate_with_logprobs(prompt)
    critique = self_critique(prompt, output)
    ext_score = external_verify(output)
 
    # 重み付き合算（経験則: Logprobs 0.3 / Critique 0.5 / External 0.2）
    # Logprobs が欠測のときは、残り 2 指標の比率（0.5 : 0.2）で重みを振り直す
    if lp_conf is None:
        composite = (critique.score * 0.5 + ext_score * 0.2) / 0.7
    else:
        composite = lp_conf * 0.3 + critique.score * 0.5 + ext_score * 0.2
 
    if critique.severity == "critical" or composite < 0.55:
        decision = "regenerate"  # 再生成または却下
    elif critique.severity in ("major",) or composite < 0.75:
        decision = "review"      # レビューキューへ
    else:
        decision = "auto_accept"
 
    return {
        "output": output,
        "decision": decision,
        "scores": {
            "logprobs": None if lp_conf is None else round(lp_conf, 3),
            "critique": critique.score,
            "external": round(ext_score, 3),
            "composite": round(composite, 3),
        },
        "severity": critique.severity,
        "issues": critique.issues,
    }
 
if __name__ == "__main__":
    result = hitl_decision("確定申告で青色申告特別控除を受ける条件を簡潔に教えてください。")
    print(result)
    # 期待される出力例:
    # {'output': '...', 'decision': 'auto_accept',
    #  'scores': {'logprobs': 0.82, 'critique': 0.91, 'external': 1.0, 'composite': 0.879},
    #  'severity': 'none', 'issues': []}

-- レビューキューのテーブル設計
CREATE TABLE review_queue (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    prompt TEXT NOT NULL,
    output TEXT NOT NULL,
    composite_score NUMERIC(4,3) NOT NULL,
    severity TEXT NOT NULL,
    issues JSONB,
    state TEXT NOT NULL DEFAULT 'pending',
        -- pending → claimed → approved / edited / rejected / escalated
    claimed_by TEXT,
    claimed_at TIMESTAMPTZ,
    decided_at TIMESTAMPTZ,
    final_output TEXT,
    sla_deadline TIMESTAMPTZ NOT NULL,
    created_at TIMESTAMPTZ NOT NULL DEFAULT now()
);
 
CREATE INDEX idx_queue_state_deadline
    ON review_queue (state, sla_deadline)
    WHERE state IN ('pending', 'claimed');

# pip install asyncpg fastapi
import asyncpg
from datetime import datetime, timedelta, timezone
 
VALID_TRANSITIONS = {
    "pending":  {"claimed", "escalated"},
    "claimed":  {"approved", "edited", "rejected", "escalated"},
    # 終端状態は遷移不可
    "approved": set(),
    "edited":   set(),
    "rejected": set(),
    "escalated":{"claimed"},
}
 
async def claim_item(pool, reviewer_id: str) -> dict | None:
    """SLA が古いものから取得（FOR UPDATE SKIP LOCKED）。"""
    async with pool.acquire() as conn:
        # トランザクションの外で FOR UPDATE を使うと、文の終了と同時にロックが外れる
        async with conn.transaction():
            row = await conn.fetchrow("""
                SELECT id, prompt, output, sla_deadline FROM review_queue
                WHERE state IN ('pending', 'escalated')
                ORDER BY (state = 'escalated') DESC, sla_deadline ASC
                LIMIT 1
                FOR UPDATE SKIP LOCKED
            """)
            if not row:
                return None
            await conn.execute("""
                UPDATE review_queue
                   SET state='claimed', claimed_by=$1, claimed_at=now()
                 WHERE id=$2
            """, reviewer_id, row["id"])
            return dict(row)
 
async def decide_item(pool, item_id: str, reviewer_id: str,
                      new_state: str, final_output: str | None = None):
    """状態遷移をアプリ層でも検証。"""
    async with pool.acquire() as conn:
        async with conn.transaction():
            row = await conn.fetchrow(
                "SELECT state, claimed_by FROM review_queue WHERE id=$1 FOR UPDATE",
                item_id,
            )
            if not row:
                raise ValueError("item not found")
            if row["claimed_by"] != reviewer_id:
                raise PermissionError("not your item")
            if new_state not in VALID_TRANSITIONS[row["state"]]:
                raise ValueError(f"invalid transition: {row['state']} -> {new_state}")
 
            await conn.execute("""
                UPDATE review_queue
                   SET state=$1, decided_at=now(), final_output=$2
                 WHERE id=$3
            """, new_state, final_output, item_id)

import asyncio
 
async def sla_watcher(pool):
    """SLA を超過した pending/claimed をエスカレーションへ移す。"""
    while True:
        try:
            async with pool.acquire() as conn:
                escalated = await conn.fetch("""
                    UPDATE review_queue
                       SET state='escalated'
                     WHERE state IN ('pending', 'claimed')
                       AND sla_deadline < now()
                  RETURNING id
                """)
                if escalated:
                    print(f"[SLA] escalated {len(escalated)} items")
                    # ここで Slack 通知などを送る
        except Exception as e:
            print(f"[SLA] watcher error: {e}")
        await asyncio.sleep(60)

async def collect_recent_edits(pool, days: int = 7) -> list[dict]:
    """過去 N 日間の edited 案件を収集する。"""
    async with pool.acquire() as conn:
        rows = await conn.fetch("""
            SELECT prompt, output AS original, final_output AS edited, issues
              FROM review_queue
             WHERE state = 'edited'
               AND decided_at > now() - INTERVAL '1 day' * $1
        """, days)
        return [dict(r) for r in rows]
 
async def synthesize_few_shots(edits: list[dict]) -> str:
    """編集パターンから Few-Shot を Gemini に作らせる。"""
    samples = "\n---\n".join(
        f"Prompt: {e['prompt']}\nBefore: {e['original']}\nAfter: {e['edited']}"
        for e in edits[:20]
    )
    prompt = f"""以下は人間レビュアーが Gemini の出力を編集した実例です。
このパターンから、今後の出力品質を上げるための Few-Shot 例を 3 つ抽出してください。
形式は ## 例1 / ## 例2 / ## 例3 のマークダウンで、各例に質問と理想回答を含めてください。
 
{samples}
"""
    resp = client.models.generate_content(
        model="gemini-2.5-pro", contents=prompt,
        config=types.GenerateContentConfig(temperature=0.2),
    )
    return resp.text

ALTER TABLE review_queue
    ADD COLUMN source TEXT NOT NULL DEFAULT 'gate';  -- 'gate' or 'shadow'

import hashlib
 
SHADOW_RATE = 0.05
 
def is_shadow_sample(request_id: str) -> bool:
    """同じ ID なら常に同じ判定になる（あとから再現できる）。"""
    h = int(hashlib.sha256(request_id.encode()).hexdigest()[:8], 16)
    return (h / 0xFFFFFFFF) < SHADOW_RATE

# 健康指標を1日1回スナップショット取得する例
async def snapshot_metrics(pool) -> dict:
    async with pool.acquire() as conn:
        row = await conn.fetchrow("""
        SELECT
          COUNT(*) FILTER (WHERE state IN ('approved','edited','rejected'))
            AS decided_count,
          COUNT(*) FILTER (WHERE state = 'rejected') AS rejected,
          COUNT(*) FILTER (WHERE state = 'approved') AS approved,
          COUNT(*) FILTER (WHERE state = 'edited') AS edited,
          PERCENTILE_CONT(0.5) WITHIN GROUP
            (ORDER BY EXTRACT(EPOCH FROM (decided_at - created_at)))
            FILTER (WHERE decided_at IS NOT NULL) AS median_review_seconds
        FROM review_queue
        WHERE created_at > now() - INTERVAL '24 hours'
        """)
        decided = row["decided_count"] or 0
        return {
            "rejection_rate": (row["rejected"] / decided) if decided else 0.0,
            "edit_rate": (row["edited"] / decided) if decided else 0.0,
            "approve_rate": (row["approved"] / decided) if decided else 0.0,
            "median_review_seconds": row["median_review_seconds"],
        }