我們如何在圖書翻譯平台中加入隨需翻譯輔助功能，串流傳送 LLM 對棘手段落的建議。

在 LectuLibre，我們的 AI 驅動圖書翻譯服務允許使用者上傳 EPUB 或 PDF 檔案，並由大型語言模型如 Claude 和 DeepSeek 產生翻譯。但我們很快發現一個痛點：自動化翻譯雖然快速，但對於文化特定用語、成語或技術術語，有時會產生生澀或模稜兩可的結果。使用者希望能在平台內即時取得這些棘手段落的上下文協助，而無需離開平台。於是我們著手建立 翻譯與轉錄求助（Translation Assistance）功能——一個互動側邊面板，讓使用者可以選取任何句子或段落，並即時從 LLM 取得替代翻譯、解釋和風格建議。

在本文中，我將帶領各位了解工程挑戰、我們選擇的架構，以及讓它能在生產環境限制下順利運作的具體程式碼與取捨。

問題：即時、具上下文感知的翻譯協助

核心需求很簡單：使用者在已翻譯的書籍中反白一段文字，點擊「取得協助」。系統應立即串流回傳多個翻譯選項、差異簡要說明，以及風格註記——所有內容皆需考量周圍上下文、作者風格與目標語言。

在底層，這意味著：

  • 低延遲：使用者期望在 2 秒內得到回應。
  • 串流：LLM 輸出可能很長，因此我們需要串流傳送產生的 token。
  • 上下文感知：我們必須包含足夠的書籍周圍文字，以為模型回應提供基礎。
  • 不阻塞：主要翻譯流程不應受到影響；協助功能應作為獨立的非同步服務存在。
  • 成本效益：避免每次使用者求助時重新處理整本書。

我們的做法：非同步 FastAPI + SSE + 速率限制

我們在 VPS 上執行 Python/FastAPI 後端，現有的翻譯流程會對 Claude 的 API 發出批次呼叫以進行整本書翻譯。針對協助功能，我們建立了獨立的端點，接受書本 ID、起始/結束位移，並透過 Server-Sent Events (SSE) 串流回傳協助。流程如下：

  1. 前端傳送包含所選文字位移的 POST 請求。
  2. 後端從 PostgreSQL 資料庫（已在擷取時分塊）取得周圍段落。
  3. 精心設計提示詞，包含所選文字、其上下文，以及多個翻譯變體的指示。
  4. 呼叫 Claude 的串流 API（messages.create(stream=True)）並將每個區塊轉發為 SSE 事件。

資料庫與擷取設定

當書籍上傳時，我們解析 EPUB/PDF、擷取文字，並將其儲存為分塊（段落），附帶位置與元資料。在 PostgreSQL 中，我們有一個 book_chunks 表格，包含 book_id、chunk_index、text 和 lang 欄位。這讓我們能透過簡單查詢快速取得包含所選位移的分塊以及幾個相鄰分塊：

async def get_context_chunks(book_id: str, chunk_index: int, context_size: int = 2):
    query = """
        SELECT chunk_index, text
        FROM book_chunks
        WHERE book_id = :book_id
          AND chunk_index BETWEEN :start_idx AND :end_idx
        ORDER BY chunk_index
    """
    values = {
        "book_id": book_id,
        "start_idx": max(0, chunk_index - context_size),
        "end_idx": chunk_index + context_size,
    }
    return await database.fetch_all(query, values)

進入全螢幕模式 退出全螢幕模式

這讓我們能在不將整本書載入記憶體的情況下取得周圍文字。

串流端點

我們使用 FastAPI 的 StreamingResponse 搭配產生器，該產生器會輸出 SSE 格式的訊息。關鍵在於 Claude SDK 的 stream=True 會回傳非同步迭代器，因此我們可以 await 每個區塊並立即輸出：

from fastapi import APIRouter, Request
from fastapi.responses import StreamingResponse
import anthropic

router = APIRouter()
client = anthropic.AsyncAnthropic()

@router.post("/assist/stream")
async def stream_assistance(request: Request):
    data = await request.json()
    book_id = data["book_id"]
    chunk_index = data["chunk_index"]
    selected_text = data["selected_text"]

    context_chunks = await get_context_chunks(book_id, chunk_index)
    context = "\n".join([c["text"] for c in context_chunks])

    prompt = build_assistance_prompt(context, selected_text)

    async def event_generator():
        try:
            async with client.messages.stream(
                model="claude-3-5-sonnet-20240620",
                max_tokens=1024,
                temperature=0.7,
                messages=[{"role": "user", "content": prompt}],
            ) as stream:
                async for text in stream.text_stream:
                    yield f"data: {text}\n\n"
                yield "data: [DONE]\n\n"
        except anthropic.APIStatusError as e:
            yield f"data: ERROR: {e.message}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

