Gemini の Function Calling に危険度別の承認ゲートを挟む実装

def run_with_approval(client, contents, tools, approve, model="gemini-flash-latest"):
    """approve(call, tier) -> bool を人間の承認手続きとして受け取る。
    green は approve を呼ばずに自動実行する。"""
    while True:
        resp = client.models.generate_content(
            model=model,
            contents=contents,
            config=types.GenerateContentConfig(tools=[tools]),
        )
        parts = resp.candidates[0].content.parts
        calls = [p.function_call for p in parts if p.function_call]
 
        # ツール呼び出しが無ければ、モデルは最終応答を出したと判断
        if not calls:
            return resp.text
 
        # モデルの発話(functionCall を含む)を履歴へ積む
        contents.append(resp.candidates[0].content)
 
        tool_parts = []
        for call in calls:
            tier = tier_of(call.name)
            args = dict(call.args)
            if tier == "green" or approve(call, tier):
                result = EXECUTORS[call.name](**args)
            else:
                # 却下も「結果」としてモデルに返す。黙って捨てない
                result = {"status": "rejected", "reason": "human declined"}
            tool_parts.append(
                types.Part.from_function_response(name=call.name, response=result)
            )
 
        # 全 functionCall への応答をまとめて返す
        contents.append(types.Content(role="tool", parts=tool_parts))

def approve_via_terminal(call, tier) -> bool:
    print(f"[{tier.upper()}] {call.name}({dict(call.args)}) を実行しますか?")
    return input("y/N > ").strip().lower() == "y"
 
client = genai.Client(api_key="YOUR_API_KEY")
contents = [types.Content(
    role="user",
    parts=[types.Part(text="空き容量を確保して。まず状況を調べてから提案して。")],
)]
answer = run_with_approval(client, contents, tools, approve_via_terminal)
print(answer)

import json
 
class ApprovalRequired(Exception):
    def __init__(self, call, tier):
        self.call, self.tier = call, tier
 
def approve_or_suspend(call, tier) -> bool:
    # 無人運用では yellow は自動、red は中断して永続化に委ねる
    if tier == "yellow":
        return True
    raise ApprovalRequired(call, tier)
 
def run_until_approval(client, contents, tools, state_path):
    try:
        return run_with_approval(client, contents, tools, approve_or_suspend)
    except ApprovalRequired as pending:
        snapshot = {
            "contents": [c.model_dump() for c in contents],
            "pending": {"name": pending.call.name, "args": dict(pending.call.args)},
        }
        with open(state_path, "w", encoding="utf-8") as f:
            json.dump(snapshot, f, ensure_ascii=False)
        return None  # 承認待ちで中断

def resume_after_approval(client, tools, state_path, approved: bool):
    with open(state_path, encoding="utf-8") as f:
        snap = json.load(f)
    contents = [types.Content.model_validate(c) for c in snap["contents"]]
    name = snap["pending"]["name"]
    args = snap["pending"]["args"]
 
    result = EXECUTORS[name](**args) if approved else {"status": "rejected"}
    contents.append(types.Content(
        role="tool",
        parts=[types.Part.from_function_response(name=name, response=result)],
    ))
    # 残りの yellow は自動、次の red でまた中断
    return run_until_approval(client, contents, tools, state_path)