Gemini Interactions API：结构化提取会议待办

用 Gemini 的 Interactions API（官方结构化输出文档现在使用的接口）把会议纪要提取成待办事项列表。 官方文档有一句很关键的话：maxLength、minLength 等计数约束不会被约束解码执行，而是被自动去掉。所以 schema 只能保证 JSON 格式正确，内容是否满足约束还要在客户端用 Pydantic 再校验一次。 此外，状态不是 completed（例如 incomplete）时，不能使用这次的输出。

先看清成品

• 本地替身上：正常提取出 2 条待办；任务超过 60 字被客户端拦下；JSON 被截断时报“输出不符合约束”；状态为 incomplete 时拒绝使用。pytest 5 项通过。
• 用无效 Key 调用真实接口：✗ 调用失败（HTTP 400）： BadRequestError ... [{'error': {'code': 400, 'message': '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 自带测试中的数据结构。

做完之后你会得到什么？

需要准备哪些环境？

关键事实

文件结构

62-google-ai-A-src/
├── action_items.py
├── fake_gemini.py
├── notes.txt
└── test_action_items.py

第 1 步：提取脚本

"""会议纪要 → 待办事项（Gemini Interactions API 结构化输出）。
- 按官方文档：response_format = {"type": "text", "mime_type": "application/json", "schema": Model.model_json_schema()}
- 文档明确说明：maxLength、minLength 等计数约束**不会**被约束解码执行（会被自动去掉），
  所以必须在客户端用 Pydantic 再校验一次
- status 不是 completed（比如 incomplete、failed、budget_exceeded）时，不使用这次输出"""
import os, sys
from typing import Optional
from pydantic import BaseModel, Field, ValidationError

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

class Item(BaseModel):
    owner: str = Field(description="负责人姓名，原文没有写就填 待定")
    task: str = Field(max_length=60, description="一句话描述的任务，不超过 60 个字")
    due: Optional[str] = Field(default=None, pattern=r"^\d{4}-\d{2}-\d{2}$", description="截止日期 YYYY-MM-DD，没有就填 null")

class Items(BaseModel):
    items: list[Item]

PROMPT = "从下面的会议纪要中提取待办事项，只依据原文：\n\n{notes}"

class ExtractError(RuntimeError):
    pass

def extract(client, notes, model=MODEL):
    it = client.interactions.create(
        model=model, input=PROMPT.format(notes=notes), store=False,          # 一次性任务，不需要在服务端保存
        response_format={"type": "text", "mime_type": "application/json", "schema": Items.model_json_schema()})
    if it.status != "completed":
        raise ExtractError(f"状态为 {it.status}，不使用这次输出")
    try:
        return Items.model_validate_json(it.output_text)
    except ValidationError as e:
        raise ExtractError(f"输出不符合约束：{e.errors()[0]['loc']} {e.errors()[0]['msg']}") from e

if __name__ == "__main__":
    from google import genai
    client = genai.Client()                                                # 读取环境变量 GEMINI_API_KEY
    notes = open(sys.argv[1], encoding="utf-8").read()
    try:
        for i in extract(client, notes).items:
            print(f"- [{i.owner}] {i.task}（截止：{i.due or '未定'}）")
    except ExtractError as e:
        print("✗", e); sys.exit(1)
    except Exception as e:
        # Interactions 的错误类不是 google.genai.errors.APIError 的子类（SDK 2.32 实测），
        # 所以统一按属性读取 HTTP 状态码
        code = getattr(e, "status_code", None) or getattr(e, "code", None)
        print(f"✗ 调用失败（HTTP {code}）：", type(e).__name__, str(e)[:300]); sys.exit(2)

第 2 步：本地替身

"""本地替身（我自己写的，不是 Google 官方程序），用于离线测试 google-genai SDK。
格式来源：SDK 2.32.0 实际发出的请求（本地抓包），以及 SDK 自带测试中的 Interaction 结构；
错误 Key 的返回照搬真实接口：/v1beta/interactions 返回的是 JSON **数组** [{"error": ...}]，
generateContent / embed 返回的是对象 {"error": ...}。"""
import hashlib, json, math, re, threading, urllib.parse
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

STATE = {"requests": [], "deleted": [], "store": {}, "next_id": 0, "scenario": "ok", "rate_429": 0}
KEY_ERR = {"error": {"code": 400, "message": "API key not valid. Please pass a valid API key.", "status": "INVALID_ARGUMENT",
                     "details": [{"@type": "type.googleapis.com/google.rpc.ErrorInfo", "reason": "API_KEY_INVALID",
                                  "domain": "googleapis.com", "metadata": {"service": "generativelanguage.googleapis.com"}}]}}

def embed_text(text, dim):
    """确定性的“词袋哈希”向量，只用于测试：共享的词越多越相似。"""
    v = [0.0] * dim
    for w in re.findall(r"[a-z0-9]+|[\u4e00-\u9fff]", text.lower()):
        if w in {"task", "search", "result", "query", "title", "none", "text"}:
            continue
        v[int(hashlib.md5(w.encode()).hexdigest(), 16) % dim] += 1.0
    n = math.sqrt(sum(x * x for x in v)) or 1.0
    return [x / n for x in v]

def reply_for(body):
    """根据输入生成模型输出；scenario 控制几种异常情况。"""
    sc = STATE["scenario"]
    inp = body.get("input")
    last = inp if isinstance(inp, str) else next(
        (c["text"] for s in reversed(inp) if s.get("type") == "user_input" for c in s["content"] if c.get("type") == "text"), "")
    if body.get("response_format"):
        items = [{"owner": "李雷", "task": "整理需求文档并发给设计组", "due": "2026-10-28"},
                 {"owner": "韩梅梅", "task": "确认测试环境", "due": None}]
        if sc == "overlong":
            items[0]["task"] = "整理" * 60
        if sc == "bad_json":
            return "{\"items\": [ {\"owner\": \"李雷\""
        return json.dumps({"items": items}, ensure_ascii=False)
    if isinstance(inp, list):                       # 无状态：根据完整历史回答
        users = [c["text"] for s in inp if s.get("type") == "user_input" for c in s["content"]]
        return f"（看到了 {len(users)} 轮用户输入）你刚才说：{users[0]}" if len(users) > 1 else f"收到：{last}"
    prev = body.get("previous_interaction_id")
    if prev:
        first = STATE["store"].get(prev, {}).get("first_user")
        return f"你之前说：{first}" if first else "（找不到上一轮，可能已删除或未存储）"
    return f"收到：{last}"

class H(BaseHTTPRequestHandler):
    def log_message(self, *a):
        pass
    def _send(self, code, obj):
        raw = json.dumps(obj, ensure_ascii=False).encode()
        self.send_response(code); self.send_header("Content-Type", "application/json"); self.end_headers(); self.wfile.write(raw)
    def _auth(self, array):
        if self.headers.get("x-goog-api-key") == "good":
            return True
        self._send(400, [KEY_ERR] if array else KEY_ERR); return False
    def do_DELETE(self):
        if not self._auth(True):
            return
        iid = urllib.parse.urlsplit(self.path).path.rsplit("/", 1)[-1]
        if STATE["store"].pop(iid, None) is None:
            return self._send(404, [{"error": {"code": 404, "message": "not found", "status": "NOT_FOUND"}}])
        STATE["deleted"].append(iid); self._send(200, {})
    def do_POST(self):
        path = urllib.parse.urlsplit(self.path).path
        body = json.loads(self.rfile.read(int(self.headers["Content-Length"])))
        STATE["requests"].append((path, body))
        if path == "/v1beta/interactions":
            if not self._auth(True):
                return
            if STATE["rate_429"]:
                STATE["rate_429"] -= 1
                return self._send(429, [{"error": {"code": 429, "message": "Resource exhausted (fake)", "status": "RESOURCE_EXHAUSTED"}}])
            STATE["next_id"] += 1
            iid = f"v1_fake{STATE['next_id']}"
            status = "incomplete" if STATE["scenario"] == "incomplete" else "completed"
            steps = [{"type": "thought", "signature": "sig", "summary": [{"type": "text", "text": "思考摘要"}]},
                     {"type": "model_output", "content": [{"type": "text", "text": reply_for(body)}]}]
            if body.get("store", True):
                inp = body["input"]
                prev = STATE["store"].get(body.get("previous_interaction_id"), {})
                STATE["store"][iid] = {"first_user": prev.get("first_user") or (inp if isinstance(inp, str) else None)}
            return self._send(200, {"id": iid, "status": status, "steps": steps, "model": body.get("model")})
        if path.endswith(":batchEmbedContents"):
            if not self._auth(False):
                return
            out = []
            for r in body["requests"]:
                text = " ".join(p.get("text", "") for p in r["content"]["parts"])   # 多个 part 聚合成一个向量
                out.append({"values": embed_text(text, r.get("outputDimensionality", 3072))})
            return self._send(200, {"embeddings": out})
        self._send(404, {"error": {"code": 404, "message": f"no route {path}", "status": "NOT_FOUND"}})

def start():
    s = ThreadingHTTPServer(("127.0.0.1", 0), H)
    threading.Thread(target=s.serve_forever, daemon=True).start()
    return s, f"http://127.0.0.1:{s.server_address[1]}"

def reset(**kw):
    STATE.update({**dict(requests=[], deleted=[], store={}, next_id=0, scenario="ok", rate_429=0), **kw})

第 3 步：测试

import pytest
from google import genai
from google.genai import types
import action_items as A, 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()

NOTES = open("notes.txt", encoding="utf-8").read()

def test_ok_and_request_shape(client):
    r = A.extract(client, NOTES)
    assert [i.owner for i in r.items] == ["李雷", "韩梅梅"] and r.items[1].due is None
    path, body = F.STATE["requests"][-1]
    assert path == "/v1beta/interactions" and body["store"] is False
    assert body["response_format"]["schema"]["properties"]["items"]["type"] == "array"

def test_overlong_rejected_client_side(client):
    F.STATE["scenario"] = "overlong"
    with pytest.raises(A.ExtractError, match="task"):
        A.extract(client, NOTES)

def test_truncated_json(client):
    F.STATE["scenario"] = "bad_json"
    with pytest.raises(A.ExtractError, match="不符合约束"):
        A.extract(client, NOTES)

def test_incomplete_status(client):
    F.STATE["scenario"] = "incomplete"
    with pytest.raises(A.ExtractError, match="incomplete"):
        A.extract(client, NOTES)

def test_bad_key_error_from_array_body(client):
    bad = genai.Client(api_key="bad", http_options=types.HttpOptions(base_url=client._api_client._http_options.base_url))
    with pytest.raises(Exception) as e:
        A.extract(bad, NOTES)
    assert "API key not valid" in str(e.value)

第 4 步：运行

$ GEMINI_API_KEY=INVALID python action_items.py notes.txt
✗ 调用失败（HTTP 400）： BadRequestError Error code: 400 - [{'error': {'code': 400, 'message': 'API key not valid. Please pass a valid API key.', 'status': 'INVALID_ARGUMENT', ...
$ python -m pytest -q
5 passed

换成有效的 Key（并且在支持的地区）后，会输出 - [李雷] 整理需求文档并发给设计组（截止：2026-10-28） 这样的列表（这一步我没有运行）。

调试示例

1. 临时创建的 Client 被提前关闭了。 我在一行里写了 genai.Client(api_key=...).interactions.create(...)，得到的错误不是 Key 无效，而是 RuntimeError('Cannot send a request, as the client has been closed.')。原因是临时的 Client 在请求发出前就被回收了。把 Client 保存到变量里之后，才拿到真实的 400 错误。

2. errors.APIError 捕获不到这个错误。 用同一个 Client 实测：generate_content 的错误能被 google.genai.errors.APIError 捕获（code 400，status INVALID_ARGUMENT），但 interactions.create 抛出的是 google.genai._gaos.lib.compat_errors.BadRequestError，不是 APIError 的子类。错误体的格式也不一样，前者是对象，后者是数组。因为不想导入以下划线开头的私有模块，代码里改为通过 status_code 或 code 属性读取 HTTP 状态码。

常见问题

模板规则

• 用 schema 约束 JSON 格式，用 Pydantic 校验内容。
• 只使用 completed 状态的输出。
• 一次性任务设置 store=False。
• 错误统一按 HTTP 状态码处理。

项目资源

格式靠 schema 约束，内容对不对要靠客户端校验。