当前位置：首页 > news >正文

KAG：通过知识增强生成提升专业领域的大型语言模型（五）

news 来源：原创 2025/5/19 10:05:19

摘要

Abstract

1 planner

2 generator

3 pipeline

4 Solver示例

总结

摘要

本周对KAG-solver中的规划器、生成器、和Solver通道的代码进行了解读，涵盖了多个组件的配置和实现。规划器通过LLM生成任务计划；生成器用于生成最终答案；Solver通道提供了一个灵活、可扩展的框架，用于处理规划和执行工作流，通过集成不同的规划器、执行器和生成器组件，可以适应多种应用场景。最后通过示例展现了KAG-Solver的执行过程。

Abstract

This week's breakdown of the code for the planner, generator, and solver channels in the KAG-solver covers the configuration and implementation of multiple components. The planner generates a task plan through the LLM; The generator is used to generate the final answer; The Solver channel provides a flexible, extensible framework for handling planning and execution workflows, adapting to a variety of use cases by integrating different planner, executor, and generator components. Finally, an example is given to show the execution process of KAG-Solver.

1 planner

planner为用于生成任务计划的规划器。下面为kag_static_planner，即使用具有查询重写功能的LLM生成任务计划的静态规划器的代码实现：

@PlannerABC.register("kag_static_planner")
class KAGStaticPlanner(PlannerABC):def __init__(self,llm: LLMClient,plan_prompt: PromptABC,rewrite_prompt: PromptABC,**kwargs,):super().__init__(**kwargs)self.llm = llmself.plan_prompt = plan_promptself.rewrite_prompt = rewrite_prompt

其中参数：

llm （LLMClient）：用于计划生成的语言模型客户端；
plan_prompt （PromptABC）：初始规划请求的提示模板；
rewrite_prompt （PromptABC）：用于查询重写作的提示模板。

format_context函数用于将父任务执行上下文格式化为结构化字典：

    def format_context(self, task: Task):formatted_context = {}# get all prvious tasks from context.for parent_task in task.parents:formatted_context[parent_task.id] = {"query": parent_task.arguments["query"],"output": str(parent_task.result),}return formatted_context

其中：

参数：task，其父上下文需要格式化的当前任务；
返回值dict：父任务ID与其执行详细信息的映射，其中包含：
- - action：使用的 Executor 和参数；
- - result：父任务的执行结果。

check_require_rewrite函数用于根据参数模式确定是否需要重写查询：

    def check_require_rewrite(self, task: Task):query = task.argumentspattern = r"\{\{\d+\.output\}\}"return bool(re.search(pattern, str(query)))

其中：

参数：task，检查重写要求的任务；
返回值bool：如果查询包含动态参数引用（例如，{{1.output}}），则为 True。

finish_judger方法通过调用语言模型来判断生成的答案是否有效：

    async def finish_judger(self, query: str, answer: str):finish_prompt = f"""# TaskThe answer is a response to a question. Please determine whether the content of this answer is invalid, such as  "UNKNOWN", "I don't know" or "Insufficient Information."  \nIf the answer is invalid, return "Yes", otherwise, return "No".\nYou output should only be "Yes" or "No".\n# Answer\n{answer}"""try:response = await self.llm.acall(prompt=finish_prompt)if response.strip().lower() == "yes":return Falsereturn Trueexcept Exception as e:print(f"Failed to run finish_judger, info: {e}")import tracebacktraceback.print_exc()return True

query_rewrite使用 LLM 和 context 执行异步查询重写：

    async def query_rewrite(self, task: Task, **kwargs):query = task.arguments# print(f"Old query: {query}")context = self.format_context(task)new_query = await self.llm.ainvoke({"input": query,"context": context,},self.rewrite_prompt,segment_name="thinker",tag_name="Rewrite query",with_json_parse=self.rewrite_prompt.is_json_format(),**kwargs,)# print(f"query rewrite context = {context}")# print(f"New query: {new_query}")return new_query

其中：

参数：task，包含要重写的查询的任务；
返回值str：使用已解析的动态引用重写的查询。

