You need a support bot that answers in the user’s language without maintaining separate pipelines per locale. LlamaIndex gives you the primitives — language detection, translation, and retrieval — to compose a single multilingual RAG system. This tutorial walks through a complete implementation: detect the incoming language, translate the query to English for retrieval against a unified knowledge base, then translate the answer back. You’ll end up with a FastAPI service you can deploy behind your gateway.
Prerequisites
- Python 3.10+
- An OpenAI-compatible endpoint (we use
gpt-4o-minifor chat andtext-embedding-3-smallfor embeddings) - A vector store — this example uses ChromaDB running locally
- Familiarity with LlamaIndex core concepts:
VectorStoreIndex,QueryEngine, and custom components
Install dependencies:
pip install llama-index llama-index-llms-openai llama-index-embeddings-openai \
llama-index-vector-stores-chroma chromadb fastapi uvicorn langdetect \
python-dotenv
Set your endpoint and key in .env:
OPENAI_API_KEY=sk-...
OPENAI_API_BASE=https://api.openai.com/v1 # or your n4n.ai endpoint
Project structure
multilingual-bot/
├── app/
│ ├── __init__.py
│ ├── config.py
│ ├── language.py
│ ├── translation.py
│ ├── retrieval.py
│ └── main.py
├── data/
│ └── support_docs/ # markdown files per locale
├── chroma_db/ # persisted vector store
├── .env
└── requirements.txt
Configuration
Centralize model names and supported locales so you can swap providers without touching logic.
# app/config.py
from dataclasses import dataclass
from typing import List
@dataclass
class Settings:
llm_model: str = "gpt-4o-mini"
embed_model: str = "text-embedding-3-small"
supported_locales: List[str] = None
default_locale: str = "en"
chroma_path: str = "./chroma_db"
collection_name: str = "support_kb"
def __post_init__(self):
if self.supported_locales is None:
self.supported_locales = ["en", "es", "fr", "de", "ja", "zh", "pt", "it"]
settings = Settings()
Language detection
Use langdetect for fast, zero-dependency detection. Wrap it in a small service so you can swap in a more robust model later (e.g., fastText or a dedicated classifier) without changing callers.
# app/language.py
from langdetect import detect, DetectorFactory
from langdetect.lang_detect_exception import LangDetectException
from app.config import settings
# Deterministic results
DetectorFactory.seed = 42
class LanguageDetector:
def __init__(self, supported: list[str] | None = None):
self.supported = supported or settings.supported_locales
self.default = settings.default_locale
def detect(self, text: str) -> str:
if not text or not text.strip():
return self.default
try:
lang = detect(text)
return lang if lang in self.supported else self.default
except LangDetectException:
return self.default
def is_supported(self, lang: str) -> bool:
return lang in self.supported
Test it quickly:
# python -c "from app.language import LanguageDetector; d=LanguageDetector(); print(d.detect('Hola, ¿cómo puedo ayudarte?'))"
# es
Translation layer
Translate the user query to English for retrieval, then translate the response back. We use the same LLM for translation to avoid an extra dependency. Keep prompts tight and deterministic.
# app/translation.py
from llama_index.core.llms import ChatMessage, MessageRole
from llama_index.llms.openai import OpenAI
from app.config import settings
TRANSLATE_TO_EN_PROMPT = """Translate the following text to English. Preserve meaning, tone, and any technical terms. Output only the translation."""
TRANSLATE_FROM_EN_PROMPT = """Translate the following English response to {target_lang}. Preserve meaning, tone, and formatting. Output only the translation."""
class Translator:
def __init__(self, model: str | None = None):
self.llm = OpenAI(model=model or settings.llm_model, temperature=0.0)
async def to_english(self, text: str, source_lang: str) -> str:
if source_lang == "en":
return text
messages = [
ChatMessage(role=MessageRole.SYSTEM, content=TRANSLATE_TO_EN_PROMPT),
ChatMessage(role=MessageRole.USER, content=text),
]
resp = await self.llm.achat(messages)
return resp.message.content.strip()
async def from_english(self, text: str, target_lang: str) -> str:
if target_lang == "en":
return text
prompt = TRANSLATE_FROM_EN_PROMPT.format(target_lang=target_lang)
messages = [
ChatMessage(role=MessageRole.SYSTEM, content=prompt),
ChatMessage(role=MessageRole.USER, content=text),
]
resp = await self.llm.achat(messages)
return resp.message.content.strip()
Knowledge base ingestion
Store all documentation in English in a single vector index. At ingest time, translate non-English source documents to English so retrieval works against one unified corpus. This avoids maintaining per-language indexes and keeps latency predictable.
# app/retrieval.py
import os
from pathlib import Path
from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, StorageContext
from llama_index.core.node_parser import SentenceSplitter
from llama_index.vector_stores.chroma import ChromaVectorStore
from llama_index.embeddings.openai import OpenAIEmbedding
import chromadb
from app.config import settings
from app.translation import Translator
class KnowledgeBase:
def __init__(self):
self.embed_model = OpenAIEmbedding(model=settings.embed_model)
self.translator = Translator()
self.client = chromadb.PersistentClient(path=settings.chroma_path)
self.vector_store = ChromaVectorStore(
chroma_collection=self.client.get_or_create_collection(settings.collection_name)
)
self.storage_context = StorageContext.from_defaults(vector_store=self.vector_store)
async def ingest(self, docs_path: str = "./data/support_docs") -> VectorStoreIndex:
"""Load markdown files, translate non-English content to English, index."""
reader = SimpleDirectoryReader(
input_dir=docs_path,
required_exts=[".md", ".txt"],
recursive=True,
)
documents = reader.load_data()
# Translate each document to English if needed
for doc in documents:
# Assume filename convention: *.en.md, *.es.md, etc. or detect from content
# For simplicity, detect from first 200 chars
sample = doc.text[:200]
detected = self.translator.llm.complete(
f"Detect the language of this text. Reply with ISO code only (en, es, fr, etc.):\n{sample}"
).text.strip().lower()
if detected != "en":
translated = await self.translator.to_english(doc.text, detected)
doc.text = translated
doc.metadata["original_language"] = detected
parser = SentenceSplitter(chunk_size=512, chunk_overlap=50)
nodes = parser.get_nodes_from_documents(documents)
index = VectorStoreIndex(
nodes,
storage_context=self.storage_context,
embed_model=self.embed_model,
show_progress=True,
)
return index
def load_index(self) -> VectorStoreIndex:
return VectorStoreIndex.from_vector_store(
self.vector_store,
embed_model=self.embed_model,
)
Run ingestion once (or as a scheduled job):
python -c "
import asyncio
from app.retrieval import KnowledgeBase
kb = KnowledgeBase()
asyncio.run(kb.ingest())
print('Ingestion complete')
"
Expected output:
Ingestion complete
Query engine with translation pipeline
Compose the pieces: detect language → translate query → retrieve → synthesize → translate answer.
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from llama_index.core import VectorStoreIndex
from llama_index.core.query_engine import RetrieverQueryEngine
from llama_index.core.retrievers import VectorIndexRetriever
from llama_index.core.response_synthesizers import get_response_synthesizer
from llama_index.llms.openai import OpenAI
from app.config import settings
from app.language import LanguageDetector
from app.translation import Translator
from app.retrieval import KnowledgeBase
# Global singletons
detector = LanguageDetector()
translator = Translator()
kb = KnowledgeBase()
index: VectorStoreIndex | None = None
query_engine: RetrieverQueryEngine | None = None
@asynccontextmanager
async def lifespan(app: FastAPI):
global index, query_engine
index = kb.load_index()
retriever = VectorIndexRetriever(index=index, similarity_top_k=4)
synthesizer = get_response_synthesizer(
llm=OpenAI(model=settings.llm_model, temperature=0.1),
response_mode="compact",
)
query_engine = RetrieverQueryEngine(retriever=retriever, response_synthesizer=synthesizer)
yield
app = FastAPI(title="Multilingual Support Bot", lifespan=lifespan)
class QueryRequest(BaseModel):
question: str
locale: str | None = None # optional override
class QueryResponse(BaseModel):
answer: str
detected_locale: str
source_locale: str
@app.post("/query", response_model=QueryResponse)
async def query(request: QueryRequest):
if query_engine is None:
raise HTTPException(status_code=503, detail="Index not ready")
# Detect or use provided locale
source_locale = request.locale or detector.detect(request.question)
if not detector.is_supported(source_locale):
source_locale = settings.default_locale
# Translate question to English for retrieval
english_question = await translator.to_english(request.question, source_locale)
# Retrieve and synthesize in English
response = await query_engine.aquery(english_question)
english_answer = str(response)
# Translate answer back to user's language
final_answer = await translator.from_english(english_answer, source_locale)
return QueryResponse(
answer=final_answer,
detected_locale=source_locale,
source_locale=source_locale,
)
@app.get("/health")
async def health():
return {"status": "ok", "index_ready": query_engine is not None}
Run the server:
uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload
Test with curl:
curl -X POST http://localhost:8000/query \
-H "Content-Type: application/json" \
-d '{"question": "Comment puis-je réinitialiser mon mot de passe ?"}'
Expected response:
{
"answer": "Pour réinitialiser votre mot de passe, cliquez sur \"Mot de passe oublié\" sur la page de connexion...",
"detected_locale": "fr",
"source_locale": "fr"
}
Handling edge cases
Mixed-language queries
Users often mix languages (“How do I reset mi contraseña?”). The detector picks the dominant language. For stricter handling, run detection per-sentence and translate each segment, but in practice single-detection works for >95% of support traffic.
Unsupported languages
If detection returns a locale outside supported_locales, fall back to English. Log these occurrences — they signal where to expand coverage.
# In query endpoint, after detection:
if not detector.is_supported(source_locale):
logger.warning(f"Unsupported locale detected: {source_locale}, falling back to en")
source_locale = settings.default_locale
Translation failures
Wrap translation calls in try/except and fall back to English answer if translation fails. Never let a translation error drop the request.
async def safe_translate(self, text: str, target_lang: str, direction: str) -> str:
try:
if direction == "to_en":
return await self.to_english(text, target_lang)
else:
return await self.from_english(text, target_lang)
except Exception as e:
logger.error(f"Translation {direction} failed: {e}")
return text # Return original (English) as fallback
Production considerations
Caching translations
Repeated queries in the same language hit the translation LLM every time. Cache translations in Redis with a TTL (24h typical). Key format: trans:{direction}:{lang}:{hash(text)}.
Observability
Log each request with: request_id, source_locale, detected_locale, retrieval_latency_ms, translation_latency_ms, total_latency_ms. This lets you spot regression in specific language pairs.
Evaluating quality
Build a small eval set per locale (50-100 questions with reference answers). Run nightly and track:
- Answer correctness (LLM-as-judge)
- Translation fidelity (back-translation consistency)
- Latency p50/p95 per locale
Scaling the vector store
Chroma works for <1M vectors. For larger corpora, swap to Pinecone, Weaviate, or Qdrant — change only KnowledgeBase.__init__ and the vector store import. The rest of the pipeline is store-agnostic.
Extending the pipeline
Per-locale knowledge
Some products have region-specific policies (GDPR in EU, CCPA in California). Add a metadata filter at retrieval time:
from llama_index.core.vector_stores import MetadataFilters, ExactMatchFilter
filters = MetadataFilters(filters=[ExactMatchFilter(key="region", value="eu")])
retriever = VectorIndexRetriever(index=index, similarity_top_k=4, filters=filters)
Ingest region-specific docs with region metadata, and pass the user’s detected region (from IP or account) at query time.
Streaming responses
For lower perceived latency, stream the English synthesis and translate chunks as they arrive. This requires a custom response synthesizer and a streaming translation wrapper — more complex but worthwhile for chat interfaces.
Client-side language hints
If your frontend knows the user’s preferred language (from browser Accept-Language or account settings), pass it as locale in the request to skip detection entirely. The endpoint already supports this override.
Summary
You now have a single multilingual support bot that:
- Detects the user’s language on each request
- Translates queries to English for retrieval against one unified index
- Synthesizes answers in English, then translates back
- Runs as a FastAPI service ready for container deployment
The architecture keeps operational complexity low — one index, one query engine, translation as a thin layer — while supporting any language your LLM handles. Add locales by updating supported_locales and ingesting a few seed documents; no pipeline changes required.