使用后台任务

有时，您的API端点需要执行一些不直接影响发送给客户端的响应的操作。例如，在成功返回预测结果后，您可能希望将请求和响应的详细信息记录到单独的文件或数据库中，发送通知，或启动清理流程。在发送响应之前执行这些操作会增加客户端不必要的延迟。

FastAPI提供了一种便捷的机制，称为后台任务，正是为了处理这些情况。后台任务是在响应已发送到客户端之后执行的函数。这确保了客户端能尽快收到响应，而服务器则独立处理这些后续操作。

可以将其比作在邮局寄送包裹。您的主要目标是获得包裹已寄出的收据（API响应）。打印一份关于包裹内容的详细内部报告（后台任务）可以在您带着收据离开后进行。

后台任务的工作方式

FastAPI通过在同一个事件循环中运行后台任务（或者为传递给后台任务的阻塞函数使用线程池）来管理它们，但重要的是，它不会在发送HTTP响应之前等待它们完成。

要使用后台任务，您需要：

1. 从fastapi导入BackgroundTasks类。
2. 在您的路径操作函数中定义一个类型提示为BackgroundTasks的参数 (parameter)。FastAPI的依赖注入系统将自动提供此类的实例。
3. 在您的端点逻辑中，调用BackgroundTasks实例上的add_task()方法。此方法将要执行的函数作为第一个参数，后面是该函数所需的任何参数和关键字参数。

让我们看一个例子。假设我们有一个预测端点，并且我们希望在返回预测结果之后，将输入特征和预测结果记录到一个文件中。

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
import time
import json

# 假设 predict_model 函数存在并返回预测结果
# def predict_model(features):
#     # 模拟模型推理
#     time.sleep(0.1)
#     prediction = sum(features) * 0.5 # 模拟预测逻辑
#     return {"prediction": prediction}

# 假设 load_model 已在其他地方调用以获取 'model' 对象（如果需要）

app = FastAPI()

class FeaturesInput(BaseModel):
    sepal_length: float
    sepal_width: float
    petal_length: float
    petal_width: float

class PredictionOutput(BaseModel):
    prediction: float

def log_prediction_details(input_data: dict, output_data: dict):
    """一个用于将预测详情记录到文件的简单函数。"""
    log_entry = {"input": input_data, "output": output_data, "timestamp": time.time()}
    try:
        with open("prediction_log.jsonl", "a") as f:
            f.write(json.dumps(log_entry) + "\n")
        print("后台任务：已记录预测详情。")
    except Exception as e:
        # 重要提示：在后台任务本身内处理错误
        print(f"后台任务错误：记录预测详情失败：{e}")


@app.post("/predict_log_later", response_model=PredictionOutput)
async def predict_and_log(
    features: FeaturesInput,
    background_tasks: BackgroundTasks # 依赖注入
):
    """
    此端点用于进行预测并在后台记录详情。
    """
    input_dict = features.dict()
    # 替换为您的实际模型预测逻辑
    # 例如：result = model.predict([list(input_dict.values())])[0]
    # 使用模拟预测进行演示
    prediction_result = {"prediction": (input_dict["sepal_length"] + input_dict["petal_length"]) * 0.8}

    # 将日志记录函数添加到后台任务中
    background_tasks.add_task(
        log_prediction_details, # 要运行的函数
        input_dict,             # 函数的参数
        prediction_result       # 更多参数
    )

    # 响应在此处发送，在 log_prediction_details 完成之前
    return prediction_result

在此示例中：

1. 我们定义了log_prediction_details，它接收输入和输出数据并将其写入文件prediction_log.jsonl。
2. predict_and_log端点依赖于BackgroundTasks。
3. 在端点内部，获取prediction_result后，我们调用background_tasks.add_task()，传入我们的日志记录函数和必要的数据（input_dict，prediction_result）。
4. 端点立即返回prediction_result。
5. FastAPI确保log_prediction_details(input_dict, prediction_result)在稍后运行，而不会阻塞响应。

后台任务在ML API中的应用场景

• 详细日志记录：如上所示，记录大量信息而不会延迟客户端。
• 通知：发送关于特定预测结果的电子邮件或Slack消息（例如，检测到高风险预测）。
• 更新监控系统：将关于预测延迟、输入数据分布或模型性能的指标推送到监控仪表板。
• 简单缓存更新：使相关缓存条目失效或更新。
• 触发异步工作流程：向队列（如RabbitMQ或Kafka）发送消息，以启动更复杂的下游处理（例如，根据实时预测启动批处理分析）。

重要事项

尽管后台任务很有用，但它们也有局限性：

• 不保证执行：它们在与FastAPI应用程序相同的进程中运行。如果服务器进程在任务完成之前崩溃或重启，任务可能会丢失。它们不适用于必须完成的关键操作。为了保证执行，请使用专用的任务队列系统，如Celery或Redis Queue (RQ)。
• 错误处理：由于响应已经发送，后台任务中的错误不会自动通知客户端。您必须在后台任务函数内部实现错误处理（例如，try...except块）和日志记录，以检测和诊断故障。
• 资源消耗：后台任务仍然会消耗服务器资源（CPU、内存、I/O）。在后台运行过多计算密集型任务仍然可能降低API服务器对后续请求的整体性能和响应能力。如果您的后台任务涉及大量计算（如模型再训练），通常最好将其卸载到单独的工作进程或系统。

后台任务提供了一种简单有效的方法，通过将非必要操作推迟到响应发送后执行，从而提高API端点的感知性能。它们在ML模型服务场景中尤其适用于日志记录、通知和触发其他非关键异步操作。请记住它们在可靠性和资源使用方面的局限性，并在核心操作不需要任务绝对成功完成时选择它们。