invoke为使用LLM同步生成任务计划、ainvoke为使用LLM异步生成任务计划：

    def invoke(self, query, **kwargs) -> List[Task]:num_iteration = kwargs.get("num_iteration", 0)return self.llm.invoke({"query": query,"executors": kwargs.get("executors", []),},self.plan_prompt,with_json_parse=self.plan_prompt.is_json_format(),segment_name="thinker",tag_name=f"Static planning {num_iteration}",**kwargs,)async def ainvoke(self, query, **kwargs) -> List[Task]:num_iteration = kwargs.get("num_iteration", 0)return await self.llm.ainvoke({"query": query,"executors": kwargs.get("executors", []),},self.plan_prompt,with_json_parse=self.plan_prompt.is_json_format(),segment_name="thinker",tag_name=f"Static planning {num_iteration}",**kwargs,)

其中：

参数：
- query：要为其生成计划的用户查询；
- **kwargs：其他参数包括：
  - - executors （list）：可用于任务规划的executor。
返回值List[Task]：生成的任务序列。

除了静态规划器之外，还有如下几种planner：

kag_iterative_planner：迭代规划器，根据上下文和可用执行者使用大型语言模型生成任务计划；
lf_kag_static_planner：静态规划器，使用具有查询重写能力的LLM生成任务计划；
mcp_planner：使用具有查询重写能力的语言模型生成任务计划的MCP规划器。

2 generator

generator为用于生成最终答案的生成器，其中主要为llm_generator方法，使用大语言模型来生成最终答案，并支持引用数据的处理：

@GeneratorABC.register("llm_generator")
class LLMGenerator(GeneratorABC):def __init__(self,llm_client: LLMClient,generated_prompt: PromptABC,chunk_reranker: RerankByVector = None,enable_ref=False,**kwargs,):super().__init__(**kwargs)self.llm_client = llm_clientself.generated_prompt = generated_promptself.chunk_reranker = chunk_reranker or RerankByVector.from_config({"type": "rerank_by_vector",})if enable_ref:self.with_out_ref_prompt = init_prompt_with_fallback("without_refer_generator_prompt", KAG_PROJECT_CONF.biz_scene)self.with_ref_prompt = init_prompt_with_fallback("refer_generator_prompt", KAG_PROJECT_CONF.biz_scene)self.enable_ref = enable_ref

其中参数：

llm_client：用于生成答案的语言模型客户端；
generated_prompt：生成答案的提示模板；
chunk_reranker：用于重排序的工具，默认为None；
enable_ref：是否启用引用功能，默认为False。

generate_answer函数使用LLM生成最终答案，并使用重试机制，最多重试3次：

    @retry(stop=stop_after_attempt(3))def generate_answer(self, query, content, refer_data, **kwargs):if not self.enable_ref:return self.llm_client.invoke({"query": query, "content": content},self.generated_prompt,segment_name="answer",tag_name="Final Answer",with_json_parse=self.generated_prompt.is_json_format(),**kwargs,)if refer_data and len(refer_data):refer_data_str = json.dumps(refer_data, ensure_ascii=False, indent=2)return self.llm_client.invoke({"query": query, "content": content, "ref": refer_data_str},self.with_ref_prompt,segment_name="answer",tag_name="Final Answer",with_json_parse=False,**kwargs,)return self.llm_client.invoke({"query": query, "content": content},self.with_out_ref_prompt,segment_name="answer",tag_name="Final Answer",with_json_parse=False,**kwargs,)

其中：

参数：
- query：用户查询；
- content：生成答案所需的内容；
- refer_data：引用数据。
返回：生成的最终答案。

invoke为生成最终答案的主调用方法。首先，提取任务结果并转换为字符串格式，如果启用了引用功能，使用重排序工具处理引用数据；然后，构建生成答案所需的内容，包括步骤和引用数据；最后调用generate_answer方法生成最终答案：

    def invoke(self, query, context, **kwargs):reporter: Optional[ReporterABC] = kwargs.get("reporter", None)results = []rerank_queries = []chunks = []graph_data = context.variables_graphfor task in context.gen_task(False):if isinstance(task.result, KAGRetrievedResponse) and self.chunk_reranker:rerank_queries.append(task.arguments.get("rewrite_query", task.arguments["query"]))chunks.append(task.result.chunk_datas)results.append(to_task_context_str(task.get_task_context()))rerank_chunks = self.chunk_reranker.invoke(query, rerank_queries, chunks)refer_data = to_reference_list(prefix_id=0, retrieved_datas=rerank_chunks)content_json = {"step": results}if reporter:reporter.add_report_line("generator", "final_generator_input", content_json, "FINISH")reporter.add_report_line("generator_reference", "reference_chunk", rerank_chunks, "FINISH")reporter.add_report_line("generator_reference_all", "reference_ref_format", refer_data, "FINISH")reporter.add_report_line("generator_reference_graphs", "reference_graph", graph_data, "FINISH")if len(refer_data) and (not self.enable_ref):content_json["reference"] = refer_datacontent = json.dumps(content_json, ensure_ascii=False, indent=2)if not self.enable_ref:refer_data = [f"Title:{x['document_name']}\n{x['content']}" for x in refer_data]refer_data = "\n\n".join(refer_data)thoughts = "\n\n".join(results)content = f"""
Docs:
{refer_data}Step by Step Analysis:
{thoughts}"""return self.generate_answer(query=query, content=content, refer_data=refer_data, **kwargs)

