CoderPath

Interactions：服务端状态、工具回填与流式完成 ​

上一节在本地保存完整 contents。对话增长后，应用要区分“我要继续哪次运行”和“我要提交哪些新输入”。Interactions 用 previous_interaction_id 指向服务端保存的前一次交互，新的 input 只携带本轮输入。这改变了状态的归属，也要求应用保留交互身份、工具调用身份和完成状态。

本节固定 Python 3.12、google-genai==1.75.0，沿用本书环境。此版本 SDK 将 Interactions 标为实验性接口；下面明确使用其 outputs 和事件形状，升级时应重新验证序列化及终止规则。先运行不联网的真实 SDK 实验，再配置有权限的模型运行在线入口。

两个 ID 解决两种关联 ​

用户同时查询 A、B 库存时，一次交互可返回两个 function_call。交互的 id 用来继续对话，每个调用的 id 则要分别写入 function_result.call_id。只返回最后一个调用的结果，会让另一项请求悬而未决；按函数名关联也不够，因为两项都叫 stock。

requires_action 表示应用必须先处理工具，不能把现有文本当最终结果。下面先验证整批调用，再执行只读库存查询，避免执行到一半才发现非法参数。零库存与没有该商品分别表示为 found=True,count=0 和 found=False,count=None。

完成事件不等于完整文本快照 ​

流式事件的 content.delta 携带增量，index 表示输出片段的位置。应用按 index 累积文本；interaction.complete 确認完成状态，它携带的 interaction 可以没有完整 outputs，因此不能在最后一步丢弃累积结果再读取最终文本。连接直接结束只证明 EOF，不证明模型完成。

下面适配器只接收文本流，工具流会拒绝并要求进入专用工具处理流程。它限制累计文本长度，并明确拒绝失败、取消、未完成和需要工具的终止状态。生产传输还要有请求超时；字符上限不能中断尚未到达的事件。

保存为 interactions_lab.py，运行 python interactions_lab.py。MockTransport 替换 HTTP 服务端，客户端的请求序列化、响应对象和 SSE 解析仍由真实 SDK 执行；无需密钥，也不会调用模型。

python
import json
import httpx
from google import genai
from google.genai import types

TOOLS=[{'type':'function','name':'stock','description':'Read inventory only',
        'parameters':{'type':'object','properties':{'sku':{'type':'string'}},
                      'required':['sku'],'additionalProperties':False}}]

def results_for(interaction):
    if interaction.status!='requires_action':raise ValueError('not waiting for tools')
    calls=[o for o in interaction.outputs or [] if o.type=='function_call']
    if not calls or len(calls)>4 or len({o.id for o in calls})!=len(calls):
        raise ValueError('missing, excessive or duplicate calls')
    # Validate the entire batch before any execution; these tools are read-only.
    for call in calls:
        if call.name!='stock' or not isinstance(call.arguments,dict) or set(call.arguments)!={'sku'}:
            raise ValueError('invalid tool')
        if type(call.arguments['sku']) is not str:raise ValueError('invalid sku')
    inventory={'A':3,'B':0}
    return [{'type':'function_result','name':call.name,'call_id':call.id,
             'result':{'sku':call.arguments['sku'],'count':inventory.get(call.arguments['sku']),
                       'found':call.arguments['sku'] in inventory}} for call in calls]

def complete_text(interaction):
    if interaction.status!='completed':raise ValueError('not completed')
    if any(o.type=='function_call' for o in interaction.outputs or []):raise ValueError('unhandled tool')
    return ''.join(o.text for o in interaction.outputs or [] if o.type=='text')

def text_stream(events):
    parts={};ended=False
    for event in events:
        if ended:raise ValueError('event after terminal')
        kind=event.event_type
        if kind=='error':raise RuntimeError('provider error')
        if kind=='content.delta':
            if event.delta.type!='text':raise ValueError('text-only stream adapter')
            parts[event.index]=parts.get(event.index,'')+event.delta.text
            if sum(map(len,parts.values()))>4000:raise ValueError('text budget')
        elif kind=='content.start':
            if event.content.type!='text':raise ValueError('text-only stream adapter')
        elif kind=='interaction.complete':
            if event.interaction.status!='completed':raise RuntimeError(event.interaction.status)
            ended=True
        elif kind=='interaction.status_update':
            if event.status in {'failed','cancelled','incomplete','requires_action'}:
                raise RuntimeError(event.status)
        elif kind not in {'interaction.start','content.stop'}:
            raise ValueError('unknown event')
    if not ended:raise RuntimeError('stream ended without completion')
    return ''.join(parts[i] for i in sorted(parts))

def record(id,status,outputs):
    return {'id':id,'status':status,'outputs':outputs,
            'created':'2026-01-01T00:00:00Z','updated':'2026-01-01T00:00:00Z'}

