Merge pull request #58 from basedosdados/feat/language-support

feat: localize the chatbot by language (pt/en/es)

Author: rdahis

alembic/versions/4b3d2fa4a75f_add_thread_language.py

@@ -0,0 +1,32 @@
+"""Add language column to threads table.
+
+Revision ID: 4b3d2fa4a75f
+Revises: f6ce7837e023
+Create Date: 2026-08-04 19:50:02.000000
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "4b3d2fa4a75f"
+down_revision: Union[str, Sequence[str], None] = "f6ce7837e023"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    # server_default backfills existing threads with the Portuguese default; new rows get
+    # their value from the application (ThreadPayload.language).
+    op.add_column(
+        "thread",
+        sa.Column("language", sa.String(), nullable=False, server_default="pt"),
+    )
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    op.drop_column("thread", "language")

app/api/routers/chatbot.py

@@ -30,6 +30,7 @@
     ResultTooLarge,
     materialize_export,
 )
+from app.i18n import DEFAULT_LANGUAGE, t
 from app.settings import settings
 from app.storage import generate_signed_url
 
@@ -118,6 +119,7 @@ async def create_thread(
     thread_create = ThreadCreate(
         title=thread_payload.title,
         user_id=user_id,
+        language=thread_payload.language,
     )
 
     return await database.create_thread(thread_create)
@@ -169,13 +171,17 @@ async def send_message(
     running_runs: RunningRuns,
     user_id: UserID,
 ) -> StreamingResponse:
-    await _authorize_thread(database, thread_id, user_id)
+    thread = await _authorize_thread(database, thread_id, user_id)
 
     run_id = str(uuid.uuid4())
 
     config = ConfigDict(
         run_id=run_id,
-        configurable={"thread_id": thread_id, "user_id": user_id},
+        configurable={
+            "thread_id": thread_id,
+            "user_id": user_id,
+            "language": thread.language,
+        },
     )
 
     message_create = MessageCreate(
@@ -196,6 +202,7 @@ async def send_message(
             thread_id=thread_id,
             user_message=message,
             model_uri=settings.MODEL_URI,
+            language=thread.language,
             queue=queue,
         ),
         name=f"run_agent:{run_id}",
@@ -220,28 +227,18 @@ def _cleanup(task: asyncio.Task):  # pragma: no cover
     )
 
 
-# User-facing details the frontend surfaces to the end user when a download fails.
-RESULTS_EXPIRED_DETAIL = "Estes resultados não estão mais disponíveis para download."
-
-RESULTS_TOO_LARGE_DETAIL = (
-    "Estes resultados são grandes demais para baixar em um único arquivo."
-)
-
-# Fallback base name when a query's slug yields nothing filesystem-safe.
-DEFAULT_EXPORT_FILENAME = "resultados"
-
-
-def _sanitize_filename(slug: str) -> str:
+def _sanitize_filename(slug: str, fallback: str) -> str:
     """Sanitize a query's slug into a safe base filename.
 
     Args:
         slug (str): The query's slug.
+        fallback (str): Base name to use when the slug yields nothing filesystem-safe.
 
     Returns:
         str: A filesystem-safe base filename, without extension.
     """
     filename = re.sub(r"[^\w-]+", "_", slug).strip("_")
-    return filename or DEFAULT_EXPORT_FILENAME
+    return filename or fallback
 
 
 @router.post("/messages/{message_id}/exports")
@@ -263,6 +260,11 @@ async def export_message_results(
 
     message = await _authorize_message(database, message_id, user_id)
 
+    # The thread's language localizes the download-failure details below. _authorize_message
+    # already validated the thread exists and is owned; re-read it to read its language.
+    thread = await database.get_thread(message.thread_id)
+    language = thread.language if thread else DEFAULT_LANGUAGE
+
     query_handle = await database.get_query_handle(message.id, query_ref)
 
     if query_handle is None:
@@ -277,18 +279,20 @@ async def export_message_results(
             query_ref=query_handle.query_ref,
             destination_table=query_handle.destination_table,
             file_format=file_format,
-            filename=_sanitize_filename(query_handle.slug),
+            filename=_sanitize_filename(
+                query_handle.slug, t("default_export_filename", language)
+            ),
             message_id=str(message.id),
         )
     except ResultTableExpired as e:
         raise HTTPException(
             status_code=status.HTTP_410_GONE,
-            detail=RESULTS_EXPIRED_DETAIL,
+            detail=t("results_expired", language),
         ) from e
     except ResultTooLarge as e:
         raise HTTPException(
             status_code=status.HTTP_400_BAD_REQUEST,
-            detail=RESULTS_TOO_LARGE_DETAIL,
+            detail=t("results_too_large", language),
         ) from e
 
     signed_url = generate_signed_url(

app/api/streaming/agent_runner.py

@@ -20,19 +20,7 @@
     QueryHandle,
 )
 from app.exports import CollectedQueryHandle, collect_query_handles
-
-
-class ErrorMessage:
-    INTERRUPTED = (
-        "A conexão com o servidor foi interrompida. Por favor, tente novamente."
-    )
-
-    MODEL_CALL_LIMIT_REACHED = (
-        "Essa pergunta gerou um raciocínio muito longo e não consegui chegar a uma conclusão. "
-        "Por favor, tente ser mais específico ou divida sua pergunta em partes menores."
-    )
-
-    UNEXPECTED = "Ocorreu um erro inesperado. Por favor, tente novamente. Se o problema persistir, avise-nos."
+from app.i18n import DEFAULT_LANGUAGE, language_directive, t
 
 
 def _truncate_json(
@@ -93,11 +81,14 @@ def _truncate_json(
     return json.dumps(data, ensure_ascii=False, indent=2)
 
 
-def _process_chunk(chunk: dict[str, Any]) -> StreamEvent | None:
+def _process_chunk(
+    chunk: dict[str, Any], language: str = DEFAULT_LANGUAGE
+) -> StreamEvent | None:
     """Process a streaming chunk from a react agent workflow into a StreamEvent.
 
     Args:
         chunk (dict[str, Any]): A raw update chunk from the agent workflow.
