前言

最近在做一个企业内部文档中台项目，需要把PDF翻译能力封装成微服务，供前端、IM机器人、定时任务等多个消费者调用。调研了一圈，发现市面上的方案要么太重（直接部署商业软件），要么太轻（纯脚本无法水平扩展）。

最终选型是：FastAPI + PDFTranslator API + Docker，轻量、异步、易部署。本文分享完整的架构设计和代码实现，读者可以直接复制使用。

环境准备

  • Python 3.10+
  • Docker & Docker Compose（可选，用于部署）
  • 依赖库：
pip install fastapi uvicorn httpx python-multipart aiofiles

架构设计

┌─────────────┐     ┌──────────────────┐     ┌─────────────────┐
│   前端/客户端  │────→│  FastAPI微服务    │────→│  PDFTranslator  │
│  /IM机器人   │     │  (翻译+任务管理)   │     │    翻译API      │
└─────────────┘     └──────────────────┘     └─────────────────┘
                            │
                            ↓
                    ┌──────────────────┐
                    │   Redis (可选)    │
                    │   任务队列/缓存    │
                    └──────────────────┘

核心设计原则：

  1. 异步处理：PDF翻译是IO密集型操作，FastAPI的异步能力可以高效处理并发
  2. 状态管理：翻译任务异步执行，客户端通过任务ID轮询进度
  3. 容错设计：网络异常时自动重试，失败任务可重新触发

实现步骤

Step 1: 项目结构

pdf-translate-service/
├── main.py              # FastAPI入口
├── translator.py        # 翻译核心逻辑
├── models.py            # Pydantic模型
├── requirements.txt
└── Dockerfile

Step 2: 数据模型定义

# models.py
from pydantic import BaseModel, Field
from typing import Optional, Literal
from datetime import datetime
from enum import Enum

