BookahTranslate / AthenaDev
AI-сервис перевода документов. PDF, EPUB, DOCX до 300+ страниц с сохранением форматирования на базе GPT-4 и Claude.
Метрики
Контекст
Перевод длинных документов (50–300 страниц) с сохранением форматирования — это реальная боль: Google Translate превращает PDF в plain text, у DeepL нет PDF-поддержки для конечных пользователей, а немногие SaaS, которые умеют PDF, либо медленные, либо дорогие, либо ломают layout.
Хотел сервис, куда бросаешь PDF/EPUB/DOCX и получаешь читабельный перевод с сохранёнными иллюстрациями, таблицами, формулами и сносками. И хотел запустить это как настоящий SaaS с подпиской, не игрушку.
Что я сделал
- Архитектура из 3 компонентов: Web App (Flask + SQLAlchemy), Translation API (отдельный Flask для долгих задач), Telegram Bot — связаны только через Redis-очередь.
- LLM-роутинг: GPT-4 / GPT-4o для художки, Anthropic Claude для технического и юридического, Google Translate для дешёвых bulk-задач. Выбор по типу документа, тарифу и объёму.
- Трёхтарифная подписочная система: page-based billing, бонусные страницы, разовые платежи, автопродление через YooKassa. Webhook с верификацией по IP и подписи.
- OAuth 2.0 через Google, VK и Telegram (deep-link). Плюс email/пароль с rate-limit на регистрацию и валидацией пароля.
- PDF-пайплайн: BabelDOC для перевода с layout, pdf2zh как fallback, ReportLab на выходе, OCR через Tesseract для сканов.
- Асинхронные задачи через Celery + Redis с прогрессом в браузере — переводит документы 300+ страниц без HTTP-таймаутов.
- Фолбэк между провайдерами на уровне вызова: таймауты, рейт-лимиты, недоступность провайдера и некорректные ответы деградируют на запасную модель, а не роняют задачу пользователя. Каждый фолбэк логируется с классом ошибки, которая его вызвала, — картина сбоев видна, а не додумывается.
- Evaluation-набор, через который проходит каждое изменение промпта или модели: точность, релевантность, уровень галлюцинаций, задержка и стоимость страницы, с пер-кейсовым сравнением с сохранённым baseline. Изменение, которое улучшает среднее, но ломает три конкретных типа документов, не выкатывается.
- Версионированные промпты и логирование запросов/ответов с трейсингом пайплайна — когда перевод вышел плохим, восстановим точный промпт, модель и промежуточные чанки, которые его дали.
- 80+ pytest-тестов (auth, биллинг, перевод, изоляция пользователей) плюс Playwright e2e полного пути «регистрация → перевод → оплата».
Технические решения
База-на-контекст: три независимых PostgreSQL-базы (translate_bot, translate_bot_staging, translate_web). Падение одной не валит остальные. Аналогично для Redis.
Защита от race conditions: списание страниц через SELECT FOR UPDATE в атомарной транзакции; активация подписки через partial unique index («одна активная подписка на юзера»).
Security: JWT с blacklist в БД (мгновенный revoke при logout), Fernet-шифрование пользовательских API-ключей, Flask-Limiter на регистрацию (5/час) и логин (10/мин) на Redis.
Cost engineering: smart-роутинг моделей снизил среднюю стоимость страницы перевода LLM примерно на 40% против «всегда GPT-4». Bulk-путь через Google Translate даёт ещё ярус экономии для тарифа «Эконом». Кэширование ответов убирает повторную оплату одинаковых чанков — это заметно на документах с типовыми разделами.
Некорректный ответ с кодом 200 — это тоже сбой. Ответы провайдера проверяются на структуру до того, как приняты: обрезанный или пустой перевод, пришедший с HTTP 200, хуже ошибки — он доходит до клиента с видом готового.
Лучше отказать, чем отдать половину: когда падают все провайдеры в цепочке, задача честно завершается ошибкой, а страница возвращается на баланс. Недопереведённый документ, выданный за готовый, разрушает доверие куда дешевле, чем честная ошибка.
Из кода
def use_pages_atomic(subscription_id: int, pages_count: int) -> tuple[bool, int]:
"""Deduct pages with row-level lock to prevent concurrent over-spend."""
sub = (
db.session.query(UserSubscription)
.with_for_update() # SELECT ... FOR UPDATE
.filter(UserSubscription.id == subscription_id)
.first()
)
if not sub or sub.status not in ("active", "grace_period"):
return False, 0
if sub.pages_remaining < pages_count:
db.session.rollback()
return False, sub.pages_remaining
sub.pages_remaining -= pages_count
db.session.add(PageTransaction(
subscription_id=sub.id,
pages_delta=-pages_count,
balance_after=sub.pages_remaining,
))
db.session.commit() # releases lock
return True, sub.pages_remaining
def pick_model(doc: Document, tier: UserTier) -> ModelChoice:
if tier == UserTier.ECONOMY and doc.kind in {DocKind.TXT, DocKind.PLAIN_PDF}:
return ModelChoice.GOOGLE_TRANSLATE
if doc.kind == DocKind.SCANNED_PDF:
return ModelChoice.GPT_4O # needs vision + OCR fusion
if doc.kind == DocKind.LEGAL or doc.has_dense_terminology:
return ModelChoice.CLAUDE_SONNET
if doc.pages > 200 and tier != UserTier.PREMIUM:
return ModelChoice.CLAUDE_HAIKU # cheap for bulk
return ModelChoice.GPT_4O
# Each failure class is retried differently: a timeout deserves another
# attempt, a 400 does not. The chain degrades to a cheaper model rather
# than returning an error to someone who already paid for the page.
RETRYABLE = (ProviderTimeout, RateLimited, ProviderUnavailable)
def translate_chunk(chunk: str, chain: list[ModelChoice]) -> Translation:
failures: list[str] = []
for model in chain:
for attempt in range(MAX_ATTEMPTS):
try:
result = call_provider(model, chunk, timeout=timeout_for(model))
if not result.is_well_formed():
# A malformed response is a failure even with HTTP 200.
raise MalformedResponse(model)
if failures:
log.warning("recovered after %s", ",".join(failures))
metrics.incr("translate.fallback", tags={"to": model.name})
return result
except RETRYABLE as exc:
failures.append(f"{model.name}:{type(exc).__name__}")
sleep(backoff(attempt)) # exponential, jittered
except MalformedResponse as exc:
failures.append(f"{model.name}:malformed")
break # retrying won't fix the prompt
except ProviderRefused:
failures.append(f"{model.name}:refused")
break # next model, not next attempt
# Every provider is down: fail loudly and refund the page, never
# hand the user a half-translated document as if it succeeded.
raise AllProvidersFailed(failures)
def gate(report: RunReport, baseline: Baseline) -> GateResult:
"""Block a deploy when quality drops. Published as llm-eval-harness."""
violations = []
if report.hallucination_rate > MAX_HALLUCINATION_RATE:
violations.append(f"hallucination {report.hallucination_rate:.1%}")
if report.mean_accuracy < MIN_ACCURACY:
violations.append(f"accuracy {report.mean_accuracy:.3f}")
if report.cost_per_page > baseline.cost_per_page * COST_CEILING:
violations.append(f"cost/page {report.cost_per_page:.4f}")
# The averages above can all pass while specific documents break.
# Per-case comparison is what actually catches a regression.
regressions = [
case.id for case in report.cases
if baseline.passed(case.id) and not case.passed
]
return GateResult(
passed=not violations and not regressions,
violations=violations,
regressions=regressions,
)
Результат
В продакшене на bookahtranslate.tech (веб) и через @BookahTranslateBot (Telegram). Построено и поддерживается в одиночку, платящие пользователи, органический рост.
Работает как живая лаборатория для AI-native product engineering: каждый PR катится в реальный сервис с реальными пользователями, реальным биллингом и реальными failure modes — не песочница.
Подходы к надёжности отсюда выложены отдельными читаемыми репозиториями: релизный гейт — llm-eval-harness (47 тестов), grounded-поиск, который отказывается вместо выдумки, — rag-grounded (40 тестов), вызов инструментов — mcp-toolserver (50 тестов). Именно работа платного сервиса показала, какие из них действительно важны.