""" title: Aliyun Manifold Pipe Author: Doubao Orginal_authors: LB7666 author_url: https://github.com/acking-you funding_url: https://github.com/open-webui version: 0.2 required_open_webui_version: 0.5.6 license: MIT """ import os import requests import json import time import re from typing import List, Union, Generator, Iterator from pydantic import BaseModel, Field from open_webui.utils.misc import pop_system_message class Pipe: class Valves(BaseModel): API_KEY: str = Field(default="") def __init__(self): self.type = "manifold" self.id = "aliyun" self.name = "aliyun/" self.valves = self.Valves(**{"API_KEY": os.getenv("API_KEY", "")}) self.data_prefix = "data:" self.thinking_state = {"thinking": -1} # 用于存储 thinking 状态 def get_deepseek_models(self): return [ {"id": "deepseek-r1", "name": "deepseek-r1"}, {"id": "deepseek-v3", "name": "deepseek-v3"}, # { # "id": "deepseek-r1-distill-qwen-1.5b", # "name": "deepseek-r1-distill-qwen-1.5b", # }, # { # "id": "deepseek-r1-distill-qwen-7b", # "name": "deepseek-r1-distill-qwen-7b", # }, # { # "id": "deepseek-r1-distill-llama-8b", # "name": "deepseek-r1-distill-llama-8b", # }, # { # "id": "deepseek-r1-distill-qwen-14b", # "name": "deepseek-r1-distill-qwen-14b", # }, # { # "id": "deepseek-r1-distill-qwen-32b", # "name": "deepseek-r1-distill-qwen-32b", # }, # { # "id": "deepseek-r1-distill-llama-70b", # "name": "deepseek-r1-distill-llama-70b", # }, ] def pipes(self) -> List[dict]: return self.get_deepseek_models() def pipe(self, body: dict) -> Union[str, Generator, Iterator]: system_message, messages = pop_system_message(body["messages"]) processed_messages = [] for message in messages: processed_content = "" if isinstance(message.get("content"), list): for item in message["content"]: if item["type"] == "text": processed_content = item["text"] else: processed_content = message.get("content", "") processed_messages.append( {"role": message["role"], "content": processed_content} ) # Ensure the system_message is coerced to a string payload = { "model": body["model"][body["model"].find(".") + 1 :], "messages": processed_messages, "stream": body.get("stream", False), } headers = {} headers["Authorization"] = f"Bearer {self.valves.API_KEY}" headers["Content-Type"] = "application/json" url = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" try: if body.get("stream", False): return self.stream_response(url, headers, payload) else: return self.non_stream_response(url, headers, payload) except requests.exceptions.RequestException as e: print(f"Request failed: {e}") return f"Error: Request failed: {e}" except Exception as e: print(f"Error in pipe method: {e}") return f"Error1: {e}" def stream_response(self, url, headers, payload): # Remove unnecessary fields from the payload unnecessary_fields = ["user", "chat_id", "title"] for field in unnecessary_fields: payload.pop(field, None) try: response = requests.post( url=url, json=payload, headers=headers, stream=True, ) if response.status_code != 200: raise Exception( f"HTTP Error {payload} {response.status_code}: {response.text}" ) # Process the streamed response thinking_content = "" # 用于存储思维链内容 normal_content = "" # 用于存储正常回答内容 in_thinking = False # 标记是否处于思维链输出状态 for chunk in response.iter_content(chunk_size=None): if chunk: decoded_chunk = chunk.decode("utf-8") lines = decoded_chunk.splitlines() for line in lines: if line.startswith(self.data_prefix): json_str = line[len(self.data_prefix):] if json_str.strip() != "[DONE]": try: data = json.loads(json_str) choice = data.get("choices", [{}])[0] state_output = self._update_thinking_state( choice.get("delta", {}) ) if state_output: if state_output == "<think>": in_thinking = True # 输出思维链开始标记 yield '<details type="thinking">\n' yield "<summary>思维链内容</summary>\n" yield state_output elif state_output == "</think>": in_thinking = False # 输出思维链内容 if thinking_content: yield thinking_content # 输出思维链结束标记 yield state_output yield "</details>\n" content = self._process_content(choice["delta"]) if content: if in_thinking: thinking_content += content yield content # 实时输出思维链内容 else: normal_content += content yield content except json.JSONDecodeError as e: continue else: normal_content += line yield line # Delay to avoid overwhelming the client time.sleep(0.01) # 如果最后还有思维链内容未输出，输出它 if thinking_content: yield thinking_content except requests.exceptions.RequestException as e: print(f"Stream request failed: {e}") yield f"Error: {e}" except Exception as e: print(f"Error in stream_response method: {e}") yield f"Error: {e}" def non_stream_response(self, url, headers, payload): try: response = requests.post( url=url, headers=headers, json=payload, timeout=(3.05, 60) ) if response.status_code != 200: raise Exception(f"HTTP Error: {response.status_code}: {response.text}") return response.json() except requests.exceptions.RequestException as e: print(f"Failed non-stream request: {e}") return f"Error5: {response.content}" except Exception as e: print(f"Error in non_stream_response method: {e}") return f"Error6: {e}" def _update_thinking_state(self, delta: dict) -> str: """更新思考状态机（简化版）""" state_output = "" # 状态转换：未开始 -> 思考中 if self.thinking_state["thinking"] == -1 and delta.get("reasoning_content"): self.thinking_state["thinking"] = 0 state_output = "<think>" # 状态转换：思考中 -> 已回答 elif ( self.thinking_state["thinking"] == 0 and not delta.get("reasoning_content") and delta.get("content") ): self.thinking_state["thinking"] = 1 state_output = "</think>" return state_output def _process_content(self, delta: dict) -> str: """直接返回处理后的内容""" return delta.get("reasoning_content", "") or delta.get("content", "")