""" title: DeepSeek R1 author: zgccruiG reviser 1: KinglyWayne reviser 2: drudream reviser 3: WolfPan 2025-02-16 description: 在OpwenWebUI中显示DeepSeek R1模型的思维链 - 仅支持0.5.6及以上版本, 修复deepseek r1生成内容缺少开头<think>标签的问题，修复联网搜索和生成标题出错的问题，通过添加处理非流式传输时不添加<think>标签来实现。 version: 1.3.4 licence: MIT """ import asyncio import json import traceback from typing import AsyncGenerator, Callable, Awaitable import httpx from pydantic import BaseModel, Field THINKING_STATE_BEFORE = 1 THINKING_STATE_INNER = 2 THINKING_STATE_AFTER = 3 THINKING_TAG_START = "<think>" THINKING_TAG_END = "</think>" 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 ", ) LOCAL_MODEL_NAME: str = Field( default="deepseek-reasoner", description="本地的模型名称，默认为 deepseek-reasoner ", ) def __init__(self): self.valves = self.Valves() self.data_prefix = "data:" def pipes(self): return [ { "id": self.valves.LOCAL_MODEL_NAME, "name": self.valves.DEEPSEEK_API_MODEL, } ] async def process_payload(self, body: dict) -> AsyncGenerator[str, None]: """核心处理函数""" model_id = body["model"].split(".", 1)[-1] payload = {**body, "model": model_id} url = f"{self.valves.DEEPSEEK_API_BASE_URL}/chat/completions" headers = { "Authorization": f"Bearer {self.valves.DEEPSEEK_API_KEY}", "Content-Type": "application/json", } params = dict(url=url, json=payload, headers=headers, timeout=300) # 处理消息以防止连续的相同角色 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 async with httpx.AsyncClient(http2=True) as client: if payload.get("stream", False): handler = self._handle_stream else: handler = self._handle_non_stream async for item in handler(client, params): yield item async def start_think(self): for part in (THINKING_TAG_START, "\n"): await asyncio.sleep(0.1) yield part async def end_think(self): for part in ("\n", THINKING_TAG_END, "\n", "\n"): await asyncio.sleep(0.1) yield part async def _handle_stream( self, client: httpx.AsyncClient, params: dict ) -> AsyncGenerator[str, None]: """处理流式响应""" async with client.stream("POST", **params) as response: if response.status_code != 200: error = await response.aread() yield self._format_error(response.status_code, error) return think_tag_added = False thinking_state = THINKING_STATE_BEFORE async for line in response.aiter_lines(): if not line.startswith(self.data_prefix): continue if line.strip() == "data: [DONE]": return try: data = json.loads(line[len(self.data_prefix) :]) except json.JSONDecodeError as e: error_detail = f"JSONDecodeError, {e}: {line}" yield self._format_error(500, error_detail) return choices = data.get("choices", []) or [{}] choice = choices[0] if choice.get("finish_reason", None): return delta = choice.get("delta", {}) reasoning: str = delta.get("reasoning_content", "") content: str = delta.get("content", "") if thinking_state == THINKING_STATE_BEFORE: if reasoning: reasoning = reasoning.removeprefix(THINKING_TAG_START) thinking_state = THINKING_STATE_INNER async for i in self.start_think(): yield i elif ( thinking_state == THINKING_STATE_INNER and not reasoning and content ): thinking_state = THINKING_STATE_AFTER content.removeprefix(THINKING_TAG_END) async for i in self.end_think(): yield i yield reasoning or content or "" async def _handle_non_stream( self, client: httpx.AsyncClient, params: dict ) -> AsyncGenerator[str, None]: """处理非流式响应""" response = await client.post(**params) if response.status_code != 200: error = response.content yield self._format_error(response.status_code, error) return data = response.json() content = data.get("choices", [{}])[0].get("message", {}).get("content", "") yield content async def pipe( self, body: dict, __event_emitter__: Callable[[dict], Awaitable[None]] = None ) -> AsyncGenerator[str, None]: """主处理管道（已移除缓冲）""" try: if not self.valves.DEEPSEEK_API_KEY: yield "[Error] API key not configured." return async for result in self.process_payload(body): yield result except httpx.ReadTimeout: yield "[Error] Request to DeepSeek service timed out, check network connection or try later" except httpx.ConnectError as e: error_msg = [ "[Error] Network connection failed between services. Possible causes:", "1. Network firewall / connectivity issues", "2. Incorrect API endpoint configuration", "3. DeepSeek service temporarily unavailable", ] yield "\n".join(error_msg) except Exception as e: tb_str = "".join(traceback.TracebackException.from_exception(e).format()) yield f"{type(e).__name__}: {str(e)}\n{tb_str}" def _format_error(self, status_code: int, error: bytes | str) -> str: if isinstance(error, bytes): error = error.decode(errors="ignore") try: err_msg = json.loads(error).get("message", error)[:200] except Exception as e: err_msg = error[:200] return f"HTTP {status_code}: {err_msg}"