""" title: Deepseek V3,R1 Manifold Pipe authors: [MCode-Team] author_url: [https://github.com/MCode-Team] funding_url: https://github.com/open-webui version: 0.1.2 license: MIT environment_variables: - DEEPSEEK_API_KEY (required) """ import os import json import time import logging import httpx import re from typing import ( List, Union, Generator, Iterator, Dict, Optional, AsyncIterator, Tuple, Awaitable, Callable, ) from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message class Pipe: REQUEST_TIMEOUT = (3.05, 60) MODEL_MAX_TOKENS = { "deepseek-chat": 8192, "deepseek-reasoner": 8192, } class Valves(BaseModel): DEEPSEEK_BASE_URL: str = Field( default=os.getenv("DEEPSEEK_BASE_URL", "https://api.deepseek.com"), description="DeepSeek API URL", ) DEEPSEEK_API_KEY: str = Field( default=os.getenv("DEEPSEEK_API_KEY", ""), description="DeepSeek API Key", ) def __init__(self): logging.basicConfig(level=logging.INFO) self.type = "manifold" self.id = "deepseek" self.name = "deepseek/" self.valves = self.Valves() self.request_id = None self.clean_pattern = re.compile(r"

.*?

\n\n", flags=re.DOTALL) self.buffer_size = 3 self.thinking_state = -1 # -1: Not started, 0: Thinking, 1: Answered @staticmethod def get_model_id(model_name: str) -> str: return model_name.replace(".", "/").split("/")[-1] def get_deepseek_models(self) -> List[Dict[str, str]]: try: headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } with httpx.Client() as client: response = client.get( f"{self.valves.DEEPSEEK_BASE_URL}/models", headers=headers, timeout=10, ) response.raise_for_status() models_data = response.json() return [ {"id": model["id"], "name": model["id"]} for model in models_data.get("data", []) ] except Exception as e: logging.error(f"Error getting models: {e}") return [] def pipes(self) -> List[dict]: return self.get_deepseek_models() async def _stream_response( self, url: str, headers: dict, payload: dict, __event_emitter__=None, model_id: str = "", ) -> AsyncIterator[str]: buffer = [] self.thinking_state = -1 last_status_time = time.time() status_dots = 0 try: async with httpx.AsyncClient() as client: async with client.stream( "POST", url, headers=headers, json=payload, timeout=self.REQUEST_TIMEOUT, ) as response: response.raise_for_status() async for line in response.aiter_lines(): line = line.strip() if not line.startswith("data: "): continue data_str = line[6:].strip() if data_str == "[DONE]": continue try: data = json.loads(data_str) except json.JSONDecodeError as e: logging.error( f"Failed to parse data line: {data_str}, error: {e}" ) continue choice = data.get("choices", [{}])[0] delta = choice.get("delta", {}) reasoning = delta.get("reasoning_content") or "" content = delta.get("content") or "" finish_reason = choice.get("finish_reason") if self.thinking_state == -1 and reasoning: self.thinking_state = 0 buffer.append( "

Thinking Process

\n\n" ) elif self.thinking_state == 0 and not reasoning and content: self.thinking_state = 1 buffer.append("\n

\n\n") if self.thinking_state == 0 and model_id == "deepseek-reasoner": current_time = time.time() if current_time - last_status_time > 1: status_dots = (status_dots % 3) + 1 last_status_time = current_time if reasoning: buffer.append(reasoning.replace("\n", "\n> ")) elif content: buffer.append(content) if finish_reason == "stop": if self.thinking_state == 0: buffer.append("\n\n\n") break if len(buffer) >= self.buffer_size or "\n" in ( reasoning + content ): yield "".join(buffer) buffer.clear() if buffer: yield "".join(buffer) except Exception as e: error_msg = f"Stream Error: {str(e)}" yield error_msg async def _regular_request( self, url: str, headers: dict, payload: dict, __event_emitter__=None, model_id: str = "", ) -> dict: try: async with httpx.AsyncClient() as client: response = await client.post( url, headers=headers, json=payload, timeout=self.REQUEST_TIMEOUT, ) response.raise_for_status() data = response.json() if "choices" in data and len(data["choices"]) > 0: choice = data["choices"][0] message = choice.get("message", {}) original_content = message.get("content", "") reasoning = message.get("reasoning_content", "") if reasoning: processed_content = ( f"

Thinking Process

\n\n" f"{reasoning}\n

\n\n{original_content}" ) processed_content = self.clean_pattern.sub( "", processed_content ).strip() data["choices"][0]["message"]["content"] = processed_content data["choices"][0]["message"]["reasoning_content"] = reasoning return data except Exception as e: error_msg = f"Error: {str(e)}" logging.error(f"Regular request failed: {error_msg}") return {"error": error_msg, "choices": []} async def pipe( self, body: Dict, __event_emitter__=None ) -> Union[AsyncIterator[str], dict]: if not self.valves.DEEPSEEK_API_KEY: return {"error": "DEEPSEEK_API_KEY is required", "choices": []} try: system_message, messages = pop_system_message(body.get("messages", [])) for msg in messages: if msg.get("role") == "assistant" and "content" in msg: msg["content"] = self.clean_pattern.sub("", msg["content"]).strip() model_id = self.get_model_id(body["model"]) max_tokens_limit = self.MODEL_MAX_TOKENS.get(model_id, 8192) if system_message: messages.insert(0, {"role": "system", "content": str(system_message)}) payload = { "model": model_id, "messages": messages, "max_tokens": min( body.get("max_tokens", max_tokens_limit), max_tokens_limit ), "temperature": float(body.get("temperature", 0.7)), "stream": body.get("stream", False), } headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } if payload["stream"]: return self._stream_response( url=f"{self.valves.DEEPSEEK_BASE_URL}/chat/completions", headers=headers, payload=payload, __event_emitter__=__event_emitter__, model_id=model_id, ) else: return await self._regular_request( url=f"{self.valves.DEEPSEEK_BASE_URL}/chat/completions", headers=headers, payload=payload, __event_emitter__=__event_emitter__, model_id=model_id, ) except Exception as e: error_msg = f"Error: {str(e)}" logging.error(f"Pipe processing failed: {error_msg}") return {"error": error_msg, "choices": []}