"""
|
DifyAI API客户端
|
"""
|
|
import requests
|
import time
|
import json
|
from typing import List, Dict, Optional, Any
|
from loguru import logger
|
from config import settings
|
|
|
class DifyClient:
|
"""DifyAI API客户端"""
|
|
def __init__(self):
|
self.base_url = settings.dify_base_url.rstrip("/")
|
self.headers = {
|
"Content-Type": "application/json",
|
"Authorization": f"Bearer {settings.dify_api_key}",
|
}
|
self.session = requests.Session()
|
self.session.headers.update(self.headers)
|
|
def send_chat_message(
|
self,
|
query: str,
|
user: str,
|
conversation_id: Optional[str] = None,
|
max_retries: int = None,
|
nick_name: Optional[str] = None,
|
) -> Optional[Dict[str, Any]]:
|
"""
|
发送对话消息（非流式模式）
|
|
Args:
|
query: 用户输入/提问内容
|
user: 用户标识
|
conversation_id: 会话ID（可选）
|
max_retries: 最大重试次数
|
nick_name: 昵称（可选，用于传递给dify）
|
|
Returns:
|
响应数据字典，失败返回None
|
"""
|
if max_retries is None:
|
max_retries = settings.max_retry_count
|
|
url = f"{self.base_url}/chat-messages"
|
|
# 构建inputs参数
|
inputs = {}
|
if nick_name:
|
inputs["nick_name"] = nick_name
|
|
payload = {
|
"query": query,
|
"response_mode": "blocking", # 使用阻塞模式（非流式）
|
"user": user,
|
"inputs": inputs,
|
}
|
|
# 如果有会话ID，添加到请求中
|
if conversation_id:
|
payload["conversation_id"] = conversation_id
|
|
retry_count = 0
|
while retry_count <= max_retries:
|
try:
|
logger.info(
|
f"发送Dify消息: user={user}, conversation_id={conversation_id}, retry={retry_count}"
|
)
|
|
response = self.session.post(url, json=payload, timeout=60)
|
response.raise_for_status()
|
|
result = response.json()
|
|
# 检查响应是否成功
|
if "answer" in result and "conversation_id" in result:
|
logger.info(
|
f"Dify消息发送成功: user={user}, conversation_id={result.get('conversation_id')}"
|
)
|
return result
|
else:
|
logger.error(f"Dify响应格式异常: user={user}, response={result}")
|
|
except requests.exceptions.Timeout:
|
logger.warning(f"Dify请求超时: user={user}, retry={retry_count}")
|
except requests.exceptions.RequestException as e:
|
logger.error(
|
f"Dify网络错误: user={user}, retry={retry_count}, error={str(e)}"
|
)
|
except Exception as e:
|
logger.error(
|
f"Dify请求异常: user={user}, retry={retry_count}, error={str(e)}"
|
)
|
|
retry_count += 1
|
if retry_count <= max_retries:
|
wait_time = settings.retry_delay * retry_count
|
logger.info(f"等待重试: user={user}, wait_time={wait_time}s")
|
time.sleep(wait_time)
|
|
logger.error(
|
f"Dify消息发送失败，已达最大重试次数: user={user}, max_retries={max_retries}"
|
)
|
return None
|
|
def send_chat_message_stream(
|
self,
|
query: str,
|
user: str,
|
conversation_id: Optional[str] = None,
|
max_retries: int = None,
|
nick_name: Optional[str] = None,
|
) -> Optional[Dict[str, Any]]:
|
"""
|
发送对话消息（流式模式）
|
|
Args:
|
query: 用户输入/提问内容
|
user: 用户标识
|
conversation_id: 会话ID（可选）
|
max_retries: 最大重试次数
|
nick_name: 昵称（可选，用于传递给dify）
|
|
Returns:
|
完整的响应数据字典，失败返回None
|
"""
|
if max_retries is None:
|
max_retries = settings.max_retry_count
|
|
url = f"{self.base_url}/chat-messages"
|
|
# 构建inputs参数
|
inputs = {}
|
if nick_name:
|
inputs["nick_name"] = nick_name
|
|
payload = {
|
"query": query,
|
"response_mode": "streaming", # 使用流式模式
|
"user": user,
|
"inputs": inputs,
|
}
|
|
# 如果有会话ID，添加到请求中
|
if conversation_id:
|
payload["conversation_id"] = conversation_id
|
|
retry_count = 0
|
while retry_count <= max_retries:
|
try:
|
logger.info(
|
f"发送Dify流式消息: user={user}, conversation_id={conversation_id}, retry={retry_count}"
|
)
|
|
response = self.session.post(
|
url,
|
json=payload,
|
timeout=settings.dify_streaming_timeout,
|
stream=True
|
)
|
response.raise_for_status()
|
|
# 处理流式响应
|
result = self._process_stream_response(response, user)
|
|
if result:
|
logger.info(
|
f"Dify流式消息发送成功: user={user}, conversation_id={result.get('conversation_id')}"
|
)
|
return result
|
else:
|
logger.error(f"Dify流式响应处理失败: user={user}")
|
|
except requests.exceptions.Timeout:
|
logger.warning(f"Dify流式请求超时: user={user}, retry={retry_count}")
|
except requests.exceptions.RequestException as e:
|
logger.error(
|
f"Dify流式网络错误: user={user}, retry={retry_count}, error={str(e)}"
|
)
|
except Exception as e:
|
logger.error(
|
f"Dify流式请求异常: user={user}, retry={retry_count}, error={str(e)}"
|
)
|
|
retry_count += 1
|
if retry_count <= max_retries:
|
wait_time = settings.retry_delay * retry_count
|
logger.info(f"等待重试: user={user}, wait_time={wait_time}s")
|
time.sleep(wait_time)
|
|
logger.error(
|
f"Dify流式消息发送失败，已达最大重试次数: user={user}, max_retries={max_retries}"
|
)
|
return None
|
|
def _process_stream_response(self, response: requests.Response, user: str) -> Optional[Dict[str, Any]]:
|
"""
|
处理流式响应
|
|
Args:
|
response: requests响应对象
|
user: 用户标识
|
|
Returns:
|
完整的响应数据字典，失败返回None
|
"""
|
try:
|
# 检查响应头
|
content_type = response.headers.get('content-type', '')
|
if 'text/event-stream' not in content_type:
|
logger.warning(f"响应不是SSE格式: user={user}, content_type={content_type}")
|
|
complete_answer = ""
|
conversation_id = ""
|
task_id = ""
|
message_id = ""
|
created_at = None
|
metadata = None
|
usage = None
|
retriever_resources = None
|
message_ended = False # 标记是否收到message_end事件
|
|
logger.info(f"开始处理流式响应: user={user}")
|
|
# 添加超时和行计数器
|
line_count = 0
|
max_empty_lines = 50 # 最大连续空行数，防止无限循环
|
|
for line in response.iter_lines(decode_unicode=True):
|
line_count += 1
|
|
if not line:
|
# 空行计数，防止无限等待
|
if line_count > max_empty_lines and not complete_answer:
|
logger.warning(f"流式响应过多空行，可能连接异常: user={user}, line_count={line_count}")
|
break
|
continue
|
|
# 跳过非数据行
|
if not line.startswith("data: "):
|
continue
|
|
# 提取JSON数据
|
data_str = line[6:] # 移除 "data: " 前缀
|
|
if not data_str.strip():
|
continue
|
|
try:
|
data = json.loads(data_str)
|
event = data.get("event", "")
|
|
if event == "message":
|
# 消息事件 - 累积答案内容
|
answer_chunk = data.get("answer", "")
|
complete_answer += answer_chunk
|
|
# 保存基本信息
|
if not conversation_id:
|
conversation_id = data.get("conversation_id", "")
|
if not task_id:
|
task_id = data.get("task_id", "")
|
if not message_id:
|
message_id = data.get("id", "")
|
if created_at is None:
|
created_at = data.get("created_at")
|
|
logger.debug(f"收到消息块: user={user}, chunk_length={len(answer_chunk)}")
|
|
elif event == "message_end":
|
# 消息结束事件 - 获取元数据
|
metadata = data.get("metadata")
|
usage = data.get("usage")
|
retriever_resources = data.get("retriever_resources")
|
message_ended = True
|
|
logger.info(f"流式响应完成: user={user}, total_length={len(complete_answer)}")
|
break
|
|
elif event == "message_file":
|
# 文件事件 - 记录但继续处理
|
logger.debug(f"收到文件事件: user={user}, file_type={data.get('type')}")
|
continue
|
|
elif event == "message_replace":
|
# 消息替换事件 - 替换答案内容
|
replace_answer = data.get("answer", "")
|
if replace_answer:
|
complete_answer = replace_answer
|
logger.info(f"消息内容被替换: user={user}, new_length={len(complete_answer)}")
|
continue
|
|
elif event in ["workflow_started", "node_started", "node_finished", "workflow_finished"]:
|
# 工作流相关事件 - 记录但继续处理
|
logger.debug(f"收到工作流事件: user={user}, event={event}")
|
continue
|
|
elif event in ["tts_message", "tts_message_end"]:
|
# TTS音频事件 - 记录但继续处理
|
logger.debug(f"收到TTS事件: user={user}, event={event}")
|
continue
|
|
elif event in ["agent_thought", "agent_message"]:
|
# Agent相关事件 - 需要特殊处理
|
logger.debug(f"收到Agent事件: user={user}, event={event}, data_keys={list(data.keys())}")
|
|
# 从agent事件中提取conversation_id（如果有的话）
|
if not conversation_id and data.get("conversation_id"):
|
conversation_id = data.get("conversation_id")
|
logger.info(f"从Agent事件获取conversation_id: user={user}, conversation_id={conversation_id}")
|
|
# 从agent事件中提取基本信息（如果有的话）
|
if not task_id and data.get("task_id"):
|
task_id = data.get("task_id")
|
logger.debug(f"从Agent事件获取task_id: user={user}, task_id={task_id}")
|
if not message_id and data.get("id"):
|
message_id = data.get("id")
|
logger.debug(f"从Agent事件获取message_id: user={user}, message_id={message_id}")
|
if created_at is None and data.get("created_at"):
|
created_at = data.get("created_at")
|
logger.debug(f"从Agent事件获取created_at: user={user}, created_at={created_at}")
|
|
# 检查agent_message是否包含answer内容
|
if event == "agent_message" and data.get("answer"):
|
agent_answer = data.get("answer", "")
|
complete_answer += agent_answer
|
logger.debug(f"从Agent消息获取内容: user={user}, chunk_length={len(agent_answer)}")
|
|
continue
|
|
elif event == "error":
|
# 错误事件
|
error_msg = data.get("message", "未知错误")
|
logger.error(f"流式响应错误: user={user}, error={error_msg}")
|
return None
|
|
elif event == "ping":
|
# ping事件 - 保持连接
|
logger.debug(f"收到ping事件: user={user}")
|
continue
|
|
else:
|
# 未知事件类型 - 记录但继续处理
|
logger.debug(f"收到未知事件: user={user}, event={event}")
|
continue
|
|
except json.JSONDecodeError as e:
|
logger.warning(f"解析流式数据JSON失败: user={user}, data={data_str}, error={str(e)}")
|
continue
|
|
# 构建完整响应
|
# 对于Agent模式，可能没有conversation_id，但有task_id
|
# 在这种情况下，我们可以使用task_id作为conversation_id的替代
|
if conversation_id or (task_id and (complete_answer or message_ended)):
|
# 如果没有conversation_id但有task_id，使用task_id
|
final_conversation_id = conversation_id or task_id
|
|
result = {
|
"event": "message",
|
"task_id": task_id,
|
"id": message_id,
|
"message_id": message_id,
|
"conversation_id": final_conversation_id,
|
"mode": "chat",
|
"answer": complete_answer, # 可能为空字符串
|
"created_at": created_at
|
}
|
|
# 添加可选字段
|
if metadata:
|
result["metadata"] = metadata
|
if usage:
|
result["usage"] = usage
|
if retriever_resources:
|
result["retriever_resources"] = retriever_resources
|
|
if complete_answer:
|
logger.info(f"流式响应处理成功: user={user}, answer_length={len(complete_answer)}, conversation_id={final_conversation_id}, message_ended={message_ended}")
|
else:
|
logger.info(f"流式响应处理成功(无内容): user={user}, conversation_id={final_conversation_id}, message_ended={message_ended}")
|
return result
|
else:
|
logger.error(f"流式响应不完整: user={user}, answer={bool(complete_answer)}, conversation_id={bool(conversation_id)}, task_id={bool(task_id)}")
|
# 记录更多调试信息
|
logger.debug(f"调试信息: task_id={task_id}, message_id={message_id}, created_at={created_at}, message_ended={message_ended}")
|
return None
|
|
except Exception as e:
|
logger.error(f"处理流式响应异常: user={user}, error={str(e)}")
|
return None
|
|
def send_message(
|
self,
|
query: str,
|
user: str,
|
conversation_id: Optional[str] = None,
|
max_retries: int = None,
|
force_streaming: Optional[bool] = None,
|
nick_name: Optional[str] = None,
|
) -> Optional[Dict[str, Any]]:
|
"""
|
发送对话消息（根据配置选择模式）
|
|
Args:
|
query: 用户输入/提问内容
|
user: 用户标识
|
conversation_id: 会话ID（可选）
|
max_retries: 最大重试次数
|
force_streaming: 强制使用流式模式（可选，覆盖配置）
|
nick_name: 昵称（可选，用于传递给dify）
|
|
Returns:
|
响应数据字典，失败返回None
|
"""
|
# 确定使用哪种模式
|
use_streaming = force_streaming if force_streaming is not None else settings.dify_streaming_enabled
|
|
if use_streaming:
|
logger.info(f"使用流式模式发送消息: user={user}")
|
return self.send_chat_message_stream(query, user, conversation_id, max_retries, nick_name)
|
else:
|
logger.info(f"使用阻塞模式发送消息: user={user}")
|
return self.send_chat_message(query, user, conversation_id, max_retries, nick_name)
|
|
def get_conversation_messages(
|
self, conversation_id: str, user: str
|
) -> Optional[List[Dict[str, Any]]]:
|
"""
|
获取对话历史消息
|
|
Args:
|
conversation_id: 会话ID
|
user: 用户标识
|
|
Returns:
|
消息列表，失败返回None
|
"""
|
try:
|
url = f"{self.base_url}/messages"
|
params = {"conversation_id": conversation_id, "user": user}
|
|
logger.info(f"获取对话历史: conversation_id={conversation_id}, user={user}")
|
|
response = self.session.get(url, params=params, timeout=30)
|
response.raise_for_status()
|
|
result = response.json()
|
|
if "data" in result:
|
logger.info(f"成功获取对话历史: conversation_id={conversation_id}")
|
return result["data"]
|
else:
|
logger.error(
|
f"获取对话历史响应格式异常: conversation_id={conversation_id}, response={result}"
|
)
|
return None
|
|
except requests.exceptions.RequestException as e:
|
logger.error(
|
f"获取对话历史网络错误: conversation_id={conversation_id}, error={str(e)}"
|
)
|
return None
|
except Exception as e:
|
logger.error(
|
f"获取对话历史异常: conversation_id={conversation_id}, error={str(e)}"
|
)
|
return None
|
|
|
# 全局DifyAI客户端实例
|
dify_client = DifyClient()
|
