""" title: DeepSeek R1 author: firstpick description: Display the reasoning chain of the DeepSeek R1 model in OpenWebUI - only supports version 0.5.6 and above. version: 1.2.10 license: MIT """ import json import httpx import re from typing import AsyncGenerator, Callable, Awaitable, Optional, Dict, Any from pydantic import BaseModel, Field import asyncio class Pipe: class Valves(BaseModel): DEEPSEEK_API_BASE_URL: str = Field( default="https://api.deepseek.com/v1", description="Base URL for DeepSeek API requests", ) DEEPSEEK_API_KEY: str = Field( default="", description="API key for authentication. Obtain it from the control panel.", ) DEEPSEEK_API_MODEL: str = Field( default="deepseek-reasoner", description="Model name for API requests (default: deepseek-reasoner)", ) def __init__(self) -> None: self.valves = self.Valves() self.data_prefix = "data:" self.emitter: Optional[Callable[[Dict[str, Any]], Awaitable[None]]] = None def pipes(self) -> list: """Return a list of available pipes with their IDs and names.""" return [ { "id": self.valves.DEEPSEEK_API_MODEL, "name": self.valves.DEEPSEEK_API_MODEL, } ] async def pipe( self, body: Dict[str, Any], __event_emitter__: Optional[Callable[[Dict[str, Any]], Awaitable[None]]] = None, ) -> AsyncGenerator[str, None]: """Main processing pipeline (buffer removed). Processes the request body and streams responses from the DeepSeek API. """ # Use a dictionary to keep track of the thinking state: -1 means uninitialized, # 0 means thinking, and 1 indicates the answer has been provided. thinking_state: Dict[str, int] = {"thinking": -1} self.emitter = __event_emitter__ # Validate configuration: ensure the API key is provided. if not self.valves.DEEPSEEK_API_KEY: yield json.dumps({"error": "API key is not configured"}, ensure_ascii=False) return # Prepare request headers and payload. headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } try: # Extract the model ID from the 'model' field. model_id = body.get("model", "").split(".", 1)[-1] payload = {**body, "model": model_id} # Process messages to prevent consecutive entries from the same role. messages = payload.get("messages", []) i = 0 while i < len(messages) - 1: if messages[i].get("role") == messages[i + 1].get("role"): # Insert a placeholder message with an alternate role. alternate_role = ( "assistant" if messages[i]["role"] == "user" else "user" ) messages.insert( i + 1, {"role": alternate_role, "content": "[Unfinished thinking]"}, ) i += 1 async with httpx.AsyncClient(http2=True) as client: async with client.stream( "POST", f"{self.valves.DEEPSEEK_API_BASE_URL}/chat/completions", json=payload, headers=headers, timeout=300, ) as response: # Handle non-200 HTTP responses. if response.status_code != 200: error_data = await response.aread() yield self._format_error(response.status_code, error_data) return # Stream the response and process it line by line. async for line in response.aiter_lines(): if not line.startswith(self.data_prefix): continue # Extract the JSON string from the line. json_str = line[len(self.data_prefix) :] # Check for the termination signal. if json_str.strip() == "[DONE]": return try: data = json.loads(json_str) except json.JSONDecodeError as e: error_detail = f"JSON parsing failed - content: {json_str}, reason: {e}" yield self._format_error( "JSONDecodeError", error_detail.encode() ) return choice = data.get("choices", [{}])[0] # Exit if a finish reason is specified. if choice.get("finish_reason"): return # Update the thinking state based on the delta data. state_output = await self._update_thinking_state( choice.get("delta", {}), thinking_state ) if state_output: yield state_output # Emit state indicator. if state_output == "<think>": yield "\n" # Process and immediately yield the response content. content = self._process_content(choice.get("delta", {})) if content: if content.startswith("<think>"): # Remove the marker and yield it separately. content = content[len("<think>") :] yield "<think>" await asyncio.sleep(0.1) yield "\n" elif content.startswith("</think>"): # Remove the closing marker and yield it. content = content[len("</think>") :] yield "</think>" await asyncio.sleep(0.1) yield "\n" yield content except Exception as e: yield self._format_exception(e) async def _update_thinking_state( self, delta: Dict[str, Any], thinking_state: Dict[str, int] ) -> str: """Update the thinking state based on the current response delta. Transitions: - Uninitialized to 'thinking' when 'reasoning_content' is detected. - 'Thinking' to 'answered' when 'reasoning_content' is absent and 'content' is present. """ state_output = "" # Transition from uninitialized to thinking. if thinking_state["thinking"] == -1 and delta.get("reasoning_content"): thinking_state["thinking"] = 0 state_output = "<think>" # Transition from thinking to answered. elif ( thinking_state["thinking"] == 0 and not delta.get("reasoning_content") and delta.get("content") ): thinking_state["thinking"] = 1 state_output = "\n</think>\n\n" return state_output def _process_content(self, delta: Dict[str, Any]) -> str: """Extract and return the processed content from the delta data.""" return delta.get("reasoning_content", "") or delta.get("content", "") def _format_error(self, status_code: int, error: bytes) -> str: """Format an HTTP error into a JSON string.""" error_str = error if isinstance(error, str) else error.decode(errors="ignore") try: err_msg = json.loads(error_str).get("message", error_str)[:200] except Exception: err_msg = error_str[:200] return json.dumps( {"error": f"HTTP {status_code}: {err_msg}"}, ensure_ascii=False ) def _format_exception(self, e: Exception) -> str: """Format a caught exception into a JSON error string.""" err_type = type(e).__name__ return json.dumps({"error": f"{err_type}: {str(e)}"}, ensure_ascii=False)