python - 如何管理第三方库的函数在一段时间后停止返回值？

Question

一个函数，即我的GPT4All的prompt函数一段时间后，对象停止返回值(我处理了一堆查询)，没有任何错误/异常。我尝试为该函数的执行设置超时，但当我将此提示函数的返回值写入 CSV 文件。

因此，我希望得到您的建议，通过简单地跳过该迭代并继续处理其他输入来克服这种情况。

这是我的脚本，我在调用中添加了注释，该注释在一段时间后停止返回值:

from nomic.gpt4all import GPT4All
import csv
from time import time   


CMU_CSV_PATH = 'data/cmu_qa.csv'

def single_chatgpt_offline(gpt, question):
    try:
        print(f'\tAsking {question}')
        resp = gpt.prompt(question)  # ----> This call stops returning a result after a while
        print(f'\tGot the response: {resp}')
        return resp
    except Exception as err:
        print(f'{str(err)}')

    return None

def find_answers_and_write_csv():
    # initialize GPT4
    gpt = GPT4All()
    gpt.open()

    with open(CMU_CSV_PATH, 'r', encoding='utf-8') as csv_src:
        csv_reader = csv.DictReader(csv_src)

        with open('data/cmu_qa_answers.csv', 'w', encoding='utf-8', newline='') as csv_target:
            fields = ['question', 'answer', 'title', 'bard', 'bard_time', 'gpt', 'gpt_time']
            csv_writer = csv.DictWriter(csv_target, fieldnames=fields)
            # write the header
            csv_writer.writeheader()

            for idx, line in enumerate(csv_reader):
                question = line['question']
                line['title'] = line['title'].replace('_', ' ')

                # GPT
                duration_gpt_start = time()
                resp_gpt = single_chatgpt_offline(gpt, question)
                if resp_gpt and '\n' in resp_gpt:
                    resp_gpt = resp_gpt.replace('\n', ' ')
                duration_gpt_end = time()
                line['gpt'] = resp_gpt
                line['gpt_time'] = round((duration_gpt_end - duration_gpt_start), NUM_DECIMAL)

                line["bard"] = None
                line["bard_time"] = None

                csv_writer.writerow(line)

Answer 1

import signal
def single_chatgpt_offline(gpt, question, timeout=10):
    try:
        print('Asking {}'.format(question))
    
        signal.alarm(timeout)
    
        resp = gpt.prompt(question)
        print('Got the response: {}'.format(resp))
        return resp
    except Exception as err:
        print(f'{str(err)}')
    finally:
        signal.alarm(0)

    return None

检查这是否有效:为提示功能添加超时。如果超时已过，您将跳转到 except block ，这允许您处理其他输入。

Finally block 禁用当前分配的超时，以便对提示函数的其他调用不受影响。