pydify 2.2.1

A Python SDK for interacting with Dify API

Pydify

Pydify 是一个用于与 Dify API 交互的 Python 客户端库。

关于 Dify

Dify 是一个开源的 LLM 应用开发平台，提供直观的界面将 AI 工作流、RAG 管道、代理能力、模型管理、可观察性功能等结合在一起，使您能够快速从原型转向生产环境。

Dify 平台主要特点：

• 🤖 AI 工作流：支持构建和部署复杂的 AI 应用流程
• 📚 RAG 管道：内置检索增强生成能力，轻松连接到您的数据
• 🧠 代理能力：支持创建自动化智能代理
• 🔄 模型管理：集成多种 LLM 模型（OpenAI、Anthropic、Gemini、LLaMA 等）
• 📊 可观察性：应用性能和使用情况的监控与分析

目前，Dify 在 GitHub 上拥有超过 82k 星标，是 LLM 应用开发领域备受欢迎的开源项目。

简介

Pydify 提供了一个简洁、易用的接口，用于访问 Dify 平台的各种功能，包括：

• 💬 Chatbot 聊天助手：多轮对话、会话管理、消息反馈、多模态交互等
• 🔄 Workflow 工作流应用：执行工作流、处理流式响应、文件上传等
• 🤖 Agent 对话型应用：迭代式规划推理、自主工具调用，直至完成任务目标的智能助手
• 📝 Text Generation 文本生成应用：单轮文本生成，适合翻译、文章写作、总结等 AI 任务
• 🔧 DifySite 管理工具：自动化管理 Dify 平台的应用、API 密钥等

安装

pip install pydify

使用方法

ChatbotClient

ChatbotClient 用于与 Dify 的 Chatbot 应用交互。Chatbot 对话应用支持会话持久化，可将之前的聊天记录作为上下文进行回答，适用于聊天/客服 AI 等场景。

基本用法

from pydify import ChatbotClient

# 初始化客户端
client = ChatbotClient(
    api_key="your_dify_api_key",
    base_url="https://your-dify-instance.com/v1"  # 可选，默认使用 DIFY_BASE_URL 环境变量
)

# 获取应用信息
app_info = client.get_app_info()
print(f"应用名称: {app_info['name']}")

# 获取应用参数（开场白、推荐问题等）
params = client.get_parameters()
print(f"开场白: {params.get('opening_statement', '')}")

# 发送消息（阻塞模式）
response = client.send_message(
    query="你好，请介绍一下自己",
    user="user_123",  # 用户唯一标识
    response_mode="blocking"  # 阻塞模式，等待回复完成
)
print(f"AI回答: {response['answer']}")
print(f"会话ID: {response['conversation_id']}")

流式响应处理

# 流式模式发送消息
stream = client.send_message(
    query="请写一首关于春天的诗",
    user="user_123",
    response_mode="streaming"  # 流式模式，实时获取回复内容
)

# 处理流式响应
for chunk in stream:
    event = chunk.get("event")
    if event == "message":
        print(chunk.get("answer", ""), end="", flush=True)
    elif event == "message_end":
        print("\n\n消息回复完成！")
        if "metadata" in chunk and "usage" in chunk["metadata"]:
            usage = chunk["metadata"]["usage"]
            print(f"Token使用: {usage}")

多轮对话

# 第一轮对话
response1 = client.send_message(
    query="你能帮我写一个Python函数吗？",
    user="user_123",
    response_mode="blocking"
)
conversation_id = response1["conversation_id"]
print(f"AI: {response1['answer']}")

# 第二轮对话（基于之前的上下文）
response2 = client.send_message(
    query="这个函数需要实现什么功能？",
    user="user_123",
    conversation_id=conversation_id,  # 使用第一轮返回的会话ID
    response_mode="blocking"
)
print(f"AI: {response2['answer']}")

消息反馈（点赞/点踩）

# 对消息进行点赞
client.message_feedback(
    message_id="message_id_from_response",
    user="user_123",
    rating="like",  # "like"或"dislike"
    content="非常有用的回答，谢谢！"  # 可选
)

# 撤销反馈
client.message_feedback(
    message_id="message_id_from_response",
    user="user_123",
    rating=None  # 撤销反馈
)

会话管理

# 获取会话列表
conversations = client.get_conversations(
    user="user_123",
    limit=10  # 获取最近10条会话
)
for conv in conversations["data"]:
    print(f"会话ID: {conv['id']}, 名称: {conv['name']}")

