Gemini 2.5 Pro 多模态AI完全教程
适用人群：AI开发者、企业技术团队、全栈工程师
前置要求：Python基础、了解REST API概念
预计学习时间：8-12小时
目录
- Gemini 2.5 Pro 模型概览
- 环境搭建与API入门
- 百万Token上下文窗口
- 多模态理解：图像
- 多模态理解：视频
- 多模态理解：音频
- 代码生成与执行
- Function Calling 函数调用
- Google Search Grounding
- Gems 自定义助手
- Vertex AI 企业部署
- API最佳实践与性能优化
- 实战项目：多模态文档分析系统
- 常见问题与解决方案
1. Gemini 2.5 Pro 模型概览
1.1 Gemini 模型家族
Gemini 是 Google DeepMind 推出的多模态AI模型家族，2.5 Pro 是目前最强大的版本。
| 模型 | 定位 | 核心优势 |
|---|---|---|
| Gemini 2.5 Pro | 旗舰模型 | 最强推理、原生多模态、100万Token上下文 |
| Gemini 2.5 Flash | 快速模型 | 速度优先、成本优化 |
| Gemini 2.0 Flash | 上一代快速模型 | 成熟稳定、广泛支持 |
1.2 Gemini 2.5 Pro 核心能力
- 原生多模态：从训练阶段就支持文本、图像、视频、音频的理解
- 超长上下文：支持 100万 tokens（约70万字），可处理数小时的视频或数千页文档
- 深度推理：支持"思考"模式，在复杂推理任务上表现卓越
- 代码能力：强大的代码生成、理解和执行能力
- 工具调用：原生支持 Function Calling 和 Google Search Grounding
- 结构化输出：原生支持JSON模式，便于程序化处理
1.3 与其他模型的对比
能力维度 GPT-4o Claude 4 Opus Gemini 2.5 Pro
────────────────────────────────────────────────────────────────
多模态（图像） ✅ ✅ ✅
多模态（视频） 有限 有限 ✅（原生）
多模态（音频） ✅ 不支持 ✅（原生）
上下文长度 128K 200K 1,000K（100万）
搜索集成 需插件 不支持 ✅ Search Grounding
代码执行 需环境 不支持 ✅ 原生支持
结构化输出 ✅ 需提示 ✅ 原生JSON模式
价格（Pro级别） $$$ $$$$ $$
2. 环境搭建与API入门
2.1 获取API密钥
- 访问 Google AI Studio
- 使用Google账号登录
- 点击 "Get API Key" 创建密钥
- 妥善保存密钥
企业用户：可通过 Google Cloud Console 创建项目并启用 Vertex AI API。
2.2 Python SDK 安装
# 创建虚拟环境
python -m venv gemini-env
source gemini-env/bin/activate # Linux/macOS
# 安装 Google Generative AI SDK
pip install google-genai
# 验证安装
python -c "import google.genai; print('SDK安装成功')"
2.3 第一次API调用
from google import genai
# 初始化客户端（自动读取 GEMINI_API_KEY 环境变量）
client = genai.Client()
# 基础对话
response = client.models.generate_content(
model="gemini-2.5-pro",
contents="用一句话解释什么是多模态AI？"
)
print(response.text)
2.4 使用环境变量管理密钥
# 设置环境变量
export GEMINI_API_KEY="your-api-key-here"
# 代码中自动读取
from google import genai
client = genai.Client() # 自动读取 GEMINI_API_KEY
2.5 流式输出
from google import genai
client = genai.Client()
# 流式输出
for chunk in client.models.generate_content_stream(
model="gemini-2.5-pro",
contents="写一首关于人工智能的现代诗"
):
print(chunk.text, end="", flush=True)
print()
2.6 多轮对话（Chat）
from google import genai
client = genai.Client()
def chat():
# 创建聊天会话
chat = client.chats.create(model="gemini-2.5-pro")
print("Gemini 2.5 Pro 对话助手（输入 'quit' 退出）")
print("-" * 40)
while True:
user_input = input("\n你: ")
if user_input.lower() == 'quit':
break
response = chat.send_message(user_input)
print(f"\nGemini: {response.text}")
if __name__ == "__main__":
chat()
2.7 配置生成参数
from google import genai
from google.genai import types
client = genai.Client()
# 使用配置参数
response = client.models.generate_content(
model="gemini-2.5-pro",
contents="解释量子计算的基本原理",
config=types.GenerateContentConfig(
temperature=0.7, # 创造性（0.0-2.0）
top_p=0.95, # 核采样
top_k=40, # Top-K采样
max_output_tokens=4096, # 最大输出长度
system_instruction="你是一个物理学教授，请用通俗易懂的语言解释概念。"
)
)
print(response.text)
3. 百万Token上下文窗口
3.1 百万Token意味着什么
Gemini 2.5 Pro 支持 100万 tokens 的上下文窗口，这意味着可以一次性处理：
- 约 700,000 个中文字
- 约 1,000,000 个英文单词
- 约 30,000 行代码
- 约 1 小时的视频
- 约 11 小时的音频
- 整本《战争与和平》（多次）
3.2 超长文档处理
from google import genai
from google.genai import types
client = genai.Client()
def analyze_long_document(file_path: str, question: str):
"""分析超长文档"""
# 读取文档
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
print(f"文档大小: {len(content)} 字符")
# Gemini可以直接处理超长文本
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""以下是一份完整文档：
<document>
{content}
</document>
请回答以下问题：
{question}
要求：
1. 直接引用文档中的相关段落
2. 给出结构化的分析
3. 如果文档中没有相关信息，明确说明""",
config=types.GenerateContentConfig(
max_output_tokens=8192
)
)
return response.text
# 使用示例
result = analyze_long_document("technical_spec.txt", "这份技术规格书的核心要求是什么？")
print(result)
3.3 大规模代码库分析
from google import genai
from google.genai import types
import os
import glob
client = genai.Client()
def analyze_codebase(directory: str, question: str):
"""分析大规模代码库"""
# 收集代码文件
extensions = ['.py', '.js', '.ts', '.java', '.go', '.rs', '.cpp', '.c', '.h']
code_files = []
for ext in extensions:
code_files.extend(glob.glob(os.path.join(directory, f"**/*{ext}"), recursive=True))
# 构建代码上下文
code_context = ""
total_chars = 0
for file_path in code_files:
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
relative_path = os.path.relpath(file_path, directory)
file_entry = f"\n--- {relative_path} ---\n{content}\n"
# 检查是否超过限制（留出余量给输出）
if total_chars + len(file_entry) > 3500000: # 约100万tokens
print(f"达到上下文限制，已加载 {len(code_files)} 个文件中的部分")
break
code_context += file_entry
total_chars += len(file_entry)
except (UnicodeDecodeError, PermissionError):
continue
print(f"已加载 {len(code_files)} 个文件，共 {total_chars} 字符")
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""请分析以下代码仓库：
{code_context}
问题：{question}
请提供详细的结构化分析。""",
config=types.GenerateContentConfig(
max_output_tokens=8192
)
)
return response.text
# 使用示例
result = analyze_codebase("./my-project", "分析项目的架构设计，识别潜在的技术债务")
print(result)
3.4 多文档对比分析
from google import genai
from google.genai import types
client = genai.Client()
def compare_documents(doc_paths: list, comparison_aspects: str):
"""对比分析多个文档"""
docs_content = []
for i, path in enumerate(doc_paths):
with open(path, 'r', encoding='utf-8') as f:
content = f.read()
docs_content.append(f"<document_{i+1} name='{os.path.basename(path)}'>\n{content}\n</document_{i+1}>")
all_docs = "\n\n".join(docs_content)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""以下是{len(doc_paths)}份文档：
{all_docs}
请从以下方面进行对比分析：
{comparison_aspects}
请以表格或结构化格式呈现对比结果。""",
config=types.GenerateContentConfig(
max_output_tokens=8192
)
)
return response.text
import os
result = compare_documents(
["spec_v1.txt", "spec_v2.txt", "spec_v3.txt"],
"1. 功能差异 2. 性能要求变化 3. API接口变化 4. 新增特性"
)
4. 多模态理解：图像
4.1 图像分析基础
Gemini 2.5 Pro 原生支持图像理解，无需额外的视觉模型。
from google import genai
from google.genai import types
from PIL import Image
client = genai.Client()
def analyze_image(image_path: str, question: str):
"""分析图像内容"""
# 使用PIL加载图像
image = Image.open(image_path)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[image, question]
)
return response.text
# 使用示例
result = analyze_image("photo.jpg", "描述这张图片的内容，包括场景、人物、动作等细节。")
print(result)
4.2 使用文件API上传图像
from google import genai
from google.genai import types
import pathlib
client = genai.Client()
def analyze_with_file_api(image_path: str, question: str):
"""使用文件API分析图像（适合大图像）"""
# 上传图像文件
file = client.files.upload(file=image_path)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Content(
role="user",
parts=[
types.Part.from_uri(
file_uri=file.uri,
mime_type=file.mime_type
),
types.Part.from_text(text=question)
]
)
]
)
return response.text
# 使用示例
result = analyze_with_file_api("architecture_diagram.png", "解读这个系统架构图，说明各组件的职责和交互关系。")
print(result)
4.3 图表数据提取
from google import genai
from google.genai import types
import json
client = genai.Client()
def extract_chart_data(image_path: str):
"""从图表中提取数据"""
image = Image.open(image_path)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
image,
"""请从这张图表中提取所有数据，以JSON格式返回。
返回格式：
{
"chart_type": "图表类型",
"title": "图表标题",
"x_axis": {"label": "X轴标签", "values": [...]},
"y_axis": {"label": "Y轴标签", "values": [...]},
"data_series": [
{"name": "系列名", "values": [...]}
],
"key_insights": ["洞察1", "洞察2"]
}"""
],
config=types.GenerateContentConfig(
response_mime_type="application/json"
)
)
return json.loads(response.text)
# 使用示例
data = extract_chart_data("sales_chart.png")
print(json.dumps(data, ensure_ascii=False, indent=2))
4.4 多图对比分析
from google import genai
from google.genai import types
from PIL import Image
client = genai.Client()
def compare_images(image_paths: list, question: str):
"""对比分析多张图像"""
content = []
for path in image_paths:
content.append(Image.open(path))
content.append(question)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=content
)
return response.text
# 使用示例
result = compare_images(
["before.png", "after.png"],
"对比这两张图片，详细说明发生了哪些变化。"
)
print(result)
4.5 OCR与文档理解
from google import genai
from google.genai import types
import json
client = genai.Client()
def extract_document_text(image_path: str, output_format: str = "text"):
"""从文档图像中提取文本"""
image = Image.open(image_path)
if output_format == "json":
prompt = """请从这张文档图片中提取所有文字内容，并以JSON格式返回：
{
"title": "文档标题",
"sections": [
{"heading": "章节标题", "content": "章节内容"}
],
"tables": [
{"headers": [...], "rows": [[...], ...]}
]
}"""
config = types.GenerateContentConfig(response_mime_type="application/json")
else:
prompt = "请提取这张文档图片中的所有文字内容，保持原始格式。"
config = None
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[image, prompt],
config=config
)
return response.text if output_format == "text" else json.loads(response.text)
# 使用示例
text = extract_document_text("receipt.png")
print(text)
5. 多模态理解：视频
5.1 视频理解概述
Gemini 2.5 Pro 的视频理解是其独特优势之一。它可以分析视频内容、理解动作、提取文字、生成描述。
5.2 使用文件API处理视频
from google import genai
from google.genai import types
import time
client = genai.Client()
def analyze_video(video_path: str, question: str):
"""分析视频内容"""
print("上传视频文件...")
video_file = client.files.upload(file=video_path)
print(f"上传完成，等待处理: {video_file.name}")
# 等待视频处理完成
while video_file.state.name == "PROCESSING":
time.sleep(5)
video_file = client.files.get(name=video_file.name)
print(f" 状态: {video_file.state.name}")
if video_file.state.name == "FAILED":
raise ValueError(f"视频处理失败: {video_file.state.name}")
print("视频处理完成，开始分析...")
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Content(
role="user",
parts=[
types.Part.from_uri(
file_uri=video_file.uri,
mime_type=video_file.mime_type
),
types.Part.from_text(text=question)
]
)
],
config=types.GenerateContentConfig(
max_output_tokens=8192
)
)
return response.text
# 使用示例
result = analyze_video(
"presentation.mp4",
"请详细描述这个演示文稿视频的内容，包括每一页幻灯片的要点和演讲者的讲解要点。"
)
print(result)
5.3 视频内容摘要
from google import genai
from google.genai import types
client = genai.Client()
def generate_video_summary(video_path: str, language: str = "中文"):
"""生成视频摘要"""
video_file = client.files.upload(file=video_path)
# 等待处理
while video_file.state.name == "PROCESSING":
time.sleep(5)
video_file = client.files.get(name=video_file.name)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Part.from_uri(file_uri=video_file.uri, mime_type=video_file.mime_type),
f"""请为这个视频生成一份详细摘要（使用{language}）：
1. **视频概述**：整体内容概括
2. **时间线**：按时间顺序列出主要内容节点
3. **关键要点**：最重要的5-10个信息点
4. **视觉元素**：重要的图表、演示、画面
5. **结论**：视频的核心结论或行动建议"""
]
)
return response.text
# 使用示例
summary = generate_video_summary("lecture.mp4")
print(summary)
5.4 视频中的特定片段分析
from google import genai
from google.genai import types
client = genai.Client()
def analyze_video_segment(video_path: str, start_time: str, end_time: str, question: str):
"""分析视频的特定片段"""
video_file = client.files.upload(file=video_path)
while video_file.state.name == "PROCESSING":
time.sleep(5)
video_file = client.files.get(name=video_file.name)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Part.from_uri(file_uri=video_file.uri, mime_type=video_file.mime_type),
f"""请重点分析视频中 {start_time} 到 {end_time} 这个时间段的内容。
{question}"""
]
)
return response.text
# 使用示例
result = analyze_video_segment(
"meeting.mp4",
"10:30", "15:45",
"这段讨论的主要结论是什么？有哪些待办事项？"
)
6. 多模态理解：音频
6.1 音频理解概述
Gemini 2.5 Pro 可以直接理解音频内容，包括语音识别、说话人区分、情感分析等。
6.2 音频转录
from google import genai
from google.genai import types
client = genai.Client()
def transcribe_audio(audio_path: str, language: str = "auto"):
"""转录音频内容"""
# 上传音频文件
audio_file = client.files.upload(file=audio_path)
# 等待处理
while audio_file.state.name == "PROCESSING":
time.sleep(5)
audio_file = client.files.get(name=audio_file.name)
prompt = "请转录这段音频的全部内容。" if language == "auto" else f"请将这段{language}音频转录为文字。"
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Part.from_uri(file_uri=audio_file.uri, mime_type=audio_file.mime_type),
prompt
]
)
return response.text
# 使用示例
transcript = transcribe_audio("interview.mp3")
print(transcript)
6.3 音频内容分析
from google import genai
from google.genai import types
client = genai.Client()
def analyze_meeting_audio(audio_path: str):
"""分析会议录音"""
audio_file = client.files.upload(file=audio_path)
while audio_file.state.name == "PROCESSING":
time.sleep(5)
audio_file = client.files.get(name=audio_file.name)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Part.from_uri(file_uri=audio_file.uri, mime_type=audio_file.mime_type),
"""请分析这段会议录音，提供：
1. **完整转录**：包含说话人标注
2. **会议摘要**：主要讨论内容
3. **关键决策**：做出的决定
4. **待办事项**：行动项及负责人（如能识别）
5. **情感分析**：讨论氛围和参与者情绪"""
],
config=types.GenerateContentConfig(
max_output_tokens=8192
)
)
return response.text
# 使用示例
analysis = analyze_meeting_audio("team_meeting.wav")
print(analysis)
6.4 音频与视频同步分析
from google import genai
from google.genai import types
client = genai.Client()
def analyze_multimedia(video_path: str, question: str):
"""同时分析视频的视觉和音频内容"""
video_file = client.files.upload(file=video_path)
while video_file.state.name == "PROCESSING":
time.sleep(5)
video_file = client.files.get(name=video_file.name)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Part.from_uri(file_uri=video_file.uri, mime_type=video_file.mime_type),
f"""请同时分析这段视频的视觉内容和音频内容：
{question}
要求：
- 结合画面和声音进行综合分析
- 指出视觉和音频之间的关联
- 如果存在不一致，特别指出"""
]
)
return response.text
# 使用示例
result = analyze_multimedia("product_demo.mp4", "评估这个产品演示的效果，包括讲解质量和视觉展示。")
7. 代码生成与执行
7.1 代码生成
from google import genai
from google.genai import types
client = genai.Client()
def generate_code(task: str, language: str = "python"):
"""生成代码"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""请用{language}编写代码完成以下任务：
{task}
要求：
1. 代码完整可运行
2. 包含必要的注释
3. 遵循语言最佳实践
4. 包含错误处理""",
config=types.GenerateContentConfig(
max_output_tokens=8192
)
)
return response.text
# 使用示例
code = generate_code("实现一个LRU缓存，支持get和put操作，时间复杂度O(1)", "python")
print(code)
7.2 原生代码执行
Gemini 2.5 Pro 支持在沙盒环境中执行代码：
from google import genai
from google.genai import types
client = genai.Client()
def execute_code_with_gemini(code: str):
"""让Gemini执行代码并返回结果"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""请执行以下Python代码并返回结果：
```python
{code}
如果代码有错误，请分析错误原因并修复后重新执行。""", config=types.GenerateContentConfig( tools=[types.Tool(code_execution=types.ToolCodeExecution())] ) )
# 解析响应
result = {"text": "", "code_results": []}
for part in response.candidates[0].content.parts:
if hasattr(part, 'text') and part.text:
result["text"] += part.text
if hasattr(part, 'executable_code'):
result["code_results"].append({
"code": part.executable_code.code,
"language": part.executable_code.language
})
if hasattr(part, 'code_execution_result'):
result["code_results"].append({
"output": part.code_execution_result.output,
"status": part.code_execution_result.outcome
})
return result
使用示例
result = execute_code_with_gemini(""" import numpy as np data = np.random.normal(100, 15, 1000) print(f"均值: {data.mean():.2f}") print(f"标准差: {data.std():.2f}") print(f"中位数: {np.median(data):.2f}") """) print(result)
### 7.3 代码解释与重构
```python
from google import genai
from google.genai import types
client = genai.Client()
def explain_code(code: str, level: str = "intermediate"):
"""解释代码"""
level_prompts = {
"beginner": "请用初学者能理解的方式解释这段代码，避免使用专业术语。",
"intermediate": "请详细解释这段代码的工作原理，包括关键概念和设计模式。",
"advanced": "请深入分析这段代码的实现细节，包括时间/空间复杂度、潜在优化点和边界情况。"
}
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""{level_prompts.get(level, level_prompts['intermediate'])}
```python
{code}
```"""
)
return response.text
def refactor_code(code: str, goals: list):
"""重构代码"""
goals_str = "\n".join([f"- {g}" for g in goals])
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""请重构以下代码：
```python
{code}
重构目标：
请提供：
重构后的完整代码
每处修改的说明
重构带来的收益""", config=types.GenerateContentConfig( max_output_tokens=8192 ) )
return response.text
使用示例
explanation = explain_code(""" def fibonacci(n): if n ⇐ 1: return n a, b = 0, 1 for _ in range(2, n + 1): a, b = b, a + b return b """, level="advanced") print(explanation)
---
## 8. Function Calling 函数调用
### 8.1 Function Calling 概述
Function Calling 允许 Gemini 调用外部函数来获取实时数据或执行操作。
### 8.2 定义函数
```python
from google import genai
from google.genai import types
import json
client = genai.Client()
# 定义函数
def get_current_weather(city: str, unit: str = "celsius") -> dict:
"""获取当前天气"""
# 模拟天气数据
weather_data = {
"北京": {"temp": 28, "condition": "晴", "humidity": 45, "wind": "北风3级"},
"上海": {"temp": 26, "condition": "多云", "humidity": 72, "wind": "东风2级"},
"广州": {"temp": 33, "condition": "雷阵雨", "humidity": 85, "wind": "南风4级"},
}
data = weather_data.get(city, {"temp": 25, "condition": "未知", "humidity": 50, "wind": "微风"})
if unit == "fahrenheit":
data = {**data, "temp": data["temp"] * 9/5 + 32}
return data
def search_restaurants(city: str, cuisine: str = None, price_range: str = None) -> list:
"""搜索餐厅"""
# 模拟餐厅数据
return [
{"name": "美味餐厅", "rating": 4.5, "price": "$$", "cuisine": "中餐"},
{"name": "寿司之神", "rating": 4.8, "price": "$$$", "cuisine": "日料"},
{"name": "Pizza Palace", "rating": 4.2, "price": "$$", "cuisine": "意大利"},
]
# 将函数声明为工具
tools = [
types.Tool(
function_declarations=[
types.FunctionDeclaration(
name="get_current_weather",
description="获取指定城市的当前天气信息",
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"city": types.Schema(
type=types.Type.STRING,
description="城市名称"
),
"unit": types.Schema(
type=types.Type.STRING,
enum=["celsius", "fahrenheit"],
description="温度单位"
)
},
required=["city"]
)
),
types.FunctionDeclaration(
name="search_restaurants",
description="搜索指定城市的餐厅",
parameters=types.Schema(
type=types.Type.OBJECT,
properties={
"city": types.Schema(
type=types.Type.STRING,
description="城市名称"
),
"cuisine": types.Schema(
type=types.Type.STRING,
description="菜系类型"
),
"price_range": types.Schema(
type=types.Type.STRING,
enum=["$", "$$", "$$$", "$$$$"],
description="价格范围"
)
},
required=["city"]
)
)
]
)
]
8.3 完整的Function Calling循环
from google import genai
from google.genai import types
import json
client = genai.Client()
# 函数映射
function_map = {
"get_current_weather": get_current_weather,
"search_restaurants": search_restaurants,
}
def run_conversation(user_message: str):
"""运行完整的Function Calling对话"""
messages = [
types.Content(
role="user",
parts=[types.Part.from_text(text=user_message)]
)
]
while True:
# 调用API
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=messages,
config=types.GenerateContentConfig(
tools=tools
)
)
# 检查是否有函数调用
candidate = response.candidates[0]
# 检查是否有函数调用
has_function_call = False
function_calls = []
for part in candidate.content.parts:
if hasattr(part, 'function_call') and part.function_call:
has_function_call = True
function_calls.append(part.function_call)
if not has_function_call:
# 没有函数调用，返回最终回答
for part in candidate.content.parts:
if hasattr(part, 'text') and part.text:
print(f"\nGemini: {part.text}")
break
# 执行函数调用
function_responses = []
for fc in function_calls:
func_name = fc.name
func_args = dict(fc.args)
print(f" → 调用函数: {func_name}({json.dumps(func_args, ensure_ascii=False)})")
# 执行函数
func = function_map.get(func_name)
if func:
result = func(**func_args)
print(f" ← 结果: {json.dumps(result, ensure_ascii=False)}")
function_responses.append(
types.Part.from_function_response(
name=func_name,
response={"result": result}
)
)
else:
function_responses.append(
types.Part.from_function_response(
name=func_name,
response={"error": f"函数 {func_name} 未找到"}
)
)
# 更新消息历史
messages.append(candidate.content)
messages.append(
types.Content(
role="user",
parts=function_responses
)
)
# 测试
run_conversation("北京今天天气怎么样？顺便推荐几家餐厅。")
8.4 强制函数调用
# 强制使用特定函数
response = client.models.generate_content(
model="gemini-2.5-pro",
contents="查一下天气",
config=types.GenerateContentConfig(
tools=tools,
tool_config=types.ToolConfig(
function_calling_config=types.FunctionCallingConfig(
mode=types.FunctionCallingConfigMode.ANY, # 强制调用函数
allowed_function_names=["get_current_weather"] # 限定可调用的函数
)
)
)
)
9. Google Search Grounding
9.1 Search Grounding 概述
Google Search Grounding 允许 Gemini 在回答时搜索互联网获取最新信息，并提供引用来源。这对于需要实时信息的任务非常有用。
9.2 基础搜索集成
from google import genai
from google.genai import types
client = genai.Client()
def search_and_answer(question: str):
"""使用Google搜索增强回答"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=question,
config=types.GenerateContentConfig(
tools=[
types.Tool(
google_search=types.GoogleSearch()
)
]
)
)
# 提取回答和来源
result = {
"answer": "",
"sources": []
}
candidate = response.candidates[0]
for part in candidate.content.parts:
if hasattr(part, 'text') and part.text:
result["answer"] += part.text
# 提取搜索来源
if hasattr(candidate, 'grounding_metadata'):
metadata = candidate.grounding_metadata
if hasattr(metadata, 'grounding_chunks'):
for chunk in metadata.grounding_chunks:
if hasattr(chunk, 'web'):
result["sources"].append({
"title": chunk.web.title,
"url": chunk.web.uri
})
return result
# 使用示例
result = search_and_answer("2025年最新的AI发展趋势是什么？")
print(f"回答：{result['answer']}")
print(f"\n来源：")
for source in result['sources']:
print(f" - {source['title']}: {source['url']}")
9.3 搜索增强的对话
from google import genai
from google.genai import types
client = genai.Client()
def chat_with_search():
"""支持搜索的对话"""
chat = client.chats.create(
model="gemini-2.5-pro",
config=types.GenerateContentConfig(
tools=[
types.Tool(google_search=types.GoogleSearch())
],
system_instruction="""你是一个知识渊博的AI助手。
当需要最新信息、实时数据或不确定的事实时，请使用搜索功能。
请在回答中标注信息来源。"""
)
)
print("搜索增强对话助手（输入 'quit' 退出）")
print("-" * 40)
while True:
user_input = input("\n你: ")
if user_input.lower() == 'quit':
break
response = chat.send_message(user_input)
print(f"\nGemini: {response.text}")
# 使用示例
# chat_with_search()
9.4 搜索结果结构化处理
from google import genai
from google.genai import types
import json
client = genai.Client()
def research_topic(topic: str) -> dict:
"""深度研究某个主题"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=f"""请深入研究以下主题：{topic}
请提供：
1. 概述和定义
2. 最新发展（使用搜索获取最新信息）
3. 关键数据和统计
4. 主要参与者/公司
5. 未来趋势
6. 相关资源链接
以JSON格式返回。""",
config=types.GenerateContentConfig(
tools=[
types.Tool(google_search=types.GoogleSearch())
],
response_mime_type="application/json"
)
)
return json.loads(response.text)
# 使用示例
research = research_topic("大语言模型的最新发展")
print(json.dumps(research, ensure_ascii=False, indent=2))
10. Gems 自定义助手
10.1 Gems 概述
Gems 是 Gemini 的自定义助手功能，允许用户创建具有特定角色、知识和行为的定制化AI助手。
10.2 通过API创建自定义助手
from google import genai
from google.genai import types
client = genai.Client()
def create_custom_assistant(name: str, system_instruction: str,
knowledge_base: str = None):
"""创建自定义助手"""
full_instruction = f"""你是一个名为"{name}"的专业AI助手。
{system_instruction}
{f'以下是你的专业知识库：{knowledge_base}' if knowledge_base else ''}
请始终保持角色设定，提供专业、准确的回答。"""
return full_instruction
# 使用示例：法律助手
legal_system = create_custom_assistant(
name="法律小助手",
system_instruction="""你是一个专业的法律顾问AI助手。
你的职责是：
1. 解释法律概念和条款
2. 分析合同中的风险点
3. 提供法律建议（仅供参考，不构成正式法律意见）
4. 回答法律相关问题
请始终提醒用户，AI建议仅供参考，重要决策请咨询专业律师。""",
knowledge_base="关注中国民法、合同法、劳动法等领域。"
)
def ask_legal_assistant(question: str):
"""向法律助手提问"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=question,
config=types.GenerateContentConfig(
system_instruction=legal_system,
temperature=0.3 # 法律问题需要更保守的回答
)
)
return response.text
# 使用示例
answer = ask_legal_assistant("租房合同中，房东提前解约需要承担什么责任？")
print(answer)
10.3 多角色工作流
from google import genai
from google.genai import types
client = genai.Client()
class MultiAgentWorkflow:
"""多角色工作流"""
def __init__(self):
self.agents = {}
def add_agent(self, name: str, system_instruction: str, temperature: float = 0.7):
"""添加角色"""
self.agents[name] = {
"system_instruction": system_instruction,
"temperature": temperature
}
def run_pipeline(self, task: str, pipeline: list) -> str:
"""运行流水线"""
current_input = task
results = []
for agent_name in pipeline:
if agent_name not in self.agents:
raise ValueError(f"角色 {agent_name} 未注册")
agent = self.agents[agent_name]
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=current_input,
config=types.GenerateContentConfig(
system_instruction=agent["system_instruction"],
temperature=agent["temperature"]
)
)
result = response.text
results.append({"agent": agent_name, "output": result})
current_input = result
return results
# 使用示例：内容创作流水线
workflow = MultiAgentWorkflow()
workflow.add_agent(
"研究员",
"你是一个专业的研究员，负责收集和整理信息。请提供全面、准确的研究资料。"
)
workflow.add_agent(
"写手",
"你是一个专业的写手，负责将研究资料转化为易读的文章。请使用清晰的结构和生动的语言。"
)
workflow.add_agent(
"编辑",
"你是一个严格的编辑，负责审查文章质量。请检查逻辑、语法、风格，并提供修改建议。"
)
# 运行流水线
results = workflow.run_pipeline(
"写一篇关于人工智能在医疗领域应用的文章",
["研究员", "写手", "编辑"]
)
for r in results:
print(f"\n=== {r['agent']} ===")
print(r['output'][:500] + "...")
11. Vertex AI 企业部署
11.1 Vertex AI 概述
Vertex AI 是 Google Cloud 的企业级AI平台，提供更高级的安全性、合规性和管理功能。
11.2 使用Vertex AI SDK
# 安装Vertex AI SDK
# pip install google-cloud-aiplatform
import vertexai
from vertexai.generative_models import GenerativeModel
# 初始化Vertex AI
vertexai.init(
project="your-project-id",
location="us-central1"
)
# 创建模型实例
model = GenerativeModel("gemini-2.5-pro")
# 生成内容
response = model.generate_content("解释机器学习的基本概念")
print(response.text)
11.3 企业级安全配置
import vertexai
from vertexai.generative_models import GenerativeModel, SafetySetting, HarmCategory
vertexai.init(project="your-project-id", location="us-central1")
# 配置安全设置
safety_settings = [
SafetySetting(
category=HarmCategory.HARM_CATEGORY_HATE_SPEECH,
threshold=SafetySetting.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE
),
SafetySetting(
category=HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
threshold=SafetySetting.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE
),
SafetySetting(
category=HarmCategory.HARM_CATEGORY_HARASSMENT,
threshold=SafetySetting.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE
),
SafetySetting(
category=HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
threshold=SafetySetting.HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE
),
]
model = GenerativeModel("gemini-2.5-pro")
response = model.generate_content(
"写一个关于网络安全的故事",
safety_settings=safety_settings
)
print(response.text)
11.4 批量处理与异步调用
import vertexai
from vertexai.generative_models import GenerativeModel
import asyncio
from concurrent.futures import ThreadPoolExecutor
vertexai.init(project="your-project-id", location="us-central1")
model = GenerativeModel("gemini-2.5-pro")
async def batch_process(prompts: list, max_workers: int = 10):
"""批量异步处理"""
def call_api(prompt):
return model.generate_content(prompt)
with ThreadPoolExecutor(max_workers=max_workers) as executor:
loop = asyncio.get_event_loop()
tasks = [
loop.run_in_executor(executor, call_api, prompt)
for prompt in prompts
]
results = await asyncio.gather(*tasks)
return [r.text for r in results]
# 使用示例
prompts = [
"总结人工智能的发展历史",
"解释深度学习的工作原理",
"描述自然语言处理的应用场景",
"介绍计算机视觉的主要任务",
]
# results = asyncio.run(batch_process(prompts))
11.5 成本监控与优化
import vertexai
from vertexai.generative_models import GenerativeModel
vertexai.init(project="your-project-id", location="us-central1")
class CostTracker:
"""成本追踪器"""
# Gemini 2.5 Pro 定价（示例，实际价格请参考官方）
PRICE_PER_INPUT_TOKEN = 0.00000125 # $1.25 per 1M tokens
PRICE_PER_OUTPUT_TOKEN = 0.000005 # $5.00 per 1M tokens
def __init__(self):
self.total_input_tokens = 0
self.total_output_tokens = 0
self.total_cost = 0
def track(self, response):
"""追踪响应的token使用"""
usage = response.usage_metadata
input_tokens = usage.prompt_token_count
output_tokens = usage.candidates_token_count
cost = (input_tokens * self.PRICE_PER_INPUT_TOKEN +
output_tokens * self.PRICE_PER_OUTPUT_TOKEN)
self.total_input_tokens += input_tokens
self.total_output_tokens += output_tokens
self.total_cost += cost
return {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
"cost": cost
}
def summary(self):
"""返回成本摘要"""
return {
"total_input_tokens": self.total_input_tokens,
"total_output_tokens": self.total_output_tokens,
"total_cost": round(self.total_cost, 4),
"total_requests": self.total_input_tokens + self.total_output_tokens
}
# 使用示例
tracker = CostTracker()
model = GenerativeModel("gemini-2.5-pro")
response = model.generate_content("解释量子计算")
usage = tracker.track(response)
print(f"本次调用: 输入{usage['input_tokens']}tokens, "
f"输出{usage['output_tokens']}tokens, "
f"成本${usage['cost']:.6f}")
12. API最佳实践与性能优化
12.1 错误处理与重试
from google import genai
from google.genai import types
import time
import logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
client = genai.Client()
def call_with_retry(contents: str, max_retries: int = 3, **kwargs):
"""带重试的API调用"""
for attempt in range(max_retries):
try:
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=contents,
**kwargs
)
return response
except Exception as e:
error_str = str(e)
if "429" in error_str or "RESOURCE_EXHAUSTED" in error_str:
# 速率限制
wait_time = 2 ** attempt * 10
logger.warning(f"速率限制，等待 {wait_time} 秒后重试...")
time.sleep(wait_time)
elif "500" in error_str or "503" in error_str:
# 服务器错误
wait_time = 2 ** attempt * 5
logger.warning(f"服务器错误，等待 {wait_time} 秒后重试...")
time.sleep(wait_time)
else:
# 其他错误，不重试
logger.error(f"API错误: {error_str}")
raise
raise Exception(f"超过最大重试次数 ({max_retries})")
12.2 提示词优化
from google import genai
from google.genai import types
client = genai.Client()
# 优化前：模糊的提示词
bad_prompt = "帮我写点关于AI的东西"
# 优化后：清晰的提示词
good_prompt = """请写一篇关于"大语言模型在企业中的应用"的技术博客文章。
要求：
- 目标读者：企业技术决策者
- 字数：1500-2000字
- 结构：引言 → 3个核心应用场景 → 实施建议 → 结论
- 风格：专业但易懂，包含具体案例
- 包含：ROI分析和实施步骤"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=good_prompt,
config=types.GenerateContentConfig(
temperature=0.7,
max_output_tokens=4096
)
)
print(response.text)
12.3 结构化输出
from google import genai
from google.genai import types
import json
client = genai.Client()
def get_structured_output(prompt: str, schema: dict):
"""获取结构化JSON输出"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=prompt,
config=types.GenerateContentConfig(
response_mime_type="application/json",
response_schema=schema
)
)
return json.loads(response.text)
# 使用示例
schema = {
"type": "object",
"properties": {
"name": {"type": "string"},
"age": {"type": "integer"},
"skills": {
"type": "array",
"items": {"type": "string"}
},
"experience": {
"type": "array",
"items": {
"type": "object",
"properties": {
"company": {"type": "string"},
"role": {"type": "string"},
"years": {"type": "integer"}
}
}
}
}
}
result = get_structured_output(
"生成一个虚构的高级软件工程师的简历信息",
schema
)
print(json.dumps(result, ensure_ascii=False, indent=2))
12.4 缓存策略
from google import genai
from google.genai import types
import hashlib
import json
import redis
client = genai.Client()
class CachedGeminiClient:
"""带缓存的Gemini客户端"""
def __init__(self, redis_url: str = "redis://localhost:6379", ttl: int = 3600):
self.client = genai.Client()
self.redis = redis.from_url(redis_url)
self.ttl = ttl
def _cache_key(self, **kwargs) -> str:
"""生成缓存键"""
content = json.dumps(kwargs, sort_keys=True)
return f"gemini:cache:{hashlib.sha256(content.encode()).hexdigest()}"
def generate(self, use_cache: bool = True, **kwargs):
"""带缓存的生成"""
if use_cache:
key = self._cache_key(**kwargs)
cached = self.redis.get(key)
if cached:
return json.loads(cached)
response = self.client.models.generate_content(**kwargs)
result = {
"text": response.text,
"usage": {
"input_tokens": response.usage_metadata.prompt_token_count,
"output_tokens": response.usage_metadata.candidates_token_count
}
}
if use_cache:
self.redis.setex(key, self.ttl, json.dumps(result))
return result
13. 实战项目：多模态文档分析系统
13.1 项目概述
构建一个完整的多模态文档分析系统，能够处理文本、图像、PDF、音频和视频等多种格式的文档。
13.2 系统架构
┌─────────────────────────────────────────────────────────┐
│ 多模态文档分析系统 │
├─────────────────────────────────────────────────────────┤
│ ┌──────────┐ ┌──────────┐ ┌──────────┐ ┌────────┐ │
│ │ 文本分析 │ │ 图像分析 │ │ 视频分析 │ │音频分析 │ │
│ └────┬─────┘ └────┬─────┘ └────┬─────┘ └───┬────┘ │
│ │ │ │ │ │
│ └──────────────┴──────────────┴────────────┘ │
│ │ │
│ ┌──────┴──────┐ │
│ │ Gemini 2.5 │ │
│ │ Pro API │ │
│ └──────┬──────┘ │
│ │ │
│ ┌──────────────────┼──────────────────┐ │
│ │ │ │ │
│ ┌────┴────┐ ┌────┴────┐ ┌────┴────┐ │
│ │结果分析 │ │报告生成 │ │知识图谱 │ │
│ └─────────┘ └─────────┘ └─────────┘ │
└─────────────────────────────────────────────────────────┘
13.3 完整代码实现
"""
多模态文档分析系统 - 基于Gemini 2.5 Pro
支持：文本、图像、PDF、音频、视频文档分析
"""
from google import genai
from google.genai import types
import os
import json
import time
from typing import Dict, List, Optional, Union
from dataclasses import dataclass, field
from enum import Enum
from pathlib import Path
client = genai.Client()
class DocumentType(Enum):
"""文档类型"""
TEXT = "text"
IMAGE = "image"
PDF = "pdf"
AUDIO = "audio"
VIDEO = "video"
UNKNOWN = "unknown"
@dataclass
class AnalysisResult:
"""分析结果"""
document_name: str
document_type: DocumentType
summary: str
key_points: List[str]
entities: List[Dict]
sentiment: str
language: str
metadata: Dict = field(default_factory=dict)
raw_analysis: str = ""
class MultimodalDocumentAnalyzer:
"""多模态文档分析器"""
# 文件类型映射
TYPE_MAP = {
'.txt': DocumentType.TEXT,
'.md': DocumentType.TEXT,
'.csv': DocumentType.TEXT,
'.json': DocumentType.TEXT,
'.xml': DocumentType.TEXT,
'.jpg': DocumentType.IMAGE,
'.jpeg': DocumentType.IMAGE,
'.png': DocumentType.IMAGE,
'.gif': DocumentType.IMAGE,
'.webp': DocumentType.IMAGE,
'.bmp': DocumentType.IMAGE,
'.pdf': DocumentType.PDF,
'.mp3': DocumentType.AUDIO,
'.wav': DocumentType.AUDIO,
'.ogg': DocumentType.AUDIO,
'.flac': DocumentType.AUDIO,
'.mp4': DocumentType.VIDEO,
'.avi': DocumentType.VIDEO,
'.mov': DocumentType.VIDEO,
'.mkv': DocumentType.VIDEO,
'.webm': DocumentType.VIDEO,
}
# MIME类型映射
MIME_MAP = {
'.jpg': 'image/jpeg',
'.jpeg': 'image/jpeg',
'.png': 'image/png',
'.gif': 'image/gif',
'.webp': 'image/webp',
'.bmp': 'image/bmp',
'.pdf': 'application/pdf',
'.mp3': 'audio/mpeg',
'.wav': 'audio/wav',
'.ogg': 'audio/ogg',
'.mp4': 'video/mp4',
'.avi': 'video/x-msvideo',
'.mov': 'video/quicktime',
}
def __init__(self, model: str = "gemini-2.5-pro"):
self.client = genai.Client()
self.model = model
def detect_type(self, file_path: str) -> DocumentType:
"""检测文档类型"""
ext = Path(file_path).suffix.lower()
return self.TYPE_MAP.get(ext, DocumentType.UNKNOWN)
def analyze(self, file_path: str, question: str = None,
language: str = "中文") -> AnalysisResult:
"""分析文档"""
doc_type = self.detect_type(file_path)
file_name = os.path.basename(file_path)
print(f"分析文档: {file_name} (类型: {doc_type.value})")
if doc_type == DocumentType.TEXT:
return self._analyze_text(file_path, file_name, question, language)
elif doc_type == DocumentType.IMAGE:
return self._analyze_image(file_path, file_name, question, language)
elif doc_type == DocumentType.PDF:
return self._analyze_pdf(file_path, file_name, question, language)
elif doc_type == DocumentType.AUDIO:
return self._analyze_audio(file_path, file_name, question, language)
elif doc_type == DocumentType.VIDEO:
return self._analyze_video(file_path, file_name, question, language)
else:
return self._analyze_generic(file_path, file_name, question, language)
def _get_analysis_prompt(self, question: str = None, language: str = "中文") -> str:
"""生成分析提示词"""
base_prompt = f"""请用{language}对这个文档进行全面分析，并以JSON格式返回结果：
{{
"summary": "文档摘要（200-300字）",
"key_points": ["要点1", "要点2", "要点3", ...],
"entities": [
{{"name": "实体名", "type": "类型", "description": "描述"}}
],
"sentiment": "整体情感（positive/negative/neutral）",
"language": "文档语言",
"topics": ["主题1", "主题2", ...],
"metadata": {{
"word_count": 字数估算,
"complexity": "难度等级（简单/中等/复杂）",
"target_audience": "目标读者"
}}
}}"""
if question:
base_prompt += f"\n\n另外，请回答以下问题：{question}"
return base_prompt
def _analyze_text(self, file_path: str, file_name: str,
question: str = None, language: str = "中文") -> AnalysisResult:
"""分析文本文档"""
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
prompt = self._get_analysis_prompt(question, language)
response = self.client.models.generate_content(
model=self.model,
contents=f"文档内容：\n\n{content}\n\n{prompt}",
config=types.GenerateContentConfig(
response_mime_type="application/json"
)
)
result_data = json.loads(response.text)
return AnalysisResult(
document_name=file_name,
document_type=DocumentType.TEXT,
summary=result_data.get("summary", ""),
key_points=result_data.get("key_points", []),
entities=result_data.get("entities", []),
sentiment=result_data.get("sentiment", "neutral"),
language=result_data.get("language", language),
metadata=result_data.get("metadata", {}),
raw_analysis=response.text
)
def _analyze_image(self, file_path: str, file_name: str,
question: str = None, language: str = "中文") -> AnalysisResult:
"""分析图像文档"""
from PIL import Image
image = Image.open(file_path)
prompt = self._get_analysis_prompt(question, language)
response = self.client.models.generate_content(
model=self.model,
contents=[image, prompt],
config=types.GenerateContentConfig(
response_mime_type="application/json"
)
)
result_data = json.loads(response.text)
return AnalysisResult(
document_name=file_name,
document_type=DocumentType.IMAGE,
summary=result_data.get("summary", ""),
key_points=result_data.get("key_points", []),
entities=result_data.get("entities", []),
sentiment=result_data.get("sentiment", "neutral"),
language=result_data.get("language", language),
metadata=result_data.get("metadata", {}),
raw_analysis=response.text
)
def _analyze_pdf(self, file_path: str, file_name: str,
question: str = None, language: str = "中文") -> AnalysisResult:
"""分析PDF文档"""
# 上传PDF文件
uploaded_file = self.client.files.upload(file=file_path)
prompt = self._get_analysis_prompt(question, language)
response = self.client.models.generate_content(
model=self.model,
contents=[
types.Part.from_uri(
file_uri=uploaded_file.uri,
mime_type=uploaded_file.mime_type
),
prompt
],
config=types.GenerateContentConfig(
response_mime_type="application/json"
)
)
result_data = json.loads(response.text)
return AnalysisResult(
document_name=file_name,
document_type=DocumentType.PDF,
summary=result_data.get("summary", ""),
key_points=result_data.get("key_points", []),
entities=result_data.get("entities", []),
sentiment=result_data.get("sentiment", "neutral"),
language=result_data.get("language", language),
metadata=result_data.get("metadata", {}),
raw_analysis=response.text
)
def _analyze_audio(self, file_path: str, file_name: str,
question: str = None, language: str = "中文") -> AnalysisResult:
"""分析音频文档"""
uploaded_file = self.client.files.upload(file=file_path)
# 等待处理
while uploaded_file.state.name == "PROCESSING":
time.sleep(5)
uploaded_file = self.client.files.get(name=uploaded_file.name)
prompt = self._get_analysis_prompt(question, language)
prompt += "\n\n请同时提供音频内容的完整转录。"
response = self.client.models.generate_content(
model=self.model,
contents=[
types.Part.from_uri(
file_uri=uploaded_file.uri,
mime_type=uploaded_file.mime_type
),
prompt
],
config=types.GenerateContentConfig(
response_mime_type="application/json"
)
)
result_data = json.loads(response.text)
return AnalysisResult(
document_name=file_name,
document_type=DocumentType.AUDIO,
summary=result_data.get("summary", ""),
key_points=result_data.get("key_points", []),
entities=result_data.get("entities", []),
sentiment=result_data.get("sentiment", "neutral"),
language=result_data.get("language", language),
metadata=result_data.get("metadata", {}),
raw_analysis=response.text
)
def _analyze_video(self, file_path: str, file_name: str,
question: str = None, language: str = "中文") -> AnalysisResult:
"""分析视频文档"""
uploaded_file = self.client.files.upload(file=file_path)
# 等待处理
while uploaded_file.state.name == "PROCESSING":
time.sleep(10)
uploaded_file = self.client.files.get(name=uploaded_file.name)
print(f" 视频处理中... 状态: {uploaded_file.state.name}")
if uploaded_file.state.name == "FAILED":
raise ValueError("视频处理失败")
prompt = self._get_analysis_prompt(question, language)
prompt += "\n\n请同时分析视频的视觉内容和音频内容。"
response = self.client.models.generate_content(
model=self.model,
contents=[
types.Part.from_uri(
file_uri=uploaded_file.uri,
mime_type=uploaded_file.mime_type
),
prompt
],
config=types.GenerateContentConfig(
response_mime_type="application/json"
)
)
result_data = json.loads(response.text)
return AnalysisResult(
document_name=file_name,
document_type=DocumentType.VIDEO,
summary=result_data.get("summary", ""),
key_points=result_data.get("key_points", []),
entities=result_data.get("entities", []),
sentiment=result_data.get("sentiment", "neutral"),
language=result_data.get("language", language),
metadata=result_data.get("metadata", {}),
raw_analysis=response.text
)
def _analyze_generic(self, file_path: str, file_name: str,
question: str = None, language: str = "中文") -> AnalysisResult:
"""通用分析"""
return AnalysisResult(
document_name=file_name,
document_type=DocumentType.UNKNOWN,
summary="不支持的文档类型",
key_points=[],
entities=[],
sentiment="neutral",
language=language
)
def batch_analyze(self, file_paths: List[str],
question: str = None) -> List[AnalysisResult]:
"""批量分析多个文档"""
results = []
for path in file_paths:
try:
result = self.analyze(path, question)
results.append(result)
print(f" ✓ {path} 分析完成")
except Exception as e:
print(f" ✗ {path} 分析失败: {str(e)}")
return results
def generate_report(self, results: List[AnalysisResult]) -> str:
"""生成分析报告"""
report = "# 多模态文档分析报告\n\n"
report += f"**分析时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}\n"
report += f"**文档数量**: {len(results)}\n\n"
# 统计信息
type_count = {}
for r in results:
type_count[r.document_type.value] = type_count.get(r.document_type.value, 0) + 1
report += "## 文档类型统计\n\n"
for doc_type, count in type_count.items():
report += f"- {doc_type}: {count} 份\n"
report += "\n## 各文档分析结果\n\n"
for i, result in enumerate(results, 1):
report += f"### {i}. {result.document_name}\n\n"
report += f"**类型**: {result.document_type.value}\n"
report += f"**语言**: {result.language}\n"
report += f"**情感**: {result.sentiment}\n\n"
report += f"**摘要**:\n{result.summary}\n\n"
if result.key_points:
report += "**要点**:\n"
for point in result.key_points:
report += f"- {point}\n"
report += "\n"
if result.entities:
report += "**实体**:\n"
for entity in result.entities[:10]:
report += f"- {entity.get('name', 'N/A')} ({entity.get('type', 'N/A')}): {entity.get('description', '')}\n"
report += "\n"
report += "---\n\n"
return report
# 使用示例
def main():
analyzer = MultimodalDocumentAnalyzer()
# 分析单个文档
result = analyzer.analyze(
"report.pdf",
question="这份报告的核心结论是什么？"
)
print(f"摘要: {result.summary}")
print(f"要点: {result.key_points}")
# 批量分析
files = ["report.pdf", "chart.png", "meeting.mp3", "demo.mp4"]
results = analyzer.batch_analyze(files, question="提取关键信息")
# 生成报告
report = analyzer.generate_report(results)
print(report)
# 保存报告
with open("analysis_report.md", "w", encoding="utf-8") as f:
f.write(report)
print("报告已保存到 analysis_report.md")
if __name__ == "__main__":
main()
13.4 运行效果示例
分析文档: report.pdf (类型: pdf)
分析文档: chart.png (类型: image)
分析文档: meeting.mp3 (类型: audio)
视频处理中... 状态: PROCESSING
分析文档: demo.mp4 (类型: video)
# 多模态文档分析报告
**分析时间**: 2025-05-29 10:30:00
**文档数量**: 4
## 文档类型统计
- pdf: 1 份
- image: 1 份
- audio: 1 份
- video: 1 份
## 各文档分析结果
### 1. report.pdf
**类型**: pdf | **语言**: 中文 | **情感**: neutral
**摘要**: 本报告分析了2025年Q1的市场表现...
**要点**:
- 营收同比增长25%
- 用户增长率达到40%
- 技术债务减少了15%
...
14. 常见问题与解决方案
14.1 API调用相关
Q1: 如何处理API配额限制？
from google import genai
from google.genai import types
import time
client = genai.Client()
def call_with_quota_management(prompt: str, max_retries: int = 5):
"""配额管理的API调用"""
for attempt in range(max_retries):
try:
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=prompt
)
return response
except Exception as e:
error_msg = str(e)
if "429" in error_msg or "RESOURCE_EXHAUSTED" in error_msg:
# 计算等待时间（指数退避 + 随机抖动）
import random
base_wait = 2 ** attempt * 10
jitter = random.uniform(0, base_wait * 0.5)
wait_time = base_wait + jitter
print(f"配额耗尽，等待 {wait_time:.1f} 秒后重试...")
time.sleep(wait_time)
elif "quota" in error_msg.lower():
print(f"配额错误: {error_msg}")
print("建议：1) 升级配额 2) 使用缓存 3) 优化请求频率")
raise
else:
raise
raise Exception("超过最大重试次数")
Q2: 如何选择合适的模型？
def select_model(task_type: str, budget: str = "medium") -> str:
"""根据任务类型选择模型"""
model_map = {
"simple": {
"low": "gemini-2.0-flash",
"medium": "gemini-2.5-flash",
"high": "gemini-2.5-pro"
},
"complex": {
"low": "gemini-2.5-flash",
"medium": "gemini-2.5-pro",
"high": "gemini-2.5-pro"
},
"multimodal": {
"low": "gemini-2.5-flash",
"medium": "gemini-2.5-pro",
"high": "gemini-2.5-pro"
}
}
return model_map.get(task_type, model_map["complex"]).get(budget, "gemini-2.5-pro")
Q3: 如何处理长视频上传超时？
from google import genai
import time
client = genai.Client()
def upload_large_video(file_path: str, timeout: int = 600):
"""上传大视频文件"""
file_size = os.path.getsize(file_path)
print(f"文件大小: {file_size / (1024*1024):.1f} MB")
# 上传
uploaded_file = client.files.upload(file=file_path)
print(f"上传完成: {uploaded_file.name}")
# 等待处理（带超时）
start_time = time.time()
while uploaded_file.state.name == "PROCESSING":
if time.time() - start_time > timeout:
raise TimeoutError(f"视频处理超时（超过{timeout}秒）")
time.sleep(10)
uploaded_file = client.files.get(name=uploaded_file.name)
elapsed = time.time() - start_time
print(f" 处理中... 已等待 {elapsed:.0f} 秒")
if uploaded_file.state.name == "FAILED":
raise ValueError("视频处理失败")
print(f"处理完成: {uploaded_file.state.name}")
return uploaded_file
14.2 多模态处理相关
Q4: 如何提高图像分析的准确性？
def analyze_image_enhanced(image_path: str, question: str):
"""增强的图像分析"""
from PIL import Image
# 预处理：确保图像质量
image = Image.open(image_path)
# 如果图像太小，提示用户
width, height = image.size
if width < 100 or height < 100:
print("警告：图像分辨率较低，可能影响分析质量")
# 使用详细的提示词
enhanced_prompt = f"""请仔细分析这张图像。
分析要求：
1. 首先描述图像的整体内容
2. 然后关注细节：文字、数字、图表、颜色、布局
3. 如果有图表，请提取数据
4. 如果有文字，请准确识别
具体问题：{question}
请提供准确、详细的分析结果。"""
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[image, enhanced_prompt]
)
return response.text
Q5: 如何处理音频质量问题？
def analyze_noisy_audio(audio_path: str):
"""分析质量较差的音频"""
uploaded_file = client.files.upload(file=audio_path)
while uploaded_file.state.name == "PROCESSING":
time.sleep(5)
uploaded_file = client.files.get(name=uploaded_file.name)
response = client.models.generate_content(
model="gemini-2.5-pro",
contents=[
types.Part.from_uri(file_uri=uploaded_file.uri, mime_type=uploaded_file.mime_type),
"""请分析这段音频。注意：
- 音频可能有噪音或质量不佳
- 请尽力识别所有可听内容
- 对于不确定的部分，请标注[不确定]
- 请提供整体内容的概括"""
]
)
return response.text
14.3 性能优化
Q6: 如何优化大批量处理的性能？
import asyncio
from concurrent.futures import ThreadPoolExecutor
from google import genai
client = genai.Client()
async def optimized_batch_process(files: list, question: str,
max_concurrent: int = 5):
"""优化的批量处理"""
# 使用信号量控制并发
semaphore = asyncio.Semaphore(max_concurrent)
async def process_one(file_path: str):
async with semaphore:
# 在线程池中执行同步API调用
loop = asyncio.get_event_loop()
result = await loop.run_in_executor(
None,
lambda: analyzer.analyze(file_path, question)
)
return result
# 并发处理所有文件
tasks = [process_one(f) for f in files]
results = await asyncio.gather(*tasks, return_exceptions=True)
# 处理结果
successful = []
failed = []
for file_path, result in zip(files, results):
if isinstance(result, Exception):
failed.append((file_path, str(result)))
else:
successful.append(result)
return successful, failed
Q7: 如何控制API成本？
class CostController:
"""成本控制器"""
# Gemini 2.5 Pro 价格（示例）
PRICING = {
"input": 0.00000125, # $1.25 / 1M tokens
"output": 0.000005, # $5.00 / 1M tokens
}
def __init__(self, daily_budget: float = 10.0):
self.daily_budget = daily_budget
self.daily_cost = 0
def check_budget(self, estimated_tokens: int) -> bool:
"""检查是否超出预算"""
estimated_cost = estimated_tokens * self.PRICING["input"] / 1000000
return self.daily_cost + estimated_cost <= self.daily_budget
def record_usage(self, response):
"""记录使用量"""
input_tokens = response.usage_metadata.prompt_token_count
output_tokens = response.usage_metadata.candidates_token_count
cost = (input_tokens * self.PRICING["input"] +
output_tokens * self.PRICING["output"]) / 1000000
self.daily_cost += cost
return cost
14.4 安全与合规
Q8: 如何确保数据安全？
import vertexai
from vertexai.generative_models import GenerativeModel, SafetySetting
# 使用Vertex AI获得企业级安全
vertexai.init(project="your-project", location="us-central1")
# 配置严格的安全设置
safety_settings = [
SafetySetting(
category=cat,
threshold=SafetySetting.HarmBlockThreshold.BLOCK_LOW_AND_ABOVE
)
for cat in [
HarmCategory.HARM_CATEGORY_HATE_SPEECH,
HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
HarmCategory.HARM_CATEGORY_HARASSMENT,
HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
]
]
# 使用VPC网络确保数据不离开企业网络
# 配置数据加密和访问控制
# 启用审计日志
Q9: 如何处理敏感内容？
class ContentFilter:
"""内容过滤器"""
SENSITIVE_PATTERNS = [
r"\b\d{18}\b", # 身份证号
r"\b\d{16,19}\b", # 银行卡号
r"1[3-9]\d{9}", # 手机号
]
@classmethod
def sanitize_input(cls, text: str) -> str:
"""清理输入中的敏感信息"""
import re
for pattern in cls.SENSITIVE_PATTERNS:
text = re.sub(pattern, "[已脱敏]", text)
return text
@classmethod
def sanitize_output(cls, text: str) -> str:
"""清理输出中的敏感信息"""
import re
for pattern in cls.SENSITIVE_PATTERNS:
text = re.sub(pattern, "[已脱敏]", text)
return text
总结
本教程全面介绍了 Gemini 2.5 Pro 的核心功能和实战应用：
- 基础能力：API调用、流式输出、多轮对话、参数配置
- 超长上下文：百万Token窗口的实战应用
- 多模态理解：图像、视频、音频的原生理解能力
- 代码能力：代码生成、执行、解释、重构
- 工具集成：Function Calling、Google Search Grounding
- 自定义助手：Gems和多角色工作流
- 企业部署：Vertex AI的安全配置和成本管理
- 性能优化：错误处理、缓存、并发、成本控制
- 实战项目：完整的多模态文档分析系统
下一步学习建议
- 完成多模态文档分析系统实战项目
- 探索 Google AI Studio 进行交互式实验
- 学习 Vertex AI 的高级功能
- 关注 Gemini 模型更新和新特性
参考资源
本教程最后更新：2025年
