""" title: OpenAI Manifold Pipe author: open-webui author_url: https://github.com/open-webui funding_url: https://github.com/open-webui version: 0.1.2 """ from pydantic import BaseModel, Field from typing import Optional, Union, Generator, Iterator from utils.misc import get_last_user_message import os import requests class Pipe: class Valves(BaseModel): NAME_PREFIX: str = Field( default="OPENAI/", description="The prefix applied before the model names.", ) OPENAI_API_BASE_URL: str = Field( default="https://api.openai.com/v1", description="The base URL for OpenAI API endpoints.", ) OPENAI_API_KEY: str = Field( default="", description="Required API key to retrieve the model list.", ) pass class UserValves(BaseModel): OPENAI_API_KEY: str = Field( default="", description="User-specific API key for accessing OpenAI services.", ) def __init__(self): self.type = "manifold" self.valves = self.Valves() pass def pipes(self): if self.valves.OPENAI_API_KEY: try: headers = {} headers["Authorization"] = f"Bearer {self.valves.OPENAI_API_KEY}" headers["Content-Type"] = "application/json" r = requests.get( f"{self.valves.OPENAI_API_BASE_URL}/models", headers=headers ) models = r.json() return [ { "id": model["id"], "name": f'{self.valves.NAME_PREFIX}{model["name"] if "name" in model else model["id"]}', } for model in models["data"] if "gpt" in model["id"] ] except Exception as e: print(f"Error: {e}") return [ { "id": "error", "name": "Could not fetch models from OpenAI, please update the API Key in the valves.", }, ] else: return [ { "id": "error", "name": "Global API Key not provided.", }, ] def pipe(self, body: dict, __user__: dict) -> Union[str, Generator, Iterator]: # This is where you can add your custom pipelines like RAG. print(f"pipe:{__name__}") print(__user__) user_valves = __user__.get("valves") if not user_valves: raise Exception("User Valves not configured.") if not user_valves.OPENAI_API_KEY: raise Exception("OPENAI_API_KEY not provided by the user.") headers = {} headers["Authorization"] = f"Bearer {user_valves.OPENAI_API_KEY}" headers["Content-Type"] = "application/json" model_id = body["model"][body["model"].find(".") + 1 :] payload = {**body, "model": model_id} print(payload) try: r = requests.post( url=f"{self.valves.OPENAI_API_BASE_URL}/chat/completions", json=payload, headers=headers, 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}"