Add model-aware system prompt transforms

README.md

@@ -200,6 +200,29 @@ groq-scout: "YOUR_GROQ_KEY"
 
 For OpenAI-compatible providers, a configured provider `api_key` takes precedence over the client's `Authorization` header. Client `Authorization` is only forwarded upstream when that provider is intentionally keyless (`api_key: null` / BYOK passthrough).
 
+### Model-aware system prompt transforms
+
+OpenAI-compatible chat-completion requests can apply an optional provider-wide or exact-model system prompt policy after model resolution and before dispatch:
+
+```yaml
+providers:
+  - name: "local-qwen"
+    type: "openai"
+    url: "http://192.168.2.216:8081"
+    system_prompt:
+      mode: append
+      content: |
+        Additional instructions for every request on this provider.
+    per_model_system_prompt:
+      "qwen38-flash-next":
+        mode: append
+        content: "Pay careful attention to dark and low-contrast objects."
+```
+
+Supported modes are `append`, `prepend`, and `replace`. An exact `per_model_system_prompt` entry takes precedence over the provider default; use `null` for a model-specific opt-out. Matching uses the resolved canonical model name, so aliases are not configuration keys. Multiple ordinary string system messages are combined in their original order with a blank line. Structured system content or system messages with extra semantic fields are left unchanged and logged as a warning. Router-owned `/no_think` markers are preserved, including when using `replace`.
+
+This v1 transform applies only to `/v1/chat/completions` text requests handled by the mediator. It does not rewrite Responses `instructions`/`input`, image requests, or audio-output/TTS requests. It is a request convenience transform, not an enforcement boundary for structured content.
+
 ## Features
 
 **Intelligent routing:**

pyproject.toml

