我使用的是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密钥,即可创建我们自己的密钥,然后回到代码中将密钥填上即可,每一个独立的工作流或者聊天流应用都有一个自己的密钥哦,如果想访问自己账号下其他的应用,也需要重新创建新的密钥。

Logo

纵情码海钱塘涌,杭州开发者创新动! 属于杭州的开发者社区!致力于为杭州地区的开发者提供学习、合作和成长的机会;同时也为企业交流招聘提供舞台!

更多推荐