前言最近在做一个企业内部文档中台项目需要把PDF翻译能力封装成微服务供前端、IM机器人、定时任务等多个消费者调用。调研了一圈发现市面上的方案要么太重直接部署商业软件要么太轻纯脚本无法水平扩展。最终选型是FastAPI PDFTranslator API Docker轻量、异步、易部署。本文分享完整的架构设计和代码实现读者可以直接复制使用。环境准备Python 3.10Docker Docker Compose可选用于部署依赖库pipinstallfastapi uvicorn httpx python-multipart aiofiles架构设计┌─────────────┐ ┌──────────────────┐ ┌─────────────────┐ │ 前端/客户端 │────→│ FastAPI微服务 │────→│ PDFTranslator │ │ /IM机器人 │ │ (翻译任务管理) │ │ 翻译API │ └─────────────┘ └──────────────────┘ └─────────────────┘ │ ↓ ┌──────────────────┐ │ Redis (可选) │ │ 任务队列/缓存 │ └──────────────────┘核心设计原则异步处理PDF翻译是IO密集型操作FastAPI的异步能力可以高效处理并发状态管理翻译任务异步执行客户端通过任务ID轮询进度容错设计网络异常时自动重试失败任务可重新触发实现步骤Step 1: 项目结构pdf-translate-service/ ├── main.py # FastAPI入口 ├── translator.py # 翻译核心逻辑 ├── models.py # Pydantic模型 ├── requirements.txt └── DockerfileStep 2: 数据模型定义# models.pyfrompydanticimportBaseModel,FieldfromtypingimportOptional,LiteralfromdatetimeimportdatetimefromenumimportEnumclassTaskStatus(str,Enum):PENDINGpendingPROCESSINGprocessingCOMPLETEDcompletedFAILEDfailedclassTranslateRequest(BaseModel):source_lang:strField(defaultauto,description源语言代码)target_lang:strField(...,description目标语言代码如 zh, en, de)webhook_url:Optional[str]Field(None,description完成后的回调URL)classTranslateTask(BaseModel):task_id:strstatus:TaskStatus source_lang:strtarget_lang:stroriginal_filename:strcreated_at:datetime completed_at:Optional[datetime]Nonedownload_url:Optional[str]Noneerror_message:Optional[str]NoneStep 3: 翻译核心逻辑# translator.pyimporthttpximportuuidimportasynciofrompathlibimportPathfromtypingimportOptional PDFTRANSLATOR_APIhttps://api.pdftranslator.org/v1/translateclassPDFTranslatorClient:PDFTranslator API 异步客户端def__init__(self,api_key:Optional[str]None):self.api_keyapi_key self.clienthttpx.AsyncClient(timeout300.0)asyncdeftranslate(self,file_path:Path,target_lang:str,source_lang:strauto)-dict: 异步翻译PDF文件 Args: file_path: PDF文件路径 target_lang: 目标语言代码 source_lang: 源语言代码auto表示自动检测 Returns: API响应字典包含翻译结果URL headers{}ifself.api_key:headers[Authorization]fBearer{self.api_key}data{source_lang:source_lang,target_lang:target_lang}withopen(file_path,rb)asf:files{file:(file_path.name,f,application/pdf)}responseawaitself.client.post(PDFTRANSLATOR_API,datadata,filesfiles,headersheaders)response.raise_for_status()returnresponse.json()asyncdefclose(self):awaitself.client.aclose()Step 4: FastAPI 主服务# main.pyimportuuidimportshutilfromdatetimeimportdatetimefrompathlibimportPathfromtypingimportDictfromfastapiimportFastAPI,File,UploadFile,HTTPException,BackgroundTasksfromfastapi.responsesimportFileResponsefrommodelsimportTranslateRequest,TranslateTask,TaskStatusfromtranslatorimportPDFTranslatorClient appFastAPI(titlePDF Translation Microservice)# 内存存储生产环境建议用Redis 持久化数据库tasks_db:Dict[str,TranslateTask]{}UPLOAD_DIRPath(./uploads)RESULT_DIRPath(./results)UPLOAD_DIR.mkdir(exist_okTrue)RESULT_DIR.mkdir(exist_okTrue)app.post(/translate,response_modelTranslateTask)asyncdefcreate_translate_task(background_tasks:BackgroundTasks,request:TranslateRequest,file:UploadFileFile(...)): 创建PDF翻译任务 - 接收PDF文件和翻译参数 - 返回任务ID客户端通过 /tasks/{task_id} 查询进度 ifnotfile.filename.endswith(.pdf):raiseHTTPException(400,detailOnly PDF files are supported)task_idstr(uuid.uuid4())upload_pathUPLOAD_DIR/f{task_id}_{file.filename}withopen(upload_path,wb)asf:contentawaitfile.read()f.write(content)taskTranslateTask(task_idtask_id,statusTaskStatus.PENDING,source_langrequest.source_lang,target_langrequest.target_lang,original_filenamefile.filename,created_atdatetime.now())tasks_db[task_id]task# 后台异步执行翻译background_tasks.add_task(process_translation,task_idtask_id,file_pathupload_path,target_langrequest.target_lang,source_langrequest.source_lang)returntaskasyncdefprocess_translation(task_id:str,file_path:Path,target_lang:str,source_lang:str):后台执行翻译任务tasktasks_db[task_id]clientPDFTranslatorClient()try:task.statusTaskStatus.PROCESSING# 调用翻译API带重试resultawaittranslate_with_retry(client,file_path,target_lang,source_lang)# 保存结果这里模拟下载翻译后的文件result_pathRESULT_DIR/f{task_id}_translated.pdfawaitdownload_result(result[download_url],result_path)task.statusTaskStatus.COMPLETED task.completed_atdatetime.now()task.download_urlf/download/{task_id}exceptExceptionase:task.statusTaskStatus.FAILED task.error_messagestr(e)finally:awaitclient.close()# 清理上传的原始文件iffile_path.exists():file_path.unlink()asyncdeftranslate_with_retry(client:PDFTranslatorClient,file_path:Path,target_lang:str,source_lang:str,max_retries:int3)-dict:带重试机制的翻译调用forattemptinrange(max_retries):try:returnawaitclient.translate(file_path,target_lang,source_lang)excepthttpx.HTTPStatusErrorase:ife.response.status_code500andattemptmax_retries-1:wait_time2**attempt# 指数退避awaitasyncio.sleep(wait_time)continueraiseraiseException(Max retries exceeded)asyncdefdownload_result(url:str,save_path:Path):下载翻译结果asyncwithhttpx.AsyncClient()asclient:responseawaitclient.get(url)response.raise_for_status()withopen(save_path,wb)asf:f.write(response.content)app.get(/tasks/{task_id},response_modelTranslateTask)asyncdefget_task_status(task_id:str):查询任务状态iftask_idnotintasks_db:raiseHTTPException(404,detailTask not found)returntasks_db[task_id]app.get(/download/{task_id})asyncdefdownload_translated_file(task_id:str):下载翻译后的PDFiftask_idnotintasks_db:raiseHTTPException(404,detailTask not found)tasktasks_db[task_id]iftask.status!TaskStatus.COMPLETED:raiseHTTPException(400,detailTask not completed yet)result_pathRESULT_DIR/f{task_id}_translated.pdfifnotresult_path.exists():raiseHTTPException(404,detailResult file not found)returnFileResponse(result_path,filenameftranslated_{task.original_filename},media_typeapplication/pdf)app.get(/health)asyncdefhealth_check():健康检查端点return{status:healthy,timestamp:datetime.now().isoformat()}if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)Step 5: Docker 部署配置# Dockerfile FROM python:3.11-slim WORKDIR /app COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt COPY . . RUN mkdir -p uploads results EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]# docker-compose.ymlversion:3.8services:pdf-translate-api:build:.ports:-8000:8000volumes:-./uploads:/app/uploads-./results:/app/resultsenvironment:-PYTHONUNBUFFERED1restart:unless-stoppedhealthcheck:test:[CMD,curl,-f,http://localhost:8000/health]interval:30stimeout:10sretries:3运行效果启动服务docker-composeup-d提交翻译任务curl-XPOSThttp://localhost:8000/translate\-Ffiledocument.pdf\-Frequest{target_lang:zh}查询任务状态curlhttp://localhost:8000/tasks/{task_id}响应示例{task_id:550e8400-e29b-41d4-a716-446655440000,status:completed,source_lang:auto,target_lang:zh,original_filename:document.pdf,created_at:2026-07-28T10:30:00,completed_at:2026-07-28T10:32:15,download_url:/download/550e8400-e29b-41d4-a716-446655440000,error_message:null}生产环境建议持久化存储用PostgreSQL替代内存字典存储任务状态消息队列用Celery Redis处理大文件翻译避免阻塞API限流用FastAPI的依赖注入实现速率限制防止滥用监控集成Prometheus Grafana监控API健康度认证添加JWT或API Key认证保护翻译接口总结这套方案的核心价值在于轻量和可扩展单服务启动仅需几十MB内存异步架构天然支持高并发Docker化部署适配K8s等容器编排平台与PDFTranslator的免费API结合企业文档中台的翻译成本可以降到接近零完整代码已在上文给出直接复制即可运行。有问题欢迎在评论区讨论。标签PDF翻译、FastAPI、Python、微服务、Docker、AI翻译
