在处理 Dify 流式响应报文时，通常会遇到的是一个持续传输的文本流，例如通过 HTTP 流（HTTP Streaming）或 Server-Sent Events（SSE）等方式传输的数据。这种流式响应的特点是数据不是一次性返回，而是分段逐步到达客户端。 要解析 Dify 的流式响应，可以使用 Python 中的 `requests` 库进行 HTTP 请求，并通过逐行读取响应内容来处理流式数据。以下是一个完整的示例代码，展示如何使用 Python 解析 Dify 流式响应报文。 ### 示例代码：解析 Dify 流式响应报文 ```python import requests def parse_dify_stream(url, headers=None): """ 解析 Dify 流式响应报文 :param url: 流式接口地址 :param headers: 请求头信息 """ response = requests.get(url, headers=headers, stream=True) if response.status_code == 200: for chunk in response.iter_lines(): if chunk: decoded_chunk = chunk.decode('utf-8') print("Received data:", decoded_chunk) # 可以在此处添加对数据的进一步处理逻辑 else: print(f"Failed to connect, status code: {response.status_code}") # 示例调用 if __name__ == "__main__": dify_api_url = "https://api.dify.ai/v1/streams" headers = { "Authorization": "Bearer YOUR_ACCESS_TOKEN", "Accept": "text/event-stream" } parse_dify_stream(dify_api_url, headers) ``` ### 代码说明： - `requests.get` 设置参数 `stream=True` 以启用流式传输。 - 使用 `response.iter_lines()` 逐行读取响应内容。 - 每个 `chunk` 是一行数据，通常为 UTF-8 编码，需使用 `decode('utf-8')` 转换为字符串。 - 可根据实际需求对 `decoded_chunk` 进行处理，例如 JSON 解析、状态更新或日志记录等操作。 ### 注意事项： - 确保网络请求具有合适的超时机制，防止长时间无响应导致程序挂起。 - 若流式响应采用特定格式（如 SSE 或 JSON Lines），需根据格式规范进行解析。 - 若流式数据包含事件标识（如 `data:`、`event:` 等），需对字符串进行分割处理。 ### 扩展功能：处理 JSON 格式的流式数据 如果 Dify 返回的是 JSON 格式的流式数据，则可以使用 `json` 模块进行解析： ```python import json # 在循环中解析 JSON 数据 try: json_data = json.loads(decoded_chunk) print("Parsed JSON:", json_data) except json.JSONDecodeError as e: print("JSON decode error:", e) ``` ### 处理多段数据格式 若响应中包含多个字段（如事件类型、数据内容、ID 等），可使用字符串分割方式提取信息： ```python if decoded_chunk.startswith("data:"): data = decoded_chunk[5:].strip() print("Data payload:", data) ``` ### 处理连接中断与重连 为提高鲁棒性，可在代码中加入重试机制： ```python import time retry_attempts = 3 for attempt in range(retry_attempts): try: parse_dify_stream(dify_api_url, headers) break except requests.exceptions.RequestException as e: print(f"Connection error: {e}, retrying in 5 seconds...") time.sleep(5) ``` 通过上述方式，可以实现对 Dify 流式响应报文的完整解析与处理。 ---