Python调用Dify Agent流式响应处理指南

避坑指南

Dify的部署流程在此不赘述。但有一个关键版本信息需要注意：当前最新的1.11版存在若干已知缺陷，为求稳定，我选择部署了更成熟的1.3.1版本。请注意，1.3.1版本不支持MCP服务。若你需要此功能，请考虑搭建1.6或更高版本。

项目背景

假设你已在Dify平台完成智能体的配置，下一步是将其集成至你的Python项目中进行调用。整个集成过程逻辑清晰，我们将分步完成。

集成步骤

1. 生成API密钥

首先，进入智能体的“访问API”页面，创建新的API密钥。生成后，请立即妥善保管。我在本地虚拟机测试环境中使用，密钥暴露风险可控，但在生产环境中必须执行严格的密钥管理策略。

2. 接口测试验证

在编写代码前，建议使用Postman或Apifox等工具手动发起请求，以验证接口连通性与参数格式。官方通常提供标准的请求示例，参照填写即可。

2.1 请求示例

以下是一个标准的cURL命令示例：

curl -X POST 'http://192.168.23.134/v1/chat-messages' \
  --header 'Authorization: Bearer {api_key}' \
  --header 'Content-Type: application/json' \
  --data-raw '{
    "inputs": {},
    "query": "What are the specs of the iPhone 13 Pro Max?",
    "response_mode": "streaming",
    "conversation_id": "",
    "user": "abc-123",
    "files": [{
        "type": "image",
        "transfer_method": "remote_url",
        "url": "https://cloud.dify.ai/logo/logo-site.png"
    }]
  }'

2.2 核心参数解析

为确保调用成功，必须理解以下几个核心参数：

关于files参数，目前主要支持图像类型（type: "image"），可通过远程URL（transfer_method: "remote_url"）或本地上传（local_file）两种方式传递。

2.3 响应模式深度解析

此处有一个技术要点：根据你选择的response_mode，返回的数据结构截然不同。

• 阻塞模式 (blocking)：直接返回一个完整的ChatCompletionResponse JSON对象，内含最终答案与相关元数据。
• 流式模式 (streaming)：返回一个SSE（服务器发送事件）流，数据以分块形式持续传输。在智能体模式下，这是唯一可用的方式，因为智能体的思考过程需要逐步呈现。

流式响应中包含多种事件类型，例如基础内容message、代表智能体推理步骤的agent_thought、标记对话结束的message_end，以及错误时的error事件。准确理解这些事件，是后续正确处理响应流的前提。

2.4 使用Apifox执行测试

通过工具实际调用接口，若配置无误，你将看到数据流被成功接收并逐步显示。返回的每一行数据，均可视为一个独立的“数据块”。请特别注意：在智能体模式下，阻塞式调用已被禁用，因此你只能处理流式数据，无法一次性获取完整响应。

3. 编写Python调用代码

进入实战开发阶段。Dify智能体返回的流式数据格式并未完全遵循OpenAI官方规范，这意味着你无法直接使用openai库或LangChain中的现成封装方法进行自动解析。需要自行编写处理逻辑来适配实际的响应格式。

以下是一段可直接使用的示例代码，核心是使用requests库处理流式响应：

import requests
import json

def call_dify_api_streaming(api_key, query, user, conversation_id=""):
    """
    调用 Dify API - 流式响应模式
    :param api_key: 你的API密钥
    :param query: 用户输入的问题
    :param user: 用户唯一标识
    :param conversation_id: (可选) 如需继续对话，传入之前的会话ID
    :return:
    """
    url = "http://192.168.23.134/v1/chat-messages"
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    payload = {
        "inputs": {},
        "query": query,
        "response_mode": "streaming",
        "conversation_id": conversation_id,
        "user": user,
    }

    response = requests.post(url, headers=headers, json=payload, stream=True)

    # 处理流式响应
    if response.status_code == 200:
        for line in response.iter_lines():
            if line:
                try:
                    line_str = line.decode('utf-8')
                    # 识别SSE格式的数据行
                    if line_str.startswith('data: '):
                        data_str = line_str[6:]  # 去掉 'data: ' 前缀
                        if data_str != '[DONE]':
                            data = json.loads(data_str)
                            # 根据事件类型提取内容
                            if 'answer' in data:
                                print(data['answer'], end='', flush=True)
                            elif 'message' in data:
                                print(data['message'], end='', flush=True)
                except Exception as e:
                    print(f"解析数据块时出错: {e}")
        print()  # 最后换行
    else:
        print(f"请求失败，状态码: {response.status_code}")
        print(response.text)

代码逻辑清晰：建立流式连接后，逐行读取响应，筛选出以data: 开头的有效数据行，随后解析JSON，并从其中提取answer或message字段的内容进行输出。

编写函数后，使用pytest进行简易测试验证：

def test_dify_agent():
    API_KEY = "你的实际API_KEY"
    USER = "你的用户标识"
    call_dify_api_streaming(api_key=API_KEY, query="你是谁", user=USER)

若一切顺利，你将在控制台看到智能体流式返回的自我介绍。至此，从创建密钥到代码集成的完整调用链路已成功打通。后续工作便是根据你的具体业务需求，对此基础函数进行定制化改造与封装。

来源：互联网