進入全螢幕模式 退出全螢幕模式

這讓前端獲得即時、逐字串流的體驗。得益於 stream.text_stream（Anthropic SDK 的核心功能），event_generator 是一個非同步產生器。

多變體的提示詞工程

為了取得實用輸出，我們設計了一個提示詞，要求提供三種不同的翻譯選項，每個選項附帶簡短說明。我們也指示模型保留上下文並提供風格指南。簡化版本如下：

def build_assistance_prompt(context: str, selected: str) -> str:
    return f"""You are a professional literary translator. Given the following context from a book:

{context}

The user has selected this passage for translation help:
"{selected}"

Provide:
1. Three alternative translations into the target language, labeled Option A, B, C.
2. For each option, a one-sentence explanation of the stylistic or semantic choice.
3. A final recommendation with justification.

Respond in JSON format with keys: options (list of dicts with translation, explanation), recommendation.
"""

進入全螢幕模式 退出全螢幕模式

接著我們在前端解析 JSON，以乾淨的方式顯示替代翻譯。串流文字會累積並在到達時顯示，給人即時回應的感覺。

取捨與挑戰

延遲與完整性

串流有助於感知效能，但對於 150-token 的上下文，第一個 token 的時間仍約為 1-2 秒。我們考慮過預先擷取上下文嵌入以加速檢索，但 LLM 呼叫的開銷占主導地位。因此我們透過以下方式優化：

  • 使用 async SDK 避免阻塞事件迴圈。
  • 在 Redis 中實作簡單快取，針對相同查詢（同一本書、同一個分塊、同一個所選文字）快取，TTL 為 5 分鐘。這減少了約 30% 的重複請求。
  • 選擇相對較小的 max_tokens (1024)，因為回應較短。

成本管理

每次協助呼叫都會消耗 token。使用 Claude 3.5 Sonnet，輸入 token 價格為 $3/MTok，輸出為 $15/MTok。我們追蹤每位使用者的使用量，並實作每日 50 次協助請求的限制。若使用者超過限制，會收到禮貌訊息。這是使用 Redis 的簡單速率限制器完成的：

import aioredis
import time

redis = aioredis.from_url("redis://localhost")

async def check_rate_limit(user_id: str, limit: int = 50, window: int = 86400) -> bool:
    key = f"assist_rate:{user_id}"
    current = await redis.incr(key)
    if current == 1:
        await redis.expire(key, window)
    return current <= limit

進入全螢幕模式 退出全螢幕模式

模型選擇

我們最初使用 DeepSeek，因為它較便宜，但文學細膩度的翻譯品質明顯較低，且串流 API 穩定性較差。Claude 的輸出更豐富，因此儘管成本較高，我們仍選擇使用它。對於非文學作品，我們可能會提供切換至 DeepSeek 的選項。

上下文視窗與 token 限制

我們將上下文限制在 3 個段落（約 500 tokens），以保持在模型 200K 上下文視窗內，同時降低提示詞成本。這通常足以讓模型理解語氣與風格。

結果與經驗教訓

部署後，「翻譯與轉錄求助」功能迅速成為 LectuLibre 使用率最高的功能之一。使用者喜愛其即時性，以及能並排查看多個翻譯選項的能力。從點擊到第一個 token 的平均回應時間為 1.4 秒，完整回應（3 個選項）約需 5-7 秒，使用者認為可以接受。

關鍵收穫：

  • 串流是即時 LLM 應用不可或缺的；務必使用 SSE 或 WebSocket。
  • 快取有回報：簡單的查詢層級快取可大幅降低負載與成本。
  • 非同步真的很重要：使用 asyncio 和非同步 SDK 可防止在負載下執行緒耗盡。
  • 提示詞結構控制品質與前端解析：要求 JSON 輸出可簡化 UI 渲染，但可能限制模型的創意；我們找到一個中間地帶，指示 JSON 格式，但允許在其中自由形式說明。

接下來呢？

我們正在考慮加入使用者回饋以微調協助模型，並可能將此功能作為獨立 API 提供給譯者。我們很想聽聽其他團隊如何處理大規模即時 LLM 串流——特別是關於批次處理請求和降低 token 成本。歡迎在評論區分享您的想法！