class TaskStatus(str, Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"

class TranslateRequest(BaseModel):
    source_lang: str = Field(default="auto", description="源语言代码")
    target_lang: str = Field(..., description="目标语言代码，如 zh, en, de")
    webhook_url: Optional[str] = Field(None, description="完成后的回调URL")

class TranslateTask(BaseModel):
    task_id: str
    status: TaskStatus
    source_lang: str
    target_lang: str
    original_filename: str
    created_at: datetime
    completed_at: Optional[datetime] = None
    download_url: Optional[str] = None
    error_message: Optional[str] = None

Step 3: 翻译核心逻辑

# translator.py
import httpx
import uuid
import asyncio
from pathlib import Path
from typing import Optional

PDFTRANSLATOR_API = "https://api.pdftranslator.org/v1/translate"

class PDFTranslatorClient:
    """PDFTranslator API 异步客户端"""
    
    def __init__(self, api_key: Optional[str] = None):
        self.api_key = api_key
        self.client = httpx.AsyncClient(timeout=300.0)
    
    async def translate(
        self,
        file_path: Path,
        target_lang: str,
        source_lang: str = "auto"
    ) -> dict:
        """
        异步翻译PDF文件
        
        Args:
            file_path: PDF文件路径
            target_lang: 目标语言代码
            source_lang: 源语言代码，auto表示自动检测
            
        Returns:
            API响应字典，包含翻译结果URL
        """
        headers = {}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        
        data = {
            "source_lang": source_lang,
            "target_lang": target_lang
        }
        
        with open(file_path, "rb") as f:
            files = {"file": (file_path.name, f, "application/pdf")}
            response = await self.client.post(
                PDFTRANSLATOR_API,
                data=data,
                files=files,
                headers=headers
            )
        
        response.raise_for_status()
        return response.json()
    
    async def close(self):
        await self.client.aclose()

Step 4: FastAPI 主服务

# main.py
import uuid
import shutil
from datetime import datetime
from pathlib import Path
from typing import Dict

from fastapi import FastAPI, File, UploadFile, HTTPException, BackgroundTasks
from fastapi.responses import FileResponse

from models import TranslateRequest, TranslateTask, TaskStatus
from translator import PDFTranslatorClient

app = FastAPI(title="PDF Translation Microservice")

# 内存存储（生产环境建议用Redis + 持久化数据库）
tasks_db: Dict[str, TranslateTask] = {}
UPLOAD_DIR = Path("./uploads")
RESULT_DIR = Path("./results")
UPLOAD_DIR.mkdir(exist_ok=True)
RESULT_DIR.mkdir(exist_ok=True)

@app.post("/translate", response_model=TranslateTask)
async def create_translate_task(
    background_tasks: BackgroundTasks,
    request: TranslateRequest,
    file: UploadFile = File(...)
):
    """
    创建PDF翻译任务
    
    - 接收PDF文件和翻译参数
    - 返回任务ID，客户端通过 /tasks/{task_id} 查询进度
    """
    if not file.filename.endswith(".pdf"):
        raise HTTPException(400, detail="Only PDF files are supported")
    
    task_id = str(uuid.uuid4())
    upload_path = UPLOAD_DIR / f"{task_id}_{file.filename}"
    
    with open(upload_path, "wb") as f:
        content = await file.read()
        f.write(content)
    
    task = TranslateTask(
        task_id=task_id,
        status=TaskStatus.PENDING,
        source_lang=request.source_lang,
        target_lang=request.target_lang,
        original_filename=file.filename,
        created_at=datetime.now()
    )
    tasks_db[task_id] = task
    
    # 后台异步执行翻译
    background_tasks.add_task(
        process_translation,
        task_id=task_id,
        file_path=upload_path,
        target_lang=request.target_lang,
        source_lang=request.source_lang
    )
    
    return task

async def process_translation(
    task_id: str,
    file_path: Path,
    target_lang: str,
    source_lang: str
):
    """后台执行翻译任务"""
    task = tasks_db[task_id]
    client = PDFTranslatorClient()
    
    try:
        task.status = TaskStatus.PROCESSING
        
        # 调用翻译API（带重试）
        result = await translate_with_retry(
            client, file_path, target_lang, source_lang
        )
        
        # 保存结果（这里模拟下载翻译后的文件）
        result_path = RESULT_DIR / f"{task_id}_translated.pdf"
        await download_result(result["download_url"], result_path)
        
        task.status = TaskStatus.COMPLETED
        task.completed_at = datetime.now()
        task.download_url = f"/download/{task_id}"
        
    except Exception as e:
        task.status = TaskStatus.FAILED
        task.error_message = str(e)
    finally:
        await client.close()
        # 清理上传的原始文件
        if file_path.exists():
            file_path.unlink()

async def translate_with_retry(
    client: PDFTranslatorClient,
    file_path: Path,
    target_lang: str,
    source_lang: str,
    max_retries: int = 3
) -> dict:
    """带重试机制的翻译调用"""
    for attempt in range(max_retries):
        try:
            return await client.translate(file_path, target_lang, source_lang)
        except httpx.HTTPStatusError as e:
            if e.response.status_code >= 500 and attempt < max_retries - 1:
                wait_time = 2 ** attempt  # 指数退避
                await asyncio.sleep(wait_time)
                continue
            raise
    raise Exception("Max retries exceeded")

async def download_result(url: str, save_path: Path):
    """下载翻译结果"""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        response.raise_for_status()
        with open(save_path, "wb") as f:
            f.write(response.content)

@app.get("/tasks/{task_id}", response_model=TranslateTask)
async def get_task_status(task_id: str):
    """查询任务状态"""
    if task_id not in tasks_db:
        raise HTTPException(404, detail="Task not found")
    return tasks_db[task_id]

@app.get("/download/{task_id}")
async def download_translated_file(task_id: str):
    """下载翻译后的PDF"""
    if task_id not in tasks_db:
        raise HTTPException(404, detail="Task not found")
    
    task = tasks_db[task_id]
    if task.status != TaskStatus.COMPLETED:
        raise HTTPException(400, detail="Task not completed yet")
    
    result_path = RESULT_DIR / f"{task_id}_translated.pdf"
    if not result_path.exists():
        raise HTTPException(404, detail="Result file not found")
    
    return FileResponse(
        result_path,
        filename=f"translated_{task.original_filename}",
        media_type="application/pdf"
    )

@app.get("/health")
async def health_check():
    """健康检查端点"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

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.yml
version: '3.8'

services:
  pdf-translate-api:
    build: .
    ports:
      - "8000:8000"
    volumes:
      - ./uploads:/app/uploads
      - ./results:/app/results
    environment:
      - PYTHONUNBUFFERED=1
    restart: unless-stopped
    healthcheck:
      test: ["CMD", "curl", "-f", "http://localhost:8000/health"]
      interval: 30s
      timeout: 10s
      retries: 3

运行效果

启动服务

docker-compose up -d

提交翻译任务

curl -X POST "http://localhost:8000/translate" \
  -F "file=@document.pdf" \
  -F 'request={"target_lang":"zh"}'

查询任务状态

curl "http://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
}

生产环境建议

  1. 持久化存储：用PostgreSQL替代内存字典存储任务状态
  2. 消息队列：用Celery + Redis处理大文件翻译，避免阻塞API
  3. 限流：用FastAPI的依赖注入实现速率限制，防止滥用
  4. 监控：集成Prometheus + Grafana监控API健康度
  5. 认证：添加JWT或API Key认证，保护翻译接口

总结

这套方案的核心价值在于"轻量"和"可扩展"：

  • 单服务启动仅需几十MB内存
  • 异步架构天然支持高并发
  • Docker化部署适配K8s等容器编排平台
  • 与PDFTranslator的免费API结合，企业文档中台的翻译成本可以降到接近零

完整代码已在上文给出，直接复制即可运行。有问题欢迎在评论区讨论。

标签：PDF翻译、FastAPI、Python、微服务、Docker、AI翻译

# 架构 # fastapi # python # docker # 微服务
Logo
DAMO开发者矩阵

DAMO开发者矩阵，由阿里巴巴达摩院和中国互联网协会联合发起，致力于探讨最前沿的技术趋势与应用成果，搭建高质量的交流与分享平台，推动技术创新与产业应用链接，围绕“人工智能与新型计算”构建开放共享的开发者生态。

更多推荐

cover

2026国产协作机器人十大品牌盘点：谁更适合规模化部署？

avatar DAMO开发者矩阵
cover

2026 烟台自动化培训基地精选，数控培训成转行新路径

avatar DAMO开发者矩阵
cover

什么是AI数字员工？一文讲清这个正在走进企业的“新同事”

avatar DAMO开发者矩阵