""" title: Groq description: https://console.groq.com author: Olof Larsson author_url: https://olof.tech/groq-for-open-webui version: 1.8.0 license: MIT """ import os import requests from typing import List, Union, Generator, Iterator, Optional from pydantic import BaseModel, Field # Learn more here: # https://olof.tech/groq-for-open-webui class Pipe: class Valves(BaseModel): API_KEY: str = Field( default=os.getenv("GROQ_API_KEY", ""), description="Get your key here: https://console.groq.com/keys", ) def __init__(self): self.type = "manifold" self.id = "groq" self.name = "groq/" self.valves = self.Valves() self.base_url = "https://api.groq.com/openai/v1" # Q: Why hard-code the model ids? # A: Because it yields the best performance. OpenWebUI is very sensitive to slowness here. It impacts page loads. def get_model_ids(self): # How to update: # curl "https://api.groq.com/openai/v1/models" -H "Authorization: Bearer ${GROQ_API_KEY}" | jq '.data | sort_by(.id) | map(.id)' models = [ "allam-2-7b", "deepseek-r1-distill-llama-70b", "deepseek-r1-distill-qwen-32b", "distil-whisper-large-v3-en", "gemma2-9b-it", "llama-3.1-8b-instant", "llama-3.2-11b-vision-preview", "llama-3.2-1b-preview", "llama-3.2-3b-preview", "llama-3.2-90b-vision-preview", "llama-3.3-70b-specdec", "llama-3.3-70b-versatile", "llama-guard-3-8b", "llama3-70b-8192", "llama3-8b-8192", "meta-llama/llama-4-maverick-17b-128e-instruct", "meta-llama/llama-4-scout-17b-16e-instruct", "mistral-saba-24b", "playai-tts", "playai-tts-arabic", "qwen-2.5-32b", "qwen-2.5-coder-32b", "qwen-qwq-32b", "whisper-large-v3", "whisper-large-v3-turbo", ] excluded = ["whisper", "playai-tts"] return [model for model in models if not any(x in model for x in excluded)] def pipes(self) -> List[dict]: model_ids = self.get_model_ids() return [{"id": model_id, "name": model_id} for model_id in model_ids] def pipe( self, body: dict, ) -> Union[str, Generator, Iterator]: body["model"] = body["model"].removeprefix(self.id + ".") try: r = requests.post( url=f"{self.base_url}/chat/completions", json=body, headers={ "Content-Type": "application/json", "Authorization": f"Bearer {self.valves.API_KEY}", }, stream=True, ) r.raise_for_status() if body["stream"]: return r.iter_lines() else: return r.json() except Exception as e: return f"Error: {e}"