""" title: DeepClaude author: kjdy description: Display the reasoning chain of the DeepSeek R1 model in OpwenWebUI - only supports version 0.5.6 and above. version: 1.6.0 licence: MIT """ import json import httpx from typing import AsyncGenerator, Callable, Awaitable from pydantic import BaseModel, Field class Pipe: class Valves(BaseModel): DEEPSEEK_API_BASE_URL: str = Field( default="https://api.deepseek.com/v1", description="Base URL for DeepSeek R1 API", ) DEEPSEEK_API_KEY: str = Field( default="", description="DeepSeek R1 API key for authentication", ) DEEPSEEK_MODEL_ID: str = Field( default="deepseek-ai/DeepSeek-R1", description="Model ID for DeepSeek R1", ) CLAUDE_API_BASE_URL: str = Field( default="https://fix.it/v1", description="Base URL for Claude (OpenAI format)", ) CLAUDE_API_KEY: str = Field( default="", description="Claude API key for authentication", ) CLAUDE_MODEL_ID: str = Field( default="claude-3-5-sonnet-20240620", description="Model ID for Claude", ) def __init__(self): self.valves = self.Valves() self.data_prefix = "data: " def pipes(self): return [{"id": "DeepClaude", "name": "DeepClaude"}] async def pipe( self, body: dict, __event_emitter__: Callable[[dict], Awaitable[None]] = None ) -> AsyncGenerator[str, None]: """Main processing pipeline (buffer removed)""" # Initialize request-specific state variables thinking = -1 # -1: Not started, 0: Processing deepseek_thinking = "" # Store the thinking process emitter = __event_emitter__ # Store the event emitter # Validate configuration if not self.valves.DEEPSEEK_API_KEY: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": "Error: DeepSeek API key not configured", "done": True, }, } ) yield json.dumps( {"error": "DeepSeek API key not configured"}, ensure_ascii=False ) return # Prepare request headers headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } try: # Extract model ID payload = {**body, "model": self.valves.DEEPSEEK_MODEL_ID} # Process messages to avoid consecutive messages with the same role messages = payload["messages"] # Calculate total context length context_length = sum(len(msg.get("content", "")) for msg in messages) payload["max_tokens"] = 8192 i = 0 while i < len(messages) - 1: if messages[i]["role"] == messages[i + 1]["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 # Make the first API request if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "Processing request with DeepSeek...", "done": False}, } ) 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=30, ) as response: # Error handling if response.status_code != 200: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Error: DeepSeek API returned {response.status_code}", "done": True, }, } ) error = await response.aread() yield self._format_error(response.status_code, error) return # Process the response in a streaming manner async for line in response.aiter_lines(): if not line.startswith(self.data_prefix): continue raw_data = line[len(self.data_prefix) :] # Special handling for [DONE] message if raw_data.strip() == "[DONE]": yield "\n" yield "</think>" yield "\n\n" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "DeepSeek processing completed, starting Claude...", "done": False}, } ) break try: data = json.loads(raw_data) except json.JSONDecodeError as e: # Only print debug info when error occurs yield json.dumps( { "debug": f"Request payload: {json.dumps(payload, ensure_ascii=False)}" }, ensure_ascii=False, ) yield json.dumps( {"debug": f"Raw response line: {line}"}, ensure_ascii=False, ) yield json.dumps( { "error": f"Failed to parse JSON: {str(e)}\nRaw data: {raw_data}" }, ensure_ascii=False, ) continue choice = data.get("choices", [{}])[0] # Check termination condition if choice.get("finish_reason"): yield "\n" yield "</think>" yield "\n\n" break # Use break to exit the loop rather than return delta = choice.get("delta", {}) # State machine processing # State transition: Not started -> Processing if thinking == -1 and delta.get("reasoning_content"): thinking = 0 yield "\n" yield "<think>" yield "\n" yield "🧠 Here is DeepSeek's thinking process:\n\n" # Process content and send immediately content = self._process_content(choice["delta"]) if content: # Store content based on state if thinking == 0: deepseek_thinking += content yield content # Add DeepSeek's thinking as an assistant message if deepseek_thinking: formatted_thinking = f"<think>\n{deepseek_thinking}\n</think>" messages.append({ "role": "assistant", "content": formatted_thinking }) if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "Processing request with Claude ...", "done": False}, } ) # Make the second API request headers = { "Authorization": f"Bearer {self.valves.CLAUDE_API_KEY}", "Content-Type": "application/json", } # Prepare the second model's request second_payload = { **payload, "model": self.valves.CLAUDE_MODEL_ID, "messages": [ { "role": "system", "content": "You have access to private thinking processes wrapped in <think> tags. The content within these tags is your private reference - users CANNOT see it. You can use insights from this content but never reveal or hint at having access to it. Your response should appear completely original." }, *messages ], "max_tokens": 4096, # Set max tokens for Claude } async with httpx.AsyncClient(http2=True) as client: async with client.stream( "POST", f"{self.valves.CLAUDE_API_BASE_URL}/chat/completions", json=second_payload, headers=headers, timeout=30, ) as response: if response.status_code != 200: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Error: Claude API returned {response.status_code}", "done": True, }, } ) error = await response.aread() yield self._format_error(response.status_code, error) return found_start = False buffer = "" async for line in response.aiter_lines(): if not line.startswith(self.data_prefix): continue data = json.loads(line[len(self.data_prefix) :]) choice = data.get("choices", [{}])[0] if choice.get("finish_reason"): if found_start: yield "\n\n" if __event_emitter__: await __event_emitter__( { "type": "status", "data": {"description": "Request completed successfully", "done": True}, } ) break delta = choice.get("delta", {}) content = self._process_content(choice["delta"]) if content: yield content if not found_start: buffer += content if "<|start_of_my_answer|>" in buffer: found_start = True yield buffer.split("<|start_of_my_answer|>")[1] else: yield content except Exception as e: if __event_emitter__: await __event_emitter__( { "type": "status", "data": { "description": f"Error: {str(e)}", "done": True, }, } ) yield self._format_exception(e) def _process_content(self, delta: dict) -> str: """Return the processed content directly""" return delta.get("reasoning_content", "") or delta.get("content", "") def _format_error(self, status_code: int, error: bytes) -> str: """Error formatting remains unchanged""" try: err_msg = json.loads(error).get("message", error.decode(errors="ignore"))[ :200 ] except: err_msg = error.decode(errors="ignore")[:200] return json.dumps( {"error": f"HTTP {status_code}: {err_msg}"}, ensure_ascii=False ) def _format_exception(self, e: Exception) -> str: """Exception formatting remains unchanged""" err_type = type(e).__name__ return json.dumps({"error": f"{err_type}: {str(e)}"}, ensure_ascii=False)