如何在用python部署的 Open WebUI 中显示 DeepSeek-R1 的思考过程

一、配置管道连接

• 在 管理员面板 -> 设置 -> 外部连接 中按 + 添加一个连接。

  点击加号输入阿里云百炼的地址与api密钥 此处输入网址与API-KEY

   

  • 网址输入https://dashscope.aliyuncs.com/compatible-mode/v1

  • 密钥输入你在阿里云百炼的api密钥

    二、创建函数

    • 在 管理员面板 -> 函数中按 + 添加一个函数。

      

      • 代码输入以下代码

        """
        title: DeepSeek R1
        author: zgccrui
        description: 在OpwenWebUI中显示DeepSeek R1模型的思维链 - 仅支持0.5.6及以上版本
        version: 1.2.10
        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",
                )
                DEEPSEEK_MODEL_DISPLAY_NAME: str = Field(
                    default="deepseek-reasoner-fix",
                    description="模型名称，默认为 deepseek-reasoner-fix",
                )
            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_MODEL_DISPLAY_NAME,
                    }
                ]
            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  str:
                """更新思考状态机（简化版）"""
                state_output = ""
                # 状态转换：未开始 -> 思考中
                if thinking_state["thinking"] == -1 and delta.get("reasoning_content"):
                    thinking_state["thinking"] = 0
                    state_output = ""
                # 状态转换：思考中 -> 已回答
                elif (
                    thinking_state["thinking"] == 0
                    and not delta.get("reasoning_content")
                    and delta.get("content")
                ):
                    thinking_state["thinking"] = 1
                    state_output = "\n\n\n"
                return state_output
            def _process_content(self, delta: dict) -> str:
                """直接返回处理后的内容"""
                return delta.get("reasoning_content", "") or delta.get("content", "")
            def _format_error(self, status_code: int, error: bytes) -> str:
                # 如果 error 已经是字符串，则无需 decode
                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 as e:
                    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)

         代码原创来自于如何在 Open WebUI 中显示 DeepSeek-R1 的思考过程 | HADB.MEhttps://hadb.me/posts/2025/display-deepseek-r1-thinking

        • 函数配置中，Api URL、Api Key、模型名字和上个教程一样，输入阿里云百炼对应的即可。

          

          • 启用该函数

            三、设置新模型

            • 前往 管理员设置 -> 模型，将新的 deepseek-r1-fix 设为启用，还可手动修改 Logo 图片等。
            • 然后就可以在对话中看到思考过程了，出现「正在思考」的时候，可以点击旁边的箭头展开思考过程。 可以看到已经实现出现思考过程的内容了