我们如何为图书翻译平台添加按需翻译辅助功能，流式传输 LLM 对棘手段落的建议。

在 LectuLibre，我们的 AI 驱动图书翻译服务允许用户上传 EPUB 或 PDF 文件，并获取由 Claude 和 DeepSeek 等大语言模型生成的翻译。但我们很快注意到一个痛点：虽然自动化翻译速度很快，但对于具有文化特性的短语、习语或技术术语，有时会产生尴尬或模棱两可的结果。用户希望有一种方式在不离开平台的情况下获得对这些棘手段落的即时、情境化的帮助。就在那时，我们开始构建 翻译与转录求助（Translation Assistance）功能——一个交互式侧边栏，用户可以选择任何句子或段落，并实时从 LLM 接收替代翻译、解释和风格建议。

在本文中，我将带你了解工程挑战、我们选择的架构，以及在生产限制下使其顺利运行的具体代码和权衡。

问题：实时、情境感知的翻译帮助

核心需求很简单：用户在翻译后的图书中高亮一段文本并点击“获取帮助”。系统应立即流式传输多个翻译选项、差异的简要解释以及风格说明——所有内容都应感知到周围的上下文、作者的风格和目标语言。

在底层，这意味着：

  • 低延迟：用户期望在 2 秒内获得响应。
  • 流式传输：LLM 输出可能很长，因此我们需要流式传输生成的 token。
  • 情境感知：我们必须包含书中足够的周围文本，以支撑模型的响应。
  • 无阻塞：主翻译流程不应受到影响；辅助功能应作为独立的异步服务存在。
  • 成本效率：避免每次用户请求帮助时都重新处理整本书。

我们的方法：异步 FastAPI + SSE + 速率限制

我们在 VPS 上运行 Python/FastAPI 后端，现有的翻译流程会批量调用 Claude 的 API 进行全书翻译。对于辅助功能，我们构建了一个单独的端点，它接受图书 ID、起始/结束偏移量，并通过服务器发送事件（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")

进入全屏模式 退出全屏模式

这为前端提供了实时、逐字的流式体验。得益于 Anthropic SDK 的核心特性 stream.text_stream，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 个 token），以保持在模型 200K 上下文窗口内，同时保持提示成本较低。这通常足以让模型理解语气和风格。

结果与经验教训

部署后，“翻译与转录求助”功能迅速成为 LectuLibre 使用率最高的部分之一。用户喜欢它的即时性以及能够并排查看多个翻译选项的能力。从点击到首 token 的平均响应时间为 1.4 秒，完整响应（3 个选项）大约需要 5-7 秒，用户认为这是可以接受的。

主要经验教训：

  • 流式传输对于实时 LLM 应用是不可或缺的；始终使用 SSE 或 WebSockets。
  • 缓存带来回报：简单的查询级缓存可以显著减少负载和成本。
  • 异步确实重要：使用 asyncio 和异步 SDK 可防止在负载下线程耗尽。
  • 提示结构控制质量和前端解析：要求 JSON 输出简化了 UI 渲染，但可能会限制模型的创造力；我们通过指示 JSON 格式但允许在其中进行自由格式解释，找到了一个折中方案。

下一步是什么？

我们正在考虑添加用户反馈以微调辅助模型，并可能将该功能作为独立 API 提供给翻译人员。我们很想了解其他团队如何在大规模实时 LLM 流式传输方面进行处理——特别是在批处理请求和降低 token 成本方面。请在评论中分享您的想法！