使用Google Bard API遇429错误的技术问题咨询

1. 出错时保存已获取的Bard响应列表 #

最可靠的方式是实时写入数据+异常触发时备份，避免因报错导致已获取的响应丢失。以下是Python实现示例：

import json
import time
import random
from bardapi import Bard

# 初始化Bard客户端和响应存储列表
bard = Bard(token="your_bard_token")
responses = []
input_sentences = ["语句1", "语句2", ...]  # 你的120条语句

for idx, sentence in enumerate(input_sentences):
    try:
        # 调用Bard API获取响应
        res = bard.get_answer(sentence)
        # 存储响应
        record = {"input": sentence, "response": res}
        responses.append(record)
        
        # 实时逐行写入JSONL文件（每行一条记录，便于后续解析）
        with open("bard_responses.jsonl", "a", encoding="utf-8") as f:
            json.dump(record, f, ensure_ascii=False)
            f.write("\n")
            
        # 正常请求后添加基础延迟
        time.sleep(random.uniform(3, 5))
        
    except Exception as e:
        print(f"处理第{idx+1}条语句失败: {str(e)}")
        # 出错时立即将已获取的完整列表备份到JSON文件
        with open("bard_responses_backup.json", "w", encoding="utf-8") as f:
            json.dump(responses, f, ensure_ascii=False, indent=2)
        # 根据需求选择：break终止循环，或continue继续处理后续语句
        break

这种方式的优势：

• 实时写入避免中途崩溃导致全部数据丢失
• 异常触发时的备份进一步保障已获取数据的安全性
• JSONL格式支持逐行读取，后续可快速恢复未完成的任务

2. Google Bard的调用限制及高效拉取方案 #

调用限制说明 #

Google官方未公开Bard API的具体每秒/每小时调用限制，且非官方封装的Bard客户端（如bardapi）的限制会动态调整，取决于账号状态、请求频率、请求内容等因素。根据实际测试经验：

• 单账号短时间内连续请求（无延迟）极易触发429限流
• 保持3-5秒的请求间隔可稳定获取大部分数据

高效拉取方案 #

方案1：分块处理+指数退避

将120条语句分成若干小批量（如每组10条），组间设置更长间隔（如30秒），组内每条保持3-5秒间隔；遇到429时采用指数退避策略（延迟时间翻倍）重试，比固定延迟更灵活：

for idx, sentence in enumerate(input_sentences):
    retry_count = 0
    max_retries = 3
    success = False
    
    while retry_count < max_retries and not success:
        try:
            res = bard.get_answer(sentence)
            record = {"input": sentence, "response": res}
            responses.append(record)
            
            with open("bard_responses.jsonl", "a", encoding="utf-8") as f:
                json.dump(record, f, ensure_ascii=False)
                f.write("\n")
                
            time.sleep(random.uniform(3, 5))
            success = True
            
        except Exception as e:
            if "429" in str(e):
                retry_count += 1
                delay = 2 ** retry_count * random.uniform(1, 2)
                print(f"触发429限流，{round(delay, 2)}秒后重试第{retry_count}次")
                time.sleep(delay)
            else:
                print(f"处理失败: {str(e)}")
                with open("bard_responses_backup.json", "w", encoding="utf-8") as f:
                    json.dump(responses, f, ensure_ascii=False, indent=2)
                break

方案2：Celery异步调度（适合大规模任务）

如果需要处理更多数据，可使用Celery将请求拆分为异步任务，通过配置任务并发数和任务间隔避免限流：

• 设置Celery worker的并发数为1（避免同时发起多个请求）
• 给每个任务添加固定延迟（如3-5秒）
• 同样结合实时写入和异常备份机制

不过对于仅120条语句的场景，分块+指数退避的方案已足够高效，无需引入Celery这类复杂工具。

内容的提问来源于stack exchange，提问作者StackSpin