# 获取会话历史消息
messages = client.get_messages(
    conversation_id="conversation_id",
    user="user_123",
    limit=20  # 获取最近20条消息
)
for msg in messages["data"]:
    sender = "用户" if "query" in msg else "AI"
    content = msg.get("query") if "query" in msg else msg.get("answer")
    print(f"{sender}: {content}")

# 重命名会话
client.rename_conversation(
    conversation_id="conversation_id",
    user="user_123",
    name="Python学习讨论"  # 手动指定名称
)
# 或自动生成名称
client.rename_conversation(
    conversation_id="conversation_id",
    user="user_123",
    auto_generate=True  # 自动生成名称
)

# 删除会话
client.delete_conversation(
    conversation_id="conversation_id",
    user="user_123"
)

文件与多模态功能

# 上传图片文件
file_result = client.upload_file(
    file_path="image.png",
    user="user_123"
)
file_id = file_result["id"]

# 发送带图片的消息
response = client.send_message(
    query="请描述这张图片的内容",
    user="user_123",
    files=[{
        "type": "image",
        "transfer_method": "local_file",
        "upload_file_id": file_id
    }],
    response_mode="blocking"
)
print(f"AI对图片的描述: {response['answer']}")

语音功能

# 语音转文字
speech_result = client.audio_to_text(
    file_path="speech.mp3",
    user="user_123"
)
print(f"识别出的文字: {speech_result['text']}")

# 文字转语音
# 从文本生成
audio_result = client.text_to_audio(
    user="user_123",
    text="这段文字将被转换为语音"
)

# 从消息ID生成
audio_result = client.text_to_audio(
    user="user_123",
    message_id="message_id_from_response"
)

其他功能

# 获取推荐问题
suggestions = client.get_suggested_questions(
    message_id="message_id_from_response",
    user="user_123"
)
for question in suggestions["data"]:
    print(f"推荐问题: {question}")

# 停止响应（针对流式模式）
client.stop_task(
    task_id="task_id_from_stream",
    user="user_123"
)

AgentClient

AgentClient 用于与 Dify 的 Agent 应用交互。Agent 是能够迭代式规划推理、自主工具调用，直至完成任务目标的智能助手。

基本用法

from pydify import AgentClient
from pydify.agent import AgentEvent  # 导入事件类型

# 初始化客户端
client = AgentClient(
    api_key="your_dify_api_key",
    base_url="https://your-dify-instance.com/v1"  # 可选，默认使用 DIFY_BASE_URL 环境变量
)

# 获取应用信息
app_info = client.get_app_info()
print(f"应用名称: {app_info['name']}")

# 获取应用参数
params = client.get_parameters()
print(f"开场白: {params.get('opening_statement', '')}")

# 发送消息（Agent应用只支持流式模式）
stream = client.send_message(
    query="帮我搜索最近一周的股市行情，并分析趋势",
    user="user_123"  # 用户唯一标识
)

# 简单处理流式响应
for chunk in stream:
    event = chunk.get("event")
    if event == "agent_message":
        print(chunk.get("answer", ""), end="", flush=True)
    elif event == "agent_thought":
        print(f"\n[思考] {chunk.get('thought')}")
        print(f"[工具] {chunk.get('tool')}")
        print(f"[结果] {chunk.get('observation')}")
    elif event == "message_end":
        print("\n\n回答完成")

流式响应处理

# 发送消息
stream = client.send_message(
    query="帮我分析最近的经济数据，预测下个季度的趋势",
    user="user_123"
)

# 简单的事件处理
for chunk in stream:
    event = chunk.get("event")
    if event == AgentEvent.AGENT_MESSAGE:
        print(chunk.get("answer", ""), end="", flush=True)
    elif event == AgentEvent.AGENT_THOUGHT:
        print(f"\n\n[思考] {chunk.get('thought')}")
        print(f"[工具] {chunk.get('tool')}")
        print(f"[观察] {chunk.get('observation')}\n")
    elif event == AgentEvent.MESSAGE_END:
        print("\n\n回答完成！")

多轮对话

# 第一轮对话
stream1 = client.send_message(
    query="帮我找出最适合初学者的编程语言",
    user="user_123"
)
# 处理第一轮对话的响应...
conversation_id = None
for chunk in stream1:
    if chunk.get("event") == "message_end":
        conversation_id = chunk.get("conversation_id")

