From df9a31542f88157841939802dcbde428ef8267a0 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Mon, 14 Jul 2025 14:54:24 +0800 Subject: [PATCH 1/4] fix naive rag --- kag/solver/pipeline/naive_rag_pipeline.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/kag/solver/pipeline/naive_rag_pipeline.py b/kag/solver/pipeline/naive_rag_pipeline.py index 35d82b799..228618e53 100644 --- a/kag/solver/pipeline/naive_rag_pipeline.py +++ b/kag/solver/pipeline/naive_rag_pipeline.py @@ -71,7 +71,7 @@ async def planning(self, query, context, **kwargs): """ tasks_dep = {} tasks_dep[0] = { - "executor": "ChunkRetriever", + "executor": "Retriever", "dependent_task_ids": [], "arguments": {"query": query}, } From 2c2ce57bb0a9e344b20421091ba9aa4d087c55e6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=94=A6=E5=91=88?= Date: Mon, 14 Jul 2025 15:59:07 +0800 Subject: [PATCH 2/4] update length splitter for \n and 2* len strict --- .../component/splitter/length_splitter.py | 42 +++++++++++++++++-- 1 file changed, 39 insertions(+), 3 deletions(-) diff --git a/kag/builder/component/splitter/length_splitter.py b/kag/builder/component/splitter/length_splitter.py index c2e83a956..13e45c26e 100644 --- a/kag/builder/component/splitter/length_splitter.py +++ b/kag/builder/component/splitter/length_splitter.py @@ -79,6 +79,7 @@ def logic_break(self, chunk): def split_sentence(self, content): """ Splits the given content into sentences based on delimiters. + Also enforces a hard length limit of 2 * split_length. Args: content (str): The content to be split into sentences. @@ -87,22 +88,57 @@ def split_sentence(self, content): List[str]: A list of sentences. """ sentence_delimiters = ( - ".。？?！!" if self.kag_project_config.language == "en" else "。？！" + ".。？?！!\n" if self.kag_project_config.language == "en" else "。？！\n" ) + max_length = self.split_length * 2 output = [] start = 0 + for idx, char in enumerate(content): if char in sentence_delimiters: end = idx tmp = content[start : end + 1].strip() if len(tmp) > 0: - output.append(tmp.strip()) + # 如果句子超过最大长度限制，强制切分 + if len(tmp) > max_length: + output.extend(self._force_split_by_length(tmp, max_length)) + else: + output.append(tmp.strip()) start = idx + 1 + elif idx - start >= max_length: + # 当前片段已经达到最大长度，强制切分 + tmp = content[start:idx].strip() + if len(tmp) > 0: + output.extend(self._force_split_by_length(tmp, max_length)) + start = idx + res = content[start:].strip() if len(res) > 0: - output.append(res) + # 如果剩余部分超过最大长度限制，强制切分 + if len(res) > max_length: + output.extend(self._force_split_by_length(res, max_length)) + else: + output.append(res) return output + def _force_split_by_length(self, text, max_length): + """ + 强制按长度切分文本，不考虑分隔符 + + Args: + text (str): 要切分的文本 + max_length (int): 最大长度 + + Returns: + List[str]: 切分后的文本片段列表 + """ + result = [] + for i in range(0, len(text), max_length): + chunk = text[i : i + max_length].strip() + if chunk: + result.append(chunk) + return result + def slide_window_chunk( self, org_chunk: Chunk, From e7d5f0407fcae34a095addedef5e29ed4fd20b8c Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Mon, 28 Jul 2025 11:00:00 +0800 Subject: [PATCH 3/4] support return without data --- .../search_api/impl/openspg_search_api.py | 37 ++++++++++++------- kag/interface/common/llm_client.py | 7 ++++ kag/interface/solver/context.py | 17 +++++++++ kag/solver/pipeline/kag_static_pipeline.py | 31 +++++++++------- kag/solver/reporter/open_spg_reporter.py | 11 +++--- 5 files changed, 70 insertions(+), 33 deletions(-) diff --git a/kag/common/tools/search_api/impl/openspg_search_api.py b/kag/common/tools/search_api/impl/openspg_search_api.py index c7a4a5455..bb7c5857e 100644 --- a/kag/common/tools/search_api/impl/openspg_search_api.py +++ b/kag/common/tools/search_api/impl/openspg_search_api.py @@ -1,9 +1,11 @@ +import logging from typing import List from kag.common.conf import KAGConstants, KAGConfigAccessor from kag.common.tools.search_api.search_api_abc import SearchApiABC from knext.search.client import SearchClient +logger = logging.getLogger(__name__) @SearchApiABC.register("openspg_search_api", as_default=True) class OpenSPGSearchAPI(SearchApiABC): @@ -25,12 +27,15 @@ def search_text( self, query_string, label_constraints=None, topk=10, params=None ) -> List: if self.sc: - return self.sc.search_text( - query_string=query_string, - label_constraints=label_constraints, - topk=topk, - params=params, - ) + try: + return self.sc.search_text( + query_string=query_string, + label_constraints=label_constraints, + topk=topk, + params=params, + ) + except Exception as e: + logger.debug(f"search_vector error {e}", exc_info=True) return [] def search_vector( @@ -38,14 +43,18 @@ def search_vector( ) -> List: if self.sc is None: return [] - res = self.sc.search_vector( - label=label, - property_key=property_key, - query_vector=query_vector, - topk=topk, - ef_search=ef_search, - params=params, - ) + try: + res = self.sc.search_vector( + label=label, + property_key=property_key, + query_vector=query_vector, + topk=topk, + ef_search=ef_search, + params=params, + ) + except Exception as e: + logger.debug(f"search_vector error {e}", exc_info=True) + res = [] if res is None: return [] return res diff --git a/kag/interface/common/llm_client.py b/kag/interface/common/llm_client.py index b9738b0e8..22390246b 100644 --- a/kag/interface/common/llm_client.py +++ b/kag/interface/common/llm_client.py @@ -326,6 +326,8 @@ def invoke( tools = kwargs.get("tools", None) if tools: with_json_parse = False + context = kwargs.get("context", None) + segment_name = kwargs.get("segment_name", None) try: self.sync_limiter.acquire() response = ( @@ -335,6 +337,11 @@ def invoke( ) if tools: return response + if context and segment_name and hasattr(context, "add_kwargs"): + try: + context.add_kwargs(segment_name, response) + except Exception as e: + logger.warning(f"Failed to add kwargs to context: {e}") result = prompt_op.parse_response(response, model=self.model, **variables) logger.debug(f"Result: {result}") return result diff --git a/kag/interface/solver/context.py b/kag/interface/solver/context.py index d0eb617d6..12793af24 100644 --- a/kag/interface/solver/context.py +++ b/kag/interface/solver/context.py @@ -14,6 +14,23 @@ def __init__(self): self.variables_graph = KgGraph() self.kwargs = {} + def add_kwargs(self, key, value): + """ + Adds a key-value pair to the context's keyword arguments dictionary. + + This method stores arbitrary key-value data in the context that can be + accessed by other components during pipeline execution. + + Args: + key: The string key to store the value under + value: The value to be stored, can be of any type + + Example: + context.add_kwargs("temperature", 0.7) + context.add_kwargs("max_tokens", 1000) + """ + self.kwargs[key] = value + def add_task(self, task: Task): """Adds a task to the context. diff --git a/kag/solver/pipeline/kag_static_pipeline.py b/kag/solver/pipeline/kag_static_pipeline.py index 26f7a323e..bbf6e384a 100644 --- a/kag/solver/pipeline/kag_static_pipeline.py +++ b/kag/solver/pipeline/kag_static_pipeline.py @@ -129,21 +129,24 @@ async def ainvoke(self, query, **kwargs): while True: context: Context = Context() tasks = await self.planning(query, context, **kwargs) + if not tasks: + think_response = context.kwargs.get("thinker", "") + answer = await self.generator.ainvoke(query + "\n" + think_response, context, **kwargs) + else: + for task in tasks: + context.add_task(task) + + for task_group in context.gen_task(group=True): + await asyncio.gather( + *[ + asyncio.create_task( + self.execute_task(query, task, context, **kwargs) + ) + for task in task_group + ] + ) - for task in tasks: - context.add_task(task) - - for task_group in context.gen_task(group=True): - await asyncio.gather( - *[ - asyncio.create_task( - self.execute_task(query, task, context, **kwargs) - ) - for task in task_group - ] - ) - - answer = await self.generator.ainvoke(query, context, **kwargs) + answer = await self.generator.ainvoke(query, context, **kwargs) from kag.common.utils import red, green, reset task_info = [] diff --git a/kag/solver/reporter/open_spg_reporter.py b/kag/solver/reporter/open_spg_reporter.py index 8d0d9c95f..6877313a6 100644 --- a/kag/solver/reporter/open_spg_reporter.py +++ b/kag/solver/reporter/open_spg_reporter.py @@ -530,8 +530,6 @@ def add_report_line(self, segment, tag_name, content, status, **kwargs): } def do_report(self): - if not self.client: - return content, status_enum, metrics = self.generate_report_data() request = TaskStreamRequest( @@ -539,9 +537,12 @@ def do_report(self): ) # logging.info(f"do_report:{request}") try: - ret = self.client.reasoner_dialog_report_completions_post( - task_stream_request=request - ) + if self.client: + ret = self.client.reasoner_dialog_report_completions_post( + task_stream_request=request + ) + else: + ret = {} if self.last_report is None: logger.info(f"begin do_report: {request} ret={ret}") self.last_report = request From 64a8cc9665afdf4c4d4183a7080d64f77dffbbb8 Mon Sep 17 00:00:00 2001 From: "peilong.zip" Date: Mon, 11 Aug 2025 16:05:30 +0800 Subject: [PATCH 4/4] format and assgin pyarrow --- kag/common/tools/search_api/impl/openspg_search_api.py | 1 + kag/solver/pipeline/kag_static_pipeline.py | 4 +++- requirements.txt | 1 + 3 files changed, 5 insertions(+), 1 deletion(-) diff --git a/kag/common/tools/search_api/impl/openspg_search_api.py b/kag/common/tools/search_api/impl/openspg_search_api.py index bb7c5857e..37eefb665 100644 --- a/kag/common/tools/search_api/impl/openspg_search_api.py +++ b/kag/common/tools/search_api/impl/openspg_search_api.py @@ -7,6 +7,7 @@ logger = logging.getLogger(__name__) + @SearchApiABC.register("openspg_search_api", as_default=True) class OpenSPGSearchAPI(SearchApiABC): def __init__(self, project_id=None, host_addr=None, **kwargs): diff --git a/kag/solver/pipeline/kag_static_pipeline.py b/kag/solver/pipeline/kag_static_pipeline.py index bbf6e384a..4ba7ce199 100644 --- a/kag/solver/pipeline/kag_static_pipeline.py +++ b/kag/solver/pipeline/kag_static_pipeline.py @@ -131,7 +131,9 @@ async def ainvoke(self, query, **kwargs): tasks = await self.planning(query, context, **kwargs) if not tasks: think_response = context.kwargs.get("thinker", "") - answer = await self.generator.ainvoke(query + "\n" + think_response, context, **kwargs) + answer = await self.generator.ainvoke( + query + "\n" + think_response, context, **kwargs + ) else: for task in tasks: context.add_task(task) diff --git a/requirements.txt b/requirements.txt index 44ed0d4b9..27a1c4a98 100644 --- a/requirements.txt +++ b/requirements.txt @@ -45,6 +45,7 @@ json_repair gitpython docstring_parser aiolimiter +pyarrow==19.0.1 pyodps==0.12.2 aliyun-log-python-sdk==0.8.8 pyvis