PlanOpticon

planopticon / video_processor / providers / cohere_provider.py
Source Blame History 123 lines
0981a08… noreply 1 """Cohere provider implementation."""
0981a08… noreply 2
0981a08… noreply 3 import logging
0981a08… noreply 4 import os
0981a08… noreply 5 from pathlib import Path
0981a08… noreply 6 from typing import Optional
0981a08… noreply 7
0981a08… noreply 8 from dotenv import load_dotenv
0981a08… noreply 9
0981a08… noreply 10 from video_processor.providers.base import BaseProvider, ModelInfo, ProviderRegistry
0981a08… noreply 11
0981a08… noreply 12 load_dotenv()
0981a08… noreply 13 logger = logging.getLogger(__name__)
0981a08… noreply 14
0981a08… noreply 15 # Curated list of Cohere models
0981a08… noreply 16 _COHERE_MODELS = [
0981a08… noreply 17 ModelInfo(
0981a08… noreply 18 id="command-r-plus",
0981a08… noreply 19 provider="cohere",
0981a08… noreply 20 display_name="Command R+",
0981a08… noreply 21 capabilities=["chat"],
0981a08… noreply 22 ),
0981a08… noreply 23 ModelInfo(
0981a08… noreply 24 id="command-r",
0981a08… noreply 25 provider="cohere",
0981a08… noreply 26 display_name="Command R",
0981a08… noreply 27 capabilities=["chat"],
0981a08… noreply 28 ),
0981a08… noreply 29 ModelInfo(
0981a08… noreply 30 id="command-light",
0981a08… noreply 31 provider="cohere",
0981a08… noreply 32 display_name="Command Light",
0981a08… noreply 33 capabilities=["chat"],
0981a08… noreply 34 ),
0981a08… noreply 35 ModelInfo(
0981a08… noreply 36 id="command-nightly",
0981a08… noreply 37 provider="cohere",
0981a08… noreply 38 display_name="Command Nightly",
0981a08… noreply 39 capabilities=["chat"],
0981a08… noreply 40 ),
0981a08… noreply 41 ]
0981a08… noreply 42
0981a08… noreply 43
0981a08… noreply 44 class CohereProvider(BaseProvider):
0981a08… noreply 45 """Cohere provider using the cohere SDK."""
0981a08… noreply 46
0981a08… noreply 47 provider_name = "cohere"
0981a08… noreply 48
0981a08… noreply 49 def __init__(self, api_key: Optional[str] = None):
0981a08… noreply 50 try:
0981a08… noreply 51 import cohere
0981a08… noreply 52 except ImportError:
0981a08… noreply 53 raise ImportError("cohere package not installed. Install with: pip install cohere")
0981a08… noreply 54
0981a08… noreply 55 self._api_key = api_key or os.getenv("COHERE_API_KEY")
0981a08… noreply 56 if not self._api_key:
0981a08… noreply 57 raise ValueError("COHERE_API_KEY not set")
0981a08… noreply 58
0981a08… noreply 59 self._client = cohere.ClientV2(api_key=self._api_key)
0981a08… noreply 60 self._last_usage = {}
0981a08… noreply 61
0981a08… noreply 62 def chat(
0981a08… noreply 63 self,
0981a08… noreply 64 messages: list[dict],
0981a08… noreply 65 max_tokens: int = 4096,
0981a08… noreply 66 temperature: float = 0.7,
0981a08… noreply 67 model: Optional[str] = None,
0981a08… noreply 68 ) -> str:
0981a08… noreply 69 model = model or "command-r-plus"
0981a08… noreply 70
0981a08… noreply 71 response = self._client.chat(
0981a08… noreply 72 model=model,
0981a08… noreply 73 messages=messages,
0981a08… noreply 74 max_tokens=max_tokens,
0981a08… noreply 75 temperature=temperature,
0981a08… noreply 76 )
0981a08… noreply 77
0981a08… noreply 78 usage = getattr(response, "usage", None)
0981a08… noreply 79 tokens = getattr(usage, "tokens", None) if usage else None
0981a08… noreply 80 self._last_usage = {
0981a08… noreply 81 "input_tokens": getattr(tokens, "input_tokens", 0) if tokens else 0,
0981a08… noreply 82 "output_tokens": getattr(tokens, "output_tokens", 0) if tokens else 0,
0981a08… noreply 83 }
0981a08… noreply 84 return response.message.content[0].text if response.message.content else ""
0981a08… noreply 85
0981a08… noreply 86 def analyze_image(
0981a08… noreply 87 self,
0981a08… noreply 88 image_bytes: bytes,
0981a08… noreply 89 prompt: str,
0981a08… noreply 90 max_tokens: int = 4096,
0981a08… noreply 91 model: Optional[str] = None,
0981a08… noreply 92 ) -> str:
0981a08… noreply 93 raise NotImplementedError(
0981a08… noreply 94 "Cohere does not currently support vision/image analysis. "
0981a08… noreply 95 "Use OpenAI, Anthropic, or Gemini for image analysis."
0981a08… noreply 96 )
0981a08… noreply 97
0981a08… noreply 98 def transcribe_audio(
0981a08… noreply 99 self,
0981a08… noreply 100 audio_path: str | Path,
0981a08… noreply 101 language: Optional[str] = None,
0981a08… noreply 102 model: Optional[str] = None,
0981a08… noreply 103 ) -> dict:
0981a08… noreply 104 raise NotImplementedError(
0981a08… noreply 105 "Cohere does not provide a transcription API. "
0981a08… noreply 106 "Use OpenAI Whisper or Gemini for transcription."
0981a08… noreply 107 )
0981a08… noreply 108
0981a08… noreply 109 def list_models(self) -> list[ModelInfo]:
0981a08… noreply 110 return list(_COHERE_MODELS)
0981a08… noreply 111
0981a08… noreply 112
0981a08… noreply 113 ProviderRegistry.register(
0981a08… noreply 114 name="cohere",
0981a08… noreply 115 provider_class=CohereProvider,
0981a08… noreply 116 env_var="COHERE_API_KEY",
0981a08… noreply 117 model_prefixes=["command-"],
0981a08… noreply 118 default_models={
0981a08… noreply 119 "chat": "command-r-plus",
0981a08… noreply 120 "vision": "",
0981a08… noreply 121 "audio": "",
0981a08… noreply 122 },
0981a08… noreply 123 )

Keyboard Shortcuts

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