""" title: DeepSeek Reasoning Processer author: Oaklight description: 在OpwenWebUI中显示DeepSeek R1模型的思维链 - 仅支持0.5.6及以上版本，同时兼顾reasoning_content与<think>模式 version: 1.2.13 licence: MIT """ import json import httpx import re from typing import AsyncGenerator, Callable, Awaitable 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="DeepSeek API的基础请求地址", ) DEEPSEEK_API_KEY: str = Field( default="", description="用于身份验证的DeepSeek API密钥，可从控制台获取", ) DEEPSEEK_API_MODEL: str = Field( default="deepseek-reasoner", description="API请求的模型名称，默认为 deepseek-reasoner", ) def __init__(self): self.valves = self.Valves() self.data_prefix = "data:" self.emitter = None def pipes(self): return [ { "id": self.valves.DEEPSEEK_API_MODEL, "name": self.valves.DEEPSEEK_API_MODEL, } ] async def pipe( self, body: dict, __event_emitter__: Callable[[dict], Awaitable[None]] = None ) -> AsyncGenerator[str, None]: """主处理管道""" thinking_state = {"thinking": -1} # 用于存储 thinking 状态 self.emitter = __event_emitter__ # 验证配置 if not self.valves.DEEPSEEK_API_KEY: yield json.dumps({"error": "未配置API密钥"}, ensure_ascii=False) return # 准备请求参数 headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } try: # 模型ID提取 model_id = body["model"].split(".", 1)[-1] payload = {**body, "model": model_id} # 处理消息以防止连续的相同角色 messages = payload["messages"] i = 0 while i < len(messages) - 1: if messages[i]["role"] == messages[i + 1]["role"]: # 插入具有替代角色的占位符消息 alternate_role = ( "assistant" if messages[i]["role"] == "user" else "user" ) messages.insert( i + 1, {"role": alternate_role, "content": "[Unfinished thinking]"}, ) i += 1 # 发起API请求 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: # 错误处理 if response.status_code != 200: error = await response.aread() yield self._format_error(response.status_code, error) return # 流式处理响应 async for line in response.aiter_lines(): if not line.startswith(self.data_prefix): continue # 截取 JSON 字符串 json_str = line[len(self.data_prefix) :] # 去除首尾空格后检查是否为结束标记 if json_str.strip() == "[DONE]": return try: data = json.loads(json_str) except json.JSONDecodeError as e: error_detail = f"解析失败 - 内容：{json_str}，原因：{e}" yield self._format_error("JSONDecodeError", error_detail) return choice = data.get("choices", [{}])[0] # 结束条件判断 if choice.get("finish_reason"): return # 状态机处理 state_output = await self._update_thinking_state( choice.get("delta", {}), thinking_state ) if state_output: yield state_output if state_output == "<think>": yield "\n" # 处理并立即发送内容 content = self._process_content(choice["delta"]) if content: # 处理思考状态标记 if content.startswith("<think>"): content = re.sub(r"^<think>", "", content) yield "<think>" await asyncio.sleep(0.1) yield "\n" elif content.startswith("</think>"): content = re.sub(r"^</think>", "", content) 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, thinking_state: dict) -> str: """更新思考状态机（简化版）""" state_output = "" if thinking_state["thinking"] == -1 and delta.get("reasoning_content"): thinking_state["thinking"] = 0 state_output = "<think>" 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: """处理并返回内容，支持 reasoning_content 和 content 中的 <think> 标签""" reasoning_content = delta.get("reasoning_content", "") content = delta.get("content", "") # Case 1: reasoning_content 存在 if reasoning_content: return reasoning_content # Case 2: content 以 <think> 开头 if content.startswith("<think>") or content.startswith("</think>"): return content # 默认情况：返回普通内容 return content def _emit_status(self, description: str, done: bool = False) -> Awaitable[None]: """发送状态更新""" if self.emitter: return self.emitter( { "type": "status", "data": { "description": description, "done": done, }, } ) return None def _format_error(self, status_code: int, error: bytes) -> str: if isinstance(error, str): error_str = error else: error_str = 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: err_type = type(e).__name__ return json.dumps({"error": f"{err_type}: {str(e)}"}, ensure_ascii=False)