scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-10 22:41:41 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 82

added qualifire prompt management

Browse source
This commit is contained in:
Dror Ivry 2026-02-08 12:05:05 +02:00
parent 4f96a3b126
commit 28fd51d3f6
No known key found for this signature in database
4 changed files with 457 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
litellm/integrations/qualifire/__init__.py Normal file
View file

@ -0,0 +1,42 @@
import os
from typing import TYPE_CHECKING, Optional
if TYPE_CHECKING:
from litellm.integrations.custom_prompt_management import CustomPromptManagement
from litellm.types.prompts.init_prompts import PromptLiteLLMParams, PromptSpec
from litellm.types.prompts.init_prompts import SupportedPromptIntegrations
from .qualifire_prompt_manager import QualifirePromptManager
def prompt_initializer(
litellm_params: "PromptLiteLLMParams", prompt_spec: "PromptSpec"
) -> "CustomPromptManagement":
"""
Initialize a prompt from Qualifire.
"""
api_key = getattr(litellm_params, "api_key", None) or os.environ.get(
"QUALIFIRE_API_KEY"
)
api_base = getattr(litellm_params, "api_base", None) or "https://api.qualifire.ai"
if not api_key:
raise ValueError(
"api_key is required for Qualifire prompt integration. "
"Set it in litellm_params or via QUALIFIRE_API_KEY environment variable."
)
try:
qualifire_prompt_manager = QualifirePromptManager(
api_key=api_key,
api_base=api_base,
)
return qualifire_prompt_manager
except Exception as e:
raise e
prompt_initializer_registry = {
SupportedPromptIntegrations.QUALIFIRE.value: prompt_initializer,
}

153
litellm/integrations/qualifire/qualifire_client.py Normal file
View file

@ -0,0 +1,153 @@
"""
HTTP client for the Qualifire Studio API.
Handles sync and async calls to the /compile endpoint.
"""
from typing import Any, Dict, List, Optional
import httpx
from litellm.llms.custom_httpx.http_handler import (
_get_httpx_client,
get_async_httpx_client,
)
from litellm.types.llms.custom_http import httpxSpecialProvider
class QualifireClient:
"""
Low-level HTTP client for the Qualifire API.
Uses the /api/v1/studio/prompts/{promptId}/compile endpoint
to compile prompts with variable substitution server-side.
"""
def __init__(
self,
api_key: str,
api_base: str = "https://api.qualifire.ai",
):
self.api_key = api_key
self.api_base = api_base.rstrip("/")
def _get_headers(self) -> Dict[str, str]:
"""Get HTTP headers for API requests."""
return {
"Content-Type": "application/json",
"Accept": "application/json",
"X-Qualifire-API-Key": self.api_key,
}
def _build_compile_url(self, prompt_id: str) -> str:
"""Build the compile endpoint URL for a given prompt ID."""
return f"{self.api_base}/api/v1/studio/prompts/{prompt_id}/compile"
def _build_request_body(
self,
variables: Optional[Dict[str, Any]] = None,
revision: Optional[str] = None,
) -> Dict[str, Any]:
"""Build the request body for the compile endpoint."""
body: Dict[str, Any] = {}
if variables:
body["variables"] = variables
if revision:
body["revision"] = revision
return body
def _handle_error_response(self, response: httpx.Response, prompt_id: str) -> None:
"""Handle HTTP error responses with specific messages."""
if response.status_code == 401:
raise Exception(
f"Authentication failed for Qualifire API. "
f"Please check your API key."
)
elif response.status_code == 403:
raise Exception(
f"Access denied to prompt '{prompt_id}'. "
f"Please check your permissions."
)
elif response.status_code == 404:
raise Exception(
f"Prompt '{prompt_id}' not found in Qualifire. "
f"Please check the prompt ID."
)
response.raise_for_status()
def compile_prompt(
self,
prompt_id: str,
variables: Optional[Dict[str, Any]] = None,
revision: Optional[str] = None,
) -> Dict[str, Any]:
"""
Compile a prompt by calling the Qualifire API synchronously.
Args:
prompt_id: The Qualifire prompt CUID
variables: Variables for template substitution
revision: Optional revision CUID to pin a specific version
Returns:
The compiled prompt response from Qualifire
"""
url = self._build_compile_url(prompt_id)
body = self._build_request_body(variables, revision)
http_client = _get_httpx_client()
try:
response = http_client.post(
url,
json=body,
headers=self._get_headers(),
)
if response.status_code >= 400:
self._handle_error_response(response, prompt_id)
return response.json()
except httpx.HTTPError as e:
raise Exception(
f"Failed to compile prompt '{prompt_id}' from Qualifire: {e}"
)
async def async_compile_prompt(
self,
prompt_id: str,
variables: Optional[Dict[str, Any]] = None,
revision: Optional[str] = None,
) -> Dict[str, Any]:
"""
Compile a prompt by calling the Qualifire API asynchronously.
Args:
prompt_id: The Qualifire prompt CUID
variables: Variables for template substitution
revision: Optional revision CUID to pin a specific version
Returns:
The compiled prompt response from Qualifire
"""
url = self._build_compile_url(prompt_id)
body = self._build_request_body(variables, revision)
http_client = get_async_httpx_client(
llm_provider=httpxSpecialProvider.PromptManagement,
)
try:
response = await http_client.post(
url,
json=body,
headers=self._get_headers(),
)
if response.status_code >= 400:
self._handle_error_response(response, prompt_id)
return response.json()
except httpx.HTTPError as e:
raise Exception(
f"Failed to compile prompt '{prompt_id}' from Qualifire: {e}"
)

