import json
|
|
from Log import logger
|
from app.config.agent_base_url import RG_CHAT_DIALOG, DF_CHAT_AGENT, DF_CHAT_PARAMETERS, RG_CHAT_SESSIONS, \
|
DF_CHAT_WORKFLOW
|
from app.config.config import settings
|
from app.config.const import *
|
from app.models import DialogModel, ApiTokenModel, UserTokenModel
|
from app.models.v2.session_model import ChatSessionDao, ChatData
|
from app.service.v2.app_driver.chat_agent import ChatAgent
|
from app.service.v2.app_driver.chat_data import ChatBaseApply
|
from app.service.v2.app_driver.chat_dialog import ChatDialog
|
from app.service.v2.app_driver.chat_workflow import ChatWorkflow
|
|
|
async def update_session_log(db, session_id: str, message: dict, conversation_id: str):
|
await ChatSessionDao(db).update_session_by_id(
|
session_id=session_id,
|
session=None,
|
message=message,
|
conversation_id=conversation_id
|
)
|
|
|
async def add_session_log(db, session_id: str, question: str, chat_id: str, user_id, event_type: str,
|
conversation_id: str):
|
try:
|
session = await ChatSessionDao(db).update_or_insert_by_id(
|
session_id=session_id,
|
name=question[:255],
|
agent_id=chat_id,
|
agent_type=1,
|
tenant_id=user_id,
|
message={"role": "user", "content": question},
|
conversation_id=conversation_id,
|
event_type=event_type
|
)
|
return session
|
except Exception as e:
|
logger.error(e)
|
return None
|
|
async def get_app_token(db, app_id):
|
app_token = db.query(UserTokenModel).filter_by(id=app_id).first()
|
if app_token:
|
return app_token.access_token
|
return ""
|
|
|
|
async def get_chat_token(db, app_id):
|
app_token = db.query(ApiTokenModel).filter_by(app_id=app_id).first()
|
if app_token:
|
return app_token.token
|
return ""
|
|
|
async def add_chat_token(db, data):
|
try:
|
api_token = ApiTokenModel(**data)
|
db.add(api_token)
|
db.commit()
|
except Exception as e:
|
logger.error(e)
|
|
|
|
async def get_chat_info(db, chat_id: str):
|
return db.query(DialogModel).filter_by(id=chat_id, status=Dialog_STATSU_ON).first()
|
|
|
async def get_chat_object(mode):
|
if mode == workflow_chat:
|
url = settings.dify_base_url + DF_CHAT_WORKFLOW
|
return ChatWorkflow(), url
|
else:
|
url = settings.dify_base_url + DF_CHAT_AGENT
|
return ChatAgent(), url
|
|
|
async def service_chat_dialog(db, chat_id: str, question: str, session_id: str, user_id, mode: str):
|
conversation_id = ""
|
token = await get_chat_token(db, rg_api_token)
|
url = settings.fwr_base_url + RG_CHAT_DIALOG.format(chat_id)
|
chat = ChatDialog()
|
session = await add_session_log(db, session_id, question, chat_id, user_id, mode, session_id)
|
if session:
|
conversation_id = session.conversation_id
|
message = {"role": "assistant", "answer": "", "reference": {}}
|
try:
|
async for ans in chat.chat_completions(url, await chat.request_data(question, conversation_id),
|
await chat.get_headers(token)):
|
data = {}
|
error = ""
|
status = http_200
|
if ans.get("code", None) == 102:
|
error = ans.get("message", "error！")
|
status = http_400
|
event = smart_message_error
|
else:
|
if isinstance(ans.get("data"), bool) and ans.get("data") is True:
|
event = smart_message_end
|
else:
|
data = ans.get("data", {})
|
# conversation_id = data.get("session_id", "")
|
if "session_id" in data:
|
del data["session_id"]
|
message = data
|
event = smart_message_cover
|
message_str = "data: " + json.dumps(
|
{"event": event, "data": data, "error": error, "status": status, "session_id": session_id},
|
ensure_ascii=False) + "\n\n"
|
for i in range(0, len(message_str), max_chunk_size):
|
chunk = message_str[i:i + max_chunk_size]
|
# print(chunk)
|
yield chunk # 发送分块消息
|
except Exception as e:
|
|
logger.error(e)
|
try:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: " + str(e), "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
except:
|
...
|
finally:
|
await update_session_log(db, session_id, message, conversation_id)
|
|
|
async def service_chat_workflow(db, chat_id: str, chat_data: ChatData, session_id: str, user_id, mode: str):
|
conversation_id = ""
|
answer_event = ""
|
answer_agent = ""
|
message_id = ""
|
task_id = ""
|
error = ""
|
files = []
|
node_list = []
|
token = await get_chat_token(db, chat_id)
|
|
chat, url = await get_chat_object(mode)
|
if hasattr(chat_data, "query"):
|
query = chat_data.query
|
else:
|
query = "start new workflow"
|
session = await add_session_log(db, session_id, query, chat_id, user_id, mode, conversation_id)
|
if session:
|
conversation_id = session.conversation_id
|
try:
|
async for ans in chat.chat_completions(url,
|
await chat.request_data(query, conversation_id, str(user_id), chat_data),
|
await chat.get_headers(token)):
|
data = {}
|
status = http_200
|
conversation_id = ans.get("conversation_id")
|
task_id = ans.get("task_id")
|
if ans.get("event") == message_error:
|
error = ans.get("message", "参数异常！")
|
status = http_400
|
event = smart_message_error
|
elif ans.get("event") == message_agent:
|
data = {"answer": ans.get("answer", ""), "id": ans.get("message_id", "")}
|
answer_agent += ans.get("answer", "")
|
message_id = ans.get("message_id", "")
|
event = smart_message_stream
|
elif ans.get("event") == message_event:
|
data = {"answer": ans.get("answer", ""), "id": ans.get("message_id", "")}
|
answer_event += ans.get("answer", "")
|
message_id = ans.get("message_id", "")
|
event = smart_message_stream
|
elif ans.get("event") == message_file:
|
data = {"url": ans.get("url", ""), "id": ans.get("id", ""),
|
"type": ans.get("type", "")}
|
files.append(data)
|
event = smart_message_file
|
elif ans.get("event") in [workflow_started, node_started, node_finished]:
|
data = ans.get("data", {})
|
data["inputs"] = []
|
data["outputs"] = []
|
data["process_data"] = ""
|
node_list.append(ans)
|
event = [smart_workflow_started, smart_node_started, smart_node_finished][
|
[workflow_started, node_started, node_finished].index(ans.get("event"))]
|
elif ans.get("event") == workflow_finished:
|
data = ans.get("data", {})
|
event = smart_workflow_finished
|
node_list.append(ans)
|
|
elif ans.get("event") == message_end:
|
event = smart_message_end
|
else:
|
continue
|
|
yield "data: " + json.dumps(
|
{"event": event, "data": data, "error": error, "status": status, "task_id": task_id,
|
"session_id": session_id},
|
ensure_ascii=False) + "\n\n"
|
|
except Exception as e:
|
logger.error(e)
|
try:
|
yield "data: " + json.dumps({"message": smart_message_error,
|
"error": "\n**ERROR**: " + str(e), "status": http_500},
|
ensure_ascii=False) + "\n\n"
|
except:
|
...
|
finally:
|
await update_session_log(db, session_id, {"role": "assistant", "answer": answer_event or answer_agent,
|
"node_list": node_list, "task_id": task_id, "id": message_id,
|
"error": error}, conversation_id)
|
|
|
async def service_chat_basic(db, chat_id: str, chat_data: ChatData, session_id: str, user_id, mode: str):
|
...
|
|
|
async def service_chat_parameters(db, chat_id, user_id):
|
chat_info = db.query(DialogModel).filter_by(id=chat_id).first()
|
if not chat_info:
|
return {}
|
return chat_info.parameters
|
# if chat_info.dialog_type == RG_TYPE:
|
# return {"retriever_resource":
|
# {
|
# "enabled": True
|
# }
|
# }
|
# elif chat_info.dialog_type == BASIC_TYPE:
|
# ...
|
# elif chat_info.dialog_type == DF_TYPE:
|
# token = await get_chat_token(db, chat_id)
|
# if not token:
|
# return {}
|
# url = settings.dify_base_url + DF_CHAT_PARAMETERS
|
# chat = ChatBaseApply()
|
# return await chat.chat_get(url, {"user": str(user_id)}, await chat.get_headers(token))
|
|
|
async def service_chat_sessions(db, chat_id, name):
|
token = await get_chat_token(db, rg_api_token)
|
if not token:
|
return {}
|
url = settings.fwr_base_url + RG_CHAT_SESSIONS.format(chat_id)
|
chat = ChatDialog()
|
return await chat.chat_sessions(url, {"name": name}, await chat.get_headers(token))
|
