feat(provider): add v1m (System One) calibrated decision engine

litellm/llms/v1m.py

@@ -0,0 +1,92 @@
+from typing import Any, Dict, List, Optional, Union
+import json
+import time
+
+from litellm.llms.base_llm.chat.transformation import BaseLLMException
+from litellm.llms.openai.chat.gpt_transformation import OpenAIGPTConfig
+
+
+class V1MError(BaseLLMException):
+    def __init__(self, status_code: int, message: str, headers: Any = None):
+        super().__init__(status_code=status_code, message=message, headers=headers)
+
+
+class V1MConfig(OpenAIGPTConfig):
+    DEFAULT_BASE_URL: str = "https://v1m.ir/v1/systemone"
+
+    def __init__(
+        self,
+        api_key: Optional[str] = None,
+        base_url: Optional[str] = None,
+        timeout: float = 15.0,
+    ) -> None:
+        self.api_key = api_key
+        self.base_url = (base_url or self.DEFAULT_BASE_URL).rstrip("/")
+        if not self.base_url.endswith("/systemone"):
+            self.base_url = f"{self.base_url}/systemone"
+        self.timeout = timeout
+
+    def transform_request(
+        self,
+        model: str,
+        messages: Optional[List[Dict[str, Any]]] = None,
+        state: Optional[Union[str, Dict[str, Any]]] = None,
+        questions: Optional[Dict[str, Any]] = None,
+        **kwargs: Any,
+    ) -> Dict[str, Any]:
+        clean_model = model.replace("v1m/", "") if model.startswith("v1m/") else model
+
+        if state is not None and questions is not None:
+            return {
+                "model": clean_model or "rev-latest",
+                "state": state,
+                "questions": questions,
+            }
+
+        extracted_state = ""
+        user_prompt = ""
+        if messages:
+            for msg in messages:
+                role = msg.get("role", "")
+                content = msg.get("content", "")
+                if role == "system":
+                    extracted_state = content
+                elif role == "user":
+                    user_prompt = content
+
+        return {
+            "model": clean_model or "rev-latest",
+            "state": extracted_state or user_prompt or "General State Assessment",
+            "questions": {
+                "decision": {
+                    "prompt": user_prompt or "Evaluate state validity.",
+                    "instructions": "Determine probabilistic confidence and classification.",
+                    "type": "choice",
+                    "criteria": {
+                        "affirmative": "Conditions met",
+                        "negative": "Conditions not met",
+                    },
+                }
+            },
+        }
+
+    def transform_response(self, raw_resp: Dict[str, Any], model: str) -> Dict[str, Any]:
+        evals = raw_resp.get("evaluations", raw_resp)
+        return {
+            "id": f"v1m-{int(time.time())}",
+            "object": "chat.completion",
+            "created": int(time.time()),
+            "model": model,
+            "choices": [
+                {
+                    "index": 0,
+                    "message": {
+                        "role": "assistant",
+                        "content": json.dumps(evals, ensure_ascii=False),
+                    },
+                    "finish_reason": "stop",
+                }
+            ],
+            "usage": raw_resp.get("usage", {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}),
+            "systemone_evaluations": evals,
+        }