""" title: DeepSeek-R1 Compatible Pipe author: Forgotmypassword (original OpenAI version by Raiyan Hasan) version: 1 - (Im never updating this) license: MIT """ import os import json import requests from pydantic import BaseModel, Field from typing import List, Union, Iterator # Set DEBUG to True to enable detailed logging DEBUG = False class Pipe: class Valves(BaseModel): openai_API_KEY: str = Field(default="") # Optional API key if needed DEFAULT_MODEL: str = Field(default="deepseek-r1") # Default model identifier def __init__(self): self.id = "deepseek" self.type = "chat" self.name = "deepseek-r1: " self.valves = self.Valves( **{ "openai_API_KEY": os.getenv("openai_API_KEY", ""), "DEFAULT_MODEL": os.getenv("openai_DEFAULT_MODEL", "deepseek-r1"), } ) def get_openai_models(self): """Return available models""" return [ { "id": "deepseek-r1", "name": "DeepSeek-R1 Distill Qwen 32B", "type": "chat", "censored": True, "description": "DeepSeek-R1 Distill Qwen 32B", "baseModel": True, "reasoning": True, } ] def pipes(self) -> List[dict]: return self.get_openai_models() def pipe(self, body: dict) -> Union[str, Iterator[str]]: try: # Use default model ID model_id = body.get("model", self.valves.DEFAULT_MODEL) messages = [] # Process messages including system, user, and assistant messages for message in body["messages"]: if isinstance(message.get("content"), list): # Join multiple content parts into a single text text_parts = [] for content in message["content"]: if content["type"] == "text": text_parts.append(content["text"]) elif content["type"] == "image_url": # Handle image references text_parts.append(f"[Image: {content['image_url']['url']}]") messages.append( {"role": message["role"], "content": " ".join(text_parts)} ) else: # Handle simple text messages messages.append( {"role": message["role"], "content": message["content"]} ) if DEBUG: print("DeepSeek API request:") print(" Model:", model_id) print(" Messages:", json.dumps(messages, indent=2)) # Prepare request parameters request_params = {"messages": messages, "model": model_id} # Add optional parameters if body.get("seed") is not None: request_params["seed"] = body["seed"] if body.get("jsonMode", False): request_params["jsonMode"] = True # Make request to Pollinations.ai endpoint response = requests.post( "https://text.pollinations.ai/", json=request_params, headers={"Content-Type": "application/json"}, ) # Try to parse JSON try: result = response.json() except json.JSONDecodeError: # If JSON parsing fails, print the raw response for debugging if DEBUG: print("Raw response:", response.text) # Return the raw text directly return response.text # Handle streaming vs non-streaming if body.get("stream", False): def stream_generator(): try: # Yield the entire content as a single chunk content = ( result.get("choices", [{}])[0] .get("message", {}) .get("content", "") ) yield content except Exception as e: if DEBUG: print(f"Streaming error: {e}") yield f"Error during streaming: {str(e)}" return stream_generator() else: # Non-streaming response # Try to get content from JSON, fallback to raw text content = ( result.get("choices", [{}])[0].get("message", {}).get("content", "") ) return content if content else response.text except requests.RequestException as e: if DEBUG: print(f"Request Error: {e}") return f"Request Error: {e}" except Exception as e: if DEBUG: print(f"Unexpected Error: {e}") return f"Error: {e}" def health_check(self) -> bool: """Check if the API is accessible""" try: # Simple health check test_response = self.pipe( {"messages": [{"role": "user", "content": "Hello"}]} ) return True except Exception as e: if DEBUG: print(f"Health check failed: {e}") return False