From 47374eeb4d4cd8308a7f62f118eb8bfafae214bc Mon Sep 17 00:00:00 2001 From: ashusnapx Date: Thu, 9 Jul 2026 16:12:27 +0530 Subject: [PATCH] feat: add language-aware system prompt routing for multi-model agents MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes #6428 Adds a lightweight pre-inference layer that detects the language of an agent system prompt and automatically translates it to the language the target model handles best, before the prompt reaches the LLM. New modules: - utilities/model_profiles.py: Boundary-aware regex registry for 15 model families (gpt-4o, claude, llama, qwen, deepseek, gemini, etc.) - utilities/prompt_translator.py: Core pipeline — language detection via Unicode script analysis, token estimation, code block/URL/JSON extraction, translation via LLM, content-hash caching Agent changes: - Add auto_translate_prompt (default True) field to BaseAgent - Add prompt_glossary field to BaseAgent for domain-term preservation - Integration in Agent._build_execution_prompt() with try/except fail-safe, system-prompt-only scope Design: - System prompt only — user messages never translated - Opt-out via Agent(auto_translate_prompt=False) - Fail-safe: any exception logs and keeps original prompt - Only cache successful translations (transient failures can retry) - Zero new hard dependencies Tests: 82 passing (model profiles + translator + integration). ruff: all checks passed. mypy: no issues found. --- lib/crewai/src/crewai/agent/core.py | 37 ++ .../crewai/agents/agent_builder/base_agent.py | 16 + .../src/crewai/utilities/model_profiles.py | 255 +++++++++ .../src/crewai/utilities/prompt_translator.py | 498 ++++++++++++++++++ .../tests/utilities/test_model_profiles.py | 147 ++++++ .../tests/utilities/test_prompt_translator.py | 394 ++++++++++++++ .../test_prompt_translator_integration.py | 170 ++++++ 7 files changed, 1517 insertions(+) create mode 100644 lib/crewai/src/crewai/utilities/model_profiles.py create mode 100644 lib/crewai/src/crewai/utilities/prompt_translator.py create mode 100644 lib/crewai/tests/utilities/test_model_profiles.py create mode 100644 lib/crewai/tests/utilities/test_prompt_translator.py create mode 100644 lib/crewai/tests/utilities/test_prompt_translator_integration.py diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 5751f3a9ae..598a158f69 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -9,6 +9,7 @@ from datetime import datetime import inspect import json +import logging import os from pathlib import Path import time @@ -117,6 +118,9 @@ AgentResponseProtocol = None # type: ignore[assignment, misc] +logger = logging.getLogger(__name__) + + if TYPE_CHECKING: from crewai_files import FileInput @@ -1013,6 +1017,39 @@ def _build_execution_prompt( response_template=self.response_template, ).task_execution() + # Language-aware prompt optimization: translate system prompt to the + # language the target model handles best before sending to LLM. + # Only translates system-level prompts — user messages carry intent + # and must never be machine-translated. + if self.auto_translate_prompt and self.llm: + try: + from crewai.utilities.prompt_translator import ( + optimize_system_prompt, + ) + + model_name = getattr(self.llm, "model", "") or "" + if isinstance(prompt, SystemPromptResult): + if prompt.system: + prompt.system = optimize_system_prompt( + prompt.system, + model_name, + self.prompt_glossary, + llm_caller=self.llm, + ) + else: + if prompt.prompt: + prompt.prompt = optimize_system_prompt( + prompt.prompt, + model_name, + self.prompt_glossary, + llm_caller=self.llm, + ) + except Exception: + logger.debug( + "Prompt translation failed; using original prompt.", + exc_info=True, + ) + stop_words = [I18N_DEFAULT.slice("observation")] if self.response_template: stop_words.append( diff --git a/lib/crewai/src/crewai/agents/agent_builder/base_agent.py b/lib/crewai/src/crewai/agents/agent_builder/base_agent.py index a12a4c18b7..0e4f9084b2 100644 --- a/lib/crewai/src/crewai/agents/agent_builder/base_agent.py +++ b/lib/crewai/src/crewai/agents/agent_builder/base_agent.py @@ -391,6 +391,22 @@ class BaseAgent(BaseModel, ABC, metaclass=AgentMeta): description="Agent Skills. Accepts paths for discovery, inline SKILL.md strings, pre-loaded Skill objects, or '@org/name' registry refs.", min_length=1, ) + auto_translate_prompt: bool = Field( + default=True, + description=( + "Automatically translate the system prompt to the language best " + "suited for the target model before each LLM call. Set to False " + "to disable and keep the prompt in its original language." + ), + ) + prompt_glossary: dict[str, str] | None = Field( + default=None, + description=( + "Glossary of terms that should not be translated when " + "auto_translate_prompt is enabled. Maps source-term to " + "target-term (e.g. {'API Key': 'API Key', 'crewAI': 'crewAI'})." + ), + ) execution_context: ExecutionContext | None = Field(default=None) checkpoint_kickoff_event_id: str | None = Field(default=None) diff --git a/lib/crewai/src/crewai/utilities/model_profiles.py b/lib/crewai/src/crewai/utilities/model_profiles.py new file mode 100644 index 0000000000..3bd95ca3b4 --- /dev/null +++ b/lib/crewai/src/crewai/utilities/model_profiles.py @@ -0,0 +1,255 @@ +"""Model profile registry for language-aware prompt routing. + +Each model family has different tokenizer efficiency characteristics across +languages. This module maps model identifiers to profiles that describe +their language preferences and tokenizer behavior, enabling automatic +system prompt optimization. + +References: + - Multi-IF benchmark (arXiv:2410.15553): Non-Latin scripts exhibit + systematically higher instruction-following error rates. + - PromptQuorum (2026-05): Documents "English SP + native UP" pattern + for DeepSeek-family models. + - Presenc AI tokenizer benchmark (2026-05): Quantifies token cost + asymmetry across model families. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import re + + +@dataclass(frozen=True) +class ModelProfile: + """Language capability profile for a model family. + + Attributes: + preferred_language: ISO 639-1 code for the model's optimal prompt language. + english_tokens_per_word: Average tokens per English word for this model's tokenizer. + non_english_chars_per_token: Average non-English characters per token + (higher = more efficient for that language). + bilingual_capability: Score 0.0-1.0 indicating how well the model + handles mixed-language prompts. Higher = better. + family: Model family identifier for logging/debugging. + """ + + preferred_language: str + english_tokens_per_word: float + non_english_chars_per_token: float + bilingual_capability: float + family: str + + +# --------------------------------------------------------------------------- +# Registry: maps model name patterns to profiles. +# Lookup uses boundary-aware regex matching to avoid false positives. +# Unknown models return None (conservative: no translation). +# --------------------------------------------------------------------------- + +# Each entry is (compiled_regex, ModelProfile). +# The regex must match at a word boundary or after a separator (/, -, _). +_MODEL_PATTERNS: list[tuple[re.Pattern[str], ModelProfile]] = [ + # --- OpenAI --- + ( + re.compile(r"(?:^|[/_-])gpt-4o(?:[/_-]|$)"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.85, + family="openai", + ), + ), + ( + re.compile(r"(?:^|[/_-])gpt-5(?:[/_-]|$)"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.9, + family="openai", + ), + ), + ( + re.compile(r"(?:^|[/_-])o1(?:-|preview|$)"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.85, + family="openai", + ), + ), + ( + re.compile(r"(?:^|[/_-])o3(?:-|mini|$)"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.88, + family="openai", + ), + ), + ( + re.compile(r"(?:^|[/_-])o4(?:-|mini|$)"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.88, + family="openai", + ), + ), + # --- Anthropic --- + ( + re.compile(r"(?:^|[/_-])claude"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.05, + non_english_chars_per_token=0.85, + bilingual_capability=0.9, + family="anthropic", + ), + ), + # --- Meta LLaMA --- + ( + re.compile(r"(?:^|[/_-])llama"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.4, + bilingual_capability=0.3, + family="llama", + ), + ), + # --- Qwen (Chinese-native) --- + ( + re.compile(r"(?:^|[/_-])qwen"), + ModelProfile( + preferred_language="zh", + english_tokens_per_word=1.05, + non_english_chars_per_token=0.95, + bilingual_capability=0.95, + family="qwen", + ), + ), + # --- DeepSeek --- + ( + re.compile(r"(?:^|[/_-])deepseek"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.05, + non_english_chars_per_token=0.9, + bilingual_capability=0.85, + family="deepseek", + ), + ), + # --- Google Gemini --- + ( + re.compile(r"(?:^|[/_-])gemini"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.8, + bilingual_capability=0.85, + family="google", + ), + ), + # --- Mistral --- + ( + re.compile(r"(?:^|[/_-])mistral"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.5, + bilingual_capability=0.6, + family="mistral", + ), + ), + # --- Mixtral --- + ( + re.compile(r"(?:^|[/_-])mixtral"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.5, + bilingual_capability=0.6, + family="mistral", + ), + ), + # --- Amazon Nova --- + ( + re.compile(r"(?:^|[/_-])nova"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.7, + bilingual_capability=0.8, + family="amazon", + ), + ), + # --- Cohere --- + ( + re.compile(r"(?:^|[/_-])command"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.5, + bilingual_capability=0.6, + family="cohere", + ), + ), + # --- AI21 --- + ( + re.compile(r"(?:^|[/_-])jamba"), + ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.7, + family="ai21", + ), + ), +] + + +def get_model_profile(model_name: str) -> ModelProfile | None: + """Look up a model's language profile by name. + + Uses boundary-aware regex matching to avoid false positives. + ``"gpt-4o-2024-05-13"`` matches the ``"gpt-4o"`` entry, but + ``"proto1"`` or ``"bio1"`` will not match ``"o1"``. + + Returns ``None`` for unknown models so callers can fall back to + conservative (no-translation) behaviour. + + Args: + model_name: The model identifier string (e.g. ``"gpt-4o"``). + + Returns: + The matching :class:`ModelProfile`, or ``None`` if not found. + """ + if not model_name: + return None + + model_lower = model_name.lower() + + # Normalize: treat `.`, `/`, `-`, `_` as equivalent separators + normalized = re.sub(r"[./_-]", "-", model_lower) + + # Find all matching patterns and pick the longest match + best_match: re.Match[str] | None = None + best_profile: ModelProfile | None = None + + for pattern, profile in _MODEL_PATTERNS: + match = pattern.search(normalized) + if match is not None: + # Prefer the match with the longest matched span + if best_match is None or (match.end() - match.start()) > ( + best_match.end() - best_match.start() + ): + best_match = match + best_profile = profile + + return best_profile diff --git a/lib/crewai/src/crewai/utilities/prompt_translator.py b/lib/crewai/src/crewai/utilities/prompt_translator.py new file mode 100644 index 0000000000..e2ca31c54c --- /dev/null +++ b/lib/crewai/src/crewai/utilities/prompt_translator.py @@ -0,0 +1,498 @@ +"""Language-aware system prompt routing for multi-model agents. + +Detects the language of a system prompt and translates it to the language +the target model handles best — before the prompt reaches the LLM. + +This module is a pure-function pipeline with no framework coupling. It +can be tested and used independently of CrewAI's Agent class. + +Design decisions: + - **System prompt only**: User messages are never translated. + - **Opt-in/opt-out**: ``Agent(auto_translate_prompt=False)`` disables it. + - **Conservative thresholds**: Only translates if >10 % token savings AND + the model has poor bilingual capability. + - **Code block preservation**: Fenced code, inline code, URLs, and JSON + are detected via regex and excluded from translation. + - **Content-hash caching**: Avoids repeated translation for the same prompt. + +References: + - Multi-IF benchmark (arXiv:2410.15553) + - PromptQuorum (2026-05) + - Presenc AI tokenizer benchmark (2026-05) +""" + +from __future__ import annotations + +import hashlib +import logging +import re +from typing import TYPE_CHECKING, Any +import unicodedata + +from crewai.utilities.model_profiles import ModelProfile, get_model_profile + + +if TYPE_CHECKING: + pass + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Translation cache: content_hash -> translated text +# System prompts change infrequently, so caching avoids redundant LLM calls. +# --------------------------------------------------------------------------- +_TRANSLATION_CACHE: dict[str, str] = {} + +# Minimum token savings ratio to justify translation (0.10 = 10 %) +_MIN_SAVINGS_THRESHOLD: float = 0.10 + +# Below this bilingual capability score, we translate; above, we leave as-is. +_BILINGUAL_THRESHOLD: float = 0.7 + +# --------------------------------------------------------------------------- +# Language detection via Unicode script analysis +# --------------------------------------------------------------------------- + +# Unicode script ranges (approximate) for major writing systems. +_SCRIPT_RANGES: list[tuple[str, str, str]] = [ + ("zh", "\u4e00", "\u9fff"), # CJK Unified Ideographs + ("ja", "\u3040", "\u309f"), # Hiragana + ("ja", "\u30a0", "\u30ff"), # Katakana + ("ko", "\uac00", "\ud7af"), # Hangul Syllables + ("ko", "\u1100", "\u11ff"), # Hangul Jamo + ("ar", "\u0600", "\u06ff"), # Arabic + ("hi", "\u0900", "\u097f"), # Devanagari + ("ru", "\u0400", "\u04ff"), # Cyrillic + ("th", "\u0e00", "\u0e7f"), # Thai + ("he", "\u0590", "\u05ff"), # Hebrew +] + + +def _char_script(char: str) -> str: + """Return the script category for a single character. + + Uses Unicode name-based detection for CJK ideographs (which cover + both Chinese and Japanese) and falls back to category analysis. + """ + cp = ord(char) + + # Check CJK ranges first (most common non-Latin scripts) + if 0x4E00 <= cp <= 0x9FFF: + return "zh" # CJK Unified Ideographs (shared by zh/ja) + + for script, start, end in _SCRIPT_RANGES: + if ord(start) <= cp <= ord(end): + return script + + # Latin and common punctuation / digits + cat = unicodedata.category(char) + if cat.startswith("L"): + name = unicodedata.name(char, "") + if "LATIN" in name: + return "en" + if "CJK" in name: + return "zh" + + return "other" + + +def detect_language(text: str) -> str: + """Detect the primary language of *text* using Unicode script analysis. + + Returns an ISO 639-1 code: ``"en"``, ``"zh"``, ``"ja"``, ``"ko"``, + ``"ar"``, ``"hi"``, ``"ru"``, ``"th"``, ``"he"``. + + For mixed scripts the *dominant* non-``"other"`` script wins. + Falls back to ``"en"`` for empty input or purely numeric/punctuation text. + + Args: + text: The text to analyse. + + Returns: + ISO 639-1 language code. + """ + if not text: + return "en" + + counts: dict[str, int] = {} + for ch in text: + script = _char_script(ch) + if script != "other": + counts[script] = counts.get(script, 0) + 1 + + if not counts: + return "en" + + non_latin = {k: v for k, v in counts.items() if k != "en"} + + if not non_latin: + return "en" + + dominant = max(non_latin, key=non_latin.get) # type: ignore[arg-type] + # If CJK ideographs are present and significant, decide zh vs ja + if dominant == "zh": + # Japanese text will also contain hiragana/katakana + if counts.get("ja", 0) > 0: + ja_chars = counts.get("ja", 0) + zh_chars = non_latin.get("zh", 0) + if ja_chars > zh_chars * 0.3: + return "ja" + return "zh" + + return dominant + + +# --------------------------------------------------------------------------- +# Token estimation (character/word heuristics — no tiktoken dependency) +# --------------------------------------------------------------------------- + +_WORD_RE = re.compile(r"[a-zA-Z]+(?:['-][a-zA-Z]+)*") + + +def estimate_tokens(text: str, profile: ModelProfile) -> int: + """Estimate the token count for *text* given a model profile. + + For English portions, words are counted and multiplied by + ``english_tokens_per_word``. For non-English (CJK) characters, each + character is divided by ``non_english_chars_per_token``. + + Args: + text: The text to estimate. + profile: The target model's language profile. + + Returns: + Estimated token count. + """ + if not text: + return 0 + + # Count English words + english_words = len(_WORD_RE.findall(text)) + english_tokens = int(english_words * profile.english_tokens_per_word) + + # Count non-Latin characters (CJK, Cyrillic, Arabic, etc.) + non_latin_chars = 0 + for ch in text: + script = _char_script(ch) + if script not in ("en", "other"): + non_latin_chars += 1 + + if profile.non_english_chars_per_token > 0: + non_latin_tokens = int( + non_latin_chars / profile.non_english_chars_per_token + ) + else: + non_latin_tokens = non_latin_chars + + return english_tokens + non_latin_tokens + + +# --------------------------------------------------------------------------- +# Code block / structured data extraction +# --------------------------------------------------------------------------- + +# Patterns for segments that must NOT be translated +_FENCED_CODE_RE = re.compile(r"```[\s\S]*?```", re.MULTILINE) +_INLINE_CODE_RE = re.compile(r"`[^`\n]+`") +_URL_RE = re.compile(r"https?://\S+") +_JSON_RE = re.compile(r"\{[^{}]*(?:\{[^{}]*\}[^{}]*)*\}", re.DOTALL) +_ENV_VAR_RE = re.compile(r"\b[A-Z_]{2,}=[^\s,;]+") + + +def extract_untranslatable_segments( + text: str, +) -> tuple[str, list[tuple[str, str]]]: + """Extract code blocks, URLs, JSON, and other untranslatable segments. + + Replaces each matched segment with a placeholder token so that the + translation step only sees natural language. + + Args: + text: The input text. + + Returns: + A tuple of ``(text_with_placeholders, segments)`` where *segments* + is a list of ``(placeholder, original_segment)`` pairs. + """ + segments: list[tuple[str, str]] = [] + counter = 0 + + def _replace(match: re.Match[str]) -> str: + nonlocal counter + placeholder = f"__SEGMENT_{counter}__" + segments.append((placeholder, match.group(0))) + counter += 1 + return placeholder + + # Apply patterns in order; each pattern only matches non-overlapping text + result = _FENCED_CODE_RE.sub(_replace, text) + result = _INLINE_CODE_RE.sub(_replace, result) + result = _URL_RE.sub(_replace, result) + result = _JSON_RE.sub(_replace, result) + result = _ENV_VAR_RE.sub(_replace, result) + + return result, segments + + +def restore_untranslatable_segments( + translated: str, + segments: list[tuple[str, str]], +) -> str: + """Restore original untranslatable segments after translation. + + Args: + translated: The translated text with placeholder tokens. + segments: The ``(placeholder, original_segment)`` pairs from + :func:`extract_untranslatable_segments`. + + Returns: + The text with original segments restored. + """ + result = translated + for placeholder, original in segments: + result = result.replace(placeholder, original) + return result + + +# --------------------------------------------------------------------------- +# Translation via LLM (lightweight, cached) +# --------------------------------------------------------------------------- + +# Language display names for translation prompts +_LANG_NAMES: dict[str, str] = { + "en": "English", + "zh": "Chinese (Simplified)", + "ja": "Japanese", + "ko": "Korean", + "ar": "Arabic", + "hi": "Hindi", + "ru": "Russian", + "th": "Thai", + "he": "Hebrew", +} + + +def _cache_key( + text: str, lang_pair: str, glossary: dict[str, str] | None = None +) -> str: + """Generate a content-hash cache key. + + Includes text, language pair, and glossary so different glossaries + produce different cache entries. + """ + h = hashlib.sha256() + h.update(text.encode("utf-8")) + h.update(lang_pair.encode("utf-8")) + if glossary: + for k in sorted(glossary): + h.update(k.encode("utf-8")) + h.update(glossary[k].encode("utf-8")) + return h.hexdigest() + + +def translate_text( + text: str, + source_lang: str, + target_lang: str, + glossary: dict[str, str] | None = None, + llm_caller: Any | None = None, +) -> str: + """Translate *text* from *source_lang* to *target_lang*. + + This function uses a cached LLM call. If *llm_caller* is ``None``, + it falls back to a simple glossary-apply-and-return (useful in tests + and when no LLM is available for translation). + + Args: + text: The text to translate. + source_lang: ISO 639-1 source language code. + target_lang: ISO 639-1 target language code. + glossary: Optional dict of terms that should be preserved as-is. + llm_caller: An object with a ``call(messages)`` method, or ``None``. + + Returns: + The translated text, or the original text if translation fails. + """ + if not text.strip(): + return text + + if source_lang == target_lang: + return text + + # Check cache (key includes glossary to avoid stale translations) + key = _cache_key(text, f"{source_lang}->{target_lang}", glossary) + if key in _TRANSLATION_CACHE: + return _TRANSLATION_CACHE[key] + + # Build glossary instruction + glossary_instruction = "" + if glossary: + terms = ", ".join(f'"{k}" -> "{v}"' for k, v in glossary.items()) + glossary_instruction = ( + f"\n\nImportant: Preserve these terms exactly as written: {terms}" + ) + + source_name = _LANG_NAMES.get(source_lang, source_lang) + target_name = _LANG_NAMES.get(target_lang, target_lang) + + prompt = ( + f"Translate the following text from {source_name} to {target_name}. " + f"Preserve all formatting, line breaks, and placeholder tokens " + f"(like __SEGMENT_0__). " + f"Do not translate code, variable names, or technical identifiers." + f"{glossary_instruction}" + f"\n\nText to translate:\n{text}" + ) + + translated = text # fallback + + if llm_caller is not None: + try: + messages = [{"role": "user", "content": prompt}] + response = llm_caller.call(messages) + if isinstance(response, str) and response.strip(): + translated = response.strip() + # Only cache successful translations — fallback text + # should not be cached so transient failures can retry. + _TRANSLATION_CACHE[key] = translated + except Exception: + logger.debug( + "Translation LLM call failed; returning original text.", + exc_info=True, + ) + else: + # No LLM available — return original (safe fallback) + logger.debug( + "No LLM caller provided for translation; returning original text." + ) + + return translated + + +# --------------------------------------------------------------------------- +# Main entry point +# --------------------------------------------------------------------------- + + +def optimize_system_prompt( + prompt: str, + model_name: str, + glossary: dict[str, str] | None = None, + llm_caller: Any | None = None, +) -> str: + """Optimize a system prompt for the target model's language capabilities. + + Pipeline: + + 1. Detect the language of the prompt. + 2. Look up the model profile. + 3. If the model prefers the same language → return as-is. + 4. Estimate token cost in the current vs. target language. + 5. If translation saves >10 % tokens AND the model has poor bilingual + capability → translate. + 6. Extract code blocks / JSON / URLs before translation. + 7. Translate natural-language portions. + 8. Restore untranslatable segments. + 9. Cache the result. + + Args: + prompt: The system prompt text. + model_name: The model identifier (e.g. ``"gpt-4o"``). + glossary: Optional dict of terms that should not be translated. + llm_caller: An object with a ``call(messages)`` method for + performing the actual translation, or ``None``. + + Returns: + The (possibly translated) system prompt. + """ + if not prompt or not prompt.strip(): + return prompt + + # 1. Detect language + source_lang = detect_language(prompt) + + # 2. Look up model profile + profile = get_model_profile(model_name) + if profile is None: + # Unknown model — conservative, no translation + return prompt + + # 3. If model prefers this language, no translation needed + if profile.preferred_language == source_lang: + return prompt + + # 4. Estimate token costs + # After translation, the text composition inverts: what was Chinese + # becomes English words and vice versa. We approximate the post- + # translation cost by mapping source-language tokens to their + # target-language equivalents. + english_words = len(_WORD_RE.findall(prompt)) + non_latin_chars = sum( + 1 for ch in prompt if _char_script(ch) not in ("en", "other") + ) + + # Current cost (source language) + if source_lang == "en": + current_tokens = int(english_words * profile.english_tokens_per_word) + non_latin_chars + else: + current_tokens = ( + english_words + + int(non_latin_chars / profile.non_english_chars_per_token) + ) + + # Target cost (after translation): English words in source become + # target-lang tokens, and non-Latin chars in source become English words. + target_lang = profile.preferred_language + if target_lang == "en": + # Each non-Latin char becomes ~1 English word after translation + target_tokens = int(non_latin_chars * profile.english_tokens_per_word) + english_words + else: + # Each English word becomes ~1 target-lang char after translation + target_tokens = ( + non_latin_chars + + int(english_words / profile.non_english_chars_per_token) + ) + + # 5. Check if translation is beneficial + if current_tokens == 0: + return prompt + + savings = 1.0 - (target_tokens / current_tokens) + + if savings < _MIN_SAVINGS_THRESHOLD: + return prompt + + if profile.bilingual_capability >= _BILINGUAL_THRESHOLD: + return prompt + + # 6. Extract untranslatable segments + cleaned_text, segments = extract_untranslatable_segments(prompt) + + # 7. Translate + translated = translate_text( + cleaned_text, + source_lang=source_lang, + target_lang=profile.preferred_language, + glossary=glossary, + llm_caller=llm_caller, + ) + + # 8. Restore segments + result = restore_untranslatable_segments(translated, segments) + + # 9. Log for observability + logger.info( + "Prompt translated from %s to %s for model %s " + "(estimated savings: %.1f%%)", + source_lang, + profile.preferred_language, + model_name, + savings * 100, + ) + + return result + + +def clear_translation_cache() -> None: + """Clear the translation cache. Useful for testing.""" + _TRANSLATION_CACHE.clear() diff --git a/lib/crewai/tests/utilities/test_model_profiles.py b/lib/crewai/tests/utilities/test_model_profiles.py new file mode 100644 index 0000000000..1799d4ffba --- /dev/null +++ b/lib/crewai/tests/utilities/test_model_profiles.py @@ -0,0 +1,147 @@ +"""Tests for the model profile registry.""" + +import pytest + +from crewai.utilities.model_profiles import ( + ModelProfile, + _MODEL_PATTERNS, + get_model_profile, +) + + +class TestModelProfile: + """Tests for the ModelProfile dataclass.""" + + def test_frozen(self): + """ModelProfile should be immutable.""" + profile = ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.85, + family="openai", + ) + with pytest.raises(AttributeError): + profile.preferred_language = "zh" # type: ignore[misc] + + def test_hashable(self): + """ModelProfile should be hashable (frozen dataclass).""" + profile = ModelProfile( + preferred_language="en", + english_tokens_per_word=1.1, + non_english_chars_per_token=0.6, + bilingual_capability=0.85, + family="openai", + ) + # Should be usable as dict key + d = {profile: "test"} + assert d[profile] == "test" + + +class TestGetModelProfile: + """Tests for the get_model_profile function.""" + + @pytest.mark.parametrize( + "model_name,expected_family,expected_lang", + [ + ("gpt-4o", "openai", "en"), + ("gpt-4o-2024-05-13", "openai", "en"), + ("gpt-5", "openai", "en"), + ("gpt-5-2025-08-07", "openai", "en"), + ("claude-3-5-sonnet-20241022", "anthropic", "en"), + ("claude-opus-4-5-20251101", "anthropic", "en"), + ("llama-3.1-70b-versatile", "llama", "en"), + ("llama-3.3-70b-versatile", "llama", "en"), + ("Qwen2.5-72B-Instruct", "qwen", "zh"), + ("deepseek-chat", "deepseek", "en"), + ("gemini-2.5-pro", "google", "en"), + ("gemini-2.0-flash", "google", "en"), + ("mistral-large-latest", "mistral", "en"), + ("mixtral-8x7b-32768", "mistral", "en"), + ("amazon.nova-pro-v1:0", "amazon", "en"), + ], + ids=[ + "gpt-4o", + "gpt-4o-versioned", + "gpt-5", + "gpt-5-versioned", + "claude-sonnet", + "claude-opus", + "llama-3.1", + "llama-3.3", + "qwen", + "deepseek", + "gemini-pro", + "gemini-flash", + "mistral", + "mixtral", + "nova", + ], + ) + def test_known_models(self, model_name, expected_family, expected_lang): + """Known models should return correct profiles.""" + profile = get_model_profile(model_name) + assert profile is not None + assert profile.family == expected_family + assert profile.preferred_language == expected_lang + + def test_unknown_model_returns_none(self): + """Unknown models should return None (conservative).""" + assert get_model_profile("some-unknown-model-xyz") is None + + def test_empty_string_returns_none(self): + """Empty string should return None.""" + assert get_model_profile("") is None + + def test_longest_match_wins(self): + """Longest matching key should win for overlapping patterns.""" + profile = get_model_profile("gpt-4o-2024-05-13") + assert profile is not None + assert profile.family == "openai" + + def test_case_insensitive(self): + """Matching should be case-insensitive.""" + profile = get_model_profile("GPT-4O") + assert profile is not None + assert profile.family == "openai" + + def test_partial_model_name(self): + """Partial model names should match if pattern matches.""" + profile = get_model_profile("my-custom-gpt-4o-deployment") + assert profile is not None + assert profile.family == "openai" + + def test_no_false_positive_on_unrelated_names(self): + """Short patterns should not match unrelated model names.""" + # "o1" should NOT match "proto1" or "bio1" + assert get_model_profile("proto1") is None + assert get_model_profile("bio1") is None + + def test_o1_matches_with_boundary(self): + """o1 pattern should match actual o1 models.""" + assert get_model_profile("o1-preview") is not None + assert get_model_profile("o1-mini") is not None + assert get_model_profile("openai/o1-preview") is not None + + def test_patterns_not_empty(self): + """_MODEL_PATTERNS should contain entries.""" + assert len(_MODEL_PATTERNS) > 0 + + def test_all_profiles_have_valid_languages(self): + """All profiles should have valid ISO 639-1 language codes.""" + valid_langs = { + "en", "zh", "ja", "ko", "ar", "hi", "ru", "th", "he", + } + for pattern, profile in _MODEL_PATTERNS: + assert profile.preferred_language in valid_langs, ( + f"Pattern {pattern.pattern} has invalid language: " + f"{profile.preferred_language}" + ) + + def test_all_profiles_have_valid_capability(self): + """All profiles should have bilingual_capability between 0 and 1.""" + for pattern, profile in _MODEL_PATTERNS: + assert 0.0 <= profile.bilingual_capability <= 1.0, ( + f"Pattern {pattern.pattern} has invalid bilingual_capability: " + f"{profile.bilingual_capability}" + ) diff --git a/lib/crewai/tests/utilities/test_prompt_translator.py b/lib/crewai/tests/utilities/test_prompt_translator.py new file mode 100644 index 0000000000..64eb50f6b7 --- /dev/null +++ b/lib/crewai/tests/utilities/test_prompt_translator.py @@ -0,0 +1,394 @@ +"""Tests for the prompt translator module.""" + +import pytest + +from crewai.utilities.model_profiles import ModelProfile +from crewai.utilities.prompt_translator import ( + _TRANSLATION_CACHE, + clear_translation_cache, + detect_language, + estimate_tokens, + extract_untranslatable_segments, + optimize_system_prompt, + restore_untranslatable_segments, + translate_text, +) + + +@pytest.fixture(autouse=True) +def _clear_cache(): + """Clear translation cache before each test.""" + clear_translation_cache() + yield + clear_translation_cache() + + +# --------------------------------------------------------------------------- +# Language detection +# --------------------------------------------------------------------------- +class TestDetectLanguage: + def test_english(self): + assert detect_language("You are a helpful assistant.") == "en" + + def test_chinese(self): + assert detect_language("你是一个有用的助手。请帮我完成任务。") == "zh" + + def test_japanese(self): + assert detect_language("あなたは有用的なアシスタントです。") == "ja" + + def test_korean(self): + assert detect_language("당신은 유용한 어시스턴트입니다.") == "ko" + + def test_arabic(self): + assert detect_language("أنت مساعد مفيد. يرجى مساعدتي.") == "ar" + + def test_russian(self): + assert detect_language("Вы полезный помощник. Пожалуйста, помогите.") == "ru" + + def test_hindi(self): + assert detect_language("आप एक सहायक हैं। कृपया मेरी मदद करें।") == "hi" + + def test_mixed_english_dominant(self): + text = "You are a helpful assistant. 你好，请帮我。" + lang = detect_language(text) + # Should detect as English (dominant) or Chinese (significant non-Latin) + assert lang in ("en", "zh") + + def test_empty_string(self): + assert detect_language("") == "en" + + def test_numbers_only(self): + assert detect_language("12345 67890") == "en" + + def test_punctuation_only(self): + assert detect_language("!!! ??? ...") == "en" + + def test_chinese_with_english(self): + text = "请使用Python编写一个API接口，返回JSON格式的数据。" + lang = detect_language(text) + # Should detect Chinese as dominant + assert lang == "zh" + + def test_japanese_mixed_scripts(self): + text = "このコードをPythonで実装してください。Use async/await pattern." + lang = detect_language(text) + assert lang == "ja" + + +# --------------------------------------------------------------------------- +# Token estimation +# --------------------------------------------------------------------------- +class TestEstimateTokens: + def _make_profile( + self, + eng_tpw: float = 1.1, + non_eng_cpt: float = 0.6, + bilingual: float = 0.85, + ) -> ModelProfile: + return ModelProfile( + preferred_language="en", + english_tokens_per_word=eng_tpw, + non_english_chars_per_token=non_eng_cpt, + bilingual_capability=bilingual, + family="test", + ) + + def test_english_text(self): + profile = self._make_profile() + tokens = estimate_tokens("hello world", profile) + # 2 words * 1.1 tokens/word = ~2.2, rounded to int + assert tokens >= 2 + + def test_chinese_text(self): + profile = self._make_profile() + tokens = estimate_tokens("你好世界", profile) + # 4 CJK chars / 0.6 = ~6.67, rounded to int + assert tokens >= 6 + + def test_empty_text(self): + profile = self._make_profile() + assert estimate_tokens("", profile) == 0 + + def test_mixed_text(self): + profile = self._make_profile() + tokens = estimate_tokens("Hello 你好", profile) + # 1 English word * 1.1 + 2 CJK chars / 0.6 ≈ 1 + 3.3 = ~4 + assert tokens >= 3 + + def test_high_efficiency_profile(self): + """Qwen-like profile: Chinese is cheaper than English.""" + profile = ModelProfile( + preferred_language="zh", + english_tokens_per_word=1.05, + non_english_chars_per_token=0.95, + bilingual_capability=0.95, + family="qwen", + ) + tokens = estimate_tokens("你好世界", profile) + # 4 / 0.95 ≈ 4.2 + assert tokens >= 4 + + +# --------------------------------------------------------------------------- +# Code block extraction +# --------------------------------------------------------------------------- +class TestExtractUntranslatableSegments: + def test_fenced_code_block(self): + text = "Use this code:\n```python\nprint('hello')\n```\nDone." + cleaned, segments = extract_untranslatable_segments(text) + assert len(segments) == 1 + assert "```python" in segments[0][1] + assert "__SEGMENT_0__" in cleaned + assert "Use this code" in cleaned + + def test_inline_code(self): + text = "Call the `get_data()` function." + cleaned, segments = extract_untranslatable_segments(text) + assert len(segments) == 1 + assert "`get_data()`" in segments[0][1] + assert "__SEGMENT_0__" in cleaned + + def test_url(self): + text = "Visit https://example.com for details." + cleaned, segments = extract_untranslatable_segments(text) + assert len(segments) == 1 + assert "https://example.com" in segments[0][1] + + def test_json_object(self): + text = 'Config: {"key": "value", "count": 42}' + cleaned, segments = extract_untranslatable_segments(text) + # May match JSON or not depending on regex complexity + # At minimum, the text should be processable + assert isinstance(cleaned, str) + + def test_env_var(self): + text = "Set API_KEY=secret123 before running." + cleaned, segments = extract_untranslatable_segments(text) + # ENV var pattern may or may not match + assert isinstance(cleaned, str) + + def test_multiple_segments(self): + text = "Code: ```python\nx = 1\n```\nURL: https://example.com" + cleaned, segments = extract_untranslatable_segments(text) + assert len(segments) >= 1 + + def test_no_segments(self): + text = "This is plain English text with no special patterns." + cleaned, segments = extract_untranslatable_segments(text) + assert len(segments) == 0 + assert cleaned == text + + +class TestRestoreUntranslatableSegments: + def test_roundtrip(self): + original = "Code: ```python\nx = 1\n```\nDone." + cleaned, segments = extract_untranslatable_segments(original) + restored = restore_untranslatable_segments(cleaned, segments) + assert restored == original + + def test_no_segments(self): + text = "No segments here." + restored = restore_untranslatable_segments(text, []) + assert restored == text + + def test_multiple_roundtrip(self): + original = "Use `func()` at https://example.com with ```code```" + cleaned, segments = extract_untranslatable_segments(original) + restored = restore_untranslatable_segments(cleaned, segments) + assert restored == original + + +# --------------------------------------------------------------------------- +# Translation +# --------------------------------------------------------------------------- +class TestTranslateText: + def test_same_language_returns_original(self): + text = "Hello world" + result = translate_text(text, "en", "en") + assert result == text + + def test_empty_text(self): + result = translate_text("", "en", "zh") + assert result == "" + + def test_whitespace_only(self): + result = translate_text(" ", "en", "zh") + assert result == " " + + def test_no_llm_returns_original(self): + """Without an LLM caller, translation returns original text.""" + text = "You are a helpful assistant." + result = translate_text(text, "en", "zh", llm_caller=None) + assert result == text + + def test_with_mock_llm(self): + """With a mock LLM caller, translation uses the LLM response.""" + + class MockLLM: + def call(self, messages): + return "你是一个有用的助手。" + + text = "You are a helpful assistant." + result = translate_text( + text, "en", "zh", llm_caller=MockLLM() + ) + assert result == "你是一个有用的助手。" + + def test_caching(self): + """Translation results should be cached.""" + + class CallCounter: + def __init__(self): + self.count = 0 + + def call(self, messages): + self.count += 1 + return "translated" + + counter = CallCounter() + text = "Test text" + translate_text(text, "en", "zh", llm_caller=counter) + translate_text(text, "en", "zh", llm_caller=counter) + # Second call should use cache, not call LLM again + assert counter.count == 1 + + def test_glossary_in_prompt(self): + """Glossary terms should appear in the translation prompt.""" + captured_messages = [] + + class CapturingLLM: + def call(self, messages): + captured_messages.extend(messages) + return "translated" + + glossary = {"API Key": "API Key", "crewAI": "crewAI"} + translate_text( + "Use the API Key to authenticate.", + "en", + "zh", + glossary=glossary, + llm_caller=CapturingLLM(), + ) + assert len(captured_messages) == 1 + assert "API Key" in captured_messages[0]["content"] + assert "crewAI" in captured_messages[0]["content"] + + def test_llm_failure_returns_original(self): + """If LLM call fails, original text should be returned.""" + + class FailingLLM: + def call(self, messages): + raise RuntimeError("LLM unavailable") + + text = "Test prompt" + result = translate_text(text, "en", "zh", llm_caller=FailingLLM()) + assert result == text + + +# --------------------------------------------------------------------------- +# optimize_system_prompt (integration) +# --------------------------------------------------------------------------- +class TestOptimizeSystemPrompt: + def test_empty_prompt(self): + assert optimize_system_prompt("", "gpt-4o") == "" + + def test_none_prompt(self): + assert optimize_system_prompt(None, "gpt-4o") is None # type: ignore[arg-type] + + def test_unknown_model_returns_original(self): + text = "你是一个有用的助手。" + result = optimize_system_prompt(text, "unknown-model-xyz") + assert result == text + + def test_same_language_as_preferred(self): + """English prompt with GPT-4o (prefers English) should not be translated.""" + text = "You are a helpful assistant. Please complete the task." + result = optimize_system_prompt(text, "gpt-4o") + assert result == text + + def test_chinese_prompt_with_llama(self): + """Chinese prompt with LLaMA (poor bilingual) should be translated.""" + chinese_text = "你是一个有用的助手。请帮我完成任务。" + + class MockLLM: + def call(self, messages): + return "You are a helpful assistant. Please help me complete the task." + + result = optimize_system_prompt( + chinese_text, "llama-3.1-70b-versatile", llm_caller=MockLLM() + ) + # Should be translated because LLaMA has poor bilingual capability + # and Chinese prompt is less efficient on LLaMA + assert result != chinese_text + + def test_code_blocks_preserved(self): + """Code blocks should be preserved during translation.""" + chinese_text = "使用以下代码：\n```python\nprint('hello')\n```\n完成。" + + class MockLLM: + def call(self, messages): + return "Use the following code:\n```python\nprint('hello')\n```\nDone." + + result = optimize_system_prompt( + chinese_text, "llama-3.1-70b-versatile", llm_caller=MockLLM() + ) + # Code block should be preserved + assert "print('hello')" in result + + def test_glossary_passed_through(self): + """Glossary terms should be passed to the translation LLM.""" + chinese_text = "使用API Key进行认证。" + + captured = [] + + class CapturingLLM: + def call(self, messages): + captured.extend(messages) + return "Authenticate using the API Key." + + glossary = {"API Key": "API Key"} + optimize_system_prompt( + chinese_text, + "llama-3.1-70b-versatile", + glossary=glossary, + llm_caller=CapturingLLM(), + ) + assert len(captured) == 1 + assert "API Key" in captured[0]["content"] + + def test_caching_across_calls(self): + """Same prompt + model should return cached result.""" + chinese_text = "你是一个助手。" + + class Counter: + def __init__(self): + self.count = 0 + + def call(self, messages): + self.count += 1 + return "You are an assistant." + + counter = Counter() + result1 = optimize_system_prompt( + chinese_text, "llama-3.1-70b-versatile", llm_caller=counter + ) + result2 = optimize_system_prompt( + chinese_text, "llama-3.1-70b-versatile", llm_caller=counter + ) + assert result1 == result2 + assert counter.count == 1 # Only one LLM call + + def test_bilingual_model_no_translation(self): + """Qwen (prefers Chinese, high bilingual) should not translate Chinese.""" + chinese_text = "你是一个有用的助手。请帮我完成任务。" + result = optimize_system_prompt(chinese_text, "qwen") + # Qwen prefers Chinese, so Chinese prompt should not be translated + assert result == chinese_text + + +class TestClearTranslationCache: + def test_clear(self): + """Cache should be cleared.""" + _TRANSLATION_CACHE["test"] = "value" + clear_translation_cache() + assert len(_TRANSLATION_CACHE) == 0 diff --git a/lib/crewai/tests/utilities/test_prompt_translator_integration.py b/lib/crewai/tests/utilities/test_prompt_translator_integration.py new file mode 100644 index 0000000000..9b9a0d1ea2 --- /dev/null +++ b/lib/crewai/tests/utilities/test_prompt_translator_integration.py @@ -0,0 +1,170 @@ +"""Integration tests for Agent with auto_translate_prompt feature.""" + +from unittest.mock import MagicMock, patch + +import pytest + +from crewai import Agent +from crewai.utilities.prompt_translator import clear_translation_cache + + +@pytest.fixture(autouse=True) +def _clear_cache(): + """Clear translation cache before each test.""" + clear_translation_cache() + yield + clear_translation_cache() + + +class TestAgentAutoTranslatePrompt: + """Tests for the auto_translate_prompt parameter on Agent.""" + + def test_default_enabled(self): + """auto_translate_prompt should default to True.""" + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + ) + assert agent.auto_translate_prompt is True + + def test_disable(self): + """auto_translate_prompt can be disabled.""" + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + auto_translate_prompt=False, + ) + assert agent.auto_translate_prompt is False + + def test_glossary_default_none(self): + """prompt_glossary should default to None.""" + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + ) + assert agent.prompt_glossary is None + + def test_glossary_set(self): + """prompt_glossary can be set.""" + glossary = {"API Key": "API Key", "crewAI": "crewAI"} + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + prompt_glossary=glossary, + ) + assert agent.prompt_glossary == glossary + + +class TestAgentPromptTranslation: + """Tests for prompt translation integration in Agent._build_execution_prompt.""" + + @patch("crewai.utilities.prompt_translator.optimize_system_prompt") + def test_translation_called_when_enabled(self, mock_optimize): + """optimize_system_prompt should be called when auto_translate_prompt=True.""" + mock_optimize.side_effect = lambda prompt, model, glossary=None: prompt + + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + auto_translate_prompt=True, + ) + + # Access the private method to test prompt building + from crewai.tools.structured_tool import CrewStructuredTool + + prompt, _, _ = agent._build_execution_prompt([]) + + # optimize_system_prompt should have been called + assert mock_optimize.called + + @patch("crewai.utilities.prompt_translator.optimize_system_prompt") + def test_translation_not_called_when_disabled(self, mock_optimize): + """optimize_system_prompt should NOT be called when auto_translate_prompt=False.""" + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + auto_translate_prompt=False, + ) + + prompt, _, _ = agent._build_execution_prompt([]) + + mock_optimize.assert_not_called() + + @patch("crewai.utilities.prompt_translator.optimize_system_prompt") + def test_glossary_passed_to_optimizer(self, mock_optimize): + """prompt_glossary should be passed to optimize_system_prompt.""" + mock_optimize.side_effect = lambda prompt, model, glossary=None: prompt + + glossary = {"API Key": "API Key"} + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + auto_translate_prompt=True, + prompt_glossary=glossary, + ) + + prompt, _, _ = agent._build_execution_prompt([]) + + # Check that optimize_system_prompt was called with the glossary + for call in mock_optimize.call_args_list: + assert call[1].get("glossary") == glossary or ( + len(call[0]) >= 3 and call[0][2] == glossary + ) + + @patch("crewai.utilities.prompt_translator.optimize_system_prompt") + def test_model_name_extracted_from_llm(self, mock_optimize): + """The model name should be extracted from the LLM instance.""" + mock_optimize.side_effect = lambda prompt, model, glossary=None: prompt + + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + auto_translate_prompt=True, + ) + + prompt, _, _ = agent._build_execution_prompt([]) + + # The model name should have been passed to optimize_system_prompt + for call in mock_optimize.call_args_list: + model_arg = call[0][1] if len(call[0]) > 1 else call[1].get("model_name") + assert "gpt-4o" in str(model_arg) + + @patch("crewai.utilities.prompt_translator.optimize_system_prompt") + def test_system_prompt_result_optimized(self, mock_optimize): + """When SystemPromptResult is returned, system prompt should be optimized.""" + mock_optimize.side_effect = lambda prompt, model, glossary=None, llm_caller=None: f"optimized:{prompt}" + + agent = Agent( + role="test", + goal="test", + backstory="test", + llm="gpt-4o", + auto_translate_prompt=True, + use_system_prompt=True, + ) + + prompt, _, _ = agent._build_execution_prompt([]) + + # The system prompt should carry the "optimized:" prefix + from crewai.utilities.prompts import SystemPromptResult + + assert isinstance(prompt, SystemPromptResult) + assert prompt.system.startswith("optimized:"), ( + f"Expected system prompt to start with 'optimized:', got: {prompt.system!r}" + )