"""
title: DIFY Manifold Pipe
authors: xuzhougeng
author_url: https://github.com/xuzhougeng
funding_url: https://github.com/open-webui
version: 0.1.2
description: 该流程用于DIFY的API接口，用于与DIFY的API进行交互
"""
import os
import requests
import json
import time
from typing import List, Union, Generator, Iterator, Optional
from pydantic import BaseModel, Field
from open_webui.utils.misc import pop_system_message
from open_webui.config import UPLOAD_DIR
import base64
import tempfile
def get_file_extension(file_name: str) -> str:
"""
获取文件的扩展名
os.path.splitext(file_name) 返回一个元组, 第一个元素是文件名, 第二个元素是扩展名
"""
return os.path.splitext(file_name)[1].strip(".")
# 黑魔法, 从__event_emitter__中获取闭包变量
def get_closure_info(func):
# 获取函数的闭包变量
if hasattr(func, "__closure__") and func.__closure__:
for cell in func.__closure__:
if isinstance(cell.cell_contents, dict):
return cell.cell_contents
return None
class Pipe:
class Valves(BaseModel):
# 环境变量的设置
DIFY_BASE_URL: str = Field(default="https://api.dify.ai/v1")
DIFY_KEY: str = Field(default="")
FILE_SERVER: str = Field(default="")
def __init__(self):
self.type = "manifold"
self.id = "dify"
self.name = "dify/"
self.valves = self.Valves(**{"DIFY_KEY": os.getenv("DIFY_KEY", "")})
# 存储格式:
# {
# "chat_id_1": {
# "dify_conversation_id": "xxx",
# "messages": [{"chat_message_id_1": "dify_message_id_1"}, ...]
# }
# }
self.chat_message_mapping = {}
# 存储格式:
# {
# "chat_id_1": "gpt-3.5-turbo",
# "chat_id_2": "gpt-4"
# }
self.dify_chat_model = {}
# 存储格式:
# {
# "chat_id_1": {
# "file_id1":{
# "local_file_path": "/path/to/file1.pdf",
# "dify_file_id": "dify_file_123",
# "file_name": "file1.pdf"
# },
# "file_id2":{
# "local_file_path": "/path/to/file2.jpg",
# "dify_file_id": "dify_file_456",
# "file_name": "file2.jpg"
# }
# }
# }
self.dify_file_list = {}
self.data_cache_dir = "data/dify"
self.load_state()
def save_state(self):
"""持久化Dify相关的状态变量到文件"""
os.makedirs(self.data_cache_dir, exist_ok=True)
# chat_message_mapping.json
chat_mapping_file = os.path.join(
self.data_cache_dir, "chat_message_mapping.json"
)
with open(chat_mapping_file, "w", encoding="utf-8") as f:
json.dump(self.chat_message_mapping, f, ensure_ascii=False, indent=2)
# chat_model.json
chat_model_file = os.path.join(self.data_cache_dir, "chat_model.json")
with open(chat_model_file, "w", encoding="utf-8") as f:
json.dump(self.dify_chat_model, f, ensure_ascii=False, indent=2)
# file_list.json
file_list_file = os.path.join(self.data_cache_dir, "file_list.json")
with open(file_list_file, "w", encoding="utf-8") as f:
json.dump(self.dify_file_list, f, ensure_ascii=False, indent=2)
def load_state(self):
"""从文件加载Dify相关的状态变量"""
try:
# chat_message_mapping.json
chat_mapping_file = os.path.join(
self.data_cache_dir, "chat_message_mapping.json"
)
if os.path.exists(chat_mapping_file):
with open(chat_mapping_file, "r", encoding="utf-8") as f:
self.chat_message_mapping = json.load(f)
else:
self.chat_message_mapping = {}
# chat_model.json
chat_model_file = os.path.join(self.data_cache_dir, "chat_model.json")
if os.path.exists(chat_model_file):
with open(chat_model_file, "r", encoding="utf-8") as f:
self.dify_chat_model = json.load(f)
else:
self.dify_chat_model = {}
# file_list.json
file_list_file = os.path.join(self.data_cache_dir, "file_list.json")
if os.path.exists(file_list_file):
with open(file_list_file, "r", encoding="utf-8") as f:
self.dify_file_list = json.load(f)
else:
self.dify_file_list = {}
except Exception as e:
print(f"加载Dify状态文件失败: {e}")
# 加载失败时使用空字典
self.chat_message_mapping = {}
self.dify_chat_model = {}
self.dify_file_list = {}
def get_models(self):
"""
获取DIFY的模型列表
"""
return [
{"id": "claude3.5-haiku", "name": "claude3.5-haiku"},
{"id": "claude3.5-sonnet", "name": "claude3.5-sonnet"},
{"id": "gpt4o", "name": "gpt4o"},
{"id": "gpt4o-mini", "name": "gpt4o-mini"},
]
def upload_file(self, user_id: str, file_path: str, mime_type: str) -> str:
"""
这个函数负责上传文件到DIFY的服务器, 并返回文件的ID
"""
import requests
url = f"{self.valves.DIFY_BASE_URL}/files/upload"
headers = {
"Authorization": f"Bearer {self.valves.DIFY_KEY}",
}
file_name = os.path.basename(file_path)
files = {
# 文件字段：(文件名, 文件对象, MIME类型)
"file": (file_name, open(file_path, "rb"), mime_type),
# 普通表单字段：(None, 值)
"user": (None, user_id),
}
response = requests.post(url, headers=headers, files=files)
file_id = response.json()["id"]
# Optional: print response
return file_id
def is_doc_file(self, file_path: str) -> bool:
"""
判断文件是否是文档文件
'TXT', 'MD', 'MARKDOWN', 'PDF', 'HTML', 'XLSX', 'XLS', 'DOCX', 'CSV', 'EML', 'MSG', 'PPTX', 'PPT', 'XML', 'EPUB'
"""
file_extension = get_file_extension(file_path).upper()
if file_extension in [
"PDF",
"XLSX",
"XLS",
"DOCX",
"EML",
"MSG",
"PPTX",
"PPT",
"XML",
"EPUB",
]:
return True
return False
def is_text_file(self, mime_type: str) -> bool:
"""
判断文件是否是文本文件
"""
if "text" in mime_type:
return True
return False
def is_audio_file(self, file_path: str) -> bool:
"""
判断文件是否是音频文件
'MP3', 'M4A', 'WAV', 'WEBM', 'AMR'.
"""
if get_file_extension(file_path).upper() in [
"MP3",
"M4A",
"WAV",
"WEBM",
"AMR",
]:
return True
return False
def is_video_file(self, file_path: str) -> bool:
"""
判断文件是否是视频文件
'MP4', 'MOV', 'MPEG', 'MPGA'
"""
if get_file_extension(file_path).upper() in ["MP4", "MOV", "MPEG", "MPGA"]:
return True
return False
def upload_text_file(self, user_id: str, file_path: str) -> str:
"""
上传文本文件到服务器，第一行添加文件名
支持类型: 纯文本文件
Args:
file_path: 文本文件路径
user_id: 用户ID
Returns:
str: 上传后的文件ID
"""
try:
# 获取原始文件名
filename = os.path.basename(file_path)
# 读取原文件内容
with open(file_path, "r", encoding="utf-8") as f:
content = f.read()
# 创建带文件名标记的新内容
new_content = f"#{filename}\n{content}"
# 创建临时文件
with tempfile.NamedTemporaryFile(
delete=False, suffix=".txt", mode="w", encoding="utf-8"
) as tmp_file:
tmp_file.write(new_content)
temp_file_path = tmp_file.name
try:
# 上传文件
file_id = self.upload_file(user_id, temp_file_path, "text/plain")
return file_id
finally:
# 清理临时文件
os.remove(temp_file_path)
except UnicodeDecodeError:
raise ValueError("文件编码不是UTF-8格式")
except Exception as e:
raise ValueError(f"处理文本文件失败: {str(e)}")
def upload_images(self, image_data_base64: str, user_id: str) -> str:
"""
上传 base64 编码的图片到 DIFY 服务器，返回图片路径
支持类型: 'JPG', 'JPEG', 'PNG', 'GIF', 'WEBP', 'SVG'
"""
try:
# Remove the data URL scheme prefix if present
if image_data_base64.startswith("data:"):
# Extract the base64 data after the comma
image_data_base64 = image_data_base64.split(",", 1)[1]
# 解码 base64 图像数据
image_data = base64.b64decode(image_data_base64)
# Create and save temporary file
with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as tmp_file:
tmp_file.write(image_data)
temp_file_path = tmp_file.name
try:
file_id = self.upload_file(user_id, temp_file_path, "image/png")
finally:
os.remove(temp_file_path)
return file_id
except Exception as e:
raise ValueError(f"Failed to process base64 image data: {str(e)}")
def pipes(self) -> List[dict]:
return self.get_models()
def pipe(self, body: dict, __event_emitter__: dict = None, __user__: Optional[dict] = None, __task__=None) -> Union[str, Generator, Iterator]:
"""处理对话请求"""
# 获取模型名称
model_name = body["model"][body["model"].find(".") + 1 :]
# 处理特殊任务
if __task__ is not None:
if __task__ == "title_generation":
return model_name
elif __task__ == "tags_generation":
return f'{{"tags":[{model_name}]}}'
# 获取当前用户
current_user = __user__["email"]
# 处理系统消息和普通消息
system_message, messages = pop_system_message(body["messages"])
print(f"system_message:{system_message}")
print(f"messages:{messages}, {len(messages)}")
# 从event_emitter获取chat_id和message_id
cell_contents = get_closure_info(__event_emitter__)
chat_id = cell_contents["chat_id"]
message_id = cell_contents["message_id"]
# 处理对话模型和上下文
parent_message_id = None
# 在pipe函数中修改对话历史的处理逻辑
if len(messages) == 1:
# 新对话逻辑保持不变
self.dify_chat_model[chat_id] = model_name
self.chat_message_mapping[chat_id] = {
"dify_conversation_id": "",
"messages": []
}
self.dify_file_list[chat_id] = {}
else:
# 检查是否存在历史记录
if chat_id in self.chat_message_mapping:
# 首先验证模型
if chat_id in self.dify_chat_model:
if self.dify_chat_model[chat_id] != model_name:
raise ValueError(f"Cannot change model in an existing conversation. This conversation was started with {self.dify_chat_model[chat_id]}")
else:
# 如果somehow没有记录模型（异常情况），记录当前模型
self.dify_chat_model[chat_id] = model_name
chat_history = self.chat_message_mapping[chat_id]["messages"]
current_msg_index = len(messages) - 1 # 当前消息的索引
# 如果不是第一条消息，获取前一条消息的dify_id作为parent
if current_msg_index > 0 and len(chat_history) >= current_msg_index:
previous_msg = chat_history[current_msg_index - 1]
parent_message_id = list(previous_msg.values())[0]
# 关键修改：截断当前位置之后的消息历史
self.chat_message_mapping[chat_id]["messages"] = chat_history[:current_msg_index]
# 获取最后一条消息作为query
message = messages[-1]
query = ""
inputs = {"model": model_name}
file_list = []
# 处理消息内容
if isinstance(message.get("content"), list):
for item in message["content"]:
if item["type"] == "text":
query += item["text"]
if item["type"] == "image_url":
upload_file_id = self.upload_images(item["image_url"]["url"], current_user)
upload_file_dict = {
"type": "image",
"transfer_method": "local_file",
"url": "",
"upload_file_id": upload_file_id
}
file_list.append(upload_file_dict)
else:
query = message.get("content", "")
# 处理文件上传
if "upload_files" in body:
for file in body["upload_files"]:
if file["type"] != "file":
continue
file_id = file["id"]
if chat_id in self.dify_file_list and file_id in self.dify_file_list[chat_id]:
file_list.append(self.dify_file_list[chat_id][file_id])
continue
# 获取文件信息并上传
if "collection_name" in file:
file_path = os.path.join(UPLOAD_DIR, file["file"]["filename"])
else:
file_path = file["file"]["path"]
file_mime_type = file["file"]["meta"]["content_type"]
upload_file_dict = {
"transfer_method": "local_file",
"url": "",
}
# 处理不同类型的文件
if self.is_doc_file(file_path):
upload_file_id = self.upload_file(current_user, file_path, file_mime_type)
upload_file_dict.update({"type": "document", "upload_file_id": upload_file_id})
elif self.is_text_file(file_mime_type):
upload_file_id = self.upload_text_file(current_user, file_path)
upload_file_dict.update({"type": "document", "upload_file_id": upload_file_id})
elif self.is_audio_file(file_path):
upload_file_id = self.upload_file(current_user, file_path, file_mime_type)
upload_file_dict.update({"type": "audio", "upload_file_id": upload_file_id})
elif self.is_video_file(file_path):
upload_file_id = self.upload_file(current_user, file_path, file_mime_type)
upload_file_dict.update({"type": "video", "upload_file_id": upload_file_id})
else:
raise ValueError(f"Unsupported file type: {file_path}")
file_list.append(upload_file_dict)
if chat_id not in self.dify_file_list:
self.dify_file_list[chat_id] = {}
self.dify_file_list[chat_id][file_id] = upload_file_dict
# 准备请求载荷
payload = {
"inputs": inputs,
"parent_message_id": parent_message_id,
"query": query,
"response_mode": "streaming" if body.get("stream", False) else "blocking",
"conversation_id": self.chat_message_mapping[chat_id].get("dify_conversation_id", ""),
"user": current_user,
"files": file_list,
}
# 设置请求头
headers = {
"Authorization": f"Bearer {self.valves.DIFY_KEY}",
"content-type": "application/json",
}
url = f"{self.valves.DIFY_BASE_URL}/chat-messages"
try:
if body.get("stream", False):
return self.stream_response(url, headers, payload, chat_id, message_id)
else:
return self.non_stream_response(url, headers, payload, chat_id, message_id)
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"Error: {e}"
def stream_response(self, url, headers, payload, chat_id, message_id):
"""处理流式响应"""
try:
with requests.post(url, headers=headers, json=payload, stream=True, timeout=(3.05, 60)) as response:
if response.status_code != 200:
raise Exception(f"HTTP Error {response.status_code}: {response.text}")
for line in response.iter_lines():
if line:
line = line.decode("utf-8")
if line.startswith("data: "):
try:
data = json.loads(line[6:])
event = data.get("event")
if event == "message":
# 处理普通文本消息
yield data.get("answer", "")
elif event == "message_file":
# 处理文件（图片）消息
pass
elif event == "message_end":
# 保存会话和消息ID映射
dify_conversation_id = data.get("conversation_id", "")
dify_message_id = data.get("message_id", "")
self.chat_message_mapping[chat_id]["dify_conversation_id"] = dify_conversation_id
self.chat_message_mapping[chat_id]["messages"].append({message_id: dify_message_id})
# 保存状态
self.save_state()
break
elif event == "error":
# 处理错误
error_msg = f"Error {data.get('status')}: {data.get('message')} ({data.get('code')})"
yield f"Error: {error_msg}"
break
time.sleep(0.01)
except json.JSONDecodeError:
print(f"Failed to parse JSON: {line}")
except KeyError as e:
print(f"Unexpected data structure: {e}")
print(f"Full data: {data}")
except requests.exceptions.RequestException as e:
print(f"Request failed: {e}")
yield f"Error: Request failed: {e}"
except Exception as e:
print(f"General error in stream_response method: {e}")
yield f"Error: {e}"
def non_stream_response(self, url, headers, payload, chat_id, message_id):
"""处理非流式响应"""
try:
response = requests.post(url, headers=headers, json=payload, timeout=(3.05, 60))
if response.status_code != 200:
raise Exception(f"HTTP Error {response.status_code}: {response.text}")
res = response.json()
# 保存会话和消息ID映射
dify_conversation_id = res.get("conversation_id", "")
dify_message_id = res.get("message_id", "")
self.chat_message_mapping[chat_id]["dify_conversation_id"] = dify_conversation_id
self.chat_message_mapping[chat_id]["messages"].append({message_id: dify_message_id})
# 保存状态
self.save_state()
return res.get("answer", "")
except requests.exceptions.RequestException as e:
print(f"Failed non-stream request: {e}")
return f"Error: {e}"