You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python限制算法调用时长并保证基线算法运行完成的工具咨询

实现方案

用Python标准库concurrent.futures即可实现需求,无需额外安装第三方库,以下是完整逻辑和示例代码:

核心逻辑说明

  • 基线算法A全程不会被中断,保证一定能运行完成
  • 10秒阈值内优先收集所有已完成算法的结果,可自行选择最优
  • 10秒超时后,仅等待基线算法A完成即可,可主动取消未完成的B、C任务释放资源

示例代码

from concurrent.futures import ProcessPoolExecutor, wait
import time

# 你的三个算法实现,这里用模拟函数替代
def algorithm_a():
    time.sleep(12)  # 模拟A运行12秒,超过10秒阈值
    return {"name": "A", "score": 0.85}

def algorithm_b():
    time.sleep(8)  # 模拟B运行8秒,在10秒内完成
    return {"name": "B", "score": 0.9}

def algorithm_c():
    time.sleep(15) # 模拟C运行15秒,超过10秒阈值
    return {"name": "C", "score": 0.92}

# 自定义的结果最优判断逻辑,你可以自行修改
def select_best_result(results):
    # 示例:按score字段选最高的
    return max(results, key=lambda x: x["score"])

if __name__ == "__main__":
    # 如果是IO密集型任务可以换成ThreadPoolExecutor
    with ProcessPoolExecutor(max_workers=3) as executor:
        # 提交三个算法任务,单独记录A的future对象
        future_a = executor.submit(algorithm_a)
        future_b = executor.submit(algorithm_b)
        future_c = executor.submit(algorithm_c)
        all_futures = {future_a, future_b, future_c}
        
        # 第一步:等待10秒,收集所有在此期间完成的任务
        done, pending = wait(all_futures, timeout=10)
        
        # 收集已完成的所有结果
        completed_results = [f.result() for f in done]
        best_result = None
        
        if len(completed_results) >= 1:
            # 10秒内有完成的任务,先选最优结果
            best_result = select_best_result(completed_results)
            print(f"10秒内已选出最优结果:{best_result}")
        
        # 第二步:强制等待基线算法A完成,保证A执行结束
        a_result = future_a.result()
        print(f"基线算法A运行完成,结果为:{a_result}")
        
        # 10秒内没选出最优的话,默认使用A的结果
        if best_result is None:
            best_result = a_result
        
        # 可选:取消还在运行的B、C任务,释放资源
        for f in pending:
            if f != future_a:
                f.cancel()
        
        print(f"最终使用的结果:{best_result}")

补充说明

  • 如果你的算法是CPU密集型,优先用ProcessPoolExecutor规避GIL带来的性能影响
  • 如果需要支持更复杂的超时重试、任务中断逻辑,也可以用第三方库timeout-decorator或者tenacity辅助实现,核心调度逻辑和上述示例一致

内容的提问来源于stack exchange,提问作者mommomonthewind

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.26 18:24:05