""" title: DeepSeek Manifold Pipe author: Siraj Florida version: 0.1.0 license: MIT """ import requests import json from typing import List, Union, Generator, Iterator from pydantic import BaseModel, Field from utils.misc import pop_system_message class Pipe: class Valves(BaseModel): DEEPSEEK_API_KEY: str = Field( default="", description="API key for DeepSeek services.", ) DEEPSEEK_API_BASE_URL: str = Field( default="https://api.deepseek.com/v1", description="Base URL for DeepSeek API endpoints.", ) NAME_PREFIX: str = Field( default="deepseek/", description="The prefix applied before the model names.", ) MAX_TOKENS: int = Field( default=2048, description="Maximum number of tokens to generate.", ) TEMPERATURE: float = Field( default=1.0, description="Sampling temperature.", ) TOP_P: float = Field( default=1.0, description="Nucleus sampling top_p value.", ) FREQUENCY_PENALTY: float = Field( default=0.0, description="Frequency penalty.", ) PRESENCE_PENALTY: float = Field( default=0.0, description="Presence penalty.", ) STREAM: bool = Field( default=False, description="Whether to stream responses.", ) def __init__(self): self.valves = self.Valves() self.type = "manifold" self.id = "deepseek" self.name = "" # Leave this as an empty string def get_deepseek_models(self) -> List[dict]: """Fetch available models from DeepSeek API""" headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Accept": "application/json", } try: response = requests.get( f"{self.valves.DEEPSEEK_API_BASE_URL}/models", headers=headers ) response.raise_for_status() models_data = response.json() # Format models according to manifold requirements models = [ { "id": model["id"], "name": f'{self.valves.NAME_PREFIX}{model["id"]}', } for model in models_data.get("data", []) ] return models except Exception as e: print(f"Error fetching models: {e}") return [] def pipes(self) -> List[dict]: """Return list of available models""" return self.get_deepseek_models() def pipe(self, body: dict) -> Union[str, Generator, Iterator]: """Process chat completion requests""" system_message, messages = pop_system_message(body.get("messages", [])) # Process messages and maintain system message processed_messages = [] if system_message: processed_messages.append( {"role": "system", "content": str(system_message)} ) # Add remaining messages for message in messages: if isinstance(message.get("content"), str): processed_messages.append( {"role": message["role"], "content": message.get("content", "")} ) else: # Handle potential structured content # Note: DeepSeek currently doesn't support image inputs like OpenAI processed_messages.append( {"role": message["role"], "content": message.get("content", "")} ) # Construct payload according to DeepSeek API specifications payload = { "model": "deepseek-chat", # Default model "messages": processed_messages, "stream": self.valves.STREAM, "max_tokens": self.valves.MAX_TOKENS, "temperature": self.valves.TEMPERATURE, "top_p": self.valves.TOP_P, "frequency_penalty": self.valves.FREQUENCY_PENALTY, "presence_penalty": self.valves.PRESENCE_PENALTY, "response_format": {"type": "text"}, "tool_choice": "none", } # Add optional stop sequences if provided if body.get("stop"): payload["stop"] = body["stop"] headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", "Accept": "application/json", } url = f"{self.valves.DEEPSEEK_API_BASE_URL}/chat/completions" try: if self.valves.STREAM: return self.stream_response(url, headers, payload) else: return self.non_stream_response(url, headers, payload) except Exception as e: print(f"Error in pipe method: {e}") return f"Error: {e}" def stream_response(self, url: str, headers: dict, payload: dict) -> Generator: """Handle streaming responses""" response = requests.post(url, headers=headers, json=payload, stream=True) if response.status_code != 200: raise Exception(f"HTTP Error {response.status_code}: {response.text}") for line in response.iter_lines(): if line: decoded_line = line.decode("utf-8") if decoded_line.startswith("data: "): data = decoded_line[6:] if data.strip() == "[DONE]": break try: data_json = json.loads(data) content = data_json["choices"][0]["delta"].get("content", "") if content: yield content except json.JSONDecodeError: print(f"Failed to parse JSON: {data}") except KeyError as e: print(f"Unexpected data structure: {e}") def non_stream_response(self, url: str, headers: dict, payload: dict) -> str: """Handle non-streaming responses""" response = requests.post(url, headers=headers, json=payload) 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 res.get("choices") else ""