+        language (str): The thread's language, for localizing server-emitted content.
 
     Returns:
         StreamEvent | None: Structured event or None if the chunk is ignored:
@@ -192,7 +183,7 @@ def _process_chunk(chunk: dict[str, Any]) -> StreamEvent | None:
         update = chunk["ModelCallLimitMiddleware.before_model"] or {}
         if update.get("jump_to") == "end":
             event_data = EventData(
-                content=ErrorMessage.MODEL_CALL_LIMIT_REACHED,
+                content=t("error_model_call_limit", language),
                 tool_calls=None,
             )
             return StreamEvent(type="model_call_limit", data=event_data)
@@ -231,6 +222,7 @@ async def run_agent(
     user_message: Message,
     model_uri: str,
     queue: asyncio.Queue[StreamEvent],
+    language: str = DEFAULT_LANGUAGE,
 ):
     """Run the agent to completion and push events onto the queue.
 
@@ -245,6 +237,8 @@ async def run_agent(
         thread_id (str): Thread unique identifier.
         user_message (Message): User message.
         model_uri (str): Model URI.
+        language (str): The thread's language; sets the response-language default and
+            localizes server-emitted error messages.
         queue (asyncio.Queue[StreamEvent]): Events queue.
     """
     events = []
@@ -253,16 +247,22 @@ async def run_agent(
     collected_handles: list[CollectedQueryHandle] = []
     status: MessageStatus | None = None
 
+    # Prepend the language directive to the model input only — the persisted user Message
+    # (created by the router) keeps the user's clean text. This sets the site's language as
+    # the default while letting the model honor a user who writes in another language.
+    # A dynamic-prompt middleware would keep it out of checkpoint history entirely; see PR notes.
+    model_input = f"{language_directive(language)}\n\n{user_message.content}"
+
     try:
         async for mode, chunk in agent.astream(  # pragma: no cover
-            input={"messages": [{"role": "user", "content": user_message.content}]},
+            input={"messages": [{"role": "user", "content": model_input}]},
             config=config,
             stream_mode=["updates", "values"],
         ):
             if mode == "values":
                 continue
 
-            event = _process_chunk(chunk)
+            event = _process_chunk(chunk, language)
 
             if event is None:
                 continue
@@ -290,12 +290,12 @@ async def run_agent(
             await queue.put(event)
     except asyncio.CancelledError:
         if status is None:
-            assistant_message = ErrorMessage.INTERRUPTED
+            assistant_message = t("error_interrupted", language)
             status = MessageStatus.INTERRUPTED
         raise
     except Exception:
         logger.exception(f"Unexpected error in run {config['run_id']}:")
-        assistant_message = ErrorMessage.UNEXPECTED
+        assistant_message = t("error_unexpected", language)
         status = MessageStatus.ERROR
         event = StreamEvent(
             type="error",

app/db/models.py

@@ -3,16 +3,26 @@
 from enum import Enum
 from typing import Any
 
-from pydantic import JsonValue, computed_field
+from pydantic import JsonValue, computed_field, field_validator
 from sqlalchemy import Enum as SAEnum
 from sqlmodel import JSON, TIMESTAMP, Column, Field, Integer, Relationship, SQLModel
 
+from app.i18n import DEFAULT_LANGUAGE, normalize_language
+
 
 # =============================================================================
 # ==                              Thread Models                              ==
 # =============================================================================
 class ThreadPayload(SQLModel):
     title: str
+    # Captured at creation from the site's locale (pt/en/es). Steers the assistant's response
+    # language and localizes server-emitted messages. Stored as a plain code; see app.i18n.
+    language: str = Field(default=DEFAULT_LANGUAGE)
+
+    @field_validator("language")
+    @classmethod
+    def _normalize_language(cls, value: str) -> str:
+        return normalize_language(value)
 
 
 class ThreadCreate(ThreadPayload):