结构化结果与显式 Function Calling

先抽取，再决定是否执行

候选的文本或函数参数可能已经完整，但服务仍因限额或安全策略终止。先保存以下 gemini_guard.py，同目录的两个在线文件都导入它；运行 python gemini_guard.py 不联网，使用真实 SDK 响应类型反证“能解析就可以执行”。

from google.genai import types

def accepted(response):
    feedback=response.prompt_feedback
    if feedback and feedback.block_reason not in (None,types.BlockedReason.BLOCKED_REASON_UNSPECIFIED):
        raise ValueError('prompt blocked')
    if not response.candidates or len(response.candidates)!=1:
        raise ValueError('expected one candidate')
    candidate=response.candidates[0]
    if candidate.finish_reason!=types.FinishReason.STOP or not candidate.content:
        raise ValueError('candidate incomplete or blocked')
    return candidate

if __name__=='__main__':
    effects=[]
    for reason in (types.FinishReason.MAX_TOKENS,types.FinishReason.SAFETY):
        r=types.GenerateContentResponse(candidates=[types.Candidate(
            finish_reason=reason,content=types.Content(role='model',parts=[
                types.Part.from_text(text='{"sku":"A","count":2}'),
                types.Part(function_call=types.FunctionCall(name='stock',args={'sku':'A'}))]))])
        try:
            accepted(r)
            effects.append('business executed')
        except ValueError:pass
    assert effects==[]
    blocked=types.GenerateContentResponse(prompt_feedback=types.GenerateContentResponsePromptFeedback(
        block_reason=types.BlockedReason.SAFETY))
    try:accepted(blocked)
    except ValueError:pass
    else:raise AssertionError('blocked prompt accepted')
    ok=types.GenerateContentResponse(candidates=[types.Candidate(finish_reason='STOP',
        content=types.Content(role='model',parts=[types.Part.from_text(text='ok')]))])
    assert accepted(ok).content.parts[0].text=='ok'
    print('normal, truncated and blocked gates passed')

这是一道宿主提交门，不预测模型质量。安全终态不自动重试同一危险请求；截断可重新设计输出预算，但旧候选不能执行。

结构化输出用 schema 描述业务数据。沿用上一章环境，保存 structured.py：

import os,json
from gemini_guard import accepted
from google import genai
from google.genai import types
schema={'type':'object','properties':{'count':{'type':'integer'},'sku':{'type':'string'}},
        'required':['count','sku'],'additionalProperties':False}
with genai.Client(api_key=os.environ['GEMINI_API_KEY']) as c:
    r=c.models.generate_content(model=os.environ['GEMINI_MODEL'],contents='要 2 本 A 图书',
        config=types.GenerateContentConfig(response_mime_type='application/json',
                                            response_json_schema=schema))
    accepted(r)
    if not r.text:
        raise RuntimeError('no structured text')
    data=json.loads(r.text)
    if type(data['count']) is not int or not 1<=data['count']<=5:
        raise ValueError('count outside business bounds')
    print(data)

原生 schema 约束与 JSON 解析都不能授权购买。完整业务还要检查候选终态、商品存在、身份与预算。为未知字段设置明确 null 分支，不能为了通过 schema 让模型编造 SKU。

不让自动执行隐藏控制权

Google SDK 可以自动执行传入的 Python 函数。教学先关闭自动函数调用，显式观察每一步。以下 tool.py 是完整在线例子，只查询静态库存，不写外部系统。

import os
from gemini_guard import accepted
from google import genai
from google.genai import types
def stock(sku: str) -> dict:
    """读取商品库存，未知商品返回 null。"""
    if not isinstance(sku,str):raise ValueError('sku must be string')
    return {'sku':sku,'count':{'A':3}.get(sku)}
config=types.GenerateContentConfig(tools=[stock],
    automatic_function_calling=types.AutomaticFunctionCallingConfig(disable=True))
contents=[types.Content(role='user',parts=[types.Part.from_text(text='查询 A 的库存')])]
with genai.Client(api_key=os.environ['GEMINI_API_KEY']) as c:
    for _ in range(4):
        r=c.models.generate_content(model=os.environ['GEMINI_MODEL'],contents=contents,config=config)
        candidate=accepted(r)
        contents.append(candidate.content)
        calls=r.function_calls or []
        if not calls:
            if candidate.finish_reason != types.FinishReason.STOP:
                raise RuntimeError(str(candidate.finish_reason))
            print(r.text)
            break
        parts=[]
        for call in calls:
            if call.name!='stock' or set(call.args)!={'sku'}:
                raise ValueError('unknown tool or arguments')
            parts.append(types.Part(function_response=types.FunctionResponse(
                id=call.id,name=call.name,response=stock(**call.args))))
        contents.append(types.Content(role='tool',parts=parts))
    else:raise RuntimeError('tool step limit')

保存模型返回的完整 content，而非重建只含函数名的片段，才能保留模型/版本所需的附加字段，例如 thought signature。对包含调用 ID 的协议版本还应保留对应身份；不要假设同名工具只会调用一次。SDK 自动调用仅适合允许直接执行的函数，写操作仍应在业务边界阻断。

流式生成用 generate_content_stream，逐块文本可以展示，但结构化 JSON 和函数参数需等完整内容后校验。网络断开后的已有文字只能作为草稿；不能用“读到了最后一个非空 chunk”冒充完成。

参阅 结构化输出 与 函数调用。

练习

给库存函数增加 B=0，保持未知商品为 None；写三个断言覆盖 A、B、未知商品。说明为什么不能用 value or "unknown" 映射结果。

先记录自己的输出与解释，再核对本章答案。

下一节进入 Interactions 的状态与事件，比较服务端交互身份与本地 contents 的维护方式。