Dify Python调用API
前期准备
在调用API之前得先拿到API KEY，具体操作如下：
文件上传
如果你的工作流的参数涉及到文件上传以及图片上传的(音频、视频也类似)，需要先调用upload方法拿到file_id,具体代码如下：
import os
import requests
import logging# 假设 logger_info 已定义好
logger_info = logging.getLogger(__name__)class YourUploaderClass:def __init__(self, base_url, your_api_key):self.base_url = base_urlself.api_key = your_api_key #随意一条工作流的API_Keydef upload_file(self, file_path):"""文件上传，支持文档和图片:param file_path: 本地文件路径:return: 返回上传成功后的文件 ID，失败返回 None"""headers = {"Authorization": f"Bearer {self.api_key}"}# 获取文件扩展名并确定 MIME 类型file_extension = os.path.splitext(file_path)[1].lower()# 定义文档类型的 MIME 映射document_mime_map = {'.txt': 'text/plain','.md': 'text/markdown','.markdown': 'text/markdown','.pdf': 'application/pdf','.html': 'text/html','.xlsx': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet','.xls': 'application/vnd.ms-excel','.docx': 'application/vnd.openxmlformats-officedocument.wordprocessingml.document','.csv': 'text/csv','.eml': 'message/rfc822','.msg': 'application/vnd.ms-outlook','.pptx': 'application/vnd.openxmlformats-officedocument.presentationml.presentation','.ppt': 'application/vnd.ms-powerpoint','.xml': 'application/xml','.epub': 'application/epub+zip'}# 定义图片类型的扩展名映射到对应的 MIME 类型image_ext_to_mime = {'.png': 'image/png','.jpg': 'image/jpeg','.jpeg': 'image/jpeg','.webp': 'image/webp','.gif': 'image/gif'}# 判断是文档还是图片mime_type = document_mime_map.get(file_extension)if not mime_type:mime_type = image_ext_to_mime.get(file_extension)# 如果文件类型不支持，返回 Noneif mime_type is None:return Nonecurrent_file_path = os.path.abspath(__file__)dir = os.path.dirname(os.path.dirname(os.path.dirname(current_file_path)))path = os.path.join(dir, file_path)try:with open(path, "rb") as f:files = {"file": (os.path.basename(file_path), f, mime_type)}data = {"user": "abc_user"}upload_url = f"{self.base_url}/files/upload"response = requests.post(upload_url, headers=headers, files=files, data=data)logger_info.info(f"dify upload_file 请求结果：{response.json()}")if response.status_code == 201:result = response.json()logger_info.info(f"dify upload_file 成功：{result}")return result.get("id")else:return Noneexcept requests.exceptions.RequestException as e:logging.error(f"dify upload_file 请求出错：{e}")return Noneexcept Exception as e:logging.error(f"上传文件时发生异常：{e}")return None流式输出 streaming
基于 SSE（Server-Sent Events）实现类似打字机输出方式的流式返回。注意，以下代码展示的是chatflow模式，所以参数inputs为空，如果是工作流，入参放在inputs即可，形式是：
"inputs": {"outline": outline
}
def user_intent(self, question, upload_file_id=""):"""第一步：用户意图分析:param question::param upload_file_id::return:"""headers = {"Authorization": f"Bearer {self.intent_api_key}","Content-Type": "application/json"}data = {"inputs": {},"query": question,"response_mode": "streaming", # 根据需求选择 streaming 或 blocking"user": "abc_user" # 与上传文件时的用户标识一致}if upload_file_id:data["inputs"]["file"] = [{"type": "document", # 根据实际文件类型修改"transfer_method": "local_file","upload_file_id": upload_file_id}]workflow_url = f"{self.base_url}/chat-messages"try:response = requests.post(workflow_url, headers=headers, json=data)if response.status_code == 200:for line in response.iter_lines():if line: # 过滤掉keep-alive新行# 解码为字符串line_str = line.decode('utf-8')# 确保是有效的 JSON 行if line_str.startswith("data:"):json_str = line_str[5:].strip() # 去掉前缀并清理空格if json_str:json_data = json.loads(json_str)#这里不一定是message，具体你得打印json_data你才知道if json_data.get("event") == "message": yield json_datalogger_info.info(f"user_intent 请求结果：{response.json()}")return response.json()else:logger.error(f"user_intent 请求出错：{response.json()}")return {}except Exception as e:logger.error(f"user_intent 请求出错：{e}")return {}
阻塞模式 blocking
阻塞模式，等待执行完毕后返回结果。（请求若流程较长可能会被中断）。 由于 Cloudflare 限制，请求会在 100 秒超时无返回后中断。注意：以下代码是chatflow模式，不是工作流，但是代码是一样的，只不过取值不一样
def judge_steps(self, question):"""判断步骤:param question::return:"""headers = {"Authorization": f"Bearer {self.judge_steps_api_key}","Content-Type": "application/json"}data = {"inputs": {},"query": question,"response_mode": "blocking", # 根据需求选择 streaming 或 blocking"user": "abc_user" # 与上传文件时的用户标识一致}#注意，我这里用的是CHATFLOW，不是工作流workflow_url = f"{self.base_url}/chat-messages"try:response = requests.post(workflow_url, headers=headers, json=data)if response.status_code == 200:#由于不是工作流，所以取出来的不一定是["answer"],方法都是一样的，到时候自己打印response.json()出来就知道了answer = response.json()["answer"]if answer:logger_info.info(f"dify judge_steps 请求结果：{response.json()}")return answerelse:return ""else:logging.error(f"judge_steps 请求出错：{response.json()}")return 0except Exception as e:print(e)logging.error(f"judge_steps 请求出错：{e}")return 0