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}" + )