Gemini Interactions API：会话存储与隐私模式配置

Interactions API 默认会在服务端保存每一次交互（store=true）：付费层保存 55 天，免费层保存 1 天。这样可以用 previous_interaction_id 接着聊，但对话内容也被保存了下来。 这篇写一个 ChatSession，提供两种模式：

• stateful：会话结束时，用 delete 删除本会话保存的每一条交互；
• stateless：store=False，服务端不保存，每轮发送完整历史（按官方文档的写法，把返回的 steps 追加到历史中）。

另外，我实测发现 SDK 遇到 429 时会自己重试，所以代码里没有再加一层重试。

先看清成品

• 本地替身上：stateful 模式第二轮带上了 previous_interaction_id，结束时删除了 2 条交互；删除已过期的交互时报告 404；stateless 模式每轮的输入是 user_input → thought → model_output → user_input，服务端不保存任何内容。
• 429：连续 2 次 429 后，SDK 自己重试成功（共 3 个请求）；一直 429 时，SDK 共发 4 个请求后抛出。pytest 5 项通过。
• 用无效 Key 调用真实接口：✗ HTTP 400：... API key not valid，清理结果是 ([], [])（因为一条交互都没有创建）。

合规提醒：Gemini API 和 Google AI Studio 只在官方支持的地区提供，官方列表中没有中国大陆和香港。不在支持地区就不要使用，不要用代理或虚假信息绕过地区限制，也不要共享账号或 API Key。按照官方条款，使用免费额度（Unpaid Services）时，提交的内容和生成的结果可能被 Google 用来改进产品，还可能有人工审阅，所以不要提交个人敏感信息或公司机密。限额按项目计算，不是按 Key 计算，多建 Key 也不会增加额度。API Key 放在环境变量 GEMINI_API_KEY 中。本文不列价格和具体限额。

运行说明：box 上没有 Gemini API Key，所以没有拿到真实的模型输出。真实运行过的部分是：用无效 Key 调用真实接口，记录真实的错误格式。成功路径用的是我自己写的本地替身（fake_gemini.py，不是 Google 官方程序），客户端是真实的 google-genai 2.32.0；替身的请求和响应格式来自本地抓取的 SDK 实际请求，以及 SDK 自带测试中的数据结构。

做完之后你会得到什么？

需要准备哪些环境？

关键事实（来自官方 Interactions 和文本生成文档）

文件结构

62-google-ai-C-src/
├── chat_session.py
├── fake_gemini.py
└── test_chat_session.py

第 1 步：会话类

"""多轮对话的两种隐私模式（Gemini Interactions API）。
官方文档要点：
- 默认 store=true，交互对象保存在服务端，可以用 previous_interaction_id 接着聊；
  保留期：付费层 55 天，免费层 1 天；可以随时用 delete 删除
- store=false 时服务端不保存，但也不能用 previous_interaction_id，需要每轮发送完整历史
本脚本：
- stateful：记录每轮的 interaction id，close() 时逐个删除（会话结束就清理）
- stateless：store=False，按文档的写法把上一轮返回的 steps（model_dump()）追加到历史里
- 429：SDK 2.32 的 Interactions 客户端**自己会重试**（实测默认共发 4 次请求，约 3.5 秒后才抛出），
  所以这里不再叠加一层重试；SDK 重试后仍然 429，就视为额度用完，直接抛出"""
import os, sys

MODEL = os.environ.get("GEMINI_MODEL", "gemini-3.8-flash")

def status_of(e):
    return getattr(e, "status_code", None) or getattr(e, "code", None)

class ChatSession:
    def __init__(self, client, mode="stateless", model=MODEL):
        assert mode in ("stateful", "stateless")
        self.client, self.mode, self.model = client, mode, model
        self.ids, self.history, self.last_id = [], [], None

    def _create(self, **kw):
        return self.client.interactions.create(model=self.model, **kw)   # 重试交给 SDK

    def ask(self, text):
        if self.mode == "stateful":
            kw = {"input": text, "store": True}
            if self.last_id:
                kw["previous_interaction_id"] = self.last_id
            it = self._create(**kw)
            self.ids.append(it.id); self.last_id = it.id
        else:
            self.history.append({"type": "user_input", "content": [{"type": "text", "text": text}]})
            it = self._create(input=self.history, store=False)
            for step in it.steps:
                self.history.append(step.model_dump(exclude_none=True))
        if it.status != "completed":
            raise RuntimeError(f"状态为 {it.status}")
        return it.output_text

    def close(self):
        """删除本会话在服务端保存的所有交互。返回 (已删除, 失败) 两个列表。"""
        ok, failed = [], []
        for iid in self.ids:
            try:
                self.client.interactions.delete(iid); ok.append(iid)
            except Exception as e:
                failed.append((iid, status_of(e)))
        self.ids, self.last_id = [], None
        return ok, failed