3 pipeline

Solver_pipeline类提供了一个灵活、可扩展的框架，用于处理规划和执行工作流。通过集成不同的规划器、执行器和生成器组件，可以适应多种应用场景。如下代码所示，为一个静态的pipeline，用于通过迭代任务处理实现静态规划和执行工作流的管道，其中：

planner：用于生成任务计划的规划器配置，采用kag_static_planner的静态规划器，使用默认的逻辑静态规划提示词default_lf_static_planning；
executors：用于执行答案检索的检索执行器配置，包括混合检索kag_hybrid_executor、基于python代码的数学推理py_code_based_math_executor、推理检索kag_deduce_executor和输出用户友好的检索结果kag_output_executor_conf；
generator：用于生成最终答案的生成器配置，采用基于大模型的生成策略。

solver_pipeline:type: kag_static_pipeline # kag/solver/pipeline/kag_static_pipeline.pyplanner:type: kag_static_planner # kag/solver/planner/kag_static_planner.pyllm: *chat_llmplan_prompt:type: default_lf_static_planning # kag/solver/prompt/lf_static_planning_prompt.pyrewrite_prompt:type: default_query_rewrite # kag/solver/prompt/query_rewrite_prompt.pyexecutors:- *kag_hybrid_executor_conf- *py_code_based_math_executor_conf- *kag_deduce_executor_conf- *kag_output_executor_confgenerator:type: llm_generator # kag/solver/generator/llm_generator.pyllm_client: *chat_llmgenerated_prompt:type: default_refer_generator_prompt # kag/solver/prompt/reference_generator.pyenable_ref: true

以下为kag_static_pipeline的代码实现：

@SolverPipelineABC.register("kag_static_pipeline")
class KAGStaticPipeline(SolverPipelineABC):def __init__(self,planner: PlannerABC,executors: List[ExecutorABC],generator: GeneratorABC,max_iteration: int = 10,):super().__init__()self.planner = plannerself.executors = executorsself.generator = generatorself.max_iteration = max_iteration

其中参数有：

planner （PlannerABC）：用于生成执行计划的任务规划组件；
executors （List[ExecutorABC]）：可用于任务执行的executor实例；
generator （GeneratorABC）：用于最终答案合成的结果生成组件；
max_iteration （int）：允许的最大执行迭代数（默认值：10）。

select_executor函数从可用执行程序中按名称选择执行程序实例，如果未找到，则为None：

    def select_executor(self, executor_name: str):for executor in self.executors:schema = executor.schema()if executor_name == schema["name"]:return executorreturn None

planning函数通过基于 LLM 的规划异步生成任务计划，并自动重试：

    @retry(stop=stop_after_attempt(3), reraise=True)async def planning(self, query, context, **kwargs):tasks = await self.planner.ainvoke(query,context=context,executors=[x.schema() for x in self.executors],**kwargs,)return tasks

其中：

参数：
- query：原始用户查询；
- context：包含先前任务结果的执行上下文；
- **kwargs：其他规划参数。
返回：List[Task]，DAG格式的计划任务序列。

execute_task函数使用查询重写和执行程序调用异步执行单个任务（task）：

    @retry(stop=stop_after_attempt(3), reraise=True)async def execute_task(self, query, task, context, **kwargs):if self.planner.check_require_rewrite(task):task.update_memory("origin_arguments", task.arguments)updated_args = await self.planner.query_rewrite(task, query=query, **kwargs)task.arguments.update(updated_args)executor = self.select_executor(task.executor)if executor:await executor.ainvoke(query, task, context, **kwargs)else:logger.warn(f"Executor not  found for task {task}")

ainvoke异步编排完整的问题解决工作流：

