解决Google Gemini Pro API多轮对话中的Blob类型错误

1. 问题重现与错误分析

最近在使用Google Gemini Pro API进行多轮对话开发时，遇到了一个典型的类型错误。这个错误表面看起来是Blob对象创建失败，但实际上涉及更深层次的API使用问题。让我们先完整重现这个场景：

python复制import google.generativeai as genai

# 初始化模型
model = genai.GenerativeModel("gemini-pro")

# 初始对话
messages = [{'role':'user', 'parts': ['hello']}]
response = model.generate_content(messages)  # 第一轮响应正常："Hello, how can I help"

# 添加响应到对话历史
messages.append(response.candidates[0].content)
messages.append({'role':'user', 'parts': ['How does quantum physics work?']})

# 第二轮对话抛出错误
response = model.generate_content(messages)  # TypeError: Could not create `Blob`

1.1 错误根源解析

这个错误的核心在于消息格式的不一致性。当我们查看Gemini Pro API的官方文档时会发现：

1. 输入消息需要严格遵循特定的协议缓冲区格式
2. response.candidates[0].content返回的是Content对象，而不是原始的消息字典
3. 直接混合使用字典和Content对象会导致序列化失败

关键提示：Gemini Pro的多轮对话要求所有消息必须保持格式一致，要么全部使用原始字典格式，要么全部使用生成的Content对象格式。

2. 正确的多轮对话实现方案

2.1 方案一：统一使用字典格式

这是最接近原始代码的修改方案，适合需要完全控制消息内容的场景：

python复制messages = [
    {'role': 'user', 'parts': ['hello']}
]
response = model.generate_content(messages)

# 正确添加AI响应的方式
messages.append({
    'role': 'model',
    'parts': [response.text]  # 使用.text获取纯文本响应
})

# 添加用户新消息
messages.append({
    'role': 'user', 
    'parts': ['How does quantum physics work?']
})

# 继续对话
response = model.generate_content(messages)

2.2 方案二：使用ChatSession简化流程

Gemini Pro其实提供了专门的ChatSession类来简化多轮对话管理：

python复制chat = model.start_chat(history=[])
response = chat.send_message("hello")  # 第一轮
print(response.text)

response = chat.send_message("How does quantum physics work?")  # 第二轮
print(response.text)

# 自动维护对话历史
print(chat.history)  # 查看完整对话记录

2.3 两种方案的对比分析

3. 深入理解消息结构

3.1 协议缓冲区基础

Gemini Pro API底层使用Protocol Buffers进行数据传输。一个标准的消息结构包含：

1. role: 只能是"user"或"model"
2. parts: 消息内容数组，每个元素可以是：
   • 纯文本字符串
   • 内联数据（如图片二进制）
   • 文件引用

3.2 内容类型转换

当需要混合使用API响应和手动构造消息时，正确的类型转换方法：

python复制# 将Content对象转换为字典
def content_to_dict(content):
    return {
        'role': 'model',
        'parts': [part.text for part in content.parts]
    }

# 使用示例
messages.append(content_to_dict(response.candidates[0].content))

4. 常见问题与解决方案

4.1 错误类型及修复方法

4.2 调试技巧

1. 打印完整消息历史：

   python复制import pprint
   pp = pprint.PrettyPrinter(indent=2)
   pp.pprint(messages)
   

2. 验证单个消息：

   python复制try:
       test_blob = genai.protos.Content(**messages[0])
   except Exception as e:
       print(f"Invalid message: {e}")
   

3. 使用类型检查：

   python复制from google.protobuf.internal import containers
   if isinstance(messages[0], containers.Message):
       print("这是Protocol Buffer对象")
   elif isinstance(messages[0], dict):
       print("这是字典对象")
   

5. 高级应用技巧

5.1 混合内容类型对话

Gemini Pro支持在对话中混合文本和图片（需使用gemini-pro-vision模型）：

python复制# 上传图片
import PIL.Image
img = PIL.Image.open('image.jpg')

# 构造混合消息
mixed_message = {
    'role': 'user',
    'parts': [
        "请分析这张图片",
        img  # 直接传入PIL图像对象
    ]
}

5.2 自定义元数据

可以通过metadata字段为消息添加额外信息：

python复制custom_message = {
    'role': 'user',
    'parts': ['常规消息'],
    'metadata': {
        'timestamp': '2024-03-15',
        'source': 'mobile_app'
    }
}

5.3 流式响应处理

对于长响应内容，可以使用流式获取：

python复制response = model.generate_content(
    messages,
    stream=True
)

for chunk in response:
    print(chunk.text)
    print("---")  # 分隔符

6. 性能优化建议

1. 历史消息截断：当对话轮次过多时，适当截断早期历史

   python复制MAX_HISTORY = 10
   if len(messages) > MAX_HISTORY:
       messages = messages[-MAX_HISTORY:]
   

2. 并行请求处理：使用异步接口提高吞吐量

   python复制import asyncio
   async def async_chat():
       model = genai.GenerativeModel("gemini-pro")
       chat = model.start_chat()
       tasks = [
           chat.send_message_async("消息1"),
           chat.send_message_async("消息2")
       ]
       return await asyncio.gather(*tasks)
   

3. 缓存机制：对常见问题响应建立本地缓存

   python复制from functools import lru_cache
   
   @lru_cache(maxsize=100)
   def get_cached_response(prompt):
       return model.generate_content([{'role':'user','parts':[prompt]}])
   

7. 安全注意事项

1. API密钥保护：

   python复制# 错误做法：硬编码密钥
   # genai.configure(api_key="your-key-here")
   
   # 正确做法：从环境变量读取
   import os
   genai.configure(api_key=os.getenv('GEMINI_API_KEY'))
   

2. 输入验证：

   python复制def sanitize_input(text):
       if len(text) > 1000:
           raise ValueError("输入过长")
       return text.strip()
   

3. 错误重试机制：

   python复制from tenacity import retry, stop_after_attempt
   
   @retry(stop=stop_after_attempt(3))
   def safe_generate(messages):
       return model.generate_content(messages)
   

在实际项目中，我建议优先使用ChatSession方案，它不仅能避免本文提到的Blob错误，还能自动处理许多底层细节。对于需要精细控制的场景，再考虑手动管理消息列表的方式。无论哪种方案，保持消息格式的一致性都是成功实现多轮对话的关键。