fix main solver pipeline

Author: royzhao

kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py

@@ -154,7 +154,7 @@ def do_retrieval(self, task_query, tag_id, task, context: Context, **kwargs) ->
                         spos = list(set(spos))  # Deduplicate
 
                         # Add report line if there are any SPOs
-                        if spos:
+                        if reporter and spos:
                             reporter.add_report_line(
                                 tag_id,
                                 f"end_sub_kag_retriever_{output.retriever_method}",

kag/solver/executor/retriever/kag_model_hybrid_executor.py

@@ -104,11 +104,11 @@ def do_main(self, task_query, tag_id, task, context, **kwargs):
 
                 break
             else:
-                messages.append({
-                    "role": "assistant",
-                    "content": subquestion_response,
-                })
                 if "<search>" in subquestion_response:
+                    messages.append({
+                        "role": "assistant",
+                        "content": subquestion_response,
+                    })
                     search = search_plan_extraction(subquestion_response)
                     # 有时候会缺失</search>训练时需要优化<search>内容，不需要直接换行
                     if len(search) == 0:
@@ -117,6 +117,9 @@ def do_main(self, task_query, tag_id, task, context, **kwargs):
                     try:
                         sub_queries, logic_forms = parse_logic_form_with_str(search)
                         logic_forms = self.logic_node_parser.parse_logic_form_set(logic_forms, sub_queries, task_query)
+                        if not logic_forms:
+                            logic_node.sub_query = search
+                            logic_forms = [logic_node]
                     except Exception as e:
                         logger.warning(f"kag model think can not extra lf from {search} {e}")
                         logic_node.sub_query = search
@@ -142,7 +145,7 @@ def do_main(self, task_query, tag_id, task, context, **kwargs):
                                 "query": target_query,
                                 "logic_form_node": logic_forms[0]
                             })
-                        retriever_output = self.do_retrieval(task_query=target_query, tag_id=tag_id, task=cur_task,
+                        retriever_output = self.do_retrieval(task_query=target_query, tag_id=cur_turn_tag_name, task=cur_task,
                                                              context=context, **kwargs)
 
                         recall_information_list = []
@@ -158,7 +161,7 @@ def do_main(self, task_query, tag_id, task, context, **kwargs):
                             "content": recall_str,
                         })
                     except Exception as e:
-                        logger.error(f"kag flow exception! {e}", exc_info=True)
+                        logger.error(f"kag flow exception! {e} search={search}", exc_info=True)
                         self.report_content(
                             reporter,
                             cur_turn_tag_name,
@@ -167,6 +170,11 @@ def do_main(self, task_query, tag_id, task, context, **kwargs):
                             "INIT",
                             step=task.name,
                         )
+                else:
+                    messages.append({
+                        "role": "assistant",
+                        "content": subquestion_response,
+                    })
 
         context.kwargs["messages"] = messages
         return retriever_output

kag/solver/executor/retriever/local_knowledge_base/kag_retriever/kag_component/kg_cs/lf_kg_retriever_template.py

@@ -5,7 +5,7 @@
 from kag.common.conf import KAGConstants, KAGConfigAccessor
 from kag.interface import LLMClient
 from kag.interface.solver.base_model import SPOEntity, LogicNode
-from kag.interface.solver.reporter_abc import ReporterABC, DotRefresher
+from kag.interface.solver.reporter_abc import ReporterABC
 from kag.interface.solver.model.one_hop_graph import KgGraph, EntityData
 from kag.common.parser.logic_node_parser import GetSPONode
 

kag/solver/main_solver.py

@@ -312,9 +312,7 @@ async def qa(task_id, query, project_id, host_addr, app_id, params={}):
             task_id=task_id,
             kb_project_ids=kb_project_ids,
         )
-
-        if answer:
-            reporter.add_report_line("answer", "Final Answer", answer, "FINISH")
+        reporter.add_report_line("answer", "Final Answer", answer, "FINISH")
 
     except Exception as e:
         logger.warning(
@@ -371,6 +369,40 @@ def invoke(
             )
         return answer
 
+    async def ainvoke(
+        self,
+        project_id: int,
+        task_id: int,
+        query: str,
+        session_id: str = "0",
+        is_report=True,
+        host_addr="http://127.0.0.1:8887",
+        params=None,
+        app_id="",
+    ):
+        answer = None
+        if params is None:
+            params = {}
+        try:
+            answer = await qa(
+                    task_id=task_id,
+                    project_id=project_id,
+                    host_addr=host_addr,
+                    query=query,
+                    params=params,
+                    app_id=app_id,
+                )
+            logger.info(f"{query} answer={answer}")
+        except Exception as e:
+            import traceback
+
+            traceback.print_exc()
+            logger.warning(
+                f"An exception occurred while processing query: {query}. Error: {str(e)}",
+                exc_info=True,
+            )
+        return answer
+
 
 if __name__ == "__main__":
     # init_kag_config(

kag/solver/reporter/open_spg_kag_model_reporter.py

@@ -1,4 +1,6 @@
 import logging
+import re
+
 from kag.common.conf import KAG_PROJECT_CONF
 from kag.common.parser.logic_node_parser import extract_steps_and_actions
 
@@ -13,6 +15,12 @@ class SafeDict(dict):
     def __missing__(self, key):
         return ""
 
+def remove_xml_tags(text):
+    # 正则表达式匹配所有 XML 标签，例如：<tag> 或 </tag> 或 <tag attr="value">
+    pattern = r'<[^>]+>'
+    # 用空字符串替换所有匹配项
+    clean_text = re.sub(pattern, '', text)
+    return clean_text
 
 def process_planning(think_str):
     result = []
@@ -67,7 +75,7 @@ def process_tag_template(text):
                 clean_text += xml_tag_template[tag_info[0]][KAG_PROJECT_CONF.language].format_map(SafeDict({
                     "content": content
                 }))
-        return clean_text
+        return remove_xml_tags(clean_text)
     return text