feat: Separate system and user prompts for all LLM providers

hydropix · hydropix · commit 912d3af409d3 · 2025-11-30T16:24:11.000+01:00
- Add PromptPair NamedTuple to return system/user prompts separately
- Update OllamaProvider to use 'system' field in payload
- Update OpenAICompatibleProvider to include system message in messages array
- Update GeminiProvider to use 'systemInstruction' field
- Refactor translator.py and subtitle_translator.py to use new prompt structure
diff --git a/prompts/__init__.py b/prompts/__init__.py
@@ -1,3 +1,14 @@
 """
 Prompts module for TranslateBookWithLLM
 """
+from prompts.prompts import (
+    PromptPair,
+    generate_translation_prompt,
+    generate_subtitle_block_prompt,
+)
+
+__all__ = [
+    "PromptPair",
+    "generate_translation_prompt",
+    "generate_subtitle_block_prompt",
+]
diff --git a/prompts/prompts.py b/prompts/prompts.py
@@ -1,7 +1,13 @@
-from typing import List, Tuple
+from typing import List, Tuple, NamedTuple
 from src.config import TRANSLATE_TAG_IN, TRANSLATE_TAG_OUT, INPUT_TAG_IN, INPUT_TAG_OUT
 
 
+class PromptPair(NamedTuple):
+    """A pair of system and user prompts for LLM translation."""
+    system: str
+    user: str
+
+
 # ============================================================================
 # SHARED PROMPT SECTIONS
 # ============================================================================
