PlanOpticon

planopticon / video_processor / providers / mistral_provider.py
Source Blame History 167 lines
0981a08… noreply 1 """Mistral AI provider implementation."""
0981a08… noreply 2
0981a08… noreply 3 import base64
0981a08… noreply 4 import logging
0981a08… noreply 5 import os
0981a08… noreply 6 from pathlib import Path
0981a08… noreply 7 from typing import Optional
0981a08… noreply 8
0981a08… noreply 9 from dotenv import load_dotenv
0981a08… noreply 10
0981a08… noreply 11 from video_processor.providers.base import BaseProvider, ModelInfo, ProviderRegistry
0981a08… noreply 12
0981a08… noreply 13 load_dotenv()
0981a08… noreply 14 logger = logging.getLogger(__name__)
0981a08… noreply 15
0981a08… noreply 16 # Curated list of Mistral models
0981a08… noreply 17 _MISTRAL_MODELS = [
0981a08… noreply 18 ModelInfo(
0981a08… noreply 19 id="mistral-large-latest",
0981a08… noreply 20 provider="mistral",
0981a08… noreply 21 display_name="Mistral Large",
0981a08… noreply 22 capabilities=["chat"],
0981a08… noreply 23 ),
0981a08… noreply 24 ModelInfo(
0981a08… noreply 25 id="mistral-medium-latest",
0981a08… noreply 26 provider="mistral",
0981a08… noreply 27 display_name="Mistral Medium",
0981a08… noreply 28 capabilities=["chat"],
0981a08… noreply 29 ),
0981a08… noreply 30 ModelInfo(
0981a08… noreply 31 id="mistral-small-latest",
0981a08… noreply 32 provider="mistral",
0981a08… noreply 33 display_name="Mistral Small",
0981a08… noreply 34 capabilities=["chat"],
0981a08… noreply 35 ),
0981a08… noreply 36 ModelInfo(
0981a08… noreply 37 id="open-mistral-nemo",
0981a08… noreply 38 provider="mistral",
0981a08… noreply 39 display_name="Mistral Nemo",
0981a08… noreply 40 capabilities=["chat"],
0981a08… noreply 41 ),
0981a08… noreply 42 ModelInfo(
0981a08… noreply 43 id="pixtral-large-latest",
0981a08… noreply 44 provider="mistral",
0981a08… noreply 45 display_name="Pixtral Large",
0981a08… noreply 46 capabilities=["chat", "vision"],
0981a08… noreply 47 ),
0981a08… noreply 48 ModelInfo(
0981a08… noreply 49 id="pixtral-12b-2409",
0981a08… noreply 50 provider="mistral",
0981a08… noreply 51 display_name="Pixtral 12B",
0981a08… noreply 52 capabilities=["chat", "vision"],
0981a08… noreply 53 ),
0981a08… noreply 54 ModelInfo(
0981a08… noreply 55 id="codestral-latest",
0981a08… noreply 56 provider="mistral",
0981a08… noreply 57 display_name="Codestral",
0981a08… noreply 58 capabilities=["chat"],
0981a08… noreply 59 ),
0981a08… noreply 60 ]
0981a08… noreply 61
0981a08… noreply 62
0981a08… noreply 63 class MistralProvider(BaseProvider):
0981a08… noreply 64 """Mistral AI provider using the mistralai SDK."""
0981a08… noreply 65
0981a08… noreply 66 provider_name = "mistral"
0981a08… noreply 67
0981a08… noreply 68 def __init__(self, api_key: Optional[str] = None):
0981a08… noreply 69 try:
0981a08… noreply 70 from mistralai import Mistral
0981a08… noreply 71 except ImportError:
0981a08… noreply 72 raise ImportError(
0981a08… noreply 73 "mistralai package not installed. Install with: pip install mistralai"
0981a08… noreply 74 )
0981a08… noreply 75
0981a08… noreply 76 self._api_key = api_key or os.getenv("MISTRAL_API_KEY")
0981a08… noreply 77 if not self._api_key:
0981a08… noreply 78 raise ValueError("MISTRAL_API_KEY not set")
0981a08… noreply 79
0981a08… noreply 80 self._client = Mistral(api_key=self._api_key)
0981a08… noreply 81 self._last_usage = {}
0981a08… noreply 82
0981a08… noreply 83 def chat(
0981a08… noreply 84 self,
0981a08… noreply 85 messages: list[dict],
0981a08… noreply 86 max_tokens: int = 4096,
0981a08… noreply 87 temperature: float = 0.7,
0981a08… noreply 88 model: Optional[str] = None,
0981a08… noreply 89 ) -> str:
0981a08… noreply 90 model = model or "mistral-large-latest"
0981a08… noreply 91
0981a08… noreply 92 response = self._client.chat.complete(
0981a08… noreply 93 model=model,
0981a08… noreply 94 messages=messages,
0981a08… noreply 95 max_tokens=max_tokens,
0981a08… noreply 96 temperature=temperature,
0981a08… noreply 97 )
0981a08… noreply 98
0981a08… noreply 99 self._last_usage = {
0981a08… noreply 100 "input_tokens": getattr(response.usage, "prompt_tokens", 0) if response.usage else 0,
0981a08… noreply 101 "output_tokens": getattr(response.usage, "completion_tokens", 0)
0981a08… noreply 102 if response.usage
0981a08… noreply 103 else 0,
0981a08… noreply 104 }
0981a08… noreply 105 return response.choices[0].message.content or ""
0981a08… noreply 106
0981a08… noreply 107 def analyze_image(
0981a08… noreply 108 self,
0981a08… noreply 109 image_bytes: bytes,
0981a08… noreply 110 prompt: str,
0981a08… noreply 111 max_tokens: int = 4096,
0981a08… noreply 112 model: Optional[str] = None,
0981a08… noreply 113 ) -> str:
0981a08… noreply 114 model = model or "pixtral-large-latest"
0981a08… noreply 115 b64 = base64.b64encode(image_bytes).decode()
0981a08… noreply 116
0981a08… noreply 117 response = self._client.chat.complete(
0981a08… noreply 118 model=model,
0981a08… noreply 119 messages=[
0981a08… noreply 120 {
0981a08… noreply 121 "role": "user",
0981a08… noreply 122 "content": [
0981a08… noreply 123 {"type": "text", "text": prompt},
0981a08… noreply 124 {
0981a08… noreply 125 "type": "image_url",
0981a08… noreply 126 "image_url": {"url": f"data:image/jpeg;base64,{b64}"},
0981a08… noreply 127 },
0981a08… noreply 128 ],
0981a08… noreply 129 }
0981a08… noreply 130 ],
0981a08… noreply 131 max_tokens=max_tokens,
0981a08… noreply 132 )
0981a08… noreply 133
0981a08… noreply 134 self._last_usage = {
0981a08… noreply 135 "input_tokens": getattr(response.usage, "prompt_tokens", 0) if response.usage else 0,
0981a08… noreply 136 "output_tokens": getattr(response.usage, "completion_tokens", 0)
0981a08… noreply 137 if response.usage
0981a08… noreply 138 else 0,
0981a08… noreply 139 }
0981a08… noreply 140 return response.choices[0].message.content or ""
0981a08… noreply 141
0981a08… noreply 142 def transcribe_audio(
0981a08… noreply 143 self,
0981a08… noreply 144 audio_path: str | Path,
0981a08… noreply 145 language: Optional[str] = None,
0981a08… noreply 146 model: Optional[str] = None,
0981a08… noreply 147 ) -> dict:
0981a08… noreply 148 raise NotImplementedError(
0981a08… noreply 149 "Mistral does not provide a transcription API. "
0981a08… noreply 150 "Use OpenAI Whisper or Gemini for transcription."
0981a08… noreply 151 )
0981a08… noreply 152
0981a08… noreply 153 def list_models(self) -> list[ModelInfo]:
0981a08… noreply 154 return list(_MISTRAL_MODELS)
0981a08… noreply 155
0981a08… noreply 156
0981a08… noreply 157 ProviderRegistry.register(
0981a08… noreply 158 name="mistral",
0981a08… noreply 159 provider_class=MistralProvider,
0981a08… noreply 160 env_var="MISTRAL_API_KEY",
0981a08… noreply 161 model_prefixes=["mistral-", "pixtral-", "codestral-", "open-mistral-"],
0981a08… noreply 162 default_models={
0981a08… noreply 163 "chat": "mistral-large-latest",
0981a08… noreply 164 "vision": "pixtral-large-latest",
0981a08… noreply 165 "audio": "",
0981a08… noreply 166 },
0981a08… noreply 167 )

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button