Python多线程生成GPT博客内容时线程内函数调用失效问题

问题现象 #

• 单线程串行执行时程序逻辑正常，但单篇文章生成耗时超过2分钟，批量生成效率极低
• 尝试使用多线程、多进程方案优化生成效率时，程序无法按预期运行
• 已尝试将全量业务逻辑整合到单函数、函数内嵌套定义子函数等方案，问题均未解决

预期逻辑与实际故障 #

设计预期：qform函数从文件读取主题列表，基于单个主题生成6组对应问题，调用GPT接口拿到生成内容后，通过wppost函数完成文章发布
实际表现：测试打印确认多线程对qform的调度本身正常，但qform内部没有按预期调用6次GPT接口，后续的文章发布函数wppost完全没有触发

原始问题代码如下（已忽略笔误类小错误）：

def wppost(title,content):
 r=post(title,content)
 print(r)

def gpt(q):
 gr=openai.completion(prompt=q)

def qform():
 with open('file) as f:
  for line in f:
   q1='what is '+f
   q2='were is '+f
   q3....q6
   gptq=gpt(q=q1+...q6)

if __name__ == '__main__':
 with open file as fm:
  with concurrent.futures.ThreadPoolExecutor() as executor:
   for i in enumerate(fm):
    executor.submit(qform)

核心问题点 #

• 并发逻辑设计错误：主程序已经打开主题文件遍历，每个提交到线程池的qform任务又重复打开同一个文件全量读取，多线程同时抢占文件IO指针，会导致读内容为空、乱序，直接打断后续执行流
• 变量引用错误：qform内循环遍历文件拿到的单行主题是line变量，实际拼接prompt时误用了文件对象f，生成的prompt完全不符合接口要求
• 调用链路断裂：代码里只调用了1次gpt函数，和设计要求的6次调用不符；且gpt函数没有返回生成结果，执行完GPT调用后也没有触发wppost发布逻辑，内容生成完直接被丢弃
• 基础语法错误：文件打开语句缺引号、缺括号、缺打开模式，运行时会直接抛异常终止任务，而线程池默认不会把子线程的异常直接抛到主线程，所以表面看是“函数没被触发”，实际是子线程跑一半报错退出了

修复参考代码 #

import concurrent.futures
from openai import OpenAI
# 提前初始化客户端，避免每个线程重复初始化开销
client = OpenAI()

def wppost(title, content):
    r = post(title, content)
    print(f"文章发布结果：{r}")

def gpt(q):
    # 接口调用加异常捕获，避免单条请求失败终止整个线程
    try:
        resp = client.completions.create(prompt=q, model="gpt-3.5-turbo-instruct")
        return resp.choices[0].text.strip()
    except Exception as e:
        print(f"GPT调用失败：{str(e)}")
        return None

def qform(topic_line):
    # 不再重复读文件，直接接收主线程传过来的单个主题
    topic = topic_line.strip()
    if not topic:
        return
    # 按设计生成6个问题
    q_list = [
        f"what is {topic}",
        f"where is {topic} applicable",
        f"why is {topic} important",
        f"how to use {topic}",
        f"what are the pros of {topic}",
        f"what are the cons of {topic}"
    ]
    # 依次调用6次GPT，拼接内容
    content_list = []
    title = f"Introduction to {topic}"
    for q in q_list:
        res = gpt(q)
        if res:
            content_list.append(res)
    full_content = "\n".join(content_list)
    # 内容生成完触发发布
    if full_content:
        wppost(title, full_content)

if __name__ == '__main__':
    # 主线程一次性读完全部主题，再把单个主题提交给线程池处理
    with open('topic_list.txt', 'r', encoding='utf-8') as fm:
        topic_list = [line for line in fm if line.strip()]
    # 线程数不要开太大，GPT接口有RPM限制，一般开3-5个即可
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as executor:
        executor.map(qform, topic_list)

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