Gemini APIのFunction Callingが無限ループする原因と止め方

# 目的: 無限ループを防ぐため、反復上限・重複検出・終了判定を組み合わせる
MAX_TURNS = 8
 
def key_of(call) -> str:
    # ツール名＋引数ハッシュで「同じ呼び出し」を判定する
    return f"{call.name}:{sorted(dict(call.args).items())}"
 
seen = {}
for turn in range(MAX_TURNS):
    resp = client.models.generate_content(
        model="gemini-2.5-pro", contents=history,
        config=types.GenerateContentConfig(tools=tools))
    cand = resp.candidates[0]
    call = next((p.function_call for p in cand.content.parts if p.function_call), None)
 
    if call is None:
        # 最終応答に到達
        print("最終応答:", resp.text)
        break
 
    k = key_of(call)
    seen[k] = seen.get(k, 0) + 1
    if seen[k] >= 3:
        # 同じ呼び出しが3回以上続いたら、モデルに方針転換を促す
        history.append(cand.content)
        history.append(types.Content(role="user", parts=[types.Part.from_text(
            text=f"同じツール呼び出し({call.name})が繰り返されています。"
                 f"現時点で得られた情報だけで要約を出してください。")]))
        continue
 
    result = search_web(**call.args)
    history.append(cand.content)
    history.append(types.Content(role="tool", parts=[
        types.Part.from_function_response(name=call.name, response=result)]))
else:
    # ループを超えた場合のフォールバック
    print("反復上限に到達したため、部分的な情報で応答を生成します")

def search_web(query: str) -> dict:
    hits = my_search_api(query)
    return {
        "results": hits[:5],
        "total_found": len(hits),
        "status": "sufficient" if len(hits) >= 3 else "insufficient",
        "next_suggestion": None if len(hits) >= 3 else "クエリをより具体的にしてください",
    }

# 目的: ループに使うモデルを固定し、要求≠応答のすり替わりを検知する
REQUESTED_MODEL = "gemini-3.5-flash"  # 世代だけでなく明示的に固定する
 
resp = client.models.generate_content(
    model=REQUESTED_MODEL, contents=history,
    config=types.GenerateContentConfig(tools=tools))
 
served = getattr(resp, "model_version", None)
if served and not served.startswith(REQUESTED_MODEL):
    # 既定モデルの差し替えやエイリアス解決で別世代に振られた可能性
    import logging
    logging.warning("model drift: requested=%s served=%s", REQUESTED_MODEL, served)

# 目的: model_version ごとの平均ツール呼び出し回数を集計し、回帰を早期検知する
from collections import defaultdict
 
def tool_call_canary(rows: list[dict]) -> dict:
    # rows: [{"model_version": str, "tool_calls": int}, ...] を当日分まとめて渡す
    agg = defaultdict(list)
    for r in rows:
        agg[r["model_version"]].append(r["tool_calls"])
    report = {}
    for model, calls in agg.items():
        avg = sum(calls) / len(calls)
        report[model] = {
            "runs": len(calls),
            "avg_tool_calls": round(avg, 2),
            "hit_cap_ratio": round(sum(c >= 8 for c in calls) / len(calls), 2),  # MAX_TURNS=8
        }
    return report

# 目的: 引数が違っても結果が進まないループを、返却値の変化量で検知する
import hashlib, json
 
def result_fingerprint(result: dict) -> str:
    # 順序やノイズを排し、結果の「中身」だけを指紋化する
    norm = json.dumps(result.get("results", result), sort_keys=True, ensure_ascii=False)
    return hashlib.sha256(norm.encode("utf-8")).hexdigest()[:16]
 
recent = []          # 直近の結果指紋
STALL_WINDOW = 3     # 連続して同じならストールとみなす
 
# ループ内、ツール実行直後に:
fp = result_fingerprint(result)
recent.append(fp)
recent = recent[-STALL_WINDOW:]
if len(recent) == STALL_WINDOW and len(set(recent)) == 1:
    # 引数は違っても結果が3回連続で不変 → 情報が増えていない
    history.append(cand.content)
    history.append(types.Content(role="user", parts=[types.Part.from_text(
        text="検索を変えても新しい情報が得られていません。"
             "今ある結果だけで結論をまとめてください。")]))
    recent.clear()
    continue