""" title: DeepSeek Manifold Pipe authors: [Mapping730] author_url: [https://github.com/Mapping730] funding_url: [Your Funding URL] version: 0.1.0 required_open_webui_version: 0.3.17 license: MIT """ import os import requests import json import time from typing import List, Union, Generator, Iterator from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message class Pipe: # Use the expected class name "Pipe" class Valves(BaseModel): DEEPSEEK_API_KEY: str = Field(default="") def __init__(self): self.type = "manifold" self.id = "deepseek" self.name = "deepseek/" self.valves = self.Valves( **{"DEEPSEEK_API_KEY": os.getenv("DEEPSEEK_API_KEY", "")} ) self.MAX_IMAGE_SIZE = 5 * 1024 * 1024 # 5MB per image def get_deepseek_models(self): return [ {"id": "deepseek-chat", "name": "DeepSeek Chat"}, {"id": "deepseek-coder", "name": "DeepSeek Coder"}, ] def pipes(self) -> List[dict]: return self.get_deepseek_models() def process_image(self, image_data): """Process image data with size validation.""" if image_data["image_url"]["url"].startswith("data:image"): mime_type, base64_data = image_data["image_url"]["url"].split(",", 1) media_type = mime_type.split(":")[1].split(";")[0] # Check base64 image size image_size = len(base64_data) * 3 / 4 # Convert base64 size to bytes if image_size > self.MAX_IMAGE_SIZE: raise ValueError( f"Image size exceeds 5MB limit: {image_size / (1024 * 1024):.2f}MB" ) return { "type": "image", "source": { "type": "base64", "media_type": media_type, "data": base64_data, }, } else: # For URL images, perform size check after fetching url = image_data["image_url"]["url"] response = requests.head(url, allow_redirects=True) content_length = int(response.headers.get("content-length", 0)) if content_length > self.MAX_IMAGE_SIZE: raise ValueError( f"Image at URL exceeds 5MB limit: {content_length / (1024 * 1024):.2f}MB" ) return { "type": "image", "source": {"type": "url", "url": url}, } def pipe(self, body: dict) -> Union[str, Generator, Iterator]: system_message, messages = pop_system_message(body["messages"]) processed_messages = [] total_image_size = 0 for message in messages: processed_content = "" if isinstance(message.get("content"), list): # Extract text content from the list for item in message["content"]: if item["type"] == "text": processed_content += item["text"] + " " # Combine text items elif item["type"] == "image_url": # Handle images (if needed) processed_image = self.process_image(item) # Append image description or placeholder to the text content processed_content += ( f"[Image: {processed_image['source']['type']}] " ) else: processed_content = message.get("content", "") # Remove trailing space processed_content = processed_content.strip() processed_messages.append( {"role": message["role"], "content": processed_content} ) payload = { "model": body["model"][body["model"].find(".") + 1 :], "messages": processed_messages, "max_tokens": body.get("max_tokens", 4096), "temperature": body.get("temperature", 0.8), "top_k": body.get("top_k", 40), "top_p": body.get("top_p", 0.9), "stop_sequences": body.get("stop", []), **({"system": str(system_message)} if system_message else {}), "stream": body.get("stream", False), } headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } url = "https://api.deepseek.com/chat/completions" try: if body.get("stream", False): return self.stream_response(url, headers, payload) else: return self.non_stream_response(url, headers, payload) except requests.exceptions.RequestException as e: print(f"Request failed: {e}") return f"Error: Request failed: {e}" except Exception as e: print(f"Error in pipe method: {e}") return f"Error: {e}" def stream_response(self, url, headers, payload): try: with requests.post( url, headers=headers, json=payload, stream=True, timeout=(3.05, 60) ) as response: if response.status_code != 200: raise Exception( f"HTTP Error {response.status_code}: {response.text}" ) buffer = "" for chunk in response.iter_content(chunk_size=None): if chunk: chunk = chunk.decode("utf-8") buffer += chunk # Process complete JSON objects in the buffer while "data: " in buffer: data_start = buffer.find("data: ") data_end = buffer.find("\n", data_start) if data_end == -1: break # Incomplete JSON object, wait for more data data_str = buffer[data_start:data_end].strip() buffer = buffer[ data_end + 1 : ] # Remove processed data from the buffer if data_str.startswith("data: "): data_str = data_str[6:] # Remove "data: " prefix # Handle the [DONE] message if data_str == "[DONE]": break try: data = json.loads(data_str) if "choices" in data and data["choices"]: yield data["choices"][0]["delta"].get("content", "") except json.JSONDecodeError as e: print(f"JSON decode error: {e}") print(f"Raw data: {data_str}") except requests.exceptions.RequestException as e: print(f"Request failed: {e}") yield f"Error: Request failed: {e}" except Exception as e: print(f"General error in stream_response method: {e}") yield f"Error: {e}" def non_stream_response(self, url, headers, payload): try: response = requests.post( url, headers=headers, json=payload, timeout=(3.05, 60) ) if response.status_code != 200: raise Exception(f"HTTP Error {response.status_code}: {response.text}") res = response.json() return ( res["choices"][0]["message"]["content"] if "choices" in res and res["choices"] else "" ) except requests.exceptions.RequestException as e: print(f"Failed non-stream request: {e}") return f"Error: {e}" except Exception as e: print(f"Error in non_stream_response method: {e}") return f"Error: {e}"