From ffe87287d34b7e7a4f194c31005ca9fb8f420aa3 Mon Sep 17 00:00:00 2001 From: zhuzhongshu123 <152354526+zhuzhongshu123@users.noreply.github.com> Date: Thu, 17 Apr 2025 17:23:52 +0800 Subject: [PATCH 01/12] feat(kag): update to v0.7 (#456) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * add think cost * update csv scanner * add final rerank * add reasoner * add iterative planner * fix dpr search * fix dpr search * add reference data * move odps import * update requirement.txt * update 2wiki * add missing file * fix markdown reader * add iterative planning * update version * update runner * update 2wiki example * update bridge * merge solver and solver_new * add cur day * writer delete * update multi process * add missing files * fix report * add chunk retrieved executor * update try in stream runner result * add path * add math executor * update hotpotqa example * remove log * fix python coder solver * update hotpotqa example * fix python coder solver * update config * fix bad * add log * remove unused code * commit with task thought * move kag model to common * add default chat llm * fix * use static planner * support chunk graph node * add args * support naive rag * llm client support tool calls * add default async * add openai * fix result * fix markdown reader * fix thinker * update asyncio interface * feat(solver): add mcp support (#444) * 上传mcp client相关代码 * 1、完成一套mcp client的调用，从pipeline到planner、executor 2、允许json中传入多个mcp_server，通过大模型进行调用并选择 3、调通baidu_map_mcp的使用 * 1、schema * bugfix:删减冗余代码 --------- Co-authored-by: wanxingyu.wxy * fix affairqa after solver refactor * fix affairqa after solver refactor * fix readme * add params * update version * update mcp executor * update mcp executor * solver add mcp executor * add missing file * add mpc executor * add executor * x * update * fix requirement * fix main llm config * fix solver * bugfix:修复invoke函数调用逻辑 * chg eva * update example * add kag layer * add step task * support dot refresh * support dot refresh * support dot refresh * support dot refresh * add retrieved num * add retrieved num * add pipelineconf * update ppr * update musique prompts * update * add to_dict for BuilderComponentData * async build * add deduce prompt * add deduce prompt * add deduce prompt * fix reader * add deduce prompt * add page thinker report * modify prmpt * add step status * add self cognition * add self cognition * add memory graph storage * add now time * update memory config * add now time * chg graph loader * 添加prqa数据集和代码 * bugfix:prqa调用逻辑修复 * optimize：优化代码逻辑，生成答案规范化 * add retry py code * update memory graph * update memory graph * fix * fix ner * add with_out_refer generator prompt * fix * close ckpt * fix query * fix query * update version * add llm checker * add llm checker * 1、上传evalutor.py以及修改gold_answer.json格式 2、优化代码逻辑 3、修改README.md文件 * update exp * update exp * rerank support * add static rewrite query * recall more chunks * fix graph load * add static rewrite query * fix bugs * add finish check * add finish check * add finish check * add finish check * 1、上传evalutor.py的结果 2、优化代码逻辑，优化readme文件 * add lf retry * add memory graph api * fix reader api * add ner * add metrics * fix bug * remove ner * add reraise fo retry * add edge prop to memory graph * add memory graph * 1、评测数据集结果修正 2、优化evaluator.py代码 3、删除结果不存在而gold_answer中有答案的问题 * 删除评测结果文件 * fix knext host addr * async eva * add lf prompt * add lf prompt * add config * add retry * add unknown check * add rc result * add rc result * add rc result * add rc result * 依据kag pipeline格式修改代码逻辑并通过测试 * bugfix:删除冗余代码 * fix report prompt * bugfix:触发重试机制 * bugfix:中文符号错误 * fix rethinker prompt * update version to 0.6.2b78 * update version * 1、修改evaluator.py，通过大模型计算准确率，符合最新调用逻辑 2、修改prompt，让没有回答的结果重复测试 * update affairqa for evaluate * update affairqa for evaluate * bugfix:修正数据集 * bugfix:修正数据集 * bugfix:修正数据集 * fix name conflict * bugfix:删除错误问题 * bugfix:文件名命名错误导致evaluator失败 * update for affairqa eval * bugfix:修改代码保持evaluate逻辑一致 * x * update for affairqa readme * remove temp eval scripts * bugfix for math deduce * merge 0.6.2_dev * merge 0.6.2_dev * fix * update client addr * updated version * update for affairqa eval * evaUtils 支持中文 * fix affairqa eval: * remove unused example * update kag config * fix default value * update readme * fix init * 注释信息修改，并添加部分class说明 * update example config * Tc 0.7.0 (#459) * 提交affairQA 代码 * fix affairqa eval --------- Co-authored-by: zhengke.gzk * fix all examples * reformat --------- Co-authored-by: peilong Co-authored-by: 锦呈 Co-authored-by: wanxingyu.wxy Co-authored-by: zhengke.gzk --- kag/open_benchmark/2wiki/kag_config.yaml | 176 +++++++++++++++++++++++ 1 file changed, 176 insertions(+) create mode 100644 kag/open_benchmark/2wiki/kag_config.yaml diff --git a/kag/open_benchmark/2wiki/kag_config.yaml b/kag/open_benchmark/2wiki/kag_config.yaml new file mode 100644 index 000000000..92bcf60a9 --- /dev/null +++ b/kag/open_benchmark/2wiki/kag_config.yaml @@ -0,0 +1,176 @@ +openie_llm: &openie_llm + type: maas + base_url: https://dashscope.aliyuncs.com/compatible-mode/v1/ + api_key: key + model: qwen2.5-72b-instruct + +chat_llm: &chat_llm + type: maas + base_url: https://dashscope.aliyuncs.com/compatible-mode/v1/ + api_key: key + model: qwen2.5-72b-instruct + stream: True + +ner_llm: &ner_llm + type: maas + base_url: https://dashscope.aliyuncs.com/compatible-mode/v1/ + api_key: key + model: qwen2.5-72b-instruct + +vectorize_model: &vectorize_model + api_key: key + base_url: https://api.siliconflow.cn/v1/ + model: BAAI/bge-m3 + type: openai + vector_dimensions: 1024 +vectorizer: *vectorize_model + +log: + level: INFO + +project: + biz_scene: default + host_addr: http://127.0.0.1:8887 + id: '2' + language: en + namespace: TwoWiki +#------------project configuration end----------------# + +#------------kag-builder configuration start----------------# +kag_builder_pipeline: + chain: + type: unstructured_builder_chain # kag.builder.default_chain.DefaultUnstructuredBuilderChain + extractor: + type: schema_free_extractor # kag.builder.component.extractor.schema_free_extractor.SchemaFreeExtractor + llm: *openie_llm + ner_prompt: + type: default_ner # kag.builder.prompt.default.ner.OpenIENERPrompt + std_prompt: + type: default_std # kag.builder.prompt.default.std.OpenIEEntitystandardizationdPrompt + triple_prompt: + type: default_triple # kag.builder.prompt.default.triple.OpenIETriplePrompt + reader: + type: dict_reader # kag.builder.component.reader.dict_reader.DictReader + post_processor: + type: kag_post_processor # kag.builder.component.postprocessor.kag_postprocessor.KAGPostProcessor + splitter: + type: length_splitter # kag.builder.component.splitter.length_splitter.LengthSplitter + split_length: 100000 + window_length: 0 + vectorizer: + type: batch_vectorizer # kag.builder.component.vectorizer.batch_vectorizer.BatchVectorizer + vectorize_model: *vectorize_model + writer: + type: kg_writer # kag.builder.component.writer.kg_writer.KGWriter + num_threads_per_chain: 1 + num_chains: 16 + scanner: + type: 2wiki_dataset_scanner # kag.builder.component.scanner.dataset_scanner.MusiqueCorpusScanner +#------------kag-builder configuration end----------------# + +#------------kag-solver configuration start----------------# +search_api: &search_api + type: openspg_search_api #kag.solver.tools.search_api.impl.openspg_search_api.OpenSPGSearchAPI + +graph_api: &graph_api + type: openspg_graph_api #kag.solver.tools.graph_api.impl.openspg_graph_api.OpenSPGGraphApi + + + + +kg_cs: + type: kg_cs_open_spg + path_select: + type: exact_one_hop_select + graph_api: *graph_api + search_api: *search_api + entity_linking: + type: entity_linking + graph_api: *graph_api + search_api: *search_api + recognition_threshold: 0.9 + exclude_types: + - "Chunk" + +kg_fr: + type: kg_fr_open_spg + top_k: 20 + path_select: + type: fuzzy_one_hop_select + llm_client: *chat_llm + graph_api: *graph_api + search_api: *search_api + ppr_chunk_retriever_tool: + type: ppr_chunk_retriever + llm_client: *ner_llm + graph_api: *graph_api + search_api: *search_api + entity_linking: + type: entity_linking + graph_api: *graph_api + search_api: *search_api + recognition_threshold: 0.8 + exclude_types: + - "Chunk" + +rc: + type: rc_open_spg + vector_chunk_retriever: + type: vector_chunk_retriever + vectorize_model: *vectorize_model + search_api: *search_api + graph_api: *graph_api + search_api: *search_api + vectorize_model: *vectorize_model + top_k: 20 + +kag_merger: + type: kg_merger + top_k: 20 + llm_module: *chat_llm + summary_prompt: + type: default_thought_then_answer + vectorize_model: *vectorize_model + graph_api: *graph_api + search_api: *search_api + +kag_hybrid_executor: &kag_hybrid_executor_conf + type: kag_hybrid_executor + lf_rewriter: + type: kag_spo_lf + llm_client: *openie_llm + lf_trans_prompt: + type: default_spo_retriever_decompose + vectorize_model: *vectorize_model + flow: | + kg_cs->kg_fr->kag_merger;rc->kag_merger + +kag_output_executor: &kag_output_executor_conf + type: kag_output_executor +kag_deduce_executor: &kag_deduce_executor_conf + type: kag_deduce_executor + +py_code_based_math_executor: &py_code_based_math_executor_conf + type: py_code_based_math_executor + llm: *openie_llm + +kag_solver_pipeline: + type: kag_static_pipeline + planner: + type: lf_kag_static_planner + llm: *chat_llm + plan_prompt: + type: default_lf_static_planning + rewrite_prompt: + type: default_rewrite_sub_task_query + executors: + - *kag_hybrid_executor_conf + - *py_code_based_math_executor_conf + - *kag_deduce_executor_conf + - *kag_output_executor_conf + generator: + type: llm_generator + llm_client: *chat_llm + generated_prompt: + type: default_multi_hop_generator +#------------kag-solver configuration end----------------# From ff19d238861d28ed2ac6637b5c0c3820346da72c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=94=A6=E5=91=88?= Date: Tue, 1 Jul 2025 14:11:31 +0800 Subject: [PATCH 02/12] update chunk metadata --- .../component/extractor/table_extractor.py | 8 ++++---- .../component/reader/markdown_reader.py | 19 +++++++------------ kag/interface/common/model/chunk.py | 6 ++++-- 3 files changed, 15 insertions(+), 18 deletions(-) diff --git a/kag/builder/component/extractor/table_extractor.py b/kag/builder/component/extractor/table_extractor.py index 2c4ce5f32..c7368c1a6 100644 --- a/kag/builder/component/extractor/table_extractor.py +++ b/kag/builder/component/extractor/table_extractor.py @@ -100,8 +100,8 @@ def _table_row_chunk(self, input_table_chunk: Chunk): file_name = input_table_chunk.kwargs.get("file_name", "") name = f"{file_name} / {input_table_chunk.name}" content = f"{name}\n{input_table_chunk.content}" - before_text = input_table_chunk.metadata.get("before_text", "") - after_text = input_table_chunk.metadata.get("after_text", "") + before_text = input_table_chunk.get("before_text", "") + after_text = input_table_chunk.get("after_text", "") sub_graph = SubGraph(nodes=[], edges=[]) sub_graph.add_node( input_table_chunk.id, @@ -155,8 +155,8 @@ def _table_context(self, input_table: Chunk): def _get_table_context_str(self, table_chunk: Chunk): file_name = table_chunk.kwargs.get("file_name", "") section = table_chunk.name - before_text = table_chunk.kwargs["metadata"].get("before_text", "") - after_text = table_chunk.kwargs["metadata"].get("after_text", "") + before_text = table_chunk.get("before_text", "") + after_text = table_chunk.get("after_text", "") return ( file_name + "\n" diff --git a/kag/builder/component/reader/markdown_reader.py b/kag/builder/component/reader/markdown_reader.py index 3fe72a23e..49c19ec11 100644 --- a/kag/builder/component/reader/markdown_reader.py +++ b/kag/builder/component/reader/markdown_reader.py @@ -787,10 +787,8 @@ def _create_table_chunks( name=f"{full_title} / Table {i+1}", content=table_content, type=ChunkTypeEnum.Table, - metadata={ - "before_text": table.get("context", {}).get("before_text", ""), - "after_text": table.get("context", {}).get("after_text", ""), - }, + before_text=table.get("context", {}).get("before_text", ""), + after_text=table.get("context", {}).get("after_text", ""), file_name=os.path.basename(file_id), ) table_chunks.append(table_chunk) @@ -807,10 +805,8 @@ def _create_table_chunks( name=f"{full_title} / Table {i+1}", content=table_content, type=ChunkTypeEnum.Table, - metadata={ - "before_text": table.get("context", {}).get("before_text", ""), - "after_text": table.get("context", {}).get("after_text", ""), - }, + before_text=table.get("context", {}).get("before_text", ""), + after_text=table.get("context", {}).get("after_text", ""), file_name=os.path.basename(file_id), ) table_chunks.append(table_chunk) @@ -862,10 +858,8 @@ def _create_comprehensive_chunk( name=f"{full_title} / Table {i+1}", content=table_content, type=ChunkTypeEnum.Table, - metadata={ - "before_text": table.get("context", {}).get("before_text", ""), - "after_text": table.get("context", {}).get("after_text", ""), - }, + before_text=table.get("context", {}).get("before_text", ""), + after_text=table.get("context", {}).get("after_text", ""), file_name=os.path.basename(id), ) outputs.append(table_chunk) @@ -1065,5 +1059,6 @@ def _invoke(self, input: Input, **kwargs) -> List[Output]: file_path = os.path.join( dir_path, "../../../../tests/unit/builder/data", "需求内容test.md" ) + file_path = "/Users/zhangxinhong.zxh/Downloads/overmemery.md" chunks = reader.invoke(file_path, write_ckpt=False) print(chunks) diff --git a/kag/interface/common/model/chunk.py b/kag/interface/common/model/chunk.py index 526fffcd5..87b459473 100644 --- a/kag/interface/common/model/chunk.py +++ b/kag/interface/common/model/chunk.py @@ -61,7 +61,7 @@ def to_dict(self): "type": ( self.type.value if isinstance(self.type, ChunkTypeEnum) else self.type ), - "properties": self.kwargs, + "properties": json.dumps(self.kwargs), } @classmethod @@ -71,7 +71,9 @@ def from_dict(cls, input_: Dict[str, Any]): name=input_.get("name"), content=input_.get("content"), type=input_.get("type"), - **input_.get("properties", {}), + **json.loads(input_.get("properties", {})) + if isinstance(input_.get("properties"), str) + else input_.get("properties", {}), ) From d50bbb17919de761c54376607db88da675ef517d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=94=A6=E5=91=88?= Date: Tue, 1 Jul 2025 14:51:52 +0800 Subject: [PATCH 03/12] update chunk metadata --- kag/builder/component/extractor/table_extractor.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/kag/builder/component/extractor/table_extractor.py b/kag/builder/component/extractor/table_extractor.py index c7368c1a6..32d1d5819 100644 --- a/kag/builder/component/extractor/table_extractor.py +++ b/kag/builder/component/extractor/table_extractor.py @@ -100,8 +100,8 @@ def _table_row_chunk(self, input_table_chunk: Chunk): file_name = input_table_chunk.kwargs.get("file_name", "") name = f"{file_name} / {input_table_chunk.name}" content = f"{name}\n{input_table_chunk.content}" - before_text = input_table_chunk.get("before_text", "") - after_text = input_table_chunk.get("after_text", "") + before_text = getattr(input_table_chunk, "before_text", "") + after_text = getattr(input_table_chunk, "after_text", "") sub_graph = SubGraph(nodes=[], edges=[]) sub_graph.add_node( input_table_chunk.id, @@ -155,8 +155,8 @@ def _table_context(self, input_table: Chunk): def _get_table_context_str(self, table_chunk: Chunk): file_name = table_chunk.kwargs.get("file_name", "") section = table_chunk.name - before_text = table_chunk.get("before_text", "") - after_text = table_chunk.get("after_text", "") + before_text = getattr(table_chunk, "before_text", "") + after_text = getattr(table_chunk, "after_text", "") return ( file_name + "\n" From 89a3496de6146a8daf11985925bdd024829b92f6 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Wed, 2 Jul 2025 14:56:14 +0800 Subject: [PATCH 04/12] add debug reporter --- kag/common/utils.py | 14 +++- kag/common/vectorize_model/openai_model.py | 2 +- .../kag_hybrid_retrieval_executor.py | 51 +++++++++++---- .../reporter/open_spg_kag_model_reporter.py | 7 +- knext/common/env.py | 17 +++-- tests/unit/common/kag_utils.py | 64 +++++++++++++++++++ 6 files changed, 128 insertions(+), 27 deletions(-) create mode 100644 tests/unit/common/kag_utils.py diff --git a/kag/common/utils.py b/kag/common/utils.py index 7614eea79..40757bf95 100644 --- a/kag/common/utils.py +++ b/kag/common/utils.py @@ -463,9 +463,17 @@ def resolve_instance( def extract_tag_content(text): - # 匹配和之间的内容，支持任意标签名 - matches = re.findall(r"<([^>]+)>(.*?)", text, flags=re.DOTALL) - return [(tag, content.strip()) for tag, content in matches] + pattern = r"<(\w+)\b[^>]*>(.*?)|<(\w+)\b[^>]*>([^<]*)|([^<]+)" + results = [] + for match in re.finditer(pattern, text, re.DOTALL): + tag1, content1, tag2, content2, raw_text = match.groups() + if tag1: + results.append((tag1, content1)) # 保留原始内容（含空格） + elif tag2: + results.append((tag2, content2)) # 保留原始内容（含空格） + elif raw_text: + results.append(("", raw_text)) # 保留原始空格 + return results def extract_specific_tag_content(text, tag): diff --git a/kag/common/vectorize_model/openai_model.py b/kag/common/vectorize_model/openai_model.py index c3a0ed8f5..2a5a417ab 100644 --- a/kag/common/vectorize_model/openai_model.py +++ b/kag/common/vectorize_model/openai_model.py @@ -54,7 +54,7 @@ def __init__( self.aclient = AsyncOpenAI(api_key=api_key, base_url=base_url) @classmethod - def generate_key(cls, base_url, api_key, model, *args, **kwargs) -> str: + def generate_key(cls, base_url, model, api_key="", *args, **kwargs) -> str: return f"{cls}_{base_url}_{api_key}_{model}" def vectorize( diff --git a/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py b/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py index 455f1eeb8..79e0e4142 100644 --- a/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py +++ b/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py @@ -42,6 +42,15 @@ logger = logging.getLogger() +def _wrapped_invoke(retriever, task, context, segment_name, kwargs): + start_time = time.time() + output = retriever.invoke( + task, context=context, segment_name=segment_name, **kwargs + ) + elapsed_time = time.time() - start_time + return output, elapsed_time + + @ExecutorABC.register("kag_hybrid_retrieval_executor") class KAGHybridRetrievalExecutor(ExecutorABC): def __init__( @@ -76,6 +85,7 @@ def __init__( self.context_select_prompt = context_select_prompt or PromptABC.from_config( {"type": "context_select_prompt"} ) + self.with_llm_select = kwargs.get("with_llm_select", True) @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1)) def context_select_call(self, variables): @@ -152,22 +162,30 @@ def do_retrieval( "FINISH", component_name=retriever.name, ) - + # Record start time before submitting the task + start_time = time.time() # Prepare function and submit to thread pool func = partial( - retriever.invoke, + _wrapped_invoke, + retriever, task, - context=context, - segment_name=tag_id, - **kwargs, + context, + tag_id, + kwargs.copy(), ) future = executor.submit(func) + # Save future, retriever, and start_time together futures.append((future, retriever)) # Collect results from each future for future, retriever in futures: try: - output = future.result() # Wait for result + output, elapsed_time = future.result() # Wait for result + + # Log the elapsed time for this retriever + logger.info( + f"Retriever {retriever.name} executed in {elapsed_time:.2f} seconds" + ) outputs.append(output) # Log data report after successful execution @@ -241,13 +259,18 @@ def do_summary( selected_rel = list(set(selected_rel)) formatted_docs = [str(rel) for rel in selected_rel] if retrieved_data.chunks: - try: - selected_chunks = self.context_select(task_query, retrieved_data.chunks) - except Exception as e: - logger.warning( - f"select context failed {e}, we use default top 10 to summary", - exc_info=True, - ) + if self.with_llm_select: + try: + selected_chunks = self.context_select( + task_query, retrieved_data.chunks + ) + except Exception as e: + logger.warning( + f"select context failed {e}, we use default top 10 to summary", + exc_info=True, + ) + selected_chunks = retrieved_data.chunks[:10] + else: selected_chunks = retrieved_data.chunks[:10] for doc in selected_chunks: formatted_docs.append(f"{doc.content}") @@ -403,7 +426,7 @@ def do_data_report( node_type=chunk.properties.get("__labels__"), ) entity_prop = dict(chunk.properties) if chunk.properties else {} - entity_prop["content"] = chunk.content + entity_prop["content"] = f"{chunk.content[:10]}..." entity_prop["score"] = chunk.score entity.prop = Prop.from_dict(entity_prop, "Chunk", None) chunk_graph.append(entity) diff --git a/kag/solver/reporter/open_spg_kag_model_reporter.py b/kag/solver/reporter/open_spg_kag_model_reporter.py index e5ceae7d2..d569a16b9 100644 --- a/kag/solver/reporter/open_spg_kag_model_reporter.py +++ b/kag/solver/reporter/open_spg_kag_model_reporter.py @@ -1,5 +1,6 @@ import logging import re +import time from kag.common.conf import KAG_PROJECT_CONF from kag.common.parser.logic_node_parser import extract_steps_and_actions @@ -72,14 +73,16 @@ def process_tag_template(text): } clean_text = "" for tag_info in all_tags: + content = tag_info[1] if tag_info[0] in xml_tag_template: - content = tag_info[1] if "search" == tag_info[0]: content = process_planning(content) clean_text += xml_tag_template[tag_info[0]][ KAG_PROJECT_CONF.language ].format_map(SafeDict({"content": content})) - return remove_xml_tags(clean_text) + else: + clean_text += content + text = remove_xml_tags(clean_text) return text diff --git a/knext/common/env.py b/knext/common/env.py index 6a9994c73..595c8d423 100644 --- a/knext/common/env.py +++ b/knext/common/env.py @@ -48,13 +48,16 @@ def __new__(cls): @property def config(self): - closest_config = self._closest_config() - if not hasattr(self, "_config_path") or self._config_path != closest_config: - self._config_path = closest_config - self._config = self.get_config() - - if self._config is None: - self._config = self.get_config() + try: + closest_config = self._closest_config() + if not hasattr(self, "_config_path") or self._config_path != closest_config: + self._config_path = closest_config + self._config = self.get_config() + + if self._config is None: + self._config = self.get_config() + except: + return {} return self._config diff --git a/tests/unit/common/kag_utils.py b/tests/unit/common/kag_utils.py new file mode 100644 index 000000000..5d18d8533 --- /dev/null +++ b/tests/unit/common/kag_utils.py @@ -0,0 +1,64 @@ +from kag.common.utils import extract_tag_content + + +def run_extra_tag(): + test_cases = [ + { + "input": "abcedsome wordother tags", + "expected": [("tag1", "abced"), ("", "some word"), ("tag2", "other tags")], + "description": "基本闭合标签与无标签文本混合", + }, + { + "input": "

Hello world this is test", + "expected": [ + ("p", "Hello "), + ("b", "world"), + ("", " this is "), + ("i", "test"), + ], + "description": "混合闭合与未闭合标签", + }, + { + "input": "plain text without any tags", + "expected": [("", "plain text without any tags")], + "description": "纯文本无标签", + }, + { + "input": "

\n Line 1\n Line 2\n Line 3\n
", + "expected": [ + ("div", "\n Line 1\n Line 2\n Line 3\n") + ], + "description": "多行内容和空白处理", + }, + { + "input": "ABC", + "expected": [("a", "A"), ("b", "B"), ("c", "C")], + "description": "连续多个闭合标签", + }, + { + "input": "My DocumentThis is the content", + "expected": [("title", "My Document"), ("content", "This is the content")], + "description": "未闭合标签（EOF结尾）", + }, + { + "input": "Error: &*^%$#@!;End of log", + "expected": [("log", "Error: &*^%$#@!;"), ("note", "End of log")], + "description": "含特殊字符的内容", + }, + { + "input": "", + "expected": [], + "description": "空字符串输入", + }, + ] + + for i, test in enumerate(test_cases): + result = extract_tag_content(test["input"]) + assert ( + result == test["expected"] + ), f"Test {i+1} failed: {test['description']}\nGot: {result}\nExpected: {test['expected']}" + print(f"Test {i+1} passed: {test['description']}") + + +if __name__ == "__main__": + run_extra_tag() From 4f4160dcefc0da87ffc916b6bf72db8c73623994 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=94=A6=E5=91=88?= Date: Wed, 2 Jul 2025 16:07:37 +0800 Subject: [PATCH 05/12] update table text --- kag/builder/component/extractor/table_extractor.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/kag/builder/component/extractor/table_extractor.py b/kag/builder/component/extractor/table_extractor.py index 32d1d5819..f42b9b4cc 100644 --- a/kag/builder/component/extractor/table_extractor.py +++ b/kag/builder/component/extractor/table_extractor.py @@ -111,8 +111,8 @@ def _table_row_chunk(self, input_table_chunk: Chunk): "id": input_table_chunk.id, "name": name, "content": content, - "beforeText": before_text, - "afterText": after_text, + "before_text": before_text, + "after_text": after_text, }, ) From 63b5b52a4624d09977eba7ab3f7fd247cf4a4c8e Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Wed, 2 Jul 2025 20:53:19 +0800 Subject: [PATCH 06/12] add server --- .../kag_hybrid_retrieval_executor.py | 115 +++++++------ kag/solver/main_solver.py | 4 +- kag/solver/reporter/open_spg_reporter.py | 138 +++++++++------ kag/solver/server/__init__.py | 0 kag/solver/server/asyn_task_manager.py | 160 ++++++++++++++++++ kag/solver/server/example/__init__.py | 0 kag/solver/server/main_server.py | 61 +++++++ kag/solver/server/model/__init__.py | 0 kag/solver/server/model/task_req.py | 109 ++++++++++++ kag/solver/server/requirement.txt | 1 + 10 files changed, 476 insertions(+), 112 deletions(-) create mode 100644 kag/solver/server/__init__.py create mode 100644 kag/solver/server/asyn_task_manager.py create mode 100644 kag/solver/server/example/__init__.py create mode 100644 kag/solver/server/main_server.py create mode 100644 kag/solver/server/model/__init__.py create mode 100644 kag/solver/server/model/task_req.py create mode 100644 kag/solver/server/requirement.txt diff --git a/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py b/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py index 79e0e4142..bbf5e69f3 100644 --- a/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py +++ b/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py @@ -303,68 +303,71 @@ def invoke(self, query, task, context: Context, **kwargs) -> RetrieverOutput: task_query = task.arguments["query"] tag_id = f"{task_query}_begin_task" - self.report_content(reporter, "thinker", tag_id, "", "FINISH", step=task.name) + self.report_content(reporter, "thinker", tag_id, "", "INIT", step=task.name) try: - retrieved_data = self.do_main(task_query, tag_id, task, context, **kwargs) - except Exception as e: - logger.warning(f"kag hybrid retrieval failed! {e}", exc_info=True) - retrieved_data = RetrieverOutput( - retriever_method=self.schema().get("name", ""), err_msg=str(e) - ) + try: + retrieved_data = self.do_main(task_query, tag_id, task, context, **kwargs) + except Exception as e: + logger.warning(f"kag hybrid retrieval failed! {e}", exc_info=True) + retrieved_data = RetrieverOutput( + retriever_method=self.schema().get("name", ""), err_msg=str(e) + ) - self.report_content( - reporter, - "reference", - f"{task_query}_kag_retriever_result", - retrieved_data, - "FINISH", - ) + self.report_content( + reporter, + "reference", + f"{task_query}_kag_retriever_result", + retrieved_data, + "FINISH", + ) - retrieved_data.task = task - logical_node = task.arguments.get("logic_form_node", None) - if ( - logical_node - and isinstance(logical_node, GetSPONode) - and retrieved_data.summary - ): - if isinstance(retrieved_data.summary, str): - target_answer = retrieved_data.summary.split("Answer:")[-1].strip() - s_entities = context.variables_graph.get_entity_by_alias( - logical_node.s.alias_name + retrieved_data.task = task + logical_node = task.arguments.get("logic_form_node", None) + if ( + logical_node + and isinstance(logical_node, GetSPONode) + and retrieved_data.summary + ): + if isinstance(retrieved_data.summary, str): + target_answer = retrieved_data.summary.split("Answer:")[-1].strip() + s_entities = context.variables_graph.get_entity_by_alias( + logical_node.s.alias_name + ) + if ( + not s_entities + and not logical_node.s.get_mention_name() + and isinstance(logical_node.s, SPOEntity) + ): + logical_node.s.entity_name = target_answer + context.kwargs[logical_node.s.alias_name] = logical_node.s + o_entities = context.variables_graph.get_entity_by_alias( + logical_node.o.alias_name + ) + if ( + not o_entities + and not logical_node.o.get_mention_name() + and isinstance(logical_node.o, SPOEntity) + ): + logical_node.o.entity_name = target_answer + context.kwargs[logical_node.o.alias_name] = logical_node.o + + context.variables_graph.add_answered_alias( + logical_node.s.alias_name.alias_name, retrieved_data.summary ) - if ( - not s_entities - and not logical_node.s.get_mention_name() - and isinstance(logical_node.s, SPOEntity) - ): - logical_node.s.entity_name = target_answer - context.kwargs[logical_node.s.alias_name] = logical_node.s - o_entities = context.variables_graph.get_entity_by_alias( - logical_node.o.alias_name + context.variables_graph.add_answered_alias( + logical_node.p.alias_name.alias_name, retrieved_data.summary + ) + context.variables_graph.add_answered_alias( + logical_node.o.alias_name.alias_name, retrieved_data.summary ) - if ( - not o_entities - and not logical_node.o.get_mention_name() - and isinstance(logical_node.o, SPOEntity) - ): - logical_node.o.entity_name = target_answer - context.kwargs[logical_node.o.alias_name] = logical_node.o - - context.variables_graph.add_answered_alias( - logical_node.s.alias_name.alias_name, retrieved_data.summary - ) - context.variables_graph.add_answered_alias( - logical_node.p.alias_name.alias_name, retrieved_data.summary - ) - context.variables_graph.add_answered_alias( - logical_node.o.alias_name.alias_name, retrieved_data.summary - ) - task.update_result(retrieved_data) - logger.debug( - f"kag hybrid retrieval {task_query} cost={time.time() - start_time}" - ) - return retrieved_data + task.update_result(retrieved_data) + logger.debug( + f"kag hybrid retrieval {task_query} cost={time.time() - start_time}" + ) + return retrieved_data + finally: + self.report_content(reporter, "thinker", tag_id, "", "FINISH", step=task.name, overwrite=False,) def schema(self) -> dict: """Function schema definition for OpenAI Function Calling diff --git a/kag/solver/main_solver.py b/kag/solver/main_solver.py index f2fb4b165..63da7bbd8 100644 --- a/kag/solver/main_solver.py +++ b/kag/solver/main_solver.py @@ -140,8 +140,6 @@ def get_pipeline_conf(use_pipeline_name, config): raise RuntimeError("mcpServers not found in config.") default_solver_pipeline["executors"] = mcp_executors - # update KAG_CONFIG - KAG_CONFIG.update_conf(default_pipeline_conf) return default_solver_pipeline @@ -339,7 +337,7 @@ class SolverMain: def invoke( self, project_id: int, - task_id: int, + task_id, query: str, session_id: str = "0", is_report=True, diff --git a/kag/solver/reporter/open_spg_reporter.py b/kag/solver/reporter/open_spg_reporter.py index ef2ae01a7..f8e96a890 100644 --- a/kag/solver/reporter/open_spg_reporter.py +++ b/kag/solver/reporter/open_spg_reporter.py @@ -159,12 +159,12 @@ def render_jinja2_template(template_str, context): """ try: template = Template(template_str, undefined=SilentUndefined) - return template.render(**context).strip() + return template.render(**context) except Exception as e: logging.error( f"Jinja2 rendering failed: {e}, Original template: {template_str}" ) - return template_str.strip() # Fallback to raw template string on failure + return template_str # Fallback to raw template string on failure @ReporterABC.register("open_spg_reporter") @@ -264,12 +264,12 @@ def __init__(self, task_id, host_addr=None, project_id=None, **kwargs): } self.tag_mapping = { "Graph Show": { - "en": "{content}", - "zh": "{content}", + "en": "{{ content }}", + "zh": "{{ content }}", }, "Rewrite query": { - "en": "Rethinking question using LLM: {content}", - "zh": "根据依赖问题重写子问题: {content}", + "en": "Rethinking question using LLM: {{ content }}", + "zh": "根据依赖问题重写子问题: {{ content }}", }, "language_setting": { "en": "", @@ -277,125 +277,153 @@ def __init__(self, task_id, host_addr=None, project_id=None, **kwargs): }, "Iterative planning": { "en": """ - + -{content} +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", "zh": """ - + -{content} +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", }, "Static planning": { "en": """ - + -{content} +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", "zh": """ - + -{content} +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", }, "begin_sub_kag_retriever": { - "en": "Starting {component_name}: {content} {desc}", - "zh": "执行{component_name}: {content} {desc}", + "en": "Starting {{component_name}}: {{content}} {{desc}}", + "zh": "执行{{component_name}}: {{content}} {{desc}}", }, "end_sub_kag_retriever": { - "en": " {content}", - "zh": " {content}", + "en": " {{ content }}", + "zh": " {{ content }}", }, "rc_retriever_rewrite": { "en": """ - + -Rewritten question:\n{content} +Rewritten question: +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", "zh": """ - + -重写问题为：\n\n{content} +重写问题为： +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", }, "rc_retriever_summary": { - "en": "Summarizing retrieved documents,{content}", - "zh": "对文档进行总结，{content}", + "en": "Summarizing retrieved documents,{{ content }}", + "zh": "对文档进行总结，{{ content }}", }, "kg_retriever_summary": { - "en": "Summarizing retrieved graph,{content}", - "zh": "对召回的知识进行总结，{content}", + "en": "Summarizing retrieved graph,{{ content }}", + "zh": "对召回的知识进行总结，{{ content }}", }, "retriever_summary": { - "en": "Summarizing retrieved documents,{content}", - "zh": "对文档进行总结，{content}", + "en": "Summarizing retrieved documents,{{ content }}", + "zh": "对文档进行总结，{{ content }}", }, "begin_summary": { - "en": "Summarizing retrieved information, {content}", - "zh": "对检索的信息进行总结, {content}", + "en": "Summarizing retrieved information, {{ content }}", + "zh": "对检索的信息进行总结, {{ content }}", }, "begin_task": { "en": """ - + -{content} +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", "zh": """ - + -{content} +{{ content }} -""", +{% if status == 'success' %} + +{% endif %}""", }, "logic_node": { "en": """Translate query to logic form expression ```json -{content} +{{ content }} ```""", "zh": """将query转换成逻辑形式表达 ```json -{content} +{{ content }} ```""", }, "kag_retriever_result": { - "en": "Retrieved documents\n\n{content}", - "zh": "检索到的文档\n\n{content}", + "en": """Retrieved documents +{{ content }}""", + "zh": """检索到的文档 +{{ content }}""", }, "failed_kag_retriever": { "en": """KAG retriever failed ```json -{content} +{{ content }} ``` """, "zh": """KAG检索失败 ```json -{content} +{{ content }} ``` """, }, "end_math_executor": { - "en": "Math executor completed\n\n{content}", - "zh": "计算结束\n\n{content}", + "en": """Math executor completed +{{ content }}""", + "zh": """计算结束 +{{ content }}""", }, "code_generator": { - "en": "Generating code\n \n{content}\n", - "zh": "正在生成代码\n \n{content}\n", + "en": """Generating code +{{ content }} + +""", + "zh": """正在生成代码 +{{ content }} + +""", }, } task_id = kwargs.get(KAGConstants.KAG_QA_TASK_CONFIG_KEY, None) @@ -425,7 +453,11 @@ def generate_content(self, report_id, tpl, datas, content_params, graph_list): if tpl: format_params = {"content": datas} format_params.update(content_params) - datas = tpl.format_map(SafeDict(format_params)) + if "{" in tpl or "%}" in tpl: + rendered = render_jinja2_template(tpl, format_params) + else: + rendered = tpl.format_map(SafeDict(format_params)) + datas = rendered elif str(datas).strip() != "": output = str(datas).strip() if output != "": diff --git a/kag/solver/server/__init__.py b/kag/solver/server/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/kag/solver/server/asyn_task_manager.py b/kag/solver/server/asyn_task_manager.py new file mode 100644 index 000000000..de525f432 --- /dev/null +++ b/kag/solver/server/asyn_task_manager.py @@ -0,0 +1,160 @@ +import concurrent.futures +import queue +import threading +import time +import uuid +import logging +from cachetools import TTLCache + +logger = logging.getLogger() + + +class AsyncTaskManager: + def __init__(self, max_workers=10, ttl=3600): + """ + Initialize async task manager + + Args: + max_workers (int): Maximum number of worker threads + ttl (int): Time-to-live for task results in seconds + """ + self.max_workers = max_workers + self.task_queue = queue.Queue() + self.result_cache = TTLCache(maxsize=1000, ttl=ttl) + self.result_cache_lock = threading.Lock() # Protect cache from race conditions + self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) + self.workers = [threading.Thread(target=self.worker, daemon=True) for _ in range(max_workers)] + for w in self.workers: + w.start() + + def worker(self): + """Worker thread main loop that processes tasks""" + while True: + try: + # Get next task from queue with timeout to allow shutdown detection + task = self.task_queue.get() + task_id, func, args, kwargs = task + logger.info(f"Processing task {task_id}") + # finish flag + if task_id is None: + self.task_queue.task_done() + break + + # Update cache with running status + with self.result_cache_lock: + self.result_cache[task_id] = { + "task_id": task_id, + "status": "running", + "result": None + } + + # Execute task + future = self.executor.submit(func, *args, **kwargs) + result = future.result() + status = "completed" + + except queue.Empty: + # Handle queue empty timeout (normal operation) + continue + + except Exception as e: + # Handle task execution errors + result = str(e) + status = "failed" + logger.error(f"Task {task_id} failed with error: {e}", exc_info=True) + + # Store final result in cache + try: + with self.result_cache_lock: + self.result_cache[task_id] = { + "task_id": task_id, + "status": status, + "result": result + } + logger.info(f"Task {task_id} completed with status: {status}") + finally: + # Always mark task as done + self.task_queue.task_done() + + def submit_task(self, func, *args, **kwargs): + """ + Submit a new task to the queue + + Args: + func: Callable function to execute + *args: Positional arguments for the function + **kwargs: Keyword arguments for the function + + Returns: + str: Unique task ID + """ + task_id = str(uuid.uuid4()) + self.task_queue.put((task_id, func, args, kwargs)) + return task_id + + def get_task_result(self, task_id): + """ + Get result for a specific task + + Args: + task_id (str): Unique task identifier + + Returns: + dict: Task result information or expired status + """ + with self.result_cache_lock: + return self.result_cache.get( + task_id, + {"task_id": task_id, "status": "failed", "result": "Result not found or expired"} + ) + + def shutdown(self): + """Gracefully shutdown all worker threads and executors""" + # Send shutdown signals + for _ in range(self.max_workers): + self.task_queue.put((None, None, (), {})) + + # Wait for queue to empty and workers to terminate + self.task_queue.join() + + # Shutdown executors + self.executor.shutdown(wait=True) + for worker in self.workers: + worker.join(timeout=5) + + +# Global async task manager instance +asyn_task = AsyncTaskManager() + +if __name__ == "__main__": + # Configure logging + logging.basicConfig(level=logging.INFO) + + # Create task manager instance + task_manager = AsyncTaskManager(max_workers=5, ttl=600) + + + # Example task function + def example_task(x, y): + time.sleep(1) # Simulate work + return x + + + # Submit test tasks + task_ids = [task_manager.submit_task(example_task, i, i + 1) for i in range(6)] + + # Monitor task progress + try: + while True: + time.sleep(1) + if all("completed" in task_manager.get_task_result(tid)["status"] for tid in task_ids): + break + except KeyboardInterrupt: + logger.info("Shutting down due to user interrupt") + + # Print results + for task_id in task_ids: + print(f"Task {task_id} result: {task_manager.get_task_result(task_id)}") + + # Clean up resources + task_manager.shutdown() diff --git a/kag/solver/server/example/__init__.py b/kag/solver/server/example/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/kag/solver/server/main_server.py b/kag/solver/server/main_server.py new file mode 100644 index 000000000..ad49d9601 --- /dev/null +++ b/kag/solver/server/main_server.py @@ -0,0 +1,61 @@ +from fastapi import FastAPI +import uvicorn + +from kag.solver.main_solver import SolverMain +from kag.solver.server.asyn_task_manager import AsyncTaskManager +from kag.solver.server.model.task_req import FeatureRequest, TaskReq + + +def run_main_solver(task: TaskReq): + return SolverMain().invoke( + project_id=task.project_id, + task_id=task.req_id, + query=task.req.query, + is_report=task.req.report, + host_addr=task.req.host_addr, + app_id=task.app_id, + params=task.config, + ) + +class KAGSolverServer: + def __init__(self, service_name: str): + """ + 初始化 FastAPI 服务实例 + + Args: + service_name (str): 服务名称，决定加载哪个路由逻辑 + """ + self.service_name = service_name + self.app = FastAPI(title=f"{service_name} API") + + # 根据服务名绑定路由 + self._setup_routes() + self.async_manager = AsyncTaskManager() + + def sync_task(self, task: TaskReq): + if task.cmd == "submit": + + return { + 'success': True, + 'status': 'init', + 'result': self.async_manager.submit_task(run_main_solver, task) + } + elif task.cmd == "query": + return self.async_manager.get_task_result(task_id=task.req_id) + else: + return { + "success": False, + "status": "failed", + "result": f"invalid input cmd {task.cmd}", + } + + def _setup_routes(self): + """根据服务名动态绑定路由""" + @self.app.post("/process") + def process(req: FeatureRequest): + return self.sync_task(task=req.features.task_req) + + def run(self, host="0.0.0.0", port=8000): + """启动服务""" + print(f"Starting {self.service_name} service on {host}:{port}") + uvicorn.run(self.app, host=host, port=port) \ No newline at end of file diff --git a/kag/solver/server/model/__init__.py b/kag/solver/server/model/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/kag/solver/server/model/task_req.py b/kag/solver/server/model/task_req.py new file mode 100644 index 000000000..f432e2508 --- /dev/null +++ b/kag/solver/server/model/task_req.py @@ -0,0 +1,109 @@ +import json +from typing import Optional + +from pydantic import BaseModel, model_validator, field_serializer + + +class ReqBody(BaseModel): + query: str = "" + report: bool = True + host_addr: str = "" + + +class TaskReq(BaseModel): + app_id: int = "" + project_id: int = 0 + req_id: str = "" + cmd: str = "" + mode: str = "" + req: str = None + config: str = "{}" + + @model_validator(mode="after") + def parse_req_to_req_body(self): + try: + import json + if isinstance(self.req, str): + req_body_dict = json.loads(self.req) + self.req = ReqBody(**req_body_dict) + if isinstance(self.config, str) and self.config: + config_dict = json.loads(self.config) + self.config = config_dict + except Exception as e: + raise ValueError(f"Failed to parse 'req' field to ReqBody: {e}") + return self + + @field_serializer("req") + def serialize_req(self, value: object) -> object: + """将 ReqBody 再次序列化为 JSON 字符串""" + if isinstance(value, ReqBody): + return value.model_dump_json() + return value # 如果仍为字符串则直接返回 + + +# 新增的 Request 模型 +class Request(BaseModel): + in_string: str + task_req: Optional[TaskReq] = None + + @model_validator(mode="after") + def parse_in_string_to_task_req(self): + try: + import json + task_req_dict = json.loads(self.in_string) + self.task_req = TaskReq(**task_req_dict) + except Exception as e: + raise ValueError(f"Invalid TaskReq JSON string: {e}") + return self + + +class FeatureRequest(BaseModel): + features: Request + + +if __name__ == "__main__": + def feature_request_parsing(): + # 构建最内层的 ReqBody JSON 字符串 + req_body = ReqBody(query="阿里巴巴财报中，2024年-截至9月30日止六个月的收入是多少？其中云智能集团收入是多少？占比是多少", report=True, host_addr="https://spg.alipay.com") + req_body_json = json.dumps(req_body.model_dump()) + + # 构建 TaskReq 字典并序列化成字符串 + task_req = TaskReq( + req_id="9400110", + cmd="submit", + mode="async", + req=req_body_json, + app_id="app_id", + project_id=4200050, + config={"timeout": 10} + ) + task_req_json = json.dumps(task_req.model_dump()) + + # 构造最终传入的 FeatureRequest JSON 字符串 + input_data = { + "features": { + "in_string": task_req_json + } + } + + # 反序列化为 FeatureRequest 模型 + feature_request = FeatureRequest(**input_data) + + # 验证 in_string 被解析为 TaskReq + assert isinstance(feature_request.features.task_req, TaskReq) + assert feature_request.features.task_req.req_id == "abc123" + assert feature_request.features.task_req.cmd == "run" + assert feature_request.features.task_req.mode == "sync" + assert feature_request.features.task_req.config == {"timeout": 10} + + # 验证 TaskReq.req 被解析为 ReqBody + req_body_parsed = feature_request.features.task_req.req + assert isinstance(req_body_parsed, ReqBody) + assert req_body_parsed.query == "What is AI?" + assert req_body_parsed.report is True + assert req_body_parsed.host_addr == "localhost" + + print("✅ All assertions passed!") + + + feature_request_parsing() diff --git a/kag/solver/server/requirement.txt b/kag/solver/server/requirement.txt new file mode 100644 index 000000000..170703df6 --- /dev/null +++ b/kag/solver/server/requirement.txt @@ -0,0 +1 @@ +fastapi \ No newline at end of file From ea5685e8df1aa801e2c3862c3ed32930780d3210 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Thu, 3 Jul 2025 10:11:08 +0800 Subject: [PATCH 07/12] fix math executor --- .../executor/math/py_based_math_executor.py | 8 ++- kag/solver/planner/kag_model_planner.py | 1 + kag/solver/prompt/expression_builder.py | 60 ++++++++++++------- kag/solver/reporter/open_spg_reporter.py | 2 +- 4 files changed, 47 insertions(+), 24 deletions(-) diff --git a/kag/solver/executor/math/py_based_math_executor.py b/kag/solver/executor/math/py_based_math_executor.py index 544ff674d..6c172bd0f 100644 --- a/kag/solver/executor/math/py_based_math_executor.py +++ b/kag/solver/executor/math/py_based_math_executor.py @@ -131,9 +131,11 @@ def invoke(self, query: str, task: Task, context: Context, **kwargs): ) parent_results = format_task_dep_context(task.parents) - parent_results = "\n".join(parent_results) + coder_content = context.kwargs.get("planner_thought", "") + "\n\n".join( + parent_results + ) - parent_results += "\n\n" + contents + coder_content += "\n\n" + contents tries = self.tries error = None @@ -141,7 +143,7 @@ def invoke(self, query: str, task: Task, context: Context, **kwargs): tries -= 1 rst, error, code = self.run_once( math_query, - parent_results, + coder_content, error, segment_name=tag_id, tag_name=f"{task_query}_code_generator", diff --git a/kag/solver/planner/kag_model_planner.py b/kag/solver/planner/kag_model_planner.py index 62bcd6c20..008ac3c1a 100644 --- a/kag/solver/planner/kag_model_planner.py +++ b/kag/solver/planner/kag_model_planner.py @@ -186,6 +186,7 @@ async def ainvoke(self, query, **kwargs) -> List[Task]: .replace("", "") .strip() ) + context.kwargs["planner_thought"] = logic_form_response sub_queries, logic_forms = parse_logic_form_with_str(logic_form_str) logic_forms = self.logic_node_parser.parse_logic_form_set( diff --git a/kag/solver/prompt/expression_builder.py b/kag/solver/prompt/expression_builder.py index 4002bb484..36c25dbd1 100644 --- a/kag/solver/prompt/expression_builder.py +++ b/kag/solver/prompt/expression_builder.py @@ -11,42 +11,49 @@ class ExpressionBuildr(PromptABC): template_zh = ( f"今天是{get_now(language='zh')}。" - + """\n# instruction + + """ +# instruction 根据给出的问题和数据，编写python代码，输出问题结果。 -为了便于理解，输出从context中提取的数据，输出中间计算过程和结果。 -注意严格根据输入内容进行编写代码,不允许进行假设 -例如伤残等级如果context中未提及,则认为没有被认定为残疾 -如果无法回答问题，直接返回：I don't know. +从context中提取的数据必须显式赋值，所有计算步骤必须用代码实现，不得隐含推断。 +必须输出中间计算过程和结果，格式为print语句。 +如果context未提供必要数据或无法计算，直接打印"I don't know." # output format -直接输出python代码，python版本为3.10，不要包含任何其他信息 +严格输出以下结构的python代码（版本3.10）： +1. 数据提取部分：代码中涉及输入的数值需要从context及question中提取，不允许进行假设 +2. 计算过程：分步实现所有数学运算，每个步骤对应独立变量 +3. 输出：每个中间变量和最终结果必须用print语句输出 # examples ## 例子1 ### input #### question -47000元按照万分之1.5一共612天，计算利息，一共多少钱？ +4百万元按照日利率万分之1.5，一共612天，计算利息，一共多少钱？ +#### context +日利率万分之1.5 ### output ```python -# 初始本金 -principal = 47000 +# 初始本金（单位：百万） +principal = 4 # 单位：百万 -# 利率（万分之1.5） -rate = 1.5 / 10000 +# 日利率计算（万分之1.5） +daily_rate = 1.5 / 10000 -# 天数 +# 计算周期 days = 612 -# 计算年利率 -annual_rate = rate * 365 +# 单日利息计算 +daily_interest = principal * daily_rate -# 计算利息 -interest = principal * (annual_rate / 365) * days +# 累计利息计算 +total_interest = daily_interest * days -# 输出总金额（本金+利息） -total_amount = principal + interest +# 总金额计算 +total_amount = principal + total_interest -print(f"总金额：{total_amount:.2f}元") +print(f"单日利息：{daily_interest:.2f}百万") +print(f"累计利息：{total_interest:.2f}百万") +print(f"总金额：{total_amount:.2f}百万") ``` ## 例子2 @@ -70,13 +77,26 @@ class ExpressionBuildr(PromptABC): print(f"2020年的预计收入为: {revenue_2020:.2f}万") ``` +## 例子3 +### input +#### question +47000元按照612天计算利息，本息一共多少钱？ +#### content + +### output +```python +# 未给出利率，无法计算 +print("未给出利率，无法计算") +``` # input ## question $question ## context $context ## error - $error""" +$error +## output +""" ) template_en = ( f"Today is {get_now(language='en')}。\n" diff --git a/kag/solver/reporter/open_spg_reporter.py b/kag/solver/reporter/open_spg_reporter.py index f8e96a890..8d0d9c95f 100644 --- a/kag/solver/reporter/open_spg_reporter.py +++ b/kag/solver/reporter/open_spg_reporter.py @@ -548,7 +548,7 @@ def do_report(self): if self.last_report.to_dict() == request.to_dict(): return logger.info( - f"do_report: {content.answer} think={content.think} status={status_enum} ret={ret}" + f"do_report: think={content.think} {content.answer} status={status_enum} ret={ret}" ) self.last_report = request From 974590c624ef1f05315b80a8a9902a2eb0bdbf5c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=94=A6=E5=91=88?= Date: Thu, 3 Jul 2025 10:22:14 +0800 Subject: [PATCH 08/12] update api-key for openai vec --- kag/common/vectorize_model/openai_model.py | 7 ++++--- requirements.txt | 2 +- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/kag/common/vectorize_model/openai_model.py b/kag/common/vectorize_model/openai_model.py index 2a5a417ab..7874c5ca3 100644 --- a/kag/common/vectorize_model/openai_model.py +++ b/kag/common/vectorize_model/openai_model.py @@ -28,7 +28,7 @@ class OpenAIVectorizeModel(VectorizeModelABC): def __init__( self, model: str = "text-embedding-3-small", - api_key: str = "", + api_key: str = None, base_url: str = "", vector_dimensions: int = None, timeout: float = None, @@ -45,7 +45,8 @@ def __init__( base_url (str, optional): The base URL for the OpenAI service. Defaults to "". vector_dimensions (int, optional): The number of dimensions for the embedding vectors. Defaults to None. """ - name = self.generate_key(api_key, base_url, model) + api_key = api_key if api_key else "abc123" + name = self.generate_key(base_url, model, api_key) super().__init__(name, vector_dimensions, max_rate, time_period) self.model = model @@ -54,7 +55,7 @@ def __init__( self.aclient = AsyncOpenAI(api_key=api_key, base_url=base_url) @classmethod - def generate_key(cls, base_url, model, api_key="", *args, **kwargs) -> str: + def generate_key(cls, base_url, model, api_key="") -> str: return f"{cls}_{base_url}_{api_key}_{model}" def vectorize( diff --git a/requirements.txt b/requirements.txt index 2085ea8dd..44ed0d4b9 100644 --- a/requirements.txt +++ b/requirements.txt @@ -22,7 +22,7 @@ numpy>=1.23.1 pypdf pandas pycryptodome -markdown +markdown==3.7 bs4 protobuf==3.20.1 neo4j From 7554bb2ffa8400eb5c8daa7c2032400602bb02a9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=94=A6=E5=91=88?= Date: Thu, 3 Jul 2025 15:09:10 +0800 Subject: [PATCH 09/12] update --- KAG_VERSION | 2 +- kag/common/vectorize_model/openai_model.py | 33 ++++++++++++++-------- 2 files changed, 22 insertions(+), 13 deletions(-) diff --git a/KAG_VERSION b/KAG_VERSION index a3df0a695..86625ca2e 100644 --- a/KAG_VERSION +++ b/KAG_VERSION @@ -1 +1 @@ -0.8.0 +0.8.0.20250703.1119 \ No newline at end of file diff --git a/kag/common/vectorize_model/openai_model.py b/kag/common/vectorize_model/openai_model.py index 7874c5ca3..cf9acb67c 100644 --- a/kag/common/vectorize_model/openai_model.py +++ b/kag/common/vectorize_model/openai_model.py @@ -51,12 +51,14 @@ def __init__( super().__init__(name, vector_dimensions, max_rate, time_period) self.model = model self.timeout = timeout - self.client = OpenAI(api_key=api_key, base_url=base_url) - self.aclient = AsyncOpenAI(api_key=api_key, base_url=base_url) + self.client = OpenAI(api_key=api_key, base_url=base_url, timeout=self.timeout) + self.aclient = AsyncOpenAI( + api_key=api_key, base_url=base_url, timeout=self.timeout + ) @classmethod def generate_key(cls, base_url, model, api_key="") -> str: - return f"{cls}_{base_url}_{api_key}_{model}" + return f"{cls}_{base_url}_{model}_{api_key}" def vectorize( self, texts: Union[str, Iterable[str]] @@ -85,7 +87,7 @@ def vectorize( return [[] for _ in texts] # Return empty vectors for all inputs results = self.client.embeddings.create( - input=filtered_texts, model=self.model, timeout=self.timeout + input=filtered_texts, model=self.model ) # Reconstruct the results with empty lists for empty strings @@ -104,9 +106,7 @@ def vectorize( elif isinstance(texts, str) and not texts.strip(): return [] # Return empty vector for empty string else: - results = self.client.embeddings.create( - input=texts, model=self.model, timeout=self.timeout - ) + results = self.client.embeddings.create(input=texts, model=self.model) except Exception as e: logger.error(f"Error: {e}") logger.error(f"input: {texts}") @@ -137,7 +137,7 @@ async def avectorize( texts = [text if text.strip() != "" else "none" for text in texts] try: results = await self.aclient.embeddings.create( - input=texts, model=self.model, timeout=self.timeout + input=texts, model=self.model ) except Exception as e: logger.error(f"Error: {e}") @@ -200,6 +200,7 @@ def __init__( api_version=api_version, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, + timeout=self.timeout, ) self.aclient = AsyncAzureOpenAI( api_key=api_key, @@ -209,6 +210,7 @@ def __init__( api_version=api_version, azure_ad_token=azure_ad_token, azure_ad_token_provider=azure_ad_token_provider, + timeout=self.timeout, ) @classmethod @@ -227,9 +229,7 @@ def vectorize( Returns: Union[EmbeddingVector, Iterable[EmbeddingVector]]: The embedding vector(s) of the text(s). """ - results = self.client.embeddings.create( - input=texts, model=self.model, timeout=self.timeout - ) + results = self.client.embeddings.create(input=texts, model=self.model) results = [item.embedding for item in results.data] if isinstance(texts, str): assert len(results) == 1 @@ -252,7 +252,7 @@ async def avectorize( """ async with self.limiter: results = await self.aclient.embeddings.create( - input=texts, model=self.model, timeout=self.timeout + input=texts, model=self.model ) results = [item.embedding for item in results.data] if isinstance(texts, str): @@ -261,3 +261,12 @@ async def avectorize( else: assert len(results) == len(texts) return results + + +if __name__ == "__main__": + vectorize_model = OpenAIVectorizeModel( + model="bge-m3", base_url="http://localhost:11434/v1" + ) + texts = ["Hello, world!", "Hello, world!", "Hello, world!"] + embeddings = vectorize_model.vectorize(texts) + print(embeddings) From 73be52bee46047a4e321b9b3cbda7ebfc16a8678 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Tue, 8 Jul 2025 16:37:35 +0800 Subject: [PATCH 10/12] fix naive rag bug --- kag/solver/main_solver.py | 7 +++++-- kag/solver/pipelineconf/naive_rag.yaml | 17 +++++++---------- 2 files changed, 12 insertions(+), 12 deletions(-) diff --git a/kag/solver/main_solver.py b/kag/solver/main_solver.py index 63da7bbd8..32120a3a2 100644 --- a/kag/solver/main_solver.py +++ b/kag/solver/main_solver.py @@ -165,8 +165,11 @@ async def do_qa_pipeline( f"Knowledge base with id {kb_project_id} not found in qa_config['kb']" ) continue - - for index_name in matched_kb.get("index_list", []): + index_list = matched_kb.get("index_list", []) + if use_pipeline in ["default_pipeline"]: + # we only use chunk index + index_list = ["chunk_index"] + for index_name in index_list: index_manager = KAGIndexManager.from_config( { "type": index_name, diff --git a/kag/solver/pipelineconf/naive_rag.yaml b/kag/solver/pipelineconf/naive_rag.yaml index 2cc249e4c..84d4b8a60 100644 --- a/kag/solver/pipelineconf/naive_rag.yaml +++ b/kag/solver/pipelineconf/naive_rag.yaml @@ -3,20 +3,17 @@ pipeline_name: default_pipeline #------------kag-solver configuration start----------------# - -chunk_retrieved_executor: &chunk_retrieved_executor_conf - type: chunk_retrieved_executor - top_k: 10 - retriever: - type: vector_chunk_retriever - score_threshold: 0.65 - vectorize_model: "{vectorize_model}" - +kag_retriever_executor: &kag_retriever_executor_conf + type: kag_hybrid_retrieval_executor + retrievers: "{retrievers}" + merger: + type: kag_merger + enable_summary: false solver_pipeline: type: naive_rag_pipeline executors: - - *chunk_retrieved_executor_conf + - *kag_retriever_executor_conf generator: type: llm_index_generator llm_client: "{chat_llm}" From d55d639bc7a5fc74cd9bc8cd385430aa2654afb3 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Tue, 8 Jul 2025 16:49:58 +0800 Subject: [PATCH 11/12] format code --- .../kag_hybrid_retrieval_executor.py | 14 +++++- kag/solver/server/asyn_task_manager.py | 22 ++++++--- kag/solver/server/main_server.py | 20 ++++---- kag/solver/server/model/task_req.py | 47 ++++++++++++------- 4 files changed, 68 insertions(+), 35 deletions(-) diff --git a/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py b/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py index bbf5e69f3..721f8323d 100644 --- a/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py +++ b/kag/solver/executor/retriever/kag_hybrid_retrieval_executor.py @@ -306,7 +306,9 @@ def invoke(self, query, task, context: Context, **kwargs) -> RetrieverOutput: self.report_content(reporter, "thinker", tag_id, "", "INIT", step=task.name) try: try: - retrieved_data = self.do_main(task_query, tag_id, task, context, **kwargs) + retrieved_data = self.do_main( + task_query, tag_id, task, context, **kwargs + ) except Exception as e: logger.warning(f"kag hybrid retrieval failed! {e}", exc_info=True) retrieved_data = RetrieverOutput( @@ -367,7 +369,15 @@ def invoke(self, query, task, context: Context, **kwargs) -> RetrieverOutput: ) return retrieved_data finally: - self.report_content(reporter, "thinker", tag_id, "", "FINISH", step=task.name, overwrite=False,) + self.report_content( + reporter, + "thinker", + tag_id, + "", + "FINISH", + step=task.name, + overwrite=False, + ) def schema(self) -> dict: """Function schema definition for OpenAI Function Calling diff --git a/kag/solver/server/asyn_task_manager.py b/kag/solver/server/asyn_task_manager.py index de525f432..069c4e899 100644 --- a/kag/solver/server/asyn_task_manager.py +++ b/kag/solver/server/asyn_task_manager.py @@ -23,7 +23,10 @@ def __init__(self, max_workers=10, ttl=3600): self.result_cache = TTLCache(maxsize=1000, ttl=ttl) self.result_cache_lock = threading.Lock() # Protect cache from race conditions self.executor = concurrent.futures.ThreadPoolExecutor(max_workers=max_workers) - self.workers = [threading.Thread(target=self.worker, daemon=True) for _ in range(max_workers)] + self.workers = [ + threading.Thread(target=self.worker, daemon=True) + for _ in range(max_workers) + ] for w in self.workers: w.start() @@ -45,7 +48,7 @@ def worker(self): self.result_cache[task_id] = { "task_id": task_id, "status": "running", - "result": None + "result": None, } # Execute task @@ -69,7 +72,7 @@ def worker(self): self.result_cache[task_id] = { "task_id": task_id, "status": status, - "result": result + "result": result, } logger.info(f"Task {task_id} completed with status: {status}") finally: @@ -105,7 +108,11 @@ def get_task_result(self, task_id): with self.result_cache_lock: return self.result_cache.get( task_id, - {"task_id": task_id, "status": "failed", "result": "Result not found or expired"} + { + "task_id": task_id, + "status": "failed", + "result": "Result not found or expired", + }, ) def shutdown(self): @@ -133,13 +140,11 @@ def shutdown(self): # Create task manager instance task_manager = AsyncTaskManager(max_workers=5, ttl=600) - # Example task function def example_task(x, y): time.sleep(1) # Simulate work return x - # Submit test tasks task_ids = [task_manager.submit_task(example_task, i, i + 1) for i in range(6)] @@ -147,7 +152,10 @@ def example_task(x, y): try: while True: time.sleep(1) - if all("completed" in task_manager.get_task_result(tid)["status"] for tid in task_ids): + if all( + "completed" in task_manager.get_task_result(tid)["status"] + for tid in task_ids + ): break except KeyboardInterrupt: logger.info("Shutting down due to user interrupt") diff --git a/kag/solver/server/main_server.py b/kag/solver/server/main_server.py index ad49d9601..755e35909 100644 --- a/kag/solver/server/main_server.py +++ b/kag/solver/server/main_server.py @@ -17,18 +17,19 @@ def run_main_solver(task: TaskReq): params=task.config, ) + class KAGSolverServer: def __init__(self, service_name: str): """ - 初始化 FastAPI 服务实例 + Initialize a FastAPI service instance Args: - service_name (str): 服务名称，决定加载哪个路由逻辑 + service_name (str): Service name, determines which routing logic to load """ self.service_name = service_name self.app = FastAPI(title=f"{service_name} API") - # 根据服务名绑定路由 + # Bind routes according to service name self._setup_routes() self.async_manager = AsyncTaskManager() @@ -36,9 +37,9 @@ def sync_task(self, task: TaskReq): if task.cmd == "submit": return { - 'success': True, - 'status': 'init', - 'result': self.async_manager.submit_task(run_main_solver, task) + "success": True, + "status": "init", + "result": self.async_manager.submit_task(run_main_solver, task), } elif task.cmd == "query": return self.async_manager.get_task_result(task_id=task.req_id) @@ -50,12 +51,13 @@ def sync_task(self, task: TaskReq): } def _setup_routes(self): - """根据服务名动态绑定路由""" + """Dynamically bind routes according to service name""" + @self.app.post("/process") def process(req: FeatureRequest): return self.sync_task(task=req.features.task_req) def run(self, host="0.0.0.0", port=8000): - """启动服务""" + """Start the service""" print(f"Starting {self.service_name} service on {host}:{port}") - uvicorn.run(self.app, host=host, port=port) \ No newline at end of file + uvicorn.run(self.app, host=host, port=port) diff --git a/kag/solver/server/model/task_req.py b/kag/solver/server/model/task_req.py index f432e2508..0a48bd3b7 100644 --- a/kag/solver/server/model/task_req.py +++ b/kag/solver/server/model/task_req.py @@ -5,12 +5,16 @@ class ReqBody(BaseModel): + """Request body model containing query parameters""" + query: str = "" report: bool = True host_addr: str = "" class TaskReq(BaseModel): + """Task request model with validation logic""" + app_id: int = "" project_id: int = 0 req_id: str = "" @@ -21,8 +25,10 @@ class TaskReq(BaseModel): @model_validator(mode="after") def parse_req_to_req_body(self): + """Parse req string to ReqBody object and process config field""" try: import json + if isinstance(self.req, str): req_body_dict = json.loads(self.req) self.req = ReqBody(**req_body_dict) @@ -35,21 +41,25 @@ def parse_req_to_req_body(self): @field_serializer("req") def serialize_req(self, value: object) -> object: - """将 ReqBody 再次序列化为 JSON 字符串""" + """Serialize ReqBody back to JSON string""" if isinstance(value, ReqBody): return value.model_dump_json() - return value # 如果仍为字符串则直接返回 + return value # Return as-is if already a string -# 新增的 Request 模型 +# Request model with TaskReq parsing capability class Request(BaseModel): + """Container model for task request data""" + in_string: str task_req: Optional[TaskReq] = None @model_validator(mode="after") def parse_in_string_to_task_req(self): + """Convert in_string JSON string to TaskReq object""" try: import json + task_req_dict = json.loads(self.in_string) self.task_req = TaskReq(**task_req_dict) except Exception as e: @@ -58,16 +68,24 @@ def parse_in_string_to_task_req(self): class FeatureRequest(BaseModel): + """Top-level request wrapper with features container""" + features: Request if __name__ == "__main__": + def feature_request_parsing(): - # 构建最内层的 ReqBody JSON 字符串 - req_body = ReqBody(query="阿里巴巴财报中，2024年-截至9月30日止六个月的收入是多少？其中云智能集团收入是多少？占比是多少", report=True, host_addr="https://spg.alipay.com") + """Demonstrate nested model parsing workflow""" + # Build innermost ReqBody JSON string + req_body = ReqBody( + query="阿里巴巴财报中，2024年-截至9月30日止六个月的收入是多少？其中云智能集团收入是多少？占比是多少", + report=True, + host_addr="https://spg.alipay.com", + ) req_body_json = json.dumps(req_body.model_dump()) - # 构建 TaskReq 字典并序列化成字符串 + # Build TaskReq dictionary and serialize to string task_req = TaskReq( req_id="9400110", cmd="submit", @@ -75,28 +93,24 @@ def feature_request_parsing(): req=req_body_json, app_id="app_id", project_id=4200050, - config={"timeout": 10} + config={"timeout": 10}, ) task_req_json = json.dumps(task_req.model_dump()) - # 构造最终传入的 FeatureRequest JSON 字符串 - input_data = { - "features": { - "in_string": task_req_json - } - } + # Construct final FeatureRequest JSON string + input_data = {"features": {"in_string": task_req_json}} - # 反序列化为 FeatureRequest 模型 + # Deserialize to FeatureRequest model feature_request = FeatureRequest(**input_data) - # 验证 in_string 被解析为 TaskReq + # Validate in_string parsed to TaskReq assert isinstance(feature_request.features.task_req, TaskReq) assert feature_request.features.task_req.req_id == "abc123" assert feature_request.features.task_req.cmd == "run" assert feature_request.features.task_req.mode == "sync" assert feature_request.features.task_req.config == {"timeout": 10} - # 验证 TaskReq.req 被解析为 ReqBody + # Validate TaskReq.req parsed to ReqBody req_body_parsed = feature_request.features.task_req.req assert isinstance(req_body_parsed, ReqBody) assert req_body_parsed.query == "What is AI?" @@ -105,5 +119,4 @@ def feature_request_parsing(): print("✅ All assertions passed!") - feature_request_parsing() From 27293e0ae2b400c3d524db5b63dd60d8abe259d5 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Tue, 8 Jul 2025 17:03:46 +0800 Subject: [PATCH 12/12] fix --- KAG_VERSION | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/KAG_VERSION b/KAG_VERSION index 86625ca2e..8adc70fdd 100644 --- a/KAG_VERSION +++ b/KAG_VERSION @@ -1 +1 @@ -0.8.0.20250703.1119 \ No newline at end of file +0.8.0 \ No newline at end of file