调用dify工作流api(包括文件输入,附代码)
介绍了如何使用Python代码调用Dify API进行文件上传和工作流执行。主要内容包括: 提供了完整的DifyAPIClient类实现,包含文件上传和运行工作流两个核心方法 详细说明了配置参数的修改方法: API_KEY:工作流API密钥 BASE_URL:Dify服务地址 FILE_PATH:待上传文件路径 WORKFLOW_INPUTS:工作流输入参数 演示了如何根据需要开启/关闭文件上传功
我使用的是docker部署的本地dify,运行之后即可将dify作为api接口进行调用。
直接上代码:
代码
import requests
import json
import os
class DifyAPIClient:
"""通用Dify API客户端"""
def __init__(self, api_key, base_url="http://127.0.0.1:70"):
self.api_key = api_key
self.base_url = base_url
self.headers = {
'Authorization': f'Bearer {api_key}',
'Content-Type': 'application/json'
}
def upload_file(self, file_path, user="user-123"):
"""上传文件到Dify"""
upload_url = f"{self.base_url}/v1/files/upload"
with open(file_path, 'rb') as f:
files = {'file': (os.path.basename(file_path), f, 'text/plain')}
data = {'user': user}
response = requests.post(upload_url, headers={'Authorization': f'Bearer {self.api_key}'},
files=files, data=data, timeout=60)
if response.status_code in [200, 201]:
result = response.json()
print(f"✅ 文件上传成功: {result['name']}")
return result['id']
else:
print(f"❌ 文件上传失败: {response.status_code}")
return None
def run_workflow(self, workflow_inputs, file_id=None, timeout=300):
"""执行工作流"""
workflow_url = f"{self.base_url}/v1/workflows/run"
# 构建请求数据
data = {
"inputs": workflow_inputs,
"response_mode": "blocking",
"user": "user-123"
}
# 如果有文件,添加到inputs中
if file_id:
data["inputs"]["file"] = {
"type": "document",
"transfer_method": "local_file",
"upload_file_id": file_id
}
print(f"🔄 调用工作流...")
print(f"输入参数: {json.dumps(data['inputs'], ensure_ascii=False, indent=2)}")
try:
response = requests.post(workflow_url, headers=self.headers,
json=data, timeout=timeout)
if response.status_code == 200:
result = response.json()
print("✅ 工作流执行成功!")
return result.get('answer', result)
else:
print(f"❌ 工作流执行失败: {response.status_code}")
print(f"错误信息: {response.text}")
return None
except requests.exceptions.Timeout:
print("❌ 请求超时")
return None
except Exception as e:
print(f"❌ 执行异常: {e}")
return None
def main():
"""主函数 - 在这里修改你的配置"""
# ========== 配置区域 ==========
API_KEY = "你自己工作流的api-key" # 你的API密钥
BASE_URL = "http://127.0.0.1:70" # Dify服务地址,根据你自己的地址和端口进行调整
FILE_PATH = "./3001.txt" # 文件路径
# 工作流输入参数 - 根据你的工作流配置修改
WORKFLOW_INPUTS = {
"day": 10, # 示例参数1
"query": "请为我总结文件内容" # 示例参数2
# 在这里添加更多参数...
}
# 是否使用文件
USE_FILE = True # True: 上传文件, False: 不使用文件
TIMEOUT = 300 # 超时时间(秒)
# ==============================
# 创建客户端
client = DifyAPIClient(API_KEY, BASE_URL)
# 上传文件(如果需要)
file_id = None
if USE_FILE and FILE_PATH:
if not os.path.exists(FILE_PATH):
print(f"❌ 文件不存在: {FILE_PATH}")
return
print(f"📤 上传文件: {FILE_PATH}")
file_id = client.upload_file(FILE_PATH)
if file_id is None:
return
# 执行工作流
print(f"🚀 执行工作流...")
result = client.run_workflow(WORKFLOW_INPUTS, file_id, TIMEOUT)
# 输出结果
if result:
print("\n" + "="*60)
print("执行结果:")
print("="*60)
print(result)
print("="*60)
else:
print("❌ 执行失败")
if __name__ == "__main__":
main()
运行之后的结果:
使用介绍:
我们找到主函数代码(大约在第77行)
1,这里将API_KEY修改为你自己的API密钥即可,关于如何获取dify的API_KEY请看本文最后一节,或者阅读dify官方文档。
2,DIfy服务地址也根据你自己的地址和端口进行修改,我是在本地docker运行的且端口号是70,所以我的BASE_URL就是 http://127.0.0.1:70
3,FILE_PATH为你自己想要上传的的文件的地址。因为我的这个代码和我的文件在同一个目录下
所以我的地址就填的是./3001.txt,你根据自己的需要进行修改即可。
WORKFLOW_INPUTS中填上你自己的工作流的输入,务必检查自己的工作流的输入和这里传递的参数类型和名称一致,不然回报错
下面两个参数USE_FILE和TIMEOUT分别表示是否使用文件,和连接超时时间
如果你的工作流中不需要使用文件,将USE_FILE设置为False即可
其他说明:
我这里使用的是工作流,所以我的地址为
如果你是聊天流
可以将这一
行的后缀修改为
v1/chat-messages
API-KEY的获取
我们进入我们的工作流的编辑页面,完成工作流的编排之后点击右上角的
点击发布,然后点击发布更新。然后点击左侧的访问API回跳转到如下页面
然后我们点击右上角的API密钥,即可创建我们自己的密钥,然后回到代码中将密钥填上即可,每一个独立的工作流或者聊天流应用都有一个自己的密钥哦,如果想访问自己账号下其他的应用,也需要重新创建新的密钥。
更多推荐
所有评论(0)