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 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_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 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": "**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": "**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, question: str, session_id: str, user_id): ... 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 {} 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_parameters(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))