# 第二轮对话（基于之前的上下文）
if conversation_id:
    stream2 = client.send_message(
        query="我想学习你推荐的第一种语言，有什么好的学习资源？",
        user="user_123",
        conversation_id=conversation_id  # 使用第一轮返回的会话ID
    )
    # 处理第二轮对话的响应...

会话管理

# 获取会话列表
conversations = client.get_conversations(
    user="user_123",
    limit=5  # 获取最近5条会话
)

# 获取会话历史消息（包含Agent思考过程）
messages = client.get_messages(
    conversation_id="conversation_id",
    user="user_123"
)

# 重命名会话
client.rename_conversation(
    conversation_id="conversation_id",
    user="user_123",
    name="编程学习助手"
)

# 删除会话
client.delete_conversation(
    conversation_id="conversation_id",
    user="user_123"
)

停止正在进行的任务

# 停止响应
client.stop_task(
    task_id="task_id_from_stream",
    user="user_123"
)

TextGenerationClient

TextGenerationClient 用于与 Dify 的 Text Generation 应用交互。Text Generation 应用无会话支持，适合用于翻译、文章写作、总结等 AI 任务。

基本用法

from pydify import TextGenerationClient
from pydify.text_generation import TextGenerationEvent  # 导入事件类型

# 初始化客户端
client = TextGenerationClient(
    api_key="your_dify_api_key",
    base_url="https://your-dify-instance.com/v1"  # 可选，默认使用 DIFY_BASE_URL 环境变量
)

# 获取应用信息
app_info = client.get_app_info()
print(f"应用名称: {app_info['name']}")

# 获取应用参数
params = client.get_parameters()
print(f"支持的功能: {params.get('features', [])}")

# 发送请求（阻塞模式）
response = client.completion(
    query="写一篇关于人工智能的短文，不少于300字",
    user="user_123",  # 用户唯一标识
    response_mode="blocking"  # 阻塞模式，等待生成完成
)
print(f"生成ID: {response['message_id']}")
print(f"生成内容: {response['answer']}")

流式响应处理

# 发送流式请求
stream = client.completion(
    query="请写一首关于春天的诗",
    user="user_123",
    response_mode="streaming"  # 流式模式，实时获取生成内容
)

# 处理流式响应
for chunk in stream:
    event = chunk.get("event")
    if event == TextGenerationEvent.MESSAGE:
        print(chunk.get("answer", ""), end="", flush=True)
    elif event == TextGenerationEvent.MESSAGE_END:
        print("\n\n生成完成！")
        if "metadata" in chunk and "usage" in chunk["metadata"]:
            usage = chunk["metadata"]["usage"]
            print(f"Token使用: {usage}")

使用自定义输入

# 假设应用定义了一些变量，如：主题(topic)、风格(style)、字数(word_count)
inputs = {
    "topic": "人工智能",        # 主题
    "style": "科普",           # 风格
    "word_count": 500          # 字数要求
}

# 发送请求，使用自定义inputs
response = client.completion(
    query="帮我写一篇文章",
    user="user_123",
    inputs=inputs,
    response_mode="blocking"
)

文件与多模态功能

# 上传图片文件
file_result = client.upload_file(
    file_path="image.png",
    user="user_123"
)
file_id = file_result["id"]

# 发送带图片的请求
response = client.completion(
    query="描述这张图片的内容",
    user="user_123",
    files=[{
        "type": "image",
        "transfer_method": "local_file",
        "upload_file_id": file_id
    }],
    response_mode="blocking"
)
print(f"图片描述: {response['answer']}")

停止生成

# 停止正在进行的生成任务
client.stop_task(
    task_id="task_id_from_stream",
    user="user_123"
)

WorkflowClient

WorkflowClient 用于与 Dify 的 Workflow 应用交互。Workflow 应用无会话支持，专注于执行预定义的工作流程。

基本用法

from pydify import WorkflowClient
from pydify.workflow import WorkflowEvent  # 导入事件类型

# 初始化客户端
client = WorkflowClient(
    api_key="your_dify_api_key",
    base_url="https://your-dify-instance.com/v1"  # 可选，默认使用 DIFY_BASE_URL 环境变量
)

# 获取应用信息
app_info = client.get_app_info()
print(f"应用名称: {app_info['name']}")

# 准备输入参数
inputs = {
    "prompt": "请写一首关于人工智能的诗",
}

# 执行工作流（阻塞模式）
result = client.run(
    inputs=inputs,
    user="user_123",  # 用户标识
    response_mode="blocking",
    timeout=30,  # 超时时间(秒)
)

print("工作流执行结果:")
print(result)