261
litellm/integrations/qualifire/qualifire_prompt_manager.py Normal file
View file

@ -0,0 +1,261 @@
"""
Qualifire prompt manager that integrates with LiteLLM's prompt management system.
Fetches compiled prompts from Qualifire Studio's /compile endpoint.
"""
from typing import Any, Dict, List, Optional, Tuple
from litellm.integrations.custom_prompt_management import CustomPromptManagement
from litellm.integrations.prompt_management_base import (
PromptManagementBase,
PromptManagementClient,
)
from litellm.types.llms.openai import AllMessageValues
from litellm.types.prompts.init_prompts import PromptSpec
from litellm.types.utils import StandardCallbackDynamicParams
from .qualifire_client import QualifireClient
# Parameters to extract from the Qualifire response
SUPPORTED_PARAMETERS = [
"temperature",
"top_p",
"max_tokens",
"frequency_penalty",
"presence_penalty",
"reasoning_effort",
]
class QualifirePromptManager(CustomPromptManagement):
"""
Qualifire prompt manager that integrates with LiteLLM's prompt management system.
Uses Qualifire's /compile endpoint which handles variable substitution server-side,
so no client-side templating is needed. Variables are POSTed to the API and compiled
messages, tools, and parameters are returned.
"""
def __init__(
self,
api_key: str,
api_base: str = "https://api.qualifire.ai",
**kwargs,
):
super().__init__(**kwargs)
self.api_key = api_key
self.api_base = api_base
self.client = QualifireClient(api_key=api_key, api_base=api_base)
@property
def integration_name(self) -> str:
return "qualifire"
def should_run_prompt_management(
self,
prompt_id: Optional[str],
prompt_spec: Optional[PromptSpec],
dynamic_callback_params: StandardCallbackDynamicParams,
) -> bool:
return True
@staticmethod
def _extract_revision(prompt_spec: Optional[PromptSpec]) -> Optional[str]:
"""Extract revision from prompt_spec's provider_specific_query_params."""
if (
prompt_spec
and prompt_spec.litellm_params.provider_specific_query_params
):
return prompt_spec.litellm_params.provider_specific_query_params.get(
"revision"
)
return None
@staticmethod
def _parse_compile_response(
prompt_id: Optional[str],
response: Dict[str, Any],
) -> PromptManagementClient:
"""
Parse the Qualifire compile response into a PromptManagementClient.
Qualifire compile response format:
{
"messages": [...],
"parameters": {
"model": "gpt-4",
"temperature": 0.7,
...
},
"tools": [...] # optional
}
"""
messages = response.get("messages", [])
parameters = response.get("parameters", {})
tools = response.get("tools")
# Extract model from parameters
model = parameters.get("model")
# Extract optional params, filtering out None values
optional_params: Dict[str, Any] = {}
for param in SUPPORTED_PARAMETERS:
value = parameters.get(param)
if value is not None:
optional_params[param] = value
# Include tools if present and non-empty
if tools:
optional_params["tools"] = tools
return PromptManagementClient(
prompt_id=prompt_id,
prompt_template=messages,
prompt_template_model=model,
prompt_template_optional_params=optional_params if optional_params else None,
completed_messages=None,
)
def _compile_prompt_helper(
self,
prompt_id: Optional[str],
prompt_spec: Optional[PromptSpec],
prompt_variables: Optional[dict],
dynamic_callback_params: StandardCallbackDynamicParams,
prompt_label: Optional[str] = None,
prompt_version: Optional[int] = None,
) -> PromptManagementClient:
"""
Compile a prompt using the Qualifire /compile endpoint (sync).
"""
if prompt_id is None:
raise ValueError("prompt_id is required for Qualifire prompt manager")
revision = self._extract_revision(prompt_spec)
try:
response = self.client.compile_prompt(
prompt_id=prompt_id,
variables=prompt_variables,
revision=revision,
)
return self._parse_compile_response(prompt_id, response)
except Exception as e:
raise ValueError(f"Error compiling prompt '{prompt_id}' from Qualifire: {e}")
async def async_compile_prompt_helper(
self,
prompt_id: Optional[str],
prompt_variables: Optional[dict],
dynamic_callback_params: StandardCallbackDynamicParams,
prompt_spec: Optional[PromptSpec] = None,
prompt_label: Optional[str] = None,
prompt_version: Optional[int] = None,
) -> PromptManagementClient:
"""
Compile a prompt using the Qualifire /compile endpoint (async).
"""
if prompt_id is None:
raise ValueError("prompt_id is required for Qualifire prompt manager")
revision = self._extract_revision(prompt_spec)
try:
response = await self.client.async_compile_prompt(
prompt_id=prompt_id,
variables=prompt_variables,
revision=revision,
)
return self._parse_compile_response(prompt_id, response)
except Exception as e:
raise ValueError(f"Error compiling prompt '{prompt_id}' from Qualifire: {e}")
def get_chat_completion_prompt(
self,
model: str,
messages: List[AllMessageValues],
non_default_params: dict,
prompt_id: Optional[str],
prompt_variables: Optional[dict],
dynamic_callback_params: StandardCallbackDynamicParams,
prompt_spec: Optional[PromptSpec] = None,
prompt_label: Optional[str] = None,
prompt_version: Optional[int] = None,
ignore_prompt_manager_model: Optional[bool] = False,
ignore_prompt_manager_optional_params: Optional[bool] = False,
) -> Tuple[str, List[AllMessageValues], dict]:
return PromptManagementBase.get_chat_completion_prompt(
self,
model,
messages,
non_default_params,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
dynamic_callback_params=dynamic_callback_params,
prompt_spec=prompt_spec,
prompt_label=prompt_label,
prompt_version=prompt_version,
ignore_prompt_manager_model=(
ignore_prompt_manager_model
or (
prompt_spec.litellm_params.ignore_prompt_manager_model
if prompt_spec
else False
)
),
ignore_prompt_manager_optional_params=(
ignore_prompt_manager_optional_params
or (
prompt_spec.litellm_params.ignore_prompt_manager_optional_params
if prompt_spec
else False
)
),
)
async def async_get_chat_completion_prompt(
self,
model: str,
messages: List[AllMessageValues],
non_default_params: dict,
prompt_id: Optional[str],
prompt_variables: Optional[dict],
dynamic_callback_params: StandardCallbackDynamicParams,
litellm_logging_obj: Any = None,
prompt_spec: Optional[PromptSpec] = None,
tools: Optional[List[Dict]] = None,
prompt_label: Optional[str] = None,
prompt_version: Optional[int] = None,
ignore_prompt_manager_model: Optional[bool] = False,
ignore_prompt_manager_optional_params: Optional[bool] = False,
) -> Tuple[str, List[AllMessageValues], dict]:
return await PromptManagementBase.async_get_chat_completion_prompt(
self,
model,
messages,
non_default_params,
prompt_id=prompt_id,
prompt_variables=prompt_variables,
litellm_logging_obj=litellm_logging_obj,
dynamic_callback_params=dynamic_callback_params,
prompt_spec=prompt_spec,
tools=tools,
prompt_label=prompt_label,
prompt_version=prompt_version,
ignore_prompt_manager_model=(
ignore_prompt_manager_model
or (
prompt_spec.litellm_params.ignore_prompt_manager_model
if prompt_spec
else False
)
),
ignore_prompt_manager_optional_params=(
ignore_prompt_manager_optional_params
or (
prompt_spec.litellm_params.ignore_prompt_manager_optional_params
if prompt_spec
else False
)
),
)

1
litellm/types/prompts/init_prompts.py
View file

@ -13,6 +13,7 @@ class SupportedPromptIntegrations(str, Enum):
GITLAB = "gitlab"
GENERIC_PROMPT_MANAGEMENT = "generic_prompt_management"
ARIZE_PHOENIX = "arize_phoenix"
QUALIFIRE = "qualifire"
class PromptInfo(BaseModel):