File size: 3,728 Bytes
0e5fdb5 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 | """Deterministic, LLM-free reply-language detection shared across agents.
The user-facing agents (help playbook + the analysis answer composer) must reply
in the user's language. Detection is marker-word based over the user's turn, and
the result is injected into the prompt as a hard `[Reply language]` directive so
replying in that language is mandatory — not a soft hint an English system prompt
+ English data context can override.
Signal priority (first hit wins):
1. the current turn (`message`);
2. the most recent human turn in `history` — covers the button path (no
`message`) AND acts as a tiebreaker when the current turn is too short to
carry a signal (e.g. "2025 vs 2026"), so a bilingual user's ambiguous turn
inherits their previous turn's language instead of snapping to the default;
3. the user-authored goal (`objective` + `business_questions`);
4. the team default (Indonesian).
"""
from __future__ import annotations
import re
from langchain_core.messages import BaseMessage
FALLBACK_LANGUAGE = "Indonesian" # team default when nothing yields a signal
# Function words + common chat shorthand/abbreviations. Content words (nouns,
# domain terms) are deliberately excluded — they're often shared across both
# languages (e.g. "data", "revenue") and would add noise.
_ID_MARKERS = frozenset({
"yang", "dan", "apa", "gimana", "bagaimana", "kenapa", "mengapa", "aku", "saya",
"tolong", "ini", "itu", "nih", "dong", "kah", "untuk", "dengan", "pada", "adalah",
"tidak", "enggak", "nggak", "bisa", "mau", "buat", "dari", "kamu", "ya",
"berapa", "kapan", "siapa", "dimana", "juga", "sudah", "belum", "akan",
# abbreviations / chat shorthand
"brp", "gmn", "yg", "gt", "gitu", "gini", "dgn", "utk", "tdk", "sdh", "blm",
"aja", "dah", "kalo", "klo", "knp", "jd", "jgn", "krn", "udah", "udh",
"ga", "gak", "gk", "engga", "trus", "trs", "sm", "kayak", "kek",
})
_EN_MARKERS = frozenset({
"the", "what", "how", "why", "please", "this", "that", "is", "are", "can", "could",
"should", "for", "with", "of", "and", "you", "do", "does", "when", "where",
"who", "which", "my", "me", "your", "have", "has", "want", "next",
})
def _last_human_text(history: list[BaseMessage] | None) -> str:
"""Return the text of the most recent human turn in history, or '' if none."""
for msg in reversed(history or []):
if getattr(msg, "type", None) == "human":
content = msg.content
return content if isinstance(content, str) else str(content)
return ""
def _score_language(text: str) -> str | None:
"""Return "Indonesian"/"English" from marker-word counts, or None if no signal."""
tokens = re.findall(r"[a-z']+", text.lower())
id_hits = sum(1 for t in tokens if t in _ID_MARKERS)
en_hits = sum(1 for t in tokens if t in _EN_MARKERS)
if en_hits > id_hits:
return "English"
if id_hits > en_hits:
return "Indonesian"
return None
def detect_reply_language(
history: list[BaseMessage] | None,
message: str | None = None,
goal_texts: list[str] | None = None,
) -> str:
"""Detect the reply language deterministically (no LLM), by signal priority.
See the module docstring for the priority order. Returns "Indonesian" or
"English".
"""
if message:
lang = _score_language(message)
if lang:
return lang
prev = _last_human_text(history)
if prev:
lang = _score_language(prev)
if lang:
return lang
goal = " ".join(t for t in (goal_texts or []) if t).strip()
if goal:
lang = _score_language(goal)
if lang:
return lang
return FALLBACK_LANGUAGE
|