""" title: HuggingFace Serverless Inference API (DeepSeek) author: version: 1.0 description: 用于调用 HuggingFace 的 DeepSeek DeepSeek-R1-Distill-Qwen-32B 无服务器推理 API 的管道，支持多密钥轮询 """ from pydantic import BaseModel, Field import time from typing import List, Optional, Generator import requests from openai import OpenAI class Pipe: class Valves(BaseModel): NAME_PREFIX: str = Field( default="DeepSeek/", description="模型名称前缀，用于在 OpenWebUI 中显示" ) API_KEYS: str = Field( default="", description="多个 API key，使用英文逗号分隔。示例：key1,key2,key3", ) BASE_URL: str = Field( default="https://api-inference.huggingface.co/v1/", description="HuggingFace API 基础URL，末尾需要带斜杠", ) MIN_INTERVAL: float = Field( default=1.0, description="API 调用最小间隔(秒)，用于控制请求频率" ) MAX_RETRIES: int = Field(default=3, description="请求失败时的最大重试次数") TEMPERATURE: float = Field( default=0.5, description="温度参数(0-2.0)，控制输出的随机性：\n" "- 较高的值(如1.0)会产生更随机的输出\n" "- 较低的值(如0.2)会产生更确定性的输出\n" "- 0.5-0.8是比较好的创造性写作范围", ge=0.0, le=2.0, ) MAX_TOKENS: int = Field( default=2048, description="输入 + 输出的总 token 上限\n" "较大的值允许更长的对话，但会增加 API 调用成本", ge=1, le=16384, ) TOP_P: float = Field( default=0.7, description="核采样阈值(0-1.0)，控制输出文本的多样性：\n" "- 较高的值会保留更多的低概率选项\n" "- 较低的值会更严格地筛选词语", ge=0.0, le=1.0, ) STREAM: bool = Field( default=True, description="是否使用流式输出：\n" "- True: 逐字输出，体验更好\n" "- False: 等待完整回复后一次性返回", ) def __init__(self): self.valves = self.Valves() self.current_key_index = 0 self.last_used_time = 0 self._api_keys = [] @property def api_keys(self) -> List[str]: if not self._api_keys and self.valves.API_KEYS: self._api_keys = [ k.strip() for k in self.valves.API_KEYS.split(",") if k.strip() ] return self._api_keys def get_next_key(self) -> Optional[str]: if not self.api_keys: return None current_time = time.time() if current_time - self.last_used_time < self.valves.MIN_INTERVAL: time.sleep(self.valves.MIN_INTERVAL - (current_time - self.last_used_time)) key = self.api_keys[self.current_key_index] self.current_key_index = (self.current_key_index + 1) % len(self.api_keys) self.last_used_time = time.time() return key def pipes(self): return [ { "id": "deepseek-r1-32b", "name": f"{self.valves.NAME_PREFIX}DeepSeek-R1-Distill-Qwen-32B", }, ] def create_chat_completion( self, messages: list, stream: Optional[bool] = None ) -> Generator: retry_count = 0 if stream is None: stream = self.valves.STREAM while retry_count < self.valves.MAX_RETRIES: api_key = self.get_next_key() if not api_key: yield "错误：未配置API密钥" return try: client = OpenAI(base_url=self.valves.BASE_URL, api_key=api_key) response = client.chat.completions.create( model="deepseek-ai/DeepSeek-R1-Distill-Qwen-32B", messages=messages, temperature=self.valves.TEMPERATURE, max_tokens=self.valves.MAX_TOKENS, top_p=self.valves.TOP_P, stream=stream, ) if stream: for chunk in response: if chunk.choices[0].delta.content: yield chunk.choices[0].delta.content else: yield response.choices[0].message.content return except Exception as e: retry_count += 1 if retry_count >= self.valves.MAX_RETRIES: yield f"错误：所有重试都失败了 - {str(e)}" return time.sleep(1) def pipe(self, body: dict) -> Generator: try: messages = body.get("messages", []) stream = body.get("stream", None) if not messages: yield "错误：未提供消息内容" return for content in self.create_chat_completion(messages, stream): yield content except Exception as e: yield f"错误：{str(e)}"