@@ -100,6 +100,7 @@ exclude = [
 
 # New lint subsection to satisfy newer ruff configuration
 [tool.ruff.lint]
+select = ["E4", "E7", "E9", "F"]
 extend-ignore = ["E203", "E501", "B009"]
 
 # Per-file-ignores for ruff (syntax is slightly different)

smolrouter/interfaces.py

@@ -76,6 +76,55 @@ def user_id(self) -> Optional[str]:
         return None
 
 
+@dataclass(frozen=True)
+class SystemPromptConfig:
+    """A provider/model-specific transformation for chat system messages."""
+
+    mode: str
+    content: str
+
+    SUPPORTED_MODES = frozenset({"append", "prepend", "replace"})
+
+    def __post_init__(self):
+        if not isinstance(self.mode, str) or not self.mode.strip():
+            raise ValueError("System prompt mode must be one of: append, prepend, replace")
+
+        normalized_mode = self.mode.strip().lower()
+        if normalized_mode not in self.SUPPORTED_MODES:
+            raise ValueError(
+                f"Unsupported system prompt mode '{self.mode}'. "
+                f"Expected one of: {', '.join(sorted(self.SUPPORTED_MODES))}"
+            )
+        object.__setattr__(self, "mode", normalized_mode)
+
+        if not isinstance(self.content, str) or not self.content.strip():
+            raise ValueError("System prompt content must be a non-empty string")
+
+
+def coerce_system_prompt_config(value: Any) -> Optional[SystemPromptConfig]:
+    """Convert a YAML-shaped system prompt policy into its validated type."""
+    if value is None or isinstance(value, SystemPromptConfig):
+        return value
+    if isinstance(value, Mapping):
+        return SystemPromptConfig(**dict(value))
+    raise TypeError("System prompt configuration must be a mapping or null")
+
+
+def coerce_per_model_system_prompt(value: Any) -> Dict[str, Optional[SystemPromptConfig]]:
+    """Convert and validate exact model-name system prompt overrides."""
+    if value is None:
+        return {}
+    if not isinstance(value, Mapping):
+        raise TypeError("per_model_system_prompt must be a mapping")
+
+    converted = {}
+    for model_name, prompt_config in value.items():
+        if not isinstance(model_name, str) or not model_name.strip() or model_name != model_name.strip():
+            raise ValueError("per_model_system_prompt keys must be non-empty model names without surrounding whitespace")
+        converted[model_name] = coerce_system_prompt_config(prompt_config)
+    return converted
+
+
 class IModelProvider(ABC):
     """Abstraction for model discovery and health checking from providers"""
 
@@ -317,6 +366,13 @@ def coerce_provider_proxy_settings(provider_config: Mapping[str, Any]) -> Dict[s
     if isinstance(proxy_pool, list):
         config["proxy_pool"] = [None if entry is None else coerce_proxy_config(entry) for entry in proxy_pool]
 
+    if "system_prompt" in config:
+        config["system_prompt"] = coerce_system_prompt_config(config["system_prompt"])
+
+    per_model_system_prompt = config.get("per_model_system_prompt")
+    if per_model_system_prompt is not None:
+        config["per_model_system_prompt"] = coerce_per_model_system_prompt(per_model_system_prompt)
+
     return config
 
 
@@ -335,6 +391,8 @@ class ProviderConfig:
     static_models: Optional[List[str]] = None
     proxy_config: Optional[ProxyConfig] = None  # Default proxy for all models
     per_model_proxy: Optional[Dict[str, ProxyConfig]] = None  # Model-specific proxy overrides
+    system_prompt: Optional[SystemPromptConfig] = None  # Default system prompt policy
+    per_model_system_prompt: Optional[Dict[str, Optional[SystemPromptConfig]]] = None
 
     def __post_init__(self):
         if self.metadata is None:
@@ -343,6 +401,8 @@ def __post_init__(self):
             self.static_models = []
         if self.per_model_proxy is None:
             self.per_model_proxy = {}
+        self.system_prompt = coerce_system_prompt_config(self.system_prompt)
+        self.per_model_system_prompt = coerce_per_model_system_prompt(self.per_model_system_prompt)
 
     def get_proxy_for_model(self, model_name: str) -> Optional[ProxyConfig]:
         """Get proxy configuration for a specific model"""
@@ -353,6 +413,13 @@ def get_proxy_for_model(self, model_name: str) -> Optional[ProxyConfig]:
         # Fall back to default proxy
         return self.proxy_config
 
+    def get_system_prompt_for_model(self, model_name: str) -> Optional[SystemPromptConfig]:
+        """Return the exact model override, falling back to the provider default."""
+        per_model_system_prompt = self.per_model_system_prompt or {}
+        if model_name in per_model_system_prompt:
+            return per_model_system_prompt[model_name]
+        return self.system_prompt
+
 
 @dataclass
 class ModelResolution:

smolrouter/mediator.py

@@ -18,8 +18,10 @@
 from .dummy_provider import DummyProvider
 from .load_balancer import model_load_balancer
 from .request_metadata import RequestMetadata
+from .system_prompts import transform_system_prompt_messages
 
 logger = logging.getLogger(__name__)
+OPENAI_CHAT_COMPLETIONS_PATH = "/v1/chat/completions"
 
 
 def _should_use_openai_completion_tokens(model_name: Any) -> bool:
@@ -364,6 +366,52 @@ def _apply_resolved_model_to_payload(self, resolved_model: ModelInfo, request_pa
         if request_payload.get("model") != resolved_model.name:
             request_payload["model"] = resolved_model.name
 
+    @staticmethod
+    def _is_audio_output_request(request_payload: Dict[str, Any]) -> bool:
+        modalities = request_payload.get("modalities")
+        if isinstance(modalities, list) and any(str(value).strip().lower() == "audio" for value in modalities):
+            return True
+
+        response_format = request_payload.get("response_format")
+        if isinstance(response_format, dict) and str(response_format.get("type", "")).strip().lower() == "audio":
+            return True
+
+        return isinstance(request_payload.get("audio"), dict)
+
+    def _apply_system_prompt_policy(
+        self,
+        provider: IModelProvider,
+        resolved_model: ModelInfo,
+        request_payload: Dict[str, Any],
+        path: str,
+    ) -> None:
+        """Apply configured prompt policy to supported chat-completion requests."""
+        if path != OPENAI_CHAT_COMPLETIONS_PATH or self._is_audio_output_request(request_payload):
+            return
+
+        messages = request_payload.get("messages")
+        if not isinstance(messages, list):
+            return
+
+        provider_config = getattr(provider, "config", None)
+        get_policy = getattr(provider_config, "get_system_prompt_for_model", None)
+        if not callable(get_policy):
+            return
+
+        policy = get_policy(resolved_model.name)
+        if policy is None:
+            return
+
+        result = transform_system_prompt_messages(messages, policy)
+        request_payload["messages"] = result.messages
+        if result.skipped_reason:
+            logger.warning(
+                "Skipped system prompt transform for provider %s model %s: %s",
+                resolved_model.provider_id,
+                resolved_model.name,
+                result.skipped_reason,
+            )
+
     @staticmethod
     def _google_error_status_code(error_message: str) -> int:
         error_lower = error_message.lower()
@@ -493,6 +541,8 @@ async def _route_request_internal(
                 self._attach_lb_instance(lb_instance, None),
             )
 
+        self._apply_system_prompt_policy(provider, resolved_model, request_payload, path)
+
         if isinstance(provider, GoogleGenAIProvider):
             return await self._handle_google_provider_request(provider, resolved_model, request_payload, path, lb_instance)
 
@@ -551,6 +601,9 @@ async def route_request(
             Tuple of (response_data, status_code, upstream_used, metadata)
         """
         client = kwargs.get("client_context") or ClientContext(ip=source_ip, headers=headers)
+        request_payload = dict(request_payload)
+        if isinstance(request_payload.get("messages"), list):
+            request_payload["messages"] = list(request_payload["messages"])
         timeout = kwargs.get("timeout")
         if timeout is None and args:
             timeout = args[0]

smolrouter/providers.py

@@ -749,7 +749,13 @@ def create_providers_from_config(cls, providers_config: List[Dict[str, Any]]) ->
                 else:
                     logger.info(f"Skipping disabled provider: {config.name}")
             except Exception:
-                logger.exception("Failed to create provider from config: %s", provider_config)
+                provider_name = provider_config.get("name") if isinstance(provider_config, dict) else None
+                provider_type = provider_config.get("type") if isinstance(provider_config, dict) else None
+                logger.exception(
+                    "Failed to create provider from config: name=%r type=%r",
+                    provider_name,
+                    provider_type,
+                )
 
         # Sort providers by priority (lower numbers first)
         providers.sort(key=lambda p: p.config.priority)

smolrouter/system_prompts.py

@@ -0,0 +1,98 @@
+"""Pure helpers for provider-specific chat system prompt transformations."""
+
+from dataclasses import dataclass
+from typing import Any, List, Optional
+
+from .interfaces import SystemPromptConfig
+
+
+_ROUTER_SYSTEM_MARKERS = frozenset({"/no_think"})
+_SYSTEM_MESSAGE_KEYS = frozenset({"role", "content"})
+
+
+@dataclass(frozen=True)
+class SystemPromptTransformResult:
+    """Result of applying a system prompt policy to a messages list."""
+
+    messages: List[Any]
+    applied: bool = False
+    skipped_reason: Optional[str] = None
+
+
+def _copy_messages(messages: List[Any]) -> List[Any]:
+    return [dict(message) if isinstance(message, dict) else message for message in messages]
+
+
+def _classify_system_messages(messages: List[Any]):
+    logical_indices = []
+    logical_contents = []
+
+    for index, message in enumerate(messages):
+        if not isinstance(message, dict):
+            continue
+        if str(message.get("role", "")).strip().lower() != "system":
+            continue
+
+        content = message.get("content")
+        if not isinstance(content, str):
+            return [], [], "system message content is not a string"
+        if set(message) - _SYSTEM_MESSAGE_KEYS:
+            return [], [], "system message contains unsupported semantic fields"
+        if content not in _ROUTER_SYSTEM_MARKERS:
+            logical_indices.append(index)
+            logical_contents.append(content)
+
+    return logical_indices, logical_contents, None
+
+
+def _compose_system_prompt(logical_content: str, policy: SystemPromptConfig) -> str:
+    if not logical_content or policy.mode == "replace":
+        return policy.content
+    if policy.mode == "prepend":
+        return f"{policy.content}\n\n{logical_content}"
+    return f"{logical_content}\n\n{policy.content}"
+
+
+def _rebuild_messages(messages: List[Any], transformed_message: dict, first_system_index: Optional[int]) -> List[Any]:
+    result: List[Any] = []
+
+    for index, message in enumerate(messages):
+        if first_system_index == index:
+            result.append(transformed_message)
+        if isinstance(message, dict) and str(message.get("role", "")).strip().lower() == "system":
+            if message.get("content") not in _ROUTER_SYSTEM_MARKERS:
+                continue
+        result.append(dict(message) if isinstance(message, dict) else message)
+
+    if first_system_index is None:
+        result.insert(0, transformed_message)
+    return result
+
+
+def transform_system_prompt_messages(
+    messages: Any, policy: SystemPromptConfig
+) -> SystemPromptTransformResult:
+    """Apply a policy without mutating the input messages or their order.
+
+    System messages with structured content or extra semantic fields are left
+    untouched because flattening them could silently discard meaning. Router-
+    owned marker messages such as ``/no_think`` are preserved separately from
+    the logical application system prompt, including for ``replace``.
+    """
+    if not isinstance(messages, list):
+        return SystemPromptTransformResult(
+            messages=messages,
+            skipped_reason="messages is not a list",
+        )
+
+    copied_messages = _copy_messages(messages)
+    logical_indices, logical_contents, skipped_reason = _classify_system_messages(messages)
+    if skipped_reason:
+        return SystemPromptTransformResult(messages=copied_messages, skipped_reason=skipped_reason)
+
+    logical_content = "\n\n".join(logical_contents)
+    transformed_content = _compose_system_prompt(logical_content, policy)
+    first_system_index = logical_indices[0] if logical_indices else None
+    transformed_message = {"role": "system", "content": transformed_content}
+    result = _rebuild_messages(messages, transformed_message, first_system_index)
+    return SystemPromptTransformResult(messages=result, applied=True)