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

Question

一个函数，即我的

GPT4All

对象的 prompt 函数在一段时间后（我处理了一堆查询）停止返回值，没有任何错误/异常。我试图为该函数的执行设置超时，但是当我将此提示函数的返回值写入

[Errno 32] Broken pipe

文件时，此控制发生了

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

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 块，它允许您处理其他输入。

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