流式模式执行工作流

# 流式执行工作流
stream = client.run(
    inputs={"prompt": "分析当前市场趋势"},
    user="user_123",
    response_mode="streaming"
)

# 处理流式响应
for chunk in stream:
    event = chunk.get("event")

    if event == WorkflowEvent.WORKFLOW_STARTED:
        print(f"工作流开始：{chunk['data']['id']}")

    elif event == WorkflowEvent.NODE_STARTED:
        print(f"节点开始：{chunk['data']['title']}")

    elif event == WorkflowEvent.NODE_FINISHED:
        node_data = chunk['data']
        print(f"节点完成：{node_data.get('id')}")
        if 'outputs' in node_data:
            print(f"  输出：{node_data['outputs']}")

    elif event == WorkflowEvent.WORKFLOW_FINISHED:
        print(f"工作流完成：{chunk['data']['id']}")
        if 'outputs' in chunk['data']:
            print(f"最终结果：{chunk['data']['outputs']}")

文件上传与使用

# 上传文件
file_result = client.upload_file("document.pdf", "user_123")
file_id = file_result["id"]

# 使用文件执行工作流
result = client.run(
    inputs={"prompt": "总结这个文档的内容"},
    user="user_123",
    response_mode="blocking",
    files=[{
        "type": "document",
        "transfer_method": "local_file",
        "upload_file_id": file_id
    }]
)

停止正在执行的任务与获取日志

# 停止任务
client.stop_task(task_id="task_id_from_stream", user="user_123")

# 获取工作流执行日志
logs = client.get_logs(limit=10)
for log in logs["data"]:
    print(f"工作流 {log['id']} 状态: {log['workflow_run']['status']}")

DifySite 管理工具

DifySite 类提供了与 Dify 平台管理 API 的直接交互能力，用于自动化管理应用、API 密钥等后台任务。与其他客户端不同，DifySite 需要使用用户账户登录凭据而非 API 密钥。

初始化与登录

from pydify.site import DifySite, DifyAppMode

# 初始化DifySite实例（初始化时会自动登录并获取令牌）
site = DifySite(
    base_url="http://your-dify-instance.com",  # Dify平台地址
    email="[e-mail]",            # 登录邮箱
    password="your-password"                   # 登录密码
)
# 登录成功后，site.access_token和site.refresh_token已自动填充

应用管理

# 获取所有应用
apps = site.fetch_apps(limit=10)  # 获取前10个应用
for app in apps['data']:
    print(f"应用: {app['name']} (ID: {app['id']}, 模式: {app['mode']})")

# 获取所有应用（自动处理分页）
all_apps = site.fetch_all_apps()
print(f"共有{len(all_apps)}个应用")

# 获取特定应用详情
app_details = site.fetch_app("your-app-id")
print(f"应用名称: {app_details['name']}")
print(f"应用模式: {app_details['mode']}")

# 创建新应用
new_app = site.create_app(
    name="测试应用",
    description="通过API创建的测试应用",
    mode=DifyAppMode.CHAT  # 聊天助手应用
)
print(f"创建成功! 应用ID: {new_app['id']}")

# 删除应用
site.delete_app("your-app-id")

DSL 配置导入导出

# 导出应用DSL配置
dsl = site.fetch_app_dsl("your-app-id")

# 将DSL保存到文件
with open("app_backup.yaml", "w") as f:
    f.write(dsl)

# 导入应用DSL配置（创建新应用）
with open("app_backup.yaml", "r") as f:
    dsl_content = f.read()

imported_app = site.import_app_dsl(dsl_content)
print(f"导入成功! 新应用ID: {imported_app['id']}")

# 导入应用DSL配置（更新现有应用）
site.import_app_dsl(dsl_content, app_id="existing-app-id")

API 密钥管理

# 获取应用的API密钥列表
api_keys = site.fetch_app_api_keys("your-app-id")
print(f"应用共有{len(api_keys)}个API密钥")

# 创建新的API密钥
new_key = site.create_app_api_key("your-app-id")
print(f"新API密钥: {new_key['token']}")

# 删除API密钥
site.delete_app_api_key("your-app-id", "api-key-id")

其他功能

# 在浏览器中打开应用
site.jump_to_app("your-app-id", DifyAppMode.CHAT)

贡献

欢迎贡献代码、报告问题或提出建议！请查看 CONTRIBUTING.md 了解更多详情。

许可证

本项目采用 MIT 许可证。详细信息请参阅 LICENSE 文件。