def stream_client(events):
    def respond(request):
        body=json.loads(request.content);assert body['stream'] is True
        wire=''.join('event: '+e['event_type']+'\ndata: '+json.dumps(e)+'\n\n' for e in events)
        return httpx.Response(200,headers={'content-type':'text/event-stream'},text=wire)
    return genai.Client(api_key='fixture-only',http_options=types.HttpOptions(
        async_client_args={"trust_env":False}, httpx_client=httpx.Client(transport=httpx.MockTransport(respond),trust_env=False)))

def main():
    requests=[]
    def respond(request):
        body=json.loads(request.content);requests.append(body)
        assert request.url.path.endswith('/interactions') and body['store'] is True
        if len(requests)==1:
            assert body['tools'][0]['name']=='stock'
            outputs=[{'type':'function_call','id':'call-'+sku,'name':'stock','arguments':{'sku':sku}}
                     for sku in ('A','B')]
            result=record('i1','requires_action',outputs)
        else:
            assert body['previous_interaction_id']=='i1'
            assert [x['call_id'] for x in body['input']]==['call-A','call-B']
            assert body['input'][1]['result']=={'sku':'B','count':0,'found':True}
            result=record('i2','completed',[{'type':'text','text':'A=3, B=0'}])
        return httpx.Response(200,json=result)
    with genai.Client(api_key='fixture-only',http_options=types.HttpOptions(
            async_client_args={"trust_env":False}, httpx_client=httpx.Client(transport=httpx.MockTransport(respond),trust_env=False))) as client:
        first=client.interactions.create(model='fixture-model',input='A and B inventory',tools=TOOLS,store=True)
        second=client.interactions.create(model='fixture-model',previous_interaction_id=first.id,
                 input=results_for(first),tools=TOOLS,store=True)
        assert complete_text(second)=='A=3, B=0' and len(requests)==2
    delta={'event_type':'content.delta','index':0,'delta':{'type':'text','text':'OK'}}
    final={'event_type':'interaction.complete','interaction':record('s1','completed',[])}
    with stream_client([delta,final]) as client:
        with client.interactions.create(model='fixture-model',input='hello',stream=True) as events:
            assert text_stream(events)=='OK'
    for frames in ([delta],[delta,{**final,'interaction':record('s1','failed',[])}]):
        with stream_client(frames) as client:
            with client.interactions.create(model='fixture-model',input='hello',stream=True) as events:
                try:text_stream(events)
                except RuntimeError:pass
                else:raise AssertionError('partial/failed accepted')
    print('two tool IDs, stored follow-up, SSE completion and failure passed')
if __name__=='__main__':main()

输出 two tool IDs, stored follow-up, SSE completion and failure passed。先观察请求一返回 i1 和两个调用 ID，再看请求二的 previous_interaction_id=i1、两个 call_id 及 B 的零库存；流式部分用同一段 OK 文本分别配合成功事件、直接断流和失败事件，只有第一种能返回结果。HTTP fixture 证明协议处理，不证明模型会稳定选择 stock。

接到真实服务时保留循环边界 ​

保存下面入口为 online_interactions.py，与上一文件同目录。配置 GEMINI_API_KEY 和当前账号可使用、支持 Interactions 的 GEMINI_INTERACTIONS_MODEL 后运行。请求会产生服务调用费用；本节的离线实验不需要执行它。

python
import os
from google import genai
from google.genai import types
from interactions_lab import TOOLS, results_for, complete_text

with genai.Client(api_key=os.environ['GEMINI_API_KEY'],
                  http_options=types.HttpOptions(timeout=30000)) as client:
    request={'model':os.environ['GEMINI_INTERACTIONS_MODEL'],
             'input':'查询 A 和 B 的库存','tools':TOOLS,'store':True}
    for _ in range(4):
        current=client.interactions.create(**request)
        if current.status=='completed':
            print(complete_text(current))
            break
        if current.status!='requires_action':
            raise RuntimeError(f'交互尚未成功完成: {current.status}')
        request={'model':request['model'],'previous_interaction_id':current.id,
                 'input':results_for(current),'tools':TOOLS,'store':True}
    else:
        raise RuntimeError('交互步数达到上限')

这里选择 store=True，使下一轮能够引用服务端状态；交互 ID 本身不是用户授权凭证。应用仍须把它绑定到本地会话和拥有者，不能允许用户提交另一个租户的 ID。存储、保留与删除策略应按部署所用服务的约定配置，不应从一个布尔值推断零留存。in_progress 若出现在异步任务中，需要另写有截止时间的查询流程；这个短入口会明确退出，不会把它当成功，也不进行无限轮询。

独立练习 ​

  1. 用实际响应类型构造两个同 ID 调用，以及 sku=True 调用，验证整批拒绝；再查询 X，验证未知与零库存可区分。
  2. 先收到 index=1 的 B，再收到 index=0 的 A，完成后应得到 AB；改成 cancelled 或 requires_action，验证不能提交这些草稿。

实现可复用正文函数，但先自己写输入与断言，再核对Interactions 答案。

读完继续缓存、流控与治理。接口的正式参数与支持范围参见 Interactions API 和 Python SDK Interactions 示例。

用心记录每一步成长

Copyright © 2026-present Twiink