添加新旧流程信息获取功能，优化问题处理逻辑，使用多线程并发提高效率，增强代码结构和可读性。

rag2_0/dify/test_dify_chatapi.py

@@ -383,6 +383,71 @@ content: "{content}"
         avg_score = total_score / valid_scores if valid_scores > 0 else 0
         return retrieve_content, max_score, min_score, avg_score
 
+    def get_new_workflow_info(self, query:str, new_message_id:str) -> dict:
+        """
+        获取新流程的问题分类
+        
+        Args:
+            query (str): 用户问题
+            new_message_id (str): 新流程的消息ID
+            
+        Returns:
+            dict: 包含问题分类结果的字典
+        """
+        try:
+            new_message_info = DifyTool.get_message_debug_info_by_id(message_id=new_message_id)
+            for workflow_node in new_message_info["workflow_node_executions_info"]:
+                if workflow_node["title"] == "知识检索结果后处理":
+                    outputs = json.loads(workflow_node["outputs"])  
+                    retrieve_content, max_score, min_score, avg_score = self.get_retrieve_info(query=query, outputs=outputs)
+                elif workflow_node["title"] == "问题优化结果解析":
+                    outputs = json.loads(workflow_node["outputs"])  
+                    rewrite_query = outputs["optimize_query"]
+                    llm_result_json = json.loads(workflow_node['inputs'])["llm_result"]
+                    json_result = json.loads(llm_result_json)
+                    vertical_classification = json_result['vertical_classification']
+                    sub_classification = json_result['sub_classification']
+        except Exception as e:
+            return None
+        return {
+            "问题改写": rewrite_query,
+            "检索词条": "\n".join(retrieve_content) if retrieve_content else "未检索知识库",
+            "问题分类": f"{vertical_classification} - {sub_classification}",
+            "检索最高分": max_score,
+            "检索最低分": min_score,
+            "检索平均分": avg_score
+        }
+
+    def get_old_workflow_info(self, query:str, old_message_id:str) -> dict:
+        """
+        获取新流程的问题分类
+        
+        Args:
+            query (str): 用户问题
+            old_message_id (str): 旧的流程的消息ID
+            
+        Returns:
+            dict: 包含问题分类结果的字典
+        """
+        try:
+            old_message_info = DifyTool.get_message_debug_info_by_id(message_id=old_message_id)
+            for workflow_node in old_message_info["workflow_node_executions_info"]:
+                if workflow_node["title"] == "知识检索结果后处理":
+                    outputs = json.loads(workflow_node["outputs"])  
+                    retrieve_content, max_score, min_score, avg_score = self.get_retrieve_info(query=query, outputs=outputs)
+                elif workflow_node["title"] == "问题优化结果解析":
+                    outputs = json.loads(workflow_node["outputs"])  
+                    rewrite_query = outputs["optimize_query"]
+        except Exception as e:
+            return None
+        return {
+            "问题改写": rewrite_query,
+            "检索词条": "\n".join(retrieve_content) if retrieve_content else "未检索知识库",
+            "检索最高分": max_score,
+            "检索最低分": min_score,
+            "检索平均分": avg_score
+        }
+
     def process_question_with_judge(self, q:str):
         """
         处理单个问题，获取新旧流程的回答并进行评判
@@ -423,36 +488,30 @@ content: "{content}"
         if judge_result is None:
             judge_result = ""
         
-        # 获取检索内容评分
+        # 并行获取新旧流程信息
-        retrieve_content = []
+        with ThreadPoolExecutor(max_workers=2) as executor:
-        max_score = 0
+            future_new = executor.submit(self.get_new_workflow_info, query=query, new_message_id=new_message_id)
-        min_score = 0
+            future_old = executor.submit(self.get_old_workflow_info, query=query, old_message_id=old_message_id)
-        avg_score = 0
+            
-        rewrite_query = ""
+            try:
-        try:
+                new_workflow_info = future_new.result()
-            new_message_info = DifyTool.get_message_debug_info_by_id(message_id=new_message_id)
+                old_workflow_info = future_old.result()
-            for workflow_node in new_message_info["workflow_node_executions_info"]:
+            except Exception as e:
-                if workflow_node["title"] == "知识检索结果后处理":
+                print(f"处理问题 '{query}' 获取工作流信息时发生错误: {str(e)}")
-                    outputs = json.loads(workflow_node["outputs"])  
+                return None
-                    retrieve_content, max_score, min_score, avg_score = self.get_retrieve_info(query=query, outputs=outputs)
+
-                elif workflow_node["title"] == "问题优化结果解析":
-                    outputs = json.loads(workflow_node["outputs"])  
-                    rewrite_query = outputs["optimize_query"]
-        except Exception as e:
-            print(f"处理问题 '{query}' 获取检索内容时发生错误: {str(e)}")
-        
         # 返回结果
         return {
             "问题": query,
-            "问题改写": rewrite_query,
+            "新问题改写": new_workflow_info["问题改写"],
-            "旧流程答案": old_answer,
+            "旧问题改写": old_workflow_info["问题改写"],
+            "新问题分类": new_workflow_info["问题分类"],
             "新流程答案": new_answer,
+            "旧流程答案": old_answer,
             "回答判断": judge_result,
             "答案词条": answer_title if answer_title else "", 
-            "检索得到词条": "\n".join(retrieve_content) if retrieve_content else "未检索知识库",
+            "新检索词条": new_workflow_info["检索词条"],
-            "检索最高分": max_score,
+            "旧检索词条": old_workflow_info["检索词条"],
-            "检索最低分": min_score,
-            "检索平均分": avg_score
         }
         
     def run_comparison(self, with_judge=False):