Created
September 2, 2024 16:57
-
-
Save shock/d81365117c7f311be8a585a663c11a10 to your computer and use it in GitHub Desktop.
OpenWebUI DeepSeek Manifold Pipeline
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| from typing import List, Union, Generator, Iterator | |
| from pydantic import BaseModel | |
| import os | |
| import requests | |
| class Pipeline: | |
| class Valves(BaseModel): | |
| DEEPSEEK_API_BASE_URL: str = "https://api.deepseek.com" | |
| DEEPSEEK_API_KEY: str = "" | |
| pass | |
| def __init__(self): | |
| self.type = "manifold" | |
| # Optionally, you can set the id and name of the pipeline. | |
| # Best practice is to not specify the id so that it can be automatically inferred from the filename, so that users can install multiple versions of the same pipeline. | |
| # The identifier must be unique across all pipelines. | |
| # The identifier must be an alphanumeric string that can include underscores or hyphens. It cannot contain spaces, special characters, slashes, or backslashes. | |
| # self.id = "deepseek_pipeline" | |
| self.name = "deepseek/" | |
| self.valves = self.Valves( | |
| **{ | |
| "DEEPSEEK_API_KEY": os.getenv( | |
| "DEEPSEEK_API_KEY", "your-openai-api-key-here" | |
| ) | |
| } | |
| ) | |
| self.pipelines = self.get_deepseek_models() | |
| pass | |
| async def on_startup(self): | |
| # This function is called when the server is started. | |
| print(f"on_startup:{__name__}") | |
| pass | |
| async def on_shutdown(self): | |
| # This function is called when the server is stopped. | |
| print(f"on_shutdown:{__name__}") | |
| pass | |
| async def on_valves_updated(self): | |
| # This function is called when the valves are updated. | |
| print(f"on_valves_updated:{__name__}") | |
| self.pipelines = self.get_deepseek_models() | |
| pass | |
| def get_deepseek_models(self): | |
| return [ | |
| {"id": "deepseek-coder", "name": "deepseek-coder"}, | |
| {"id": "deepseek-chat", "name": "deepseek-chat"}, | |
| ] | |
| def pipe( | |
| self, user_message: str, model_id: str, messages: List[dict], body: dict | |
| ) -> Union[str, Generator, Iterator]: | |
| # This is where you can add your custom pipelines like RAG. | |
| print(f"pipe:{__name__}") | |
| print(messages) | |
| print(user_message) | |
| headers = {} | |
| headers["Authorization"] = f"Bearer {self.valves.DEEPSEEK_API_KEY}" | |
| headers["Content-Type"] = "application/json" | |
| payload = {**body, "model": model_id} | |
| if "user" in payload: | |
| del payload["user"] | |
| if "chat_id" in payload: | |
| del payload["chat_id"] | |
| if "title" in payload: | |
| del payload["title"] | |
| print(payload) | |
| try: | |
| r = requests.post( | |
| url=f"{self.valves.DEEPSEEK_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}" |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment