書籍翻訳プラットフォームにオンデマンド翻訳ヘルプ機能を追加し、難解な箇所に対する LLM の提案をストリーミング配信する方法。
LectuLibre では、AI を活用した書籍翻訳サービスにより、ユーザーは EPUB や PDF ファイルをアップロードして、Claude や DeepSeek などの大規模言語モデルが生成した翻訳を入手できます。しかし、自動翻訳は高速である一方、文化特有の表現や熟語、専門用語などで不自然または曖昧な結果を生むことがあることがすぐに分かりました。ユーザーはプラットフォームを離れることなく、こうした難解な箇所に対する即時的かつ文脈を考慮した支援を求めていました。そこで私たちは、ユーザーが任意の文や段落を選択すると、代替翻訳や解説、文体に関する提案を LLM からリアルタイムで受け取れるインタラクティブなサイドパネル「翻訳与转录求助（翻訳アシスタンス）」機能を構築しました。
本記事では、技術的課題、アーキテクチャの選択、そして本番環境の制約下でスムーズに動作させるための具体的なコードとトレードオフについて解説します。
課題：リアルタイムで文脈を考慮した翻訳ヘルプ
基本要件はシンプルです。ユーザーが翻訳済みの書籍内でテキストをハイライトし、「支援を取得」をクリックすると、システムは即座に複数の翻訳オプション、差異の簡単な解説、文体に関する注記を、周辺文脈や著者のスタイル、ターゲット言語を考慮してストリーミング配信します。
内部的には以下を意味します。
- 低遅延：ユーザーは 2 秒未満の応答を期待します。
- ストリーミング：LLM の出力は長くなり得るため、生成されたトークンをストリーミングする必要があります。
- 文脈認識：モデルの応答を根拠づけるため、書籍から十分な周辺テキストを含める必要があります。
- ノンブロッキング：メインの翻訳パイプラインに影響を与えず、アシスタンス機能は独立した非同期サービスとして存在する必要があります。
- コスト効率：ユーザーが支援を求めるたびに書籍全体を再処理しない。
アプローチ：Async FastAPI + SSE + レート制限
Python/FastAPI バックエンドを VPS 上で運用し、既存の翻訳パイプラインは書籍全体の翻訳のために Claude API へのバッチ呼び出しを行います。アシスタンス機能では、book ID、開始/終了オフセットを受け取り、Server-Sent Events (SSE) で支援内容をストリーミング配信する別のエンドポイントを構築しました。処理の流れは以下の通りです。
- フロントエンドが選択されたテキストのオフセットを含む POST リクエストを送信。
- バックエンドが PostgreSQL データベース（取り込み時に既にチャンク化済み）から周辺段落を取得。
- 選択テキスト、周辺文脈、複数翻訳バリアントの指示を含む、慎重に設計されたプロンプトを作成。
- 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)
Enter fullscreen mode Exit fullscreen mode
これにより、書籍全体をメモリに読み込むことなく周辺テキストを取得できます。
ストリーミングエンドポイント
FastAPI の StreamingResponse と、SSE 形式のメッセージを yield するジェネレーターを使用しました。ポイントは、Claude SDK の stream=True が非同期イテレーターを返すため、各チャンクを await して即座に yield できることです。
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")
Enter fullscreen mode Exit fullscreen mode
これにより、フロントエンドは単語単位でリアルタイムにストリーミングされる体験を得られます。event_generator は stream.text_stream のおかげで非同期ジェネレーターとなっています。これは Anthropic SDK のコア機能です。
複数バリアントのためのプロンプトエンジニアリング
有用な出力を得るため、3 つの異なる翻訳オプションをそれぞれ簡単な注記付きで要求するプロンプトを作成しました。また、文脈を保持し、スタイルガイドを提供するよう指示しています。簡略化したバージョンは以下の通りです。
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.
"""
Enter fullscreen mode Exit fullscreen mode
その後、フロントエンドで JSON をパースして選択肢をきれいに表示します。ストリーミングテキストは到着次第蓄積・表示されるため、即時応答のような印象を与えます。
トレードオフと課題
遅延と完全性のトレードオフ
ストリーミングは体感パフォーマンスを向上させますが、150 トークンの文脈でも初回トークン到達まで 1〜2 秒程度かかります。検索を高速化するため文脈埋め込みの事前取得を検討しましたが、LLM 呼び出しのオーバーヘッドが支配的でした。そこで以下の最適化を行いました。
-
asyncSDK を使用してイベントループのブロックを回避。 - 同一クエリ（同じ書籍、同じチャンク、同じ選択テキスト）に対するシンプルな Redis キャッシュを 5 分の TTL で実装。これにより繰り返しリクエストを約 30% 削減。
- 応答が短いため、比較的小さい
max_tokens（1024）を選択。
コスト管理
各アシスタンス呼び出しでトークンが消費されます。Claude 3.5 Sonnet では入力トークン $3/MTok、出力トークン $15/MTok です。ユーザーごとの利用状況を追跡し、1 日あたり 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
Enter fullscreen mode Exit fullscreen mode
モデルの選択
当初は DeepSeek を使用していましたが、文学的なニュアンスに対する翻訳品質が明らかに低く、ストリーミング API の安定性も劣っていました。Claude の出力の方が豊かだったため、コストは高いものの最終的に採用しました。非文学作品向けには DeepSeek を切り替え可能にするオプションを検討しています。
コンテキストウィンドウとトークン制限
文脈を 3 段落程度（約 500 トークン）に制限し、モデルの 200K コンテキストウィンドウ内に収めつつプロンプトコストを低く抑えています。通常、これでトーンやスタイルを理解するのに十分です。
結果と学んだ教訓
デプロイ後、「翻訳与转录求助」機能は LectuLibre で最も利用される機能の 1 つになりました。ユーザーは即時性と、複数の翻訳オプションを並べて確認できる点を高く評価しています。クリックから初回トークンまでの平均応答時間は 1.4 秒で、完全な応答（3 つのオプション）は約 5〜7 秒で、ユーザーは許容範囲と判断しました。
主なポイント：
- ストリーミングは必須：リアルタイム LLM アプリでは常に SSE または WebSocket を使用する。
- キャッシュは効果的：シンプルなクエリレベルのキャッシュで負荷とコストを大幅に削減可能。
-
非同期が重要：
asyncioと非同期 SDK を使用することで、負荷時のスレッド枯渇を防ぐ。 - プロンプト構造が品質とフロントエンドのパースに影響：JSON 出力を要求すると UI レンダリングは簡素化されるが、モデルの創造性を制約する可能性があるため、JSON 形式を指示しつつ自由形式の説明を許容する中間的なアプローチを採用。
今後の展望
ユーザーからのフィードバックを活用してアシスタンスモデルをファインチューニングすることや、本機能を翻訳者向けのスタンドアロン API として公開することを検討しています。他のチームが大規模環境でリアルタイム LLM ストリーミングをどのように扱っているか、特にリクエストのバッチ処理やトークンコスト削減について、ぜひコメントでお聞かせください！
0 Comments
Log in to join the conversation.No comments yet. Be the first to share your thoughts.