zbparse/flask_app/test_case/test_qianwen_long.py

import concurrent.futures
from flask_app.general.通义千问long import qianwen_long, upload_file


def multi_threaded_calls(file_id, user_query, num_threads=1):
    """
    使用多线程同时调用 qianwen_long 函数。
    参数：
    - file_id: 上传文件的 ID
    - user_query: 用户查询
    - num_threads: 并发线程数量（默认 10）
    """
    results = []

    # 定义一个辅助函数来调用 qianwen_long 并返回结果
    def call_function(thread_id):
        print(f"线程 {thread_id} 开始调用 qianwen_long")
        result = qianwen_long(file_id, user_query)
        # print(f"线程 {thread_id} 调用完成，结果长度：{len(result)}")
        return result

    # 使用 ThreadPoolExecutor 来管理线程池
    with concurrent.futures.ThreadPoolExecutor(max_workers=num_threads) as executor:
        # 提交所有任务
        futures = {executor.submit(call_function, i + 1): i + 1 for i in range(num_threads)}

        # 处理完成的任务
        for future in concurrent.futures.as_completed(futures):
            thread_id = futures[future]
            try:
                data = future.result()
                results.append(data)
                print(f"线程 {thread_id} 完成，返回数据：{data[:50]}...")  # 只打印前50个字符
            except Exception as exc:
                print(f"线程 {thread_id} 生成异常：{exc}")

    return results


if __name__ == "__main__":
    # 定义参数
    file_path = r"D:\flask_project\flask_app\static\output\output1\8bb07ee1-bcbb-4244-9d1e-367a783f1e40\invalid_del.docx"
    file_id = upload_file(file_path)
    # file_id = "file-fe-ah0F0SJUY2cEzPx5nONRBvwd"
    user_query = "该招标文件的项目概况是？项目基本情况是？请按json格式给我提供信息，键名分别为'项目概况','项目基本情况'，若存在嵌套信息，嵌套内容键名以文件中对应字段命名，而嵌套键值必须与原文保持一致，若存在未知信息，在对应的键值中填'未知'。"
    num_threads = 1  # 并发线程数量
    # 执行多线程调用
    responses = multi_threaded_calls(file_id, user_query, num_threads)
    # 打印所有响应
    for idx, response in enumerate(responses, start=1):
        print(f"响应 {idx}: {response}")