""" Mistral-Thinker needs this hack to properly initiate the thinking process. - The filter adds a think-tag by default. - It attempts to prevent repeated insertion, but this works with moderate success. - Since Mistral-Thinker doesn't respect system prompts very well, the filter converts the chat's system prompt into a user input. - The filter can also be used to influence Mistral's thinking process. To do this, change the valve think_guide to a sentence that guides the beginning of the thinking process. - Example valve setting for this: 'Okay, ich antworte auf deutsch.' to make Mistral-Thinker respond in german """ from pydantic import BaseModel, Field from typing import Optional, Dict, Any, List class Filter: prefix_added = False class Valves(BaseModel): prefix_text: str = Field( default="<think>", description="The think-tag (<think>) that is needed. Newline will be appended automatically", ) think_guide: str = Field( default="Okay, ich antworte auf deutsch.", description="A sentence that is added to the beginning of the thinking process. Like: Okay, ich antworte auf deutsch", ) pass def __init__(self): self.valves = self.Valves() self.prefix_added = False self.prefix_content = self.valves.prefix_text + "\n" + self.valves.think_guide pass def inlet(self, body: dict, __user__: Optional[dict] = None) -> dict: # Reset the flag for new conversations self.prefix_added = False # Check all existing messages if "messages" in body: for message in body["messages"]: if message.get("role") == "system": message["role"] = "user" if self.prefix_content in message.get("content", ""): self.prefix_added = True break # Add prefix if not found if not self.prefix_added: context_message = {"role": "assistant", "content": self.prefix_content} body.setdefault("messages", []).append(context_message) print(f"inlet called: {body}") # Debug Output. Disable if not wanted return body def outlet(self, body: dict, __user__: Optional[dict] = None) -> dict: # Handle non-streaming responses (if any) return body def stream(self, event: dict) -> dict: for choice in event.get("choices", []): delta = choice.get("delta", {}) if "content" in delta: if ( self.prefix_content not in delta["content"] and not self.prefix_added ): delta["content"] = self.prefix_content + delta["content"] self.prefix_added = True return event