From d961b5e9290edef3bee5cd6adc4a636af209c1e7 Mon Sep 17 00:00:00 2001
From: zhaoqingang <zhaoqg0118@163.com>
Date: 星期四, 09 一月 2025 18:52:21 +0800
Subject: [PATCH] tmp
---
app/service/v2/chat.py | 236 +++++++++++++++++++++++++++++++++++++++++++++++++----------
1 files changed, 196 insertions(+), 40 deletions(-)
diff --git a/app/service/v2/chat.py b/app/service/v2/chat.py
index 982ed43..5273f9b 100644
--- a/app/service/v2/chat.py
+++ b/app/service/v2/chat.py
@@ -1,70 +1,226 @@
import json
from Log import logger
-from app.config.agent_base_url import RG_CHAT_DIALOG
+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 max_chunk_size
-from app.models.v2.session_model import ChatSessionDao
+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 service_chat_dialog(db, chat_id:str, question: str, session_id: str, user_id):
- token = "ragflow-YzMzE1NDRjYzMyZjExZWY5ZjkxMDI0Mm"
- url = settings.fwr_base_url+RG_CHAT_DIALOG.format(chat_id)
- chat = ChatDialog(token)
- request_data = {
- "question": question,
- "stream": True,
- "session_id": session_id
- }
- headers = {
- 'Content-Type': 'application/json',
- 'Authorization': f"Bearer {token}"
- }
+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:
- await ChatSessionDao(db).update_or_insert_by_id(
+ 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=session_id,
- event_type="message"
+ 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:
- message = {"role": "assistant","answer":"", "reference": {}}
- async for ans in chat.chat_completions(url, request_data, headers):
+ 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", "璇疯緭鍏ヤ綘鐨勯棶棰橈紒")
- data = {"answer":error}
- event = "message"
+ 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:
- data = {}
- event = "message_end"
+ event = smart_message_end
else:
data = ans.get("data", {})
- message = ans.get("data", {})
- event = "message"
- message_str = "data: " + json.dumps({"event": event, "data": data}, ensure_ascii=False) + "\n\n"
+ # 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 # 鍙戦�佸垎鍧楁秷鎭�
- await ChatSessionDao(db).update_session_by_id(
- session_id=session_id,
- session=None,
- message=message
- )
+ 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)
- yield "data: " + json.dumps({"message": "message",
- "data": {"answer": "**ERROR**: " + str(e), "reference": []}},
- ensure_ascii=False) + "\n\n"
+ 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)
- yield "data: " + json.dumps({"message": "message_end",
- "data": {}},
- ensure_ascii=False) + "\n\n"
\ No newline at end of file
+
+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))
--
Gitblit v1.8.0