diff --git a/rag2_0/dify/test_dify_chatapi.py b/rag2_0/dify/test_dify_chatapi.py index 7969f37..fe3e881 100755 --- a/rag2_0/dify/test_dify_chatapi.py +++ b/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 = [] - max_score = 0 - min_score = 0 - avg_score = 0 - rewrite_query = "" - 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"] - except Exception as e: - print(f"处理问题 '{query}' 获取检索内容时发生错误: {str(e)}") - + # 并行获取新旧流程信息 + with ThreadPoolExecutor(max_workers=2) as executor: + future_new = executor.submit(self.get_new_workflow_info, query=query, new_message_id=new_message_id) + future_old = executor.submit(self.get_old_workflow_info, query=query, old_message_id=old_message_id) + + try: + new_workflow_info = future_new.result() + old_workflow_info = future_old.result() + except Exception as e: + print(f"处理问题 '{query}' 获取工作流信息时发生错误: {str(e)}") + return None + # 返回结果 return { "问题": query, - "问题改写": rewrite_query, - "旧流程答案": old_answer, + "新问题改写": new_workflow_info["问题改写"], + "旧问题改写": 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 "未检索知识库", - "检索最高分": max_score, - "检索最低分": min_score, - "检索平均分": avg_score + "新检索词条": new_workflow_info["检索词条"], + "旧检索词条": old_workflow_info["检索词条"], } def run_comparison(self, with_judge=False):