首先，生成初始任务DAG；
然后，并行批处理执行任务；
最后，从执行上下文生成最终答案。

    async def ainvoke(self, query, **kwargs):num_retry = 1while True:context: Context = Context()tasks = await self.planning(query, context, **kwargs)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)from kag.common.utils import red, green, resettask_info = []for task in context.gen_task(group=False):task_info.append({"task": task.arguments,# "memory": task.memory,"result": task.result,})if "unknown" in answer.lower():finished = Falseelse:finished = await self.planner.finish_judger(query, answer)if not finished:if num_retry == 0:print(f"{red}Failed to answer quesion: {query}\nTasks:{task_info}\n{reset}\n{answer}")else:num_retry -= 1continueprint(f"{green}Input Query: {query}\n\nTasks:\n\n{task_info}\n\nFinal Answer: {answer}\nGold Answer: {kwargs.get('gold')}{reset}")return answer

除了静态的pipeline之外，还有如下几种pipeline：

kag_iterative_pipeline：迭代的pipeline，实现迭代问题解决流程，逐步分解和分析问题；
mcp_pipeline：基于模型上下文协议的pipeline，实现与外部数据源和符号求解器的集成；
naive_generation_pipeline：使用大型语言模型直接生成答案的pipeline；
naive_rag_pipeline：使用朴素RAG的管道来检索相关文档并生成答案的pipeline；
self_cognition_pipeline：能够自我认知的pipeline。

4 Solver示例

如下，使用KAG-Builder提取关于周星驰的百科信息：

然后，使用KAG-Solver进行提问："周星驰和万梓良有什么关系"，得到以下日志内容：

2025-05-17 14:51:08 - INFO - root - found config file: D:\KAG\KAG\kag\examples\baike\kag_config.yaml
2025-05-17 14:51:08 - INFO - root - Done init config from local file
2025-05-17 14:51:13 - INFO - root - Parsing flow string: kg_cs->kg_fr->kag_merger;rc->kag_merger
2025-05-17 14:51:13 - INFO - root - Processing path: kg_cs->kg_fr->kag_merger
2025-05-17 14:51:13 - INFO - root - Adding edges from ['kg_cs'] to ['kg_fr']
2025-05-17 14:51:14 - INFO - root - Adding edges from ['kg_fr'] to ['kag_merger']
2025-05-17 14:51:14 - INFO - root - Processing path: rc->kag_merger
2025-05-17 14:51:14 - INFO - root - Adding edges from ['rc'] to ['kag_merger']
2025-05-17 14:51:18 - INFO - root -  查询周星驰和万梓良的关系 begin kag hybrid executor
2025-05-17 14:51:18 - INFO - root - Initializing response container for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root - Response container initialized in 0.00 seconds for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root - Converting query to logical form for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root - Query converted to logical form in 0.00 seconds for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root - Creating KAGFlow for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root - KAGFlow created in 0.00 seconds for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root - Executing KAGFlow for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:18 - INFO - root -  查询周星驰和万梓良的关系 Starting KAGFlow execution
2025-05-17 14:51:18 - INFO - root -  查询周星驰和万梓良的关系 Topological order retrieved: ['kg_cs', 'rc', 'kg_fr', 'kag_merger']
2025-05-17 14:51:18 - INFO - root -  查询周星驰和万梓良的关系 Current level nodes to execute: ['kg_cs', 'rc']
2025-05-17 14:51:18 - INFO - root -  查询周星驰和万梓良的关系 Executing node: kg_cs
2025-05-17 14:51:18 - INFO - root -  查询周星驰和万梓良的关系 Executing node: rc
2025-05-17 14:51:18 - INFO - root - `周星驰` Vectorization completed in 0.24 seconds.
2025-05-17 14:51:18 - INFO - root - `周星驰` Vector-based search completed in 0.63 seconds. Found 10 nodes.
2025-05-17 14:51:19 - INFO - root - ` 查询周星驰和万梓良的关系` Similarity scores calculation completed in 0.97 seconds.
2025-05-17 14:51:19 - INFO - root -  查询周星驰和万梓良的关系 Node rc executed in 0.97 seconds
2025-05-17 14:51:19 - INFO - root - Node rc executed successfully
2025-05-17 14:51:19 - INFO - root - `周星驰` Content-based vector search completed in 0.25 seconds. Found 0 nodes.
2025-05-17 14:51:19 - INFO - root - `周星驰`  rerank_sematic_type completed in 0.00 seconds. Found 10 nodes.
2025-05-17 14:51:19 - INFO - root - `万梓良` Vectorization completed in 0.08 seconds.
2025-05-17 14:51:19 - INFO - root - `万梓良` Vector-based search completed in 0.19 seconds. Found 10 nodes.
2025-05-17 14:51:19 - INFO - root - `万梓良` Content-based vector search completed in 0.12 seconds. Found 0 nodes.
2025-05-17 14:51:19 - INFO - root - `万梓良`  rerank_sematic_type completed in 0.00 seconds. Found 10 nodes.
2025-05-17 14:51:21 - INFO - root - run gql cost 1.6614246368408203
2025-05-17 14:51:21 - INFO - root - _exact_match_spo total cost=1.6614246368408203 cost=0.0 selected_rels=0
2025-05-17 14:51:21 - INFO - root -  查询周星驰和万梓良的关系 Node kg_cs executed in 3.17 seconds
2025-05-17 14:51:21 - INFO - root - Node kg_cs executed successfully
2025-05-17 14:51:21 - INFO - root -  查询周星驰和万梓良的关系 Current level nodes to execute: ['kg_fr']
2025-05-17 14:51:21 - INFO - root -  查询周星驰和万梓良的关系 Executing node: kg_fr
2025-05-17 14:51:25 - INFO - root - get_entity_one_hop  cost =4.280318975448608
2025-05-17 14:51:25 - INFO - root - get_entity_one_hop  cost =0.21115851402282715
2025-05-17 14:51:25 - INFO - root - recall_one_hop_graph cost 4.498719930648804
2025-05-17 14:51:30 - INFO - root - _fuzzy_match_spo total cost=9.613630056381226 cost=5.113976240158081 selected_rels=6
2025-05-17 14:51:30 - INFO - root - Starting invoke method with queries: [' 查询周星驰和万梓良的关系'], start_entities: [万梓良 [BaiKe.Person], 周星驰 [BaiKe.Person]], top_k: 20
2025-05-17 14:51:30 - INFO - root - Extracting candidate entities using NER for queries: [' 查询周星驰和万梓良的关系']
2025-05-17 14:51:32 - INFO - root - NER completed in 1.29 seconds. Found 2 unique entities.
2025-05-17 14:51:32 - INFO - root - Performing entity linking (EL) for 2 candidate entities.
2025-05-17 14:51:32 - INFO - root - Entity linking completed in 1.63 seconds. Found 2 unique entities.
2025-05-17 14:51:33 - INFO - root - PageRank calculation completed in 1.33 seconds.
2025-05-17 14:51:34 - INFO - root - ` 查询周星驰和万梓良的关系`  Retrieved chunks num: 47
2025-05-17 14:51:34 - INFO - root -  查询周星驰和万梓良的关系 Node kg_fr executed in 12.90 seconds
2025-05-17 14:51:34 - INFO - root - Node kg_fr executed successfully
2025-05-17 14:51:34 - INFO - root -  查询周星驰和万梓良的关系 Current level nodes to execute: ['kag_merger']
2025-05-17 14:51:34 - INFO - root -  查询周星驰和万梓良的关系 Executing node: kag_merger
2025-05-17 14:51:37 - INFO - root -  查询周星驰和万梓良的关系 Node kag_merger executed in 3.22 seconds
2025-05-17 14:51:37 - INFO - root - Node kag_merger executed successfully
2025-05-17 14:51:37 - INFO - root -  查询周星驰和万梓良的关系 KAGFlow execution completed in 19.30 seconds
2025-05-17 14:51:37 - INFO - root - KAGFlow executed in 19.30 seconds for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:37 - INFO - root - Processing logic nodes for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:37 - INFO - root - Logic nodes processed in 0.00 seconds for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:37 - INFO - root - Summary Question  查询周星驰和万梓良的关系 : 思考: 根据提供的文档，周星驰和万梓良的关系始于1988年的电影《捕风汉子》，这是周星驰出演的第一部电影。在这部电影中，万梓良是他的合作演员。之后，万梓良继续与周星驰合作了时装商战剧《他来自江湖》。这表明两人在演艺事业上有一定的合作关系。结论: 周星驰和万梓良是合作演员关系。
2025-05-17 14:51:37 - INFO - root - Storing results for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:37 - INFO - root - Results stored in 0.00 seconds for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:37 - INFO - root - Completed storing results for task:  查询周星驰和万梓良的关系
2025-05-17 14:51:37 - INFO - root -  查询周星驰和万梓良的关系 end kag hybrid executor
2025-05-17 14:51:42 - INFO - kag.open_benchmark.utils.eval_qa - so the answer for '周星驰和万梓良有什么关系' is: 周星驰和万梓良是合作演员关系。1988年，周星驰与万梓良共同出演了动作片《捕风汉子》，这是周星驰出演的第一部电影。在这部电影中，万梓良是他的合作演员。之后，他们又合作了时装商战剧《他来自江湖》<reference id="chunk:0_0"></reference>。Input Query: 周星驰和万梓良有什么关系Tasks:[{'task': {'query': ' 查询周星驰和万梓良的关系', 'logic_form_node': Retriever(s=s1:人物[周星驰] ,p=p1:与...有关,o=o1:人物[万梓良] ,sub_query= 查询周星驰和万梓良的关系,init_query=), 'is_need_rewrite': False}, 'result': {"retrieved_task": "Task<0>\n\texecutor: Retriever\n\targuments {'query': ' 查询周星驰和万梓良的关系', 'logic_form_node': Retriever(s=s1:人物[周星驰] ,p=p1:与...有关,o=o1:人物[万梓良] ,sub_query= 查询周星驰和万梓良的关系,init_query=), 'is_need_rewrite': False}\n", "summary": "思考: 根据提供的文档，周星驰和万梓良的关系始于1988年的电影《捕风汉子》，这是周星驰出演的第一部电影。在这部电影中，万梓良是他的合作演员。之后，万梓良继续与周星驰合作了时装商战剧《他来自江湖》。这表明两人在演艺事业上有一定的合作关系。结论: 周星驰和万梓良是合作演员关系。"}}, {'task': {'query': ' 推断周星驰和万梓良的关系', 'logic_form_node': Deduce(alias_name=res,op=['entailment'],content=[`s1`],target=`周星驰和万梓良有什么关系`,sub_query= 推断周星驰和万梓良的关系,init_query=), 'is_need_rewrite': True}, 'result': '周星驰和万梓良是合作演员关系。'}, {'task': {'query': ' 输出res', 'logic_form_node': Output(alias_name=res,alias_name_set=[res],sub_query= 输出res,init_query=), 'is_need_rewrite': False}, 'result': [[' 推断周星驰和万梓良的关系\n周星驰和万梓良是合作演员关系。']]}]Final Answer: 周星驰和万梓良是合作演员关系。1988年，周星驰与万梓良共同出演了动作片《捕风汉子》，这是周星驰出演的第一部电影。在这部电影中，万梓良是他的合作演员。之后，他们又合作了时装商战剧《他来自江湖》<reference id="chunk:0_0"></reference>。
Gold Answer: 
周星驰和万梓良有什么关系 is 周星驰和万梓良是合作演员关系。1988年，周星驰与万梓良共同出演了动作片《捕风汉子》，这是周星驰出演的第一部电影。在这部电影中，万梓良是他的合作演员。之后，他们又合作了时装商战剧《他来自江湖》<reference id="chunk:0_0"></reference>。

通过日志可以看出，系统成功地检索并整合了周星驰和万梓良之间的合作关系信息。

sub_retrieved_set在这个过程中发挥了关键作用，它存储了每个子问题的检索结果，这些结果随后被整合到最终的答案中。

最终的答案表明，周星驰和万梓良是合作演员关系，并且列举了他们合作的作品，如电影《捕风汉子》和电视剧《他来自江湖》。

总结

本周继续对KAG-solver中的planner、generator、和Solver-pipeline的代码进行了解读：

planner：利用 LLM 进行任务规划，能够生成任务计划，通过调用 LLM，将复杂问题转换为多个任务的有向无环图（DAG），并根据依赖关系逐步求解；
generator：使用LLM来生成最终答案。支持引用数据的处理，可以生成包含引用信息的答案。使用重排序工具对检索结果进行排序，提高答案的质量。通过灵活的提示模板，可以根据需要选择不同的提示。在生成答案时使用重试机制，确保生成过程的可靠性；
Solver_pipeline：通过迭代任务处理实现规划和执行。可以配置多种执行器和任务类型，处理复杂的多步骤任务。优化了推理阶段的知识分层机制，能够根据业务场景对知识精准性的要求进行分层检索和推理。

最后通过使用Solver，更加深入了解了KAG-Solver的执行过程、整体架构和核心功能，包括如何通过规划器生成任务计划，如何利用生成器合成最终答案，以及如何通过不同类型的管道协调各个组件完成复杂任务的求解过程。同时，也对各个组件之间的交互关系和工作流程有了更清晰的认识。