在业务系统里集成 PDF 翻译能力最常见的需求不是翻译一句话而是上传一个 PDF异步拿到翻译后的 PDF。这篇文章用 Python FastAPI 搭建一个最小可用的 PDF 翻译微服务覆盖文件上传、任务队列、翻译调用、结果回调四个核心环节。一、系统架构整个服务分为三层API 层FastAPI 提供/upload和/status/{task_id}接口。任务层后台线程池处理翻译任务避免阻塞请求。存储层本地目录保存上传文件和翻译结果。Client → FastAPI → Task Queue → Translator → Callback/Status API为了简化演示翻译环节调用一个模拟的翻译接口。生产环境中可以替换为真实的 PDF 翻译 API 或本地模型。二、环境准备Python 3.10FastAPI Uvicornaiofiles异步文件操作requests调用翻译 APIpipinstallfastapi uvicorn aiofiles requests python-multipart三、项目结构pdf_translate_service/ ├── main.py # FastAPI 主服务 ├── worker.py # 后台翻译任务 ├── storage/ # 上传文件和结果存储 └── callbacks/ # 回调日志四、核心代码实现1. 任务模型与内存状态# main.pyimportosimportuuidimportshutilfromdatetimeimportdatetimefromtypingimportOptionalfromfastapiimportFastAPI,File,UploadFile,BackgroundTasksfrompydanticimportBaseModelimportworker appFastAPI(titlePDF Translation Microservice)UPLOAD_DIRstorage/uploadsRESULT_DIRstorage/resultsos.makedirs(UPLOAD_DIR,exist_okTrue)os.makedirs(RESULT_DIR,exist_okTrue)# 内存中的任务状态生产环境建议用 Redistasks{}classTaskStatus(BaseModel):task_id:strstatus:str# pending / processing / completed / failedoriginal:Optional[str]Noneresult:Optional[str]Nonemessage:Optional[str]Nonecreated_at:strupdated_at:str2. 文件上传接口app.post(/upload,response_modelTaskStatus)asyncdefupload_pdf(background_tasks:BackgroundTasks,file:UploadFileFile(...),callback_url:Optional[str]None):ifnotfile.filename.endswith(.pdf):return{error:Only PDF files are supported}task_idstr(uuid.uuid4())upload_pathos.path.join(UPLOAD_DIR,f{task_id}.pdf)withopen(upload_path,wb)asf:shutil.copyfileobj(file.file,f)nowdatetime.utcnow().isoformat()tasks[task_id]{task_id:task_id,status:pending,original:upload_path,result:None,message:File uploaded, waiting for processing,created_at:now,updated_at:now,}background_tasks.add_task(worker.process_translation,task_id,upload_path,callback_url)returnTaskStatus(**tasks[task_id])3. 状态查询接口app.get(/status/{task_id},response_modelTaskStatus)asyncdefget_status(task_id:str):iftask_idnotintasks:return{error:Task not found}returnTaskStatus(**tasks[task_id])4. 后台翻译任务 worker.py# worker.pyimportosimporttimeimportrequestsfromdatetimeimportdatetime RESULT_DIRstorage/resultsdefprocess_translation(task_id:str,file_path:str,callback_url:strNone):模拟异步翻译任务frommainimporttasks tasks[task_id][status]processingtasks[task_id][message]Translating PDF...tasks[task_id][updated_at]datetime.utcnow().isoformat()try:# 模拟翻译耗时time.sleep(3)# 生产环境替换为真实翻译 API# response requests.post(# https://api.pdftranslator.org/v1/translate,# files{file: open(file_path, rb)},# data{target_lang: zh}# )# response.raise_for_status()result_pathos.path.join(RESULT_DIR,f{task_id}_translated.pdf)# 这里用原文件占位真实场景保存翻译后的 PDFwithopen(file_path,rb)assrc,open(result_path,wb)asdst:dst.write(src.read())tasks[task_id][status]completedtasks[task_id][result]result_path tasks[task_id][message]Translation completedifcallback_url:send_callback(callback_url,tasks[task_id])exceptExceptionase:tasks[task_id][status]failedtasks[task_id][message]str(e)tasks[task_id][updated_at]datetime.utcnow().isoformat()defsend_callback(url:str,payload:dict):try:requests.post(url,jsonpayload,timeout10)exceptrequests.RequestExceptionase:print(fCallback failed:{e})五、运行与测试启动服务uvicorn main:app--host0.0.0.0--port8000--reload上传 PDFcurl-XPOSThttp://localhost:8000/upload\-Ffilesample.pdf\-Fcallback_urlhttp://your-callback-server.com/notify查询状态curlhttp://localhost:8000/status/{task_id}六、生产环境建议任务队列内存tasks仅适合演示生产环境用 Celery Redis 或 RQ。文件存储上传到对象存储S3/OSS避免本地磁盘容量限制。并发控制翻译 API 通常有 QPS 限制用 Semaphore 或令牌桶限流。错误重试网络抖动时自动重试记录失败任务便于人工介入。回调可靠性回调失败时持久化到数据库支持手动补偿。七、总结这套 FastAPI 微服务的核心思路是把翻译一个 PDF拆成接收文件 → 异步处理 → 查询/回调结果三个标准化步骤。接口简单、易于集成也便于后续替换不同的翻译后端。如果你正在把 PDF 翻译能力接入自己的业务系统这个骨架应该能帮你快速跑通第一个版本。标签FastAPI、Python、PDF翻译、微服务、AI翻译