Python调用Dify Agent，并处理流式响应

避坑

dify的搭建这里不多说了，需要注意的是目前的最新版本1.11有些bug，为了稳定性我使用的是老版本1.3.1，1.3.1不能使用MCP服务，如果想使用MCP服务可以搭建1.6以上的版本

背景

在 dify 中自定义了一个 agent，需要在python项目中集成并调用

步骤

1、创建API key

在 agent 的"访问API"页面中先创建一个API key并把它保存好，由于我是在本地虚拟机中搭建的，所以暴露了也无所谓

2、通过接口工具测试

可以根据请求示例先通过接口工具测试下能否正常调用：

2.1、请求示例

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、接口文档

请求参数 (Request Body)

files 对象详细说明

• type: 支持类型，目前仅支持 image。
• transfer_method: 传递方式。remote_url (远程地址) 或 local_file (上传文件)。
• url: 图片地址（仅当 remote_url 时必填）。
• upload_file_id: 上传文件 ID（仅当 local_file 时必填）。

响应说明 (Response)

根据 response_mode 的不同，返回的对象也不同：

• blocking: 返回 ChatCompletionResponse 对象。
• streaming: 返回 ChunkChatCompletionResponse 流式序列。

ChatCompletionResponse (阻塞模式)

Content-Type: application/json

ChunkChatCompletionResponse (流式响应)

Content-Type: text/event-stream

每个流式块以 data: 开头，以 nn 分隔：

data: {"event": "message", "task_id": "900bbd43-dc0b-4383-a372-aa6e6c414227", "id": "663c5084-a254-4040-8ad3-51f2a3c1a77c", "answer": "Hi", "created_at": 1705398420}nn

流式响应包含多种事件类型 (event)：

2.3、使用Apifox进行请求测试

可以看到已经能成功接收响应了，返回中的每一行数据（流式块）可以看作openai规范中的一个chunk，注意agent模式下不允许使用阻塞模式（blocking），所以得自己处理这些流式数据

3、使用Python调用接口

dify的agent返回的流式数据并没有遵守一般的openai规范，所以无法通过openai或者langchain封装好的方法自动处理，只能自己观察响应结果并编写处理方法，下面是我编写的处理方法可以直接拿去用，用到了第三方库requests：

def call_dify_api_streaming(api_key, query, user, conversation_id=""):
    """
    调用 Dify - 流式响应
    :param api_key: api_key
    :param query: 用户输入/提问内容
    :param user: 用户标识，用于定义终端用户的身份
    :param conversation_id: （选填）会话 ID，需要基于之前的聊天记录继续对话，必须传之前消息的 conversation_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')
                    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)

使用pytest进行测试，成功：

def test_dify_agent():
    API_KEY = "app-YT7jsB1dRjV8x1qmIsD62jE1"
    USER = "wu"

    call_dify_api_streaming(api_key=API_KEY, query="你是谁", user=USER)