客户端、Parts 与多模态输入

同一个 SDK，不同认证入口

本书选择 Gemini Developer API，读取 GEMINI_API_KEY；Vertex AI 使用项目、区域和 Google Cloud 凭证，不应把 API key 例子直接当作生产云身份方案。两种入口支持能力和区域条件需要分别核对。

在空目录用 Python 3.12 建虚拟环境，安装现代 google-genai 包。旧 google-generativeai 的 GenerativeModel 是另一套接口，不能混用导入。

python3.12 -m venv .venv
source .venv/bin/activate
python -m pip install 'google-genai==1.75.0'
python -m pip freeze > requirements.lock

设置 GEMINI_API_KEY、GEMINI_MODEL，保存 text.py 并运行 python text.py。这是计费/配额相关的在线实验，本次未调用真实模型。

import os
from google import genai
from google.genai import types
with genai.Client(api_key=os.environ['GEMINI_API_KEY']) as client:
    r=client.models.generate_content(model=os.environ['GEMINI_MODEL'],
        contents=[types.Content(role='user',parts=[types.Part.from_text(text='解释 HTTP 状态码 429')])],
        config=types.GenerateContentConfig(max_output_tokens=300))
    print(r.text)
    print(r.prompt_feedback)
    for c in r.candidates or []:
        print(c.finish_reason)
    print(r.usage_metadata)

Content 组织角色，Part 表示文本、图像、函数调用等片段。r.text 是便利提取，不保证每个响应都有文字；安全拦截、工具意图或非文本输出都可能没有文本。先检查候选与 finish_reason，再决定是否可进入业务处理。

内联图像与文件服务

将自己有权上传的 sample.png 放在目录中，下面完整代码保存 image.py：

import os
from pathlib import Path
from google import genai
from google.genai import types
with genai.Client(api_key=os.environ['GEMINI_API_KEY']) as c:
    image=types.Part.from_bytes(data=Path('sample.png').read_bytes(),mime_type='image/png')
    r=c.models.generate_content(model=os.environ['GEMINI_MODEL'],
        contents=['描述图中的可见物体',image])
    print(r.text)

小文件可内联，反复使用或较大文件可走 Files API：uploaded=c.files.upload(file='sample.png') 后把 uploaded 放进 contents；处理结束调用 c.files.delete(name=uploaded.name)，并用 try/finally 保证清理。音视频文件可能先处于处理状态，应查询状态到可用后再生成，FAILED 时不要持续等候。上传对象不会永久有效，也不能跨认证入口随意复用。

多模态预算不等于字符串长度：图像分辨率、音视频时长、文件处理方式都会改变计算。长上下文只是容纳更多材料，不能保证准确定位；给文件编号、问题目标与证据位置，保留原材料供核对。

参阅 Gen AI SDK 与 文件输入。

文件处理的等待、失败与清理

保存files.py。同一Developer API客户端上传文件后，只有ACTIVE才可以生成；PROCESSING继续查询，FAILED直接结束。wait_file是完整可导入函数，sleep注入便于离线验证，不需要为了测试等待真实时间。

import os,time
from google import genai
from google.genai import types

def wait_file(files,item,attempts=6,sleep=time.sleep):
    if type(attempts) is not int or attempts<0:raise ValueError('invalid query budget')
    for query_count in range(attempts+1):
        if item.state==types.FileState.ACTIVE:return item
        if item.state!=types.FileState.PROCESSING:raise ValueError('file not usable')
        if query_count==attempts:raise TimeoutError('file processing budget exhausted')
        sleep(1)
        item=files.get(name=item.name)

if __name__=='__main__':
    with genai.Client(api_key=os.environ['GEMINI_API_KEY'],http_options=types.HttpOptions(timeout=30000)) as c:
        uploaded=c.files.upload(file='sample.mp4',config=types.UploadFileConfig(mime_type='video/mp4'))
        try:
            ready=wait_file(c.files,uploaded)
            r=c.models.generate_content(model=os.environ['GEMINI_MODEL'],contents=[ready,'概括可见场景并给时间位置'])
            if not r.candidates or r.candidates[0].finish_reason!=types.FinishReason.STOP:
                raise RuntimeError('video answer incomplete')
            print(r.text)
        finally:c.files.delete(name=uploaded.name)

该在线入口需授权的sample.mp4、支持视频的模型和账户配额，本次未上传。attempts 限定 files.get 查询次数：先检查上传返回的初始状态，每次查询后立即检查，包括最后一次；0 表示只检查初始状态。它不是严格总墙钟时间；每次HTTP也有超时。异常时finally清理文件，如果删除失败应进入清理任务记录，不把原业务失败改称成功。

以下check_file.py离线使用真实types.File对象，只替换文件服务的返回值。它检验等待策略，不代表远端转码性能。

from google.genai import types
from files import wait_file
class Files:
    def __init__(self,state):self.state=state;self.calls=0
    def get(self,name):
        self.calls+=1
        return types.File(name=name,state=self.state)
start=types.File(name='files/demo',state='PROCESSING')
assert wait_file(Files('ACTIVE'),start,sleep=lambda _:None).state==types.FileState.ACTIVE
for state,error in [('FAILED',ValueError),('PROCESSING',TimeoutError)]:
    service=Files(state)
    try:wait_file(service,start,attempts=1,sleep=lambda _:None)
    except error:pass
    else:raise AssertionError('unusable file accepted')
service=Files('ACTIVE')
failed=types.File(name='files/demo',state='FAILED')
try:wait_file(service,failed,attempts=1,sleep=lambda _:None)
except ValueError:assert service.calls==0
else:raise AssertionError('initial failure ignored')
service=Files('ACTIVE')
assert wait_file(service,start,attempts=1,sleep=lambda _:None).state==types.FileState.ACTIVE
assert service.calls==1
print('file readiness boundary passed')

练习

离线编写候选判定：没有 candidates 返回 no_candidate；首候选 finish_reason 为 STOP 且有文字返回 text；其余返回 inspect。对空列表和 MAX_TOKENS 各写一个断言。

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