Gemini Code Execution の計算結果とモデルの文章がずれる — 実行結果だけを信頼する検証ゲートの実装

PROMPT_SUFFIX = (
    "計算の最後に、最終結果だけを次の形式で1行だけ出力してください。"
    "他の行にこの形式を使わないこと: RESULT_JSON={\"weighted_ecpm\": <値>}"
)
 
response = client.models.generate_content(
    model="gemini-2.5-pro",
    contents=user_prompt + "\n" + PROMPT_SUFFIX,
    config=types.GenerateContentConfig(
        tools=[types.Tool(code_execution=types.ToolCodeExecution())],
    ),
)
 
import json, re
 
def extract_executed_value(parts, key):
    """code_execution_result の標準出力から RESULT_JSON 行を拾い、
    指定キーの値を返す。見つからなければ None。"""
    marker = re.compile(r'RESULT_JSON=(\{.*\})')
    found = None
    for part in parts:
        if part.code_execution_result is None:
            continue
        for line in part.code_execution_result.output.splitlines():
            m = marker.search(line)
            if m:
                try:
                    found = json.loads(m.group(1)).get(key)
                except json.JSONDecodeError:
                    continue
    return found  # 複数回実行された場合は最後の RESULT_JSON を採用
 
executed = extract_executed_value(response.candidates[0].content.parts, "weighted_ecpm")
print("実行された最終値:", executed)
# 実行された最終値: 288.42105263157896

def resolve_execution(parts, key):
    """成功した実行結果からのみ最終値を採る。
    失敗・タイムアウトは理由付きで返し、静かに握りつぶさない。"""
    marker = re.compile(r'RESULT_JSON=(\{.*\})')
    ok_value = None
    failures = []
    for part in parts:
        r = part.code_execution_result
        if r is None:
            continue
        outcome = getattr(r.outcome, "name", str(r.outcome))
        if outcome != "OUTCOME_OK":
            failures.append((outcome, r.output[:200]))
            continue
        for line in r.output.splitlines():
            m = marker.search(line)
            if m:
                try:
                    ok_value = json.loads(m.group(1)).get(key)
                except json.JSONDecodeError:
                    pass
    if ok_value is None:
        return {"status": "no_value", "failures": failures}
    return {"status": "ok", "value": ok_value, "failures": failures}
 
result = resolve_execution(response.candidates[0].content.parts, "weighted_ecpm")
print(result)
# {'status': 'ok', 'value': 288.42105263157896, 'failures': []}

def numbers_in_text(text):
    """文章中の数値（カンマ・小数対応）を集合で返す。"""
    raw = re.findall(r'-?\d[\d,]*\.?\d*', text)
    vals = set()
    for r in raw:
        try:
            vals.add(float(r.replace(",", "")))
        except ValueError:
            pass
    return vals
 
def verify_prose(executed_value, prose_text, rel_tol=0.01):
    """文章中に、実行結果と相対誤差1%以内で一致する数値が
    含まれているかを確認する。無ければ矛盾とみなす。"""
    for v in numbers_in_text(prose_text):
        if executed_value == 0:
            if v == 0:
                return True
        elif abs(v - executed_value) / abs(executed_value) <= rel_tol:
            return True
    return False
 
executed_value = result["value"]
prose_text = "".join(prose)
 
if verify_prose(executed_value, prose_text):
    final_answer = prose_text        # 文章も実行結果と整合している
    trusted_number = executed_value  # 判断に使う数字は必ず実行結果
else:
    # 文章がずれている → 表示を差し替える
    final_answer = f"加重平均eCPMは {executed_value:.1f} 円です（自動整形）。"
    trusted_number = executed_value
    # 監視: モデルの転記ずれを記録
    print("⚠ prose mismatch:", prose_text[:80])
 
print("表示:", final_answer)
print("判断に使う値:", trusted_number)