@@ -88,7 +94,7 @@ def generate_translation_prompt(
     translate_tag_in: str = TRANSLATE_TAG_IN,
     translate_tag_out: str = TRANSLATE_TAG_OUT,
     fast_mode: bool = False
-) -> str:
+) -> PromptPair:
     """
     Generate the translation prompt with all contextual elements.
 
@@ -104,7 +110,7 @@ def generate_translation_prompt(
         fast_mode: If True, excludes placeholder preservation instructions (for pure text translation)
 
     Returns:
-        str: The complete prompt formatted for translation
+        PromptPair: A named tuple with 'system' and 'user' prompts
     """
     # Get target-language-specific example text for output format
     example_texts = {
@@ -134,8 +140,8 @@ def generate_translation_prompt(
         example_format=example_format_text
     )
 
-    # PROMPT - can be edited for custom usages
-    role_and_instructions_block = f"""You are a professional {target_language} translator and writer.
+    # SYSTEM PROMPT - Role and instructions (stable across requests)
+    system_prompt = f"""You are a professional {target_language} translator and writer.
 
 # CRITICAL: TARGET LANGUAGE IS {target_language.upper()}
 
@@ -169,23 +175,20 @@ def generate_translation_prompt(
 Your entire translation output must be written in {target_language}.
 Do NOT write in {source_language} or any other language - ONLY {target_language.upper()}.
 
-{output_format_section}
-"""
+{output_format_section}"""
 
+    # USER PROMPT - Context and content to translate (varies per request)
     previous_translation_block_text = ""
     if previous_translation_context and previous_translation_context.strip():
-        previous_translation_block_text = f"""
-
-# CONTEXT - Previous Paragraph
+        previous_translation_block_text = f"""# CONTEXT - Previous Paragraph
 
 For consistency and natural flow, here's what came immediately before:
 
 {previous_translation_context}
 
 """
 
-    text_to_translate_block = f"""
-# TEXT TO TRANSLATE
+    user_prompt = f"""{previous_translation_block_text}# TEXT TO TRANSLATE
 
 {INPUT_TAG_IN}
 {main_content}
@@ -200,13 +203,7 @@ def generate_translation_prompt(
 
 Provide your translation now:"""
 
-    parts = [part.strip() for part in [
-        role_and_instructions_block,
-        previous_translation_block_text,
-        text_to_translate_block
-    ] if part]
-
-    return "\n\n".join(parts).strip()
+    return PromptPair(system=system_prompt.strip(), user=user_prompt.strip())
 
 
 def generate_subtitle_block_prompt(
@@ -217,7 +214,7 @@ def generate_subtitle_block_prompt(
     translate_tag_in: str = TRANSLATE_TAG_IN,
     translate_tag_out: str = TRANSLATE_TAG_OUT,
     custom_instructions: str = ""
-) -> str:
+) -> PromptPair:
     """
     Generate translation prompt for multiple subtitle blocks with index markers.
 
@@ -231,7 +228,7 @@ def generate_subtitle_block_prompt(
         custom_instructions: Additional custom translation instructions
 
     Returns:
-        str: The complete prompt formatted for subtitle block translation
+        PromptPair: A named tuple with 'system' and 'user' prompts
     """
     # Build the output format section outside the f-string to avoid backslash issues in Python 3.11
     subtitle_additional_rules = "\n6. Each subtitle has an index marker: [index]text - PRESERVE these markers exactly\n7. Maintain line breaks between indexed subtitles"
@@ -255,8 +252,8 @@ def generate_subtitle_block_prompt(
 {custom_instructions.strip()}
 """
 
-    # Enhanced instructions for subtitle translation
-    role_and_instructions_block = f"""You are a professional {target_language} subtitle translator and dialogue adaptation specialist.
+    # SYSTEM PROMPT - Role and instructions for subtitle translation
+    system_prompt = f"""You are a professional {target_language} subtitle translator and dialogue adaptation specialist.
 
 # CRITICAL: TARGET LANGUAGE IS {target_language.upper()}
 
@@ -286,15 +283,12 @@ def generate_subtitle_block_prompt(
 Your entire subtitle translation must be written in {target_language}.
 Do NOT write in {source_language} or any other language - ONLY {target_language.upper()}.
 
-{subtitle_output_format_section}
-"""
+{subtitle_output_format_section}"""
 
-    # Previous translation context
+    # USER PROMPT - Context and subtitles to translate
     previous_translation_block_text = ""
     if previous_translation_block and previous_translation_block.strip():
-        previous_translation_block_text = f"""
-
-# CONTEXT - Previous Subtitle Block
+        previous_translation_block_text = f"""# CONTEXT - Previous Subtitle Block
 
 For continuity and consistency, here's the previous subtitle block:
 
@@ -308,8 +302,7 @@ def generate_subtitle_block_prompt(
     # Join subtitles outside f-string to avoid Python 3.11 backslash issues
     formatted_subtitles_text = "\n".join(formatted_subtitles)
 
-    text_to_translate_block = f"""
-# SUBTITLES TO TRANSLATE
+    user_prompt = f"""{previous_translation_block_text}# SUBTITLES TO TRANSLATE
 
 {INPUT_TAG_IN}
 {formatted_subtitles_text}
@@ -325,10 +318,4 @@ def generate_subtitle_block_prompt(
 
 Provide your translation now:"""
 
-    parts = [part.strip() for part in [
-        role_and_instructions_block,
-        previous_translation_block_text,
-        text_to_translate_block
-    ] if part]
-
-    return "\n".join(parts).strip()
+    return PromptPair(system=system_prompt.strip(), user=user_prompt.strip())
diff --git a/src/core/llm_client.py b/src/core/llm_client.py
@@ -29,29 +29,30 @@ def _get_provider(self) -> LLMProvider:
             self._provider = create_llm_provider(self.provider_type, **self.provider_kwargs)
         return self._provider
     
-    async def make_request(self, prompt: str, model: Optional[str] = None, 
-                    timeout: int = None) -> Optional[str]:
+    async def make_request(self, prompt: str, model: Optional[str] = None,
+                    timeout: int = None, system_prompt: Optional[str] = None) -> Optional[str]:
         """
         Make a request to the LLM API with error handling and retries
-        
+
         Args:
-            prompt: The prompt to send
+            prompt: The user prompt to send (content to process)
             model: Model to use (defaults to instance model)
             timeout: Request timeout in seconds
-            
+            system_prompt: Optional system prompt (role/instructions)
+
         Returns:
             Raw response text or None if failed
         """
         provider = self._get_provider()
-        
+
         # Update model if specified
         if model:
             provider.model = model
-            
+
         if timeout:
-            return await provider.generate(prompt, timeout)
+            return await provider.generate(prompt, timeout, system_prompt=system_prompt)
         else:
-            return await provider.generate(prompt)
+            return await provider.generate(prompt, system_prompt=system_prompt)
     
     def extract_translation(self, response: str) -> Optional[str]:
         """
diff --git a/src/core/llm_providers.py b/src/core/llm_providers.py
@@ -22,15 +22,15 @@ class ContextOverflowError(Exception):
 
 class LLMProvider(ABC):
     """Abstract base class for LLM providers"""
-    
+
     def __init__(self, model: str):
         self.model = model
         self._compiled_regex = re.compile(
-            rf"{re.escape(TRANSLATE_TAG_IN)}(.*?){re.escape(TRANSLATE_TAG_OUT)}", 
+            rf"{re.escape(TRANSLATE_TAG_IN)}(.*?){re.escape(TRANSLATE_TAG_OUT)}",
             re.DOTALL
         )
         self._client = None
-    
+
     async def _get_client(self) -> httpx.AsyncClient:
         """Get or create a persistent HTTP client with connection pooling"""
         if self._client is None:
@@ -39,16 +39,27 @@ async def _get_client(self) -> httpx.AsyncClient:
                 timeout=httpx.Timeout(REQUEST_TIMEOUT)
             )
         return self._client
-    
+
     async def close(self):
         """Close the HTTP client"""
         if self._client:
             await self._client.aclose()
             self._client = None
-    
+
     @abstractmethod
-    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT) -> Optional[str]:
-        """Generate text from prompt"""
+    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT,
+                      system_prompt: Optional[str] = None) -> Optional[str]:
+        """
+        Generate text from prompt.
+
+        Args:
+            prompt: The user prompt (content to process)
+            timeout: Request timeout in seconds
+            system_prompt: Optional system prompt (role/instructions)
+
+        Returns:
+            Generated text or None if failed
+        """
         pass
     
     def extract_translation(self, response: str) -> Optional[str]:
@@ -134,9 +145,20 @@ def __init__(self, api_endpoint: str = API_ENDPOINT, model: str = DEFAULT_MODEL,
         self.api_endpoint = api_endpoint
         self.context_window = context_window
         self.log_callback = log_callback
-    
-    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT) -> Optional[str]:
-        """Generate text using Ollama API"""
+
+    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT,
+                      system_prompt: Optional[str] = None) -> Optional[str]:
+        """
+        Generate text using Ollama API.
+
+        Args:
+            prompt: The user prompt (content to translate)
+            timeout: Request timeout in seconds
+            system_prompt: Optional system prompt (role/instructions)
+
+        Returns:
+            Generated text or None if failed
+        """
         payload = {
             "model": self.model,
             "prompt": prompt,
@@ -147,6 +169,10 @@ async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT) -> Optiona
             }
         }
 
+        # Add system prompt if provided (Ollama supports 'system' field)
+        if system_prompt:
+            payload["system"] = system_prompt
+
         client = await self._get_client()
         for attempt in range(MAX_TRANSLATION_ATTEMPTS):
             try:
@@ -254,21 +280,38 @@ async def get_model_context_size(self) -> int:
 
 class OpenAICompatibleProvider(LLMProvider):
     """OpenAI compatible API provider"""
-    
+
     def __init__(self, api_endpoint: str, model: str, api_key: Optional[str] = None):
         super().__init__(model)
         self.api_endpoint = api_endpoint
         self.api_key = api_key
-    
-    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT) -> Optional[str]:
-        """Generate text using an OpenAI compatible API"""
+
+    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT,
+                      system_prompt: Optional[str] = None) -> Optional[str]:
+        """
+        Generate text using an OpenAI compatible API.
+
+        Args:
+            prompt: The user prompt (content to translate)
+            timeout: Request timeout in seconds
+            system_prompt: Optional system prompt (role/instructions)
+
+        Returns:
+            Generated text or None if failed
+        """
         headers = {"Content-Type": "application/json"}
         if self.api_key:
             headers["Authorization"] = f"Bearer {self.api_key}"
 
+        # Build messages array with optional system prompt
+        messages = []
+        if system_prompt:
+            messages.append({"role": "system", "content": system_prompt})
+        messages.append({"role": "user", "content": prompt})
+
         payload = {
             "model": self.model,
-            "messages": [{"role": "user", "content": prompt}],
+            "messages": messages,
             "stream": False,
         }
         
@@ -372,15 +415,27 @@ async def get_available_models(self) -> list[dict]:
             print(f"Error fetching Gemini models: {e}")
             return []
     
-    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT) -> Optional[str]:
-        """Generate text using Gemini API"""
+    async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT,
+                      system_prompt: Optional[str] = None) -> Optional[str]:
+        """
+        Generate text using Gemini API.
+
+        Args:
+            prompt: The user prompt (content to translate)
+            timeout: Request timeout in seconds
+            system_prompt: Optional system prompt (role/instructions)
+
+        Returns:
+            Generated text or None if failed
+        """
         headers = {
             "Content-Type": "application/json",
             "x-goog-api-key": self.api_key
         }
-        
+
         payload = {
             "contents": [{
+                "role": "user",
                 "parts": [{
                     "text": prompt
                 }]
@@ -390,6 +445,14 @@ async def generate(self, prompt: str, timeout: int = REQUEST_TIMEOUT) -> Optiona
                 "maxOutputTokens": 2048
             }
         }
+
+        # Add system instruction if provided (Gemini API supports systemInstruction field)
+        if system_prompt:
+            payload["systemInstruction"] = {
+                "parts": [{
+                    "text": system_prompt
+                }]
+            }
         
         # Debug logs removed - uncomment if needed for troubleshooting
         # print(f"[DEBUG] Gemini API URL: {self.api_endpoint}")
diff --git a/src/core/subtitle_translator.py b/src/core/subtitle_translator.py
diff --git a/src/core/translator.py b/src/core/translator.py
