趋近智
有时,您的API端点需要执行一些不直接影响发送给客户端的响应的操作。例如,在成功返回预测结果后,您可能希望将请求和响应的详细信息记录到单独的文件或数据库中,发送通知,或启动清理流程。在发送响应之前执行这些操作会增加客户端不必要的延迟。
FastAPI提供了一种便捷的机制,称为后台任务,正是为了处理这些情况。后台任务是在响应已发送到客户端之后执行的函数。这确保了客户端能尽快收到响应,而服务器则独立处理这些后续操作。
可以将其比作在邮局寄送包裹。您的主要目标是获得包裹已寄出的收据(API响应)。打印一份关于包裹内容的详细内部报告(后台任务)可以在您带着收据离开后进行。
FastAPI通过在同一个事件循环中运行后台任务(或者为传递给后台任务的阻塞函数使用线程池)来管理它们,但重要的是,它不会在发送HTTP响应之前等待它们完成。
要使用后台任务,您需要:
fastapi导入BackgroundTasks类。BackgroundTasks的参数 (parameter)。FastAPI的依赖注入系统将自动提供此类的实例。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
在此示例中:
log_prediction_details,它接收输入和输出数据并将其写入文件prediction_log.jsonl。predict_and_log端点依赖于BackgroundTasks。prediction_result后,我们调用background_tasks.add_task(),传入我们的日志记录函数和必要的数据(input_dict,prediction_result)。prediction_result。log_prediction_details(input_dict, prediction_result)在稍后运行,而不会阻塞响应。尽管后台任务很有用,但它们也有局限性:
try...except块)和日志记录,以检测和诊断故障。后台任务提供了一种简单有效的方法,通过将非必要操作推迟到响应发送后执行,从而提高API端点的感知性能。它们在ML模型服务场景中尤其适用于日志记录、通知和触发其他非关键异步操作。请记住它们在可靠性和资源使用方面的局限性,并在核心操作不需要任务绝对成功完成时选择它们。
这部分内容有帮助吗?
asyncio的部分,对于理解FastAPI的事件循环很有帮助。© 2026 ApX Machine LearningAI伦理与透明度•