From b7ad7ae90cd832281afe6d743776ad87f1e56e07 Mon Sep 17 00:00:00 2001
From: zhaoqingang <zhaoqg0118@163.com>
Date: 星期五, 03 一月 2025 16:55:47 +0800
Subject: [PATCH] 工作流日志

---
 app/models/session_model.py |    2 
 app/config/config.py        |    1 
 app/api/chat.py             |  247 ++++++++++++++++++++++++----------------
 app/service/session.py      |    9 +
 app/api/agent.py            |   49 +++++++-
 5 files changed, 197 insertions(+), 111 deletions(-)

diff --git a/app/api/agent.py b/app/api/agent.py
index 22ce8ef..786eb2f 100644
--- a/app/api/agent.py
+++ b/app/api/agent.py
@@ -223,17 +223,50 @@
             for i in session.log_to_json().get("message", []):
                 if i.get("role") == "user":
                     tmp_data["question"] = i.get("content")
+                    if "type" in i.get("content"):
+                        tmp_data["type"] =  i.get("content")["type"]
                 elif i.get("role") == "assistant":
                     if isinstance(i.get("content"), dict):
-                        tmp_data["answer"] = i.get("content", {}).get("answer")
-                        if "file_name" in i.get("content", {}):
-                            tmp_data["files"] = [{"file_name": i.get("content", {}).get("file_name"),
-                                                  "file_url": i.get("content", {}).get("file_url")}]
+                        content = i.get("content", {})
+                        tmp_data["answer"] = content.get("answer")
+                        if "file_name" in content:
+                            tmp_data["files"] = [{"file_name": content.get("file_name"),
+                                                  "file_url": content.get("file_url")}]
                         if "images" in i.get("content", {}):
-                            tmp_data["images"] = i.get("content", {}).get("images")
+                            tmp_data["images"] = content.get("images")
 
-                        if "download_url" in i.get("content", {}):
-                            tmp_data["download_url"] = i.get("content", {}).get("download_url")
+                        if "download_url" in content:
+                            tmp_data["download_url"] = content.get("download_url")
+
+                        if "node_list" in content:
+                            node_dict = {
+                                "node_data": [],   # {"title": "鍘婚櫎鍐椾綑",  # 鑺傜偣鍚嶇О "status": "succeeded",  # 鑺傜偣鐘舵��"created_at": 1735817337,  # 寮�濮嬫椂闂�"finished_at": 1735817337,  # 缁撴潫鏃堕棿"error": ""  # 閿欒鏃ュ織}
+                                "total_tokens": 0,  # 鑺辫垂token鏁�
+                                "created_at": 0,  # 寮�濮嬫椂闂�
+                                "finished_at": 0,  # 缁撴潫鏃堕棿
+                                "status": "succeeded",  # 宸ヤ綔娴佺姸鎬�
+                                "error": "",  # 閿欒鏃ュ織
+                            }
+                            for node in content["node_list"]:
+                                if node.get("event") == "node_finished":
+                                    node_dict["node_data"].append({
+                                        "title": node.get("data", {}).get("title", ""),
+                                        "status": node.get("data", {}).get("status", ""),
+                                        "created_at":node.get("data", {}).get("created_at", 0),
+                                        "finished_at":node.get("data", {}).get("finished_at", 0),
+                                        "node_type":node.get("data", {}).get("node_type", 0),
+                                        "elapsed_time":node.get("data", {}).get("elapsed_time", 0),
+                                        "error":node.get("data", {}).get("error", ""),
+                                    })
+                                elif node.get("event") == "workflow_finished":
+                                    node_dict["total_tokens"] = node.get("data", {}).get("total_tokens", 0)
+                                    node_dict["created_at"] = node.get("data", {}).get("created_at", 0)
+                                    node_dict["finished_at"] = node.get("data", {}).get("finished_at", 0)
+                                    node_dict["status"] = node.get("data", {}).get("status", "")
+                                    node_dict["error"] = node.get("data", {}).get("error", "")
+                                    node_dict["elapsed_time"] = node.get("data", {}).get("elapsed_time", 0)
+                            tmp_data["workflow"] = node_dict
+
                     else:
                         tmp_data["answer"] = i.get("content")
                     data.append(tmp_data)
@@ -255,3 +288,5 @@
         return Response(code=404, msg="Agent not found")
 
     return Response(code=200, msg="", data={"chat_id": uuid.uuid4().hex})
