Stream translate

Paste text, pick a target language, watch the translation stream in. Streaming pays off whenever the output is long enough that waiting for the full blob would feel slow.

Run

bash bootstrap-secrets.sh              # reads ../../../../.env, writes secrets/
docker compose up --build              # default: PROVIDER=openai

Open http://localhost:3000.

To run against Gemini instead:

PROVIDER=gemini docker compose up --build

What's where

• backend/ai_openai.py — parses target: <lang>\n---\n<text>, streams translation
• backend/ai_gemini.py — same shape, other SDK
• backend/main.py — FastAPI: /api/ai (final) + /api/ai/stream (SSE)
• frontend/app/page.tsx — language picker + textarea, types output as it arrives

Stop

docker compose down

unzip stream-translate.zip
cd stream-translate
bash bootstrap-secrets.sh   # one-time: pulls API keys into ./secrets
docker compose up --build   # default provider: openai
# or:  PROVIDER=gemini docker compose up --build

The same modules the Run button hits. The whole project (frontend, Dockerfile, compose) is in the ZIP under README.

backend/ai_openai.py

"""Week 3 - Showcase 2 (OpenAI): stream-translate.

Translates pasted text into a target language and streams the
translation back. The input carries the target language in a small
header:

    target: French
    ---
    The text to translate.

`run_stream` yields tokens; `run` accumulates them for the contract.
"""
from typing import Iterator

from openai import OpenAI

_client = OpenAI()


def _parse(payload: str) -> tuple[str, str]:
    """Split the input into (target language, text).

    The header `target: <language>` ends at a `---` separator. Anything
    that doesn't match the header defaults to English so the showcase
    never fails closed on bad input — it just translates to English.
    """
    parts = payload.split("---", 1)
    if len(parts) != 2:
        return ("English", payload.strip())
    header, text = parts
    target = "English"
    for line in header.splitlines():
        s = line.strip()
        if s.lower().startswith("target:"):
            target = s.split(":", 1)[1].strip() or target
            break
    return (target, text.strip())


def _prompt(target: str, text: str) -> str:
    return (
        f"Translate the following text into {target}. Preserve meaning, "
        "tone, and paragraph structure. Return only the translation; no "
        "commentary, no explanations."
        f"\n\n---\n{text}\n---"
    )


def run_stream(payload: str) -> Iterator[str]:
    target, text = _parse(payload)
    # Cap output length to prevent runaway responses. 2048 tokens
    # (~6k chars) is enough headroom for translations of inputs at
    # the showcase's 8k-char ceiling.
    stream = _client.responses.create(
        model="gpt-5.4-nano", input=_prompt(target, text), stream=True,
        max_output_tokens=2048,
    )
    for event in stream:
        if event.type == "response.output_text.delta":
            yield event.delta


def run(payload: str) -> str:
    return "".join(run_stream(payload))

backend/ai_gemini.py

"""Week 3 - Showcase 2 (Gemini): stream-translate.

Same input shape and contract as ai_openai.py — see that file for the
input format. Differences are mechanical: Gemini's streaming method and
chunk shape.
"""
import os
from typing import Iterator

from google import genai
from google.genai import types

_client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])


def _parse(payload: str) -> tuple[str, str]:
    parts = payload.split("---", 1)
    if len(parts) != 2:
        return ("English", payload.strip())
    header, text = parts
    target = "English"
    for line in header.splitlines():
        s = line.strip()
        if s.lower().startswith("target:"):
            target = s.split(":", 1)[1].strip() or target
            break
    return (target, text.strip())


def _prompt(target: str, text: str) -> str:
    return (
        f"Translate the following text into {target}. Preserve meaning, "
        "tone, and paragraph structure. Return only the translation; no "
        "commentary, no explanations."
        f"\n\n---\n{text}\n---"
    )


def run_stream(payload: str) -> Iterator[str]:
    target, text = _parse(payload)
    # Cap output length to prevent runaway responses. 2048 tokens
    # (~6k chars) is enough headroom for translations of inputs at
    # the showcase's 8k-char ceiling.
    stream = _client.models.generate_content_stream(
        model="gemini-3.1-flash-lite", contents=_prompt(target, text),
        config=types.GenerateContentConfig(max_output_tokens=2048),
    )
    for chunk in stream:
        if chunk.text:
            yield chunk.text


def run(payload: str) -> str:
    return "".join(run_stream(payload))

Project files

• .gitignore
• README.md
• backend/Dockerfile
• backend/ai_gemini.py
• backend/ai_openai.py
• backend/main.py
• backend/requirements.txt
• bootstrap-secrets.sh
• docker-compose.yml
• frontend/Dockerfile
• frontend/app/layout.tsx
• frontend/app/page.tsx
• frontend/next.config.ts
• frontend/package.json
• frontend/tsconfig.json