Python 批量调用大模型API,一键搞定100条用户反馈摘要(可直接运行)

python                  pip install openai

python                  import openai                  import time                  import json                  from typing import List, Dict                  # 设置你的 API 密钥（生产环境建议从环境变量读取，更安全）                  openai.api_key = "your-api-key"                  def generate_summary(feedback_text: str, retries: int = 3) -> str:                  """                  调用 OpenAI API 生成单条反馈的摘要                  内置失败自动重试机制，指数退避规避高频报错                  """                  prompt = f"请用一句话概括以下用户反馈的核心内容：\n{feedback_text}\n摘要："                  for attempt in range(retries):                  try:                  response = openai.ChatCompletion.create(                  model="gpt-3.5-turbo",# 可替换为 gpt-4                  messages=[                  {"role": "system", "content": "你是一个专业的文本摘要助手，只输出简洁的摘要。不要添加额外评论。"},                  {"role": "user", "content": prompt}                  ],                  temperature=0.3,# 低温度，保证摘要结果稳定、一致                  max_tokens=100                  )                  summary = response.choices[0].message.content.strip()                  return summary                  except Exception as e:                  print(f"第 {attempt+1} 次调用失败：{e}")                  # 未达到最大重试次数，等待后重试                  if attempt < retries - 1:                  time.sleep(2 ** attempt)# 指数退避策略：1s、2s、4s 递增等待                  else:                  return f"摘要生成失败：{str(e)}"                  return ""                  def batch_summarize(feedbacks: List[str], delay: float = 0.5) -> List[Dict]:                  """                  批量处理反馈列表，生成结构化摘要结果                  :param feedbacks: 原始用户反馈文本列表                  :param delay: 单次请求间隔，控制调用频率，规避限流                  :return: 包含原文、摘要的结果列表                  """                  results = []                  total = len(feedbacks)                  for idx, fb in enumerate(feedbacks, 1):                  print(f"正在处理第 {idx}/{total} 条反馈...")                  summary = generate_summary(fb)                  results.append({                  "original": fb,                  "summary": summary                  })                  time.sleep(delay)# 匀速限流，保护API配额                  return results                  # 模拟数据加载（实际使用替换为CSV/数据库读取逻辑）                  def load_feedbacks() -> List[str]:                  """加载用户反馈数据，测试用生成100条模拟数据，生产环境自行替换"""                  base_feedbacks = [                  "这个 APP 很好用，但是偶尔会闪退。",                  "客服响应太慢了，等了半小时都没人回复。",                  "喜欢新版本的设计，功能很实用。",                  "希望增加夜间模式。",                  "支付流程太复杂，建议简化。"                  ]                  # 批量生成100条测试数据                  return (base_feedbacks * 20)[:100]                  if __name__ == "__main__":                  # 1. 加载原始反馈数据                  feedback_list = load_feedbacks()                  print(f"共加载 {len(feedback_list)} 条反馈。")                  # 2. 批量生成AI摘要                  start_time = time.time()                  summarized_data = batch_summarize(feedback_list, delay=0.5)                  elapsed = time.time() - start_time                  # 3. 结果持久化：保存为JSON文件，方便后续分析                  with open("feedback_summaries.json", "w", encoding="utf-8") as f:                  json.dump(summarized_data, f, ensure_ascii=False, indent=2)                  # 4. 输出任务统计信息                  print(f"\n处理完成！共 {len(summarized_data)} 条，耗时 {elapsed:.2f} 秒。")                  print("结果已保存到 feedback_summaries.json")

python                  import pandas as pd                  def load_feedbacks() -> List[str]:                  df = pd.read_csv("user_feedback.csv")                  return df["feedback_content"].tolist()

python                  import os                  openai.api_key = os.getenv("OPENAI_API_KEY")