+
+
diff --git a/app/api/chat.py b/app/api/chat.py
index 2993b2a..eab8801 100644
--- a/app/api/chat.py
+++ b/app/api/chat.py
@@ -451,7 +451,9 @@
                                 title if title else title_query,
                                 agent_id,
                                 AgentType.DIFY,
-                                current_user.id
+                                current_user.id,
+                                {"role": "user", "content": title if title else title_query, "type": workflow_type, "is_clean":is_clean},
+                                workflow_type
                             )
                             conversation_id = session.conversation_id
                         except Exception as e:
@@ -475,12 +477,11 @@
                                     "upload_file_id": ""
                                 })
                         inputs_list = []
-                        token_list = []
                         if workflow_type == 1:
                             inputs["input_files"] = files
-                            inputs_list.append(inputs)
-                            token_list.append(token)
-                        elif workflow_type == 2 and is_clean == 0:
+                            inputs["Completion_of_main_indicators"] = title
+                            inputs_list.append({"inputs": inputs, "token": token, "workflow_type": workflow_type})
+                        elif workflow_type == 2:
                             inputs["file_list"] = files
                             inputs["Completion_of_main_indicators"] = title
                             inputs["sub_titles"] = sub_titles
@@ -488,9 +489,8 @@
                             if not token:
                                 await websocket.send_json(
                                     {"message": "Invalid token document_to_report", "type": "error"})
-                            inputs_list.append(inputs)
-                            token_list.append(token)
-                        elif workflow_type == 3:
+                            inputs_list.append({"inputs": inputs, "token": token, "workflow_type": workflow_type})
+                        elif workflow_type == 3 and is_clean == 0 and tokens < max_token:
                             inputs["file_list"] = files
                             inputs["number_of_title"] = title_number
                             inputs["title_style"] = title_style
@@ -498,28 +498,32 @@
                             if not token:
                                 await websocket.send_json(
                                     {"message": "Invalid token document_to_title", "type": "error"})
-                            # inputs_list.append(inputs)
-                            # token_list.append(token)
-                        elif workflow_type == 2 and is_clean == 1:
-                            # inputs["input_files"] = files
-                            inputs_list.append(inputs)
-                            token_list.append(token)
+                            inputs_list.append({"inputs": inputs, "token": token, "workflow_type": workflow_type})
+                        elif workflow_type == 3 and is_clean == 1 or tokens >= max_token:
+                            inputs["input_files"] = files
+                            inputs["Completion_of_main_indicators"] = title
+                            inputs_list.append({"inputs": inputs, "token": token, "workflow_type": 1})
                             inputs1 = {}
-                            # inputs1["file_list"] = files
-                            inputs1["Completion_of_main_indicators"] = title
-                            inputs1["sub_titles"] = sub_titles
-                            token = DfTokenDao(db).get_token_by_id(DOCUMENT_TO_REPORT_TITLE)
+                            inputs1["file_list"] = files
+                            inputs1["number_of_title"] = title_number
+                            inputs1["title_style"] = title_style
+                            token = DfTokenDao(db).get_token_by_id(DOCUMENT_TO_TITLE)
                             if not token:
                                 await websocket.send_json(
                                     {"message": "Invalid token document_to_report", "type": "error"})
-                            inputs_list.append(inputs1)
-                            token_list.append(token)
+                            inputs_list.append({"inputs": inputs, "token": token, "workflow_type": 3})
 
-                        complete_response = ""
-                        if workflow_type == 1 or workflow_type == 2:
-                            for inputs in inputs_list:
-                                inputs["input_files"] = files
-                                async for rag_response in dify_service.workflow(token, current_user.id, inputs):
+                        # print(inputs_list)
+                        for input in inputs_list:
+                            i = input["inputs"]
+                            if "file_list" in i:
+                                i["file_list"] = files
+                            node_list = []
+                            complete_response = ""
+                            workflow_list = []
+                            workflow_dict = {}
+                            if input["workflow_type"] == 1 or input["workflow_type"] == 2:
+                                async for rag_response in dify_service.workflow(input["token"], current_user.id, i):
                                     # print(rag_response)
                                     try:
                                         if rag_response[:5] == "data:":
@@ -532,8 +536,11 @@
                                             complete_response += rag_response
                                         try:
                                             data = json.loads(complete_response)
+                                            # print(data)
+                                            node_list.append(data)
                                             complete_response = ""
-                                            if data.get("event") == "node_started" or data.get("event") == "node_finished":  # "event": "message_end"
+                                            if data.get("event") == "node_started":  # "event": "message_end"
+
                                                 if "data" not in data or not data["data"]:  # 淇℃伅杩囨护
                                                     logger.error("闈炴硶鏁版嵁--------------------")
                                                     logger.error(data)
