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="your-email@example.com",            # 登录邮箱
    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 文件。

Name		Name	Last commit message	Last commit date
Latest commit History 37 Commits
.github/workflows		.github/workflows
dify-panel		dify-panel
examples		examples
pydify		pydify
tests		tests
.env.examples		.env.examples
.gitignore		.gitignore
CHANGELOG.md		CHANGELOG.md
DIFY站点API文档.md		DIFY站点API文档.md
LICENSE		LICENSE
MANIFEST.in		MANIFEST.in
PUBLISH_GUIDE.md		PUBLISH_GUIDE.md
README.md		README.md
RELEASE.md		RELEASE.md
build_and_test.sh		build_and_test.sh
example.txt		example.txt
pyproject.toml		pyproject.toml
requirements.txt		requirements.txt
setup.py		setup.py
test_image.png		test_image.png
test_install.py		test_install.py

License

jayscoder/pydify

Folders and files

Latest commit

History

Repository files navigation

Pydify

关于 Dify

简介

安装

使用方法

ChatbotClient

基本用法

流式响应处理

多轮对话

消息反馈（点赞/点踩）

会话管理

文件与多模态功能

语音功能

其他功能

AgentClient

基本用法

流式响应处理

多轮对话

会话管理

停止正在进行的任务

TextGenerationClient

基本用法

流式响应处理

使用自定义输入

文件与多模态功能

停止生成

WorkflowClient

基本用法

流式模式执行工作流

文件上传与使用

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

DifySite 管理工具

初始化与登录

应用管理

DSL 配置导入导出

API 密钥管理

其他功能

贡献

许可证

About

Resources

License

Uh oh!

Stars

Watchers

Forks

Releases

Packages 0

Uh oh!

Languages

Packages