if __name__ == "__main__":
    from google import genai
    client = genai.Client()
    s = ChatSession(client, mode=sys.argv[1] if len(sys.argv) > 1 else "stateless")
    try:
        print(s.ask("我家有 2 只狗。")); print(s.ask("我家一共有几只爪子？"))
    except Exception as e:
        print(f"✗ HTTP {status_of(e)}：{str(e)[:200]}"); sys.exit(2)
    finally:
        if s.mode == "stateful":
            print("清理：", s.close())

第 2 步：测试

import pytest
from google import genai
from google.genai import types
import chat_session as C, fake_gemini as F

@pytest.fixture
def client():
    F.reset(); s, u = F.start()
    yield genai.Client(api_key="good", http_options=types.HttpOptions(base_url=u)); s.shutdown()

def test_stateful_uses_previous_id_and_cleans_up(client):
    s = C.ChatSession(client, "stateful")
    s.ask("我家有 2 只狗。")
    assert s.ask("我刚才说了什么？") == "你之前说：我家有 2 只狗。"
    assert F.STATE["requests"][-1][1]["previous_interaction_id"] == "v1_fake1"
    ok, failed = s.close()
    assert ok == ["v1_fake1", "v1_fake2"] and failed == [] and F.STATE["store"] == {}

def test_cleanup_reports_failures(client):
    s = C.ChatSession(client, "stateful"); s.ask("hi")
    F.STATE["store"].clear()                          # 模拟已经过了保留期
    assert s.close() == ([], [("v1_fake1", 404)])

def test_stateless_sends_full_history_and_stores_nothing(client):
    s = C.ChatSession(client, "stateless")
    s.ask("我家有 2 只狗。")
    assert s.ask("一共几只爪子？").startswith("（看到了 2 轮用户输入）你刚才说：我家有 2 只狗。")
    body = F.STATE["requests"][-1][1]
    assert body["store"] is False and "previous_interaction_id" not in body
    assert [x["type"] for x in body["input"]] == ["user_input", "thought", "model_output", "user_input"]
    assert F.STATE["store"] == {}

def test_sdk_retries_429_itself(client):
    F.STATE["rate_429"] = 2
    assert C.ChatSession(client).ask("hi") == "收到：hi"
    assert len(F.STATE["requests"]) == 3              # 2 次 429 + 1 次成功，都是 SDK 自己重试的

def test_persistent_429_raises(client):
    F.STATE["rate_429"] = 99
    with pytest.raises(Exception) as e:
        C.ChatSession(client).ask("hi")
    assert C.status_of(e.value) == 429 and len(F.STATE["requests"]) == 4

fake_gemini.py 和 62A 相同。

第 3 步：运行

$ GEMINI_API_KEY=INVALID python chat_session.py stateful
✗ HTTP 400：Error code: 400 - [{'error': {'code': 400, 'message': 'API key not valid. Please pass a valid API key.', ...
清理： ([], [])
$ python -m pytest -q
5 passed

调试示例

1. 我写的 429 重试从来没有执行过。 第一版在 _create 里自己做指数退避（2 秒、4 秒、8 秒）。测试时替身先返回 2 次 429，我断言自己的代码会等待 [2, 4]，实际的等待列表却是空的；连续 9 次 429 的测试也没有抛出异常，而且跑了 10 秒。我统计了替身收到的请求数：在 SDK 默认设置下，一次调用共发出 4 个请求，约 3.5 秒后才抛出 429。也就是说，SDK 的 Interactions 客户端自己会重试。我还试过 HttpRetryOptions(attempts=1)，仍然发出了 2 个请求，看来 Interactions 有自己独立的一层重试。最后我删掉了自己的重试，由 SDK 负责；如果 SDK 重试之后仍然是 429，就当作额度用完处理。

2. 测试替身的 reset() 参数冲突。 我调用 F.reset(rate_429=9) 时报错 dict.update() got multiple values for keyword argument 'rate_429'，原因是默认值和传入的参数同名，在同一个 update 调用里冲突了。改成先合并两个字典再更新。

常见问题

模板规则

• 默认使用 stateless 模式。
• 需要 stateful 时，会话结束一定要删除交互。
• 只保留一层重试。
• 不要在 Gemini 中处理个人敏感信息。

项目资源

用 Interactions API 时，要清楚对话内容保存在哪里、保存多久、怎么删除。