@@ -549,6 +556,16 @@
                                                         message = answer.get("title", "")
 
                                                     result = {"message": message, "type": "system"}
+                                            elif data.get("event") == "node_finished":
+                                                workflow_list.append({
+                                                    "title": data.get("data", {}).get("title", ""),
+                                                    "status": data.get("data", {}).get("status", ""),
+                                                    "created_at":data.get("data", {}).get("created_at", 0),
+                                                    "finished_at":data.get("data", {}).get("finished_at", 0),
+                                                    "node_type":data.get("data", {}).get("node_type", 0),
+                                                    "elapsed_time":data.get("data", {}).get("elapsed_time", 0),
+                                                    "error":data.get("data", {}).get("error", ""),
+                                                })
                                             elif data.get("event") == "workflow_finished":
                                                 answer = data.get("data", "")
                                                 if isinstance(answer, str):
@@ -570,12 +587,22 @@
                                                             "url": download_url,
                                                             "upload_file_id": ""
                                                         }]
+                                                    workflow_dict = {
+                                                        "node_data": workflow_list,
+                                                        "total_tokens": answer.get("total_tokens", 0),
+                                                        "created_at": answer.get("created_at", 0),
+                                                        "finished_at": answer.get("finished_at", 0),
+                                                        "status": answer.get("status", ""),
+                                                        "error": answer.get("error", ""),
+                                                        "elapsed_time": answer.get("elapsed_time", 0)
+                                                    }
                                                     result = {"message": message, "type": "message", "download_url": download_url}
                                                     try:
                                                         SessionService(db).update_session(chat_id,
                                                                                           message={"role": "assistant",
                                                                                                    "content": {
                                                                                                        "answer": message,
+                                                                                                       "node_list": node_list,
                                                                                                        "download_url": download_url}},
                                                                                           conversation_id=data.get(
                                                                                               "conversation_id"))
@@ -587,7 +614,8 @@
                                                     except Exception as e:
                                                         logger.error(e)
                                                         logger.error("杩斿洖瀹㈡埛绔秷鎭紓甯�!")
-                                                    result = {"message": "", "type": "close", "download_url": ""}
+
+                                                    result = {"message": "", "type": "close", "workflow": workflow_dict}
 
 
                                             else:
@@ -605,88 +633,107 @@
                                         result = {"message": f"鍐呴儴閿欒锛� {e2}", "type": "close"}
                                         await websocket.send_json(result)
                                         print(f"Error process message of ragflow: {e2}")
-                        elif workflow_type == 3:
-                            image_list = []
-                            # print(inputs)
-                            complete_response = ""
-                            async for rag_response in dify_service.chat(token, current_user.id, title_query, [],
-                                                                        conversation_id, inputs):
-                                print(rag_response)
-                                try:
-                                    if rag_response[:5] == "data:":
-                                        # 濡傛灉鏄紝鍒欐埅鍙栨帀鍓�5涓瓧绗︼紝骞跺幓闄ら灏剧┖鐧界
-                                        complete_response = rag_response[5:].strip()
-                                    elif "event: ping" in rag_response:
-                                        continue
-                                    else:
-                                        # 鍚﹀垯锛屼繚鎸佸師鏍�
-                                        complete_response += rag_response
+                            elif input["workflow_type"] == 3:
+                                image_list = []
+                                # print(inputs)
+                                complete_response = ""
+                                async for rag_response in dify_service.chat(input["token"], current_user.id, title_query, [],
+                                                                            conversation_id, i):
+                                    # print(rag_response)
                                     try:
-                                        data = json.loads(complete_response)
-                                        complete_response = ""
-                                        if data.get("event") == "node_started" or data.get(
-                                                "event") == "node_finished":  # "event": "message_end"
-                                            if "data" not in data or not data["data"]:  # 淇℃伅杩囨护
-                                                logger.error("闈炴硶鏁版嵁--------------------")
-                                                logger.error(data)
-                                                continue
-                                            else:  # 姝e父杈撳嚭
-                                                answer = data.get("data", "")
-                                                if isinstance(answer, str):
-                                                    logger.error("----------------鏈煡鏁版嵁--------------------")
+                                        if rag_response[:5] == "data:":
+                                            # 濡傛灉鏄紝鍒欐埅鍙栨帀鍓�5涓瓧绗︼紝骞跺幓闄ら灏剧┖鐧界
+                                            complete_response = rag_response[5:].strip()
+                                        elif "event: ping" in rag_response:
+                                            continue
+                                        else:
+                                            # 鍚﹀垯锛屼繚鎸佸師鏍�
+                                            complete_response += rag_response
+                                        try:
+                                            data = json.loads(complete_response)
+                                            node_list.append(data)
+                                            complete_response = ""
+                                            if data.get("event") == "node_started":  # "event": "message_end"
+                                                if "data" not in data or not data["data"]:  # 淇℃伅杩囨护
+                                                    logger.error("闈炴硶鏁版嵁--------------------")
                                                     logger.error(data)
                                                     continue
-                                                elif isinstance(answer, dict):
+                                                else:  # 姝e父杈撳嚭
+                                                    answer = data.get("data", "")
+                                                    if isinstance(answer, str):
+                                                        logger.error("----------------鏈煡鏁版嵁--------------------")
+                                                        logger.error(data)
+                                                        continue
+                                                    elif isinstance(answer, dict):
 
-                                                    message = answer.get("title", "")
+                                                        message = answer.get("title", "")
 
-                                                result = {"message": message, "type": "system"}
-                                        elif data.get("event") == "message":
-                                            message = data.get("answer", "")
-                                            # try:
-                                            #     msg_dict = json.loads(answer)
-                                            #     message = msg_dict.get("output",  "")
-                                            # except Exception as e:
-                                            #     print(e)
-                                            #     continue
-                                            result = {"message": message, "type": "message",
-                                                      "download_url": ""}
+                                                    result = {"message": message, "type": "system"}
+                                            elif data.get("event") == "node_finished":
+                                                workflow_list.append({
+                                                    "title": data.get("data", {}).get("title", ""),
+                                                    "status": data.get("data", {}).get("status", ""),
+                                                    "created_at":data.get("data", {}).get("created_at", 0),
+                                                    "finished_at":data.get("data", {}).get("finished_at", 0),
+                                                    "node_type":data.get("data", {}).get("node_type", 0),
+                                                    "elapsed_time":data.get("data", {}).get("elapsed_time", 0),
+                                                    "error":data.get("data", {}).get("error", ""),
+                                                })
+                                            elif data.get("event") == "message":
+                                                message = data.get("answer", "")
+                                                # try:
+                                                #     msg_dict = json.loads(answer)
+                                                #     message = msg_dict.get("output",  "")
+                                                # except Exception as e:
+                                                #     print(e)
+                                                #     continue
+                                                result = {"message": message, "type": "message",
+                                                          "download_url": ""}
+                                                try:
+                                                    SessionService(db).update_session(chat_id,
+                                                                                      message={"role": "assistant",
+                                                                                               "content": {
+                                                                                                   "answer": message,
+                                                                                                   "node_list": node_list,
+                                                                                                   "download_url": ""}},
+                                                                                      conversation_id=data.get(
+                                                                                          "conversation_id"))
+                                                except Exception as e:
+                                                    logger.error("淇濆瓨dify鐨勪細璇濆紓甯革紒")
+                                                    logger.error(e)
+                                                # try:
+                                                #     await websocket.send_json(result)
+                                                # except Exception as e:
+                                                #     logger.error(e)
+                                                #     logger.error("杩斿洖瀹㈡埛绔秷鎭紓甯�!")
+                                            elif  data.get("event") == "workflow_finished":
+                                                workflow_dict = {
+                                                    "node_data": workflow_list,
+                                                    "total_tokens": answer.get("total_tokens", 0),
+                                                    "created_at": answer.get("created_at", 0),
+                                                    "finished_at": answer.get("finished_at", 0),
+                                                    "status": answer.get("status", ""),
+                                                    "error": answer.get("error", ""),
+                                                    "elapsed_time": answer.get("elapsed_time", 0)
+                                                }
+                                            elif data.get("event") == "message_end":
+                                                result = {"message": "", "type": "close", "workflow": workflow_dict}
+                                            else:
+                                                continue
                                             try:
-                                                SessionService(db).update_session(chat_id,
-                                                                                  message={"role": "assistant",
-                                                                                           "content": {
-                                                                                               "answer": message,
-                                                                                               "download_url": ""}},
-                                                                                  conversation_id=data.get(
-                                                                                      "conversation_id"))
+                                                await websocket.send_json(result)
                                             except Exception as e:
-                                                logger.error("淇濆瓨dify鐨勪細璇濆紓甯革紒")
                                                 logger.error(e)
-                                            # try:
-                                            #     await websocket.send_json(result)
-                                            # except Exception as e:
-                                            #     logger.error(e)
-                                            #     logger.error("杩斿洖瀹㈡埛绔秷鎭紓甯�!")
-
-                                        elif data.get("event") == "message_end":
-                                            result = {"message": "", "type": "close", "download_url": ""}
-                                        else:
-                                            continue
-                                        try:
-                                            await websocket.send_json(result)
-                                        except Exception as e:
-                                            logger.error(e)
-                                            logger.error("dify杩斿洖瀹㈡埛绔秷鎭紓甯�!")
-                                        complete_response = ""
-                                    except json.JSONDecodeError as e:
-                                        print(f"Error decoding JSON: {e}")
-                                        # print(f"Response text: {text}")
-                                except Exception as e2:
-                                    result = {"message": f"鍐呴儴閿欒锛� {e2}", "type": "close"}
-                                    await websocket.send_json(result)
-                                    print(f"Error process message of ragflow: {e2}")
+                                                logger.error("dify杩斿洖瀹㈡埛绔秷鎭紓甯�!")
+                                            complete_response = ""
+                                        except json.JSONDecodeError as e:
+                                            print(f"Error decoding JSON: {e}")
+                                            # print(f"Response text: {text}")
+                                    except Exception as e2:
+                                        result = {"message": f"鍐呴儴閿欒锛� {e2}", "type": "close"}
+                                        await websocket.send_json(result)
+                                        print(f"Error process message of ragflow: {e2}")
                 elif agent.type == "documentIa":
-                    print(122112)
                     token = DfTokenDao(db).get_token_by_id(DOCUMENT_IA_QUESTIONS)
                     # print(token)
                     if not token:
diff --git a/app/config/config.py b/app/config/config.py
index e925d4b..c5a10a0 100644
--- a/app/config/config.py
+++ b/app/config/config.py
@@ -23,7 +23,6 @@
     dify_workflow_clean: str = ''
     dify_workflow_report: str = ''
     postgresql_database_url: str = ''
-    max_report_tokens: int = 100000
     def __init__(self, **kwargs):
         # 鏇挎崲閰嶇疆涓殑IP鍦板潃
         host_ip = os.getenv('HOST_IP', '127.0.0.1')
diff --git a/app/models/session_model.py b/app/models/session_model.py
index fd513d2..fd8bec6 100644
--- a/app/models/session_model.py
+++ b/app/models/session_model.py
@@ -18,6 +18,7 @@
     tenant_id = Column(Integer)  # 鍒涘缓浜�
     message = Column(TEXT)  # 璇存槑
     conversation_id = Column(String(64))
+    # workflow = Column(Integer, default=0)
 
     # to_dict 鏂规硶
     def to_dict(self):
@@ -26,6 +27,7 @@
             'name': self.name,
             'agent_type': self.agent_type,
             'agent_id': self.agent_id,
+            # 'workflow': self.workflow,
             'create_date': self.create_date.strftime("%Y-%m-%d %H:%M:%S"),
             'update_date': self.update_date.strftime("%Y-%m-%d %H:%M:%S"),
         }
diff --git a/app/service/session.py b/app/service/session.py
index e00bd3e..bf96515 100644
--- a/app/service/session.py
+++ b/app/service/session.py
@@ -12,7 +12,7 @@
     def __init__(self, db: Session):
         self.db = db
 
-    def create_session(self, session_id: str, name: str, agent_id: str, agent_type: AgentType, user_id: int) -> Type[
+    def create_session(self, session_id: str, name: str, agent_id: str, agent_type: AgentType, user_id: int, message:dict=None, workflow_type: int=0) -> Type[
                                                                                                                     SessionModel] | SessionModel:
         """
         鍒涘缓涓�涓柊鐨勪細璇濊褰曘��
@@ -26,9 +26,11 @@
         杩斿洖:
             SessionModel: 鏂板垱寤虹殑浼氳瘽妯″瀷瀹炰緥锛屽鏋滀細璇滻D宸插瓨鍦ㄥ垯杩斿洖None銆�
         """
+        if not message:
+            message = {"role": "user", "content": name}
         existing_session = self.get_session_by_id(session_id)
         if existing_session:
-            existing_session.add_message({"role": "user", "content": name})
+            existing_session.add_message(message)
             existing_session.update_date = current_time()
             self.db.commit()
             self.db.refresh(existing_session)
@@ -40,7 +42,8 @@
             agent_id=agent_id,
             agent_type=agent_type,
             tenant_id=user_id,
-            message=json.dumps([{"role": "user", "content": name}])
+            # workflow=workflow_type,
+            message=json.dumps([message])
         )
         self.db.add(new_session)
         self.db.commit()

--
Gitblit v1.8.0