kotaemon 自定义索引与推理流水线开发指南从 BaseComponent 到 flowsettings 注册的完整实现【免费下载链接】kotaemonAn open-source RAG-based tool for chatting with your documents.项目地址: https://gitcode.com/GitHub_Trending/kot/kotaemon导读本文是 kotaemon开源 RAG 文档问答工具扩展开发的核心指南讲解如何为应用添加全新的索引indexing流水线与推理reasoning流水线。读完本文你将掌握用BaseComponent定义流水线类、遵循固定的run签名契约、在 flowsettings.py 中注册、通过get_user_settings/get_info/get_pipeline三个类方法接入应用设置与实例化流程、实现流式输出、复用应用级 LLM 与 Embedding 资源从而把自定义问答流程无缝嵌入 kotaemon 的 Gradio 界面。文章将结合仓库内 ktem/reasoning 与 ktem/index 的真实实现加以印证。一、总体思路两步接入新流水线在高层面上向 kotaemon 添加一条新的索引或推理流水线只需要两步定义将你的索引或推理流水线实现为一个继承自BaseComponent的类声明在设置文件flowsettings.py中声明该类写入KH_REASONINGS或KH_INDEX_TYPES。之后执行python app.py启动应用时kotaemon 会动态加载这些流水线。加载逻辑位于 ktem/app.py应用遍历settings.KH_REASONINGS用import_dotted_string(value, safeFalse)反序列化出类对象调用get_info()[id]作为唯一标识存入reasonings字典并调用get_user_settings()将配置项合并进应用默认设置。索引侧的对应逻辑在 ktem/index/manager.py 中遍历settings.KH_INDEX_TYPES完成同样的注册。因此你只需要修改 flowsettings.py不需要改动应用内核代码即可让新流水线出现在应用的流水线候选列表中。二、把流水线定义成一个类本质上一条流水线就是继承自kotaemon.base.BaseComponent的类。每条流水线由两个部分组成所有声明式参数与子流水线类属性声明流水线内部逻辑run方法。一个最小示例from kotaemon.base import BaseComponent class SoSimple(BaseComponent): arg1: int arg2: str def run(self, arg3: str): return self.arg1 * self.arg2 arg3这个示例只为演示而简化实际流水线可以携带大量参数、把其他流水线作为参数子节点并在run中实现复杂逻辑。索引流水线与推理流水线本质都是这样一个继承自BaseComponent的类。更详细的组件编写规范请参考 Creating a Component其中明确了组件的四个编写步骤继承BaseComponent、声明带类型注解的初始化参数、声明带类型注解的节点其他组件、在run中实现处理逻辑并强调“流水线本身就是嵌套组件”组件之间通过类型注解自动形成依赖关系。从源码看 BaseComponent 的契约BaseComponent定义在 libs/kotaemon/kotaemon/base/component.py它继承自theflow的Function并强调两个设计原则容忍多种输入类型例如str、Document、list[str]、list[Document]强制单一输出类型输出类型应尽可能通用。基类提供了run抽象方法必须实现、invoke/ainvoke同步/异步单次调用、stream/astream同步/异步流式调用以及report_output与set_output_queue两个用于流式输出的基础设施方法。同时Param、Node、lazy也被导出供组件声明参数与子节点使用。三、run 方法的签名契约注意本节内容在编写时尚属“暂定”状态原文档标注会在早期定稿但仓库现有实现均遵循下述约定可作为实现标准。3.1 索引流水线def run( self, file_paths: str | Path | list[str | Path], reindex: bool False, **kwargs, ): Index files to intermediate representation (e.g. vector, database...) Args: file_paths: the list of paths to files reindex: if True, files in file_paths that already exists in database should be reindex. 仓库中 ktem/index/file/base.py 的BaseFileIndexIndexing是这一契约的具体化run(file_paths, *args, **kwargs)返回(indexed_file_ids, error_messages)元组每个元素对应一个输入文件失败时相应位置为None同时定义了stream(file_paths, ...)用于向 UI 产出channel index或debug的Document最终返回(file_ids, errors, indexed_documents)。基类还提供了copy_to_filestorage辅助方法——以 SHA-256 哈希命名文件并复制到文件存储目录保证文件去重与安全存放。3.2 推理流水线def run(self, question: str, history: list, **kwargs) - Document: Answer the question Args: question: the user input history: the chat history [(user_msg1, bot_msg1), (user_msg2, bot_msg2)...] Returns: kotaemon.base.Document: the final answer 推理流水线返回kotaemon.base.Document。Document定义于 libs/kotaemon/kotaemon/base/schema.py继承自 llama_index 的Document额外携带content任意类型的原始内容source文档来源 id可选channel文档展示通道可选chat聊天消息、info信息面板、index索引面板、debug调试面板、plot图表。RetrievedDocument带score与retrieval_metadata则用于检索结果传递。应用端 ktem/pages/chat/init.py 在消费流水线输出时正是依据response.channel决定将内容渲染到聊天区还是信息面板因此推理流水线的每一步输出都应显式设置channel。仓库内置推理流水线的基类BaseReasoningktem/reasoning/base.py还声明了get_info必须实现、get_user_settings默认空字典、get_pipeline(user_settings, state, retrievers)默认cls()三个类方法并提供了run(message, conv_id, history, **kwargs)的默认签名其中conv_id是会话 id——这是与上述文档签名略有扩展的实际调用形态聊天页调用链 ktem/pages/chat/init.py 会依次传入llm_query、conversation_id、chat_history。四、把流水线注册到 ktem注册方式是在flowsettings.py中声明类路径。该文件位于你启动 ktem 的当前工作目录下默认即为仓库根目录的 flowsettings.pyKH_REASONINGS [python.module.path.to.the.reasoning.class] KH_INDEX_TYPES python.module.path.to.the.indexing.class注意两点通过填充KH_REASONINGS列表可以注册多条推理流水线用户可以在应用的设置Settings页中自由选择使用哪一条目前KH_INDEX_TYPES支持注册多种索引类型仓库当前的 flowsettings.py 即同时注册了ktem.index.file.FileIndex与若干 GraphRAG 索引。确保你的类能够被 Python 正常发现即模块路径可导入、类可被import_dotted_string反序列化。仓库内的真实注册示例当前仓库 flowsettings.py 默认注册了四条推理流水线KH_REASONINGS [ ktem.reasoning.simple.FullQAPipeline, ktem.reasoning.simple.FullDecomposeQAPipeline, ktem.reasoning.react.ReactAgentPipeline, ktem.reasoning.rewoo.RewooAgentPipeline, ]索引类型则在 flowsettings.py 中定义为KH_INDEX_TYPES列表每一项附带名称与supported_file_types等配置例如ktem.index.file.FileIndexFile Collection以及可选的GraphRAGIndex/LightRAGIndex等集合。同一文件还定义了KH_LLMS、KH_EMBEDDINGS、KH_RERANKINGS、KH_DOCSTORE、KH_VECTORSTORE等运行时资源它们是自定义流水线可复用的全局依赖。五、允许用户在应用设置中定制你的流水线要让用户能在设置页配置你的流水线需要在流水线类中声明两个类方法get_user_settings(cls) - dict返回设置字典ktem 会把它们并入应用设置get_info(cls) - dict返回流水线元信息用于标识与展示。示例class SoSimple(BaseComponent): ... # as above classmethod def get_user_settings(cls) - dict: The settings to the user return { setting_1: { name: Human-friendly name, value: Default value, choices: [(Human-friendly Choice 1, choice1-id), (HFC 2, choice2-id)], # optional component: Which Gradio UI component to render, can be: text, number, checkbox, dropdown, radio, checkboxgroup }, setting_2: { # follow the same rule as above } } classmethod def get_info(cls) - dict: Pipeline information for bookkeeping purpose return { id: a unique id to differentiate this pipeline from other pipeline, name: Human-friendly name of the pipeline, description: Can be a short description of this pipeline }设置项字段说明每个设置项是一个以设置 id 为键的字典支持以下字段字段是否必填说明name是显示在设置页的人类可读名称value是默认值choices否可选列表元素为(显示文本, 存储值)元组用于 dropdown / radio 等组件component推荐Gradio 组件类型text、number、checkbox、dropdown、radio、checkboxgroupinfo否对该设置的补充说明文案仓库内置流水线广泛使用special_type否特殊类型标记如llm只要给流水线类添加了这两个方法ktem 会在启动时自动提取并合并进设置见上文register_reasonings中对get_user_settings的调用。源码级范例内置 FullQAPipeline 的设置声明ktem/reasoning/simple.py 中FullQAPipeline.get_user_settings返回了完整设置集是本文档语法的真实落地llmdropdown 类型choices动态取自llms.options().keys()带special_type: llm与说明文案highlight_citationradio三选一highlight/inline/off并读取环境变量USE_LOW_LLM_REQUESTS决定默认值create_mindmap、create_citation_viz、use_multimodalcheckbox 类型system_prompt、qa_prompttext 类型后者占位符为{context}、{question}、{lang}默认值即 citation_qa.py 中的DEFAULT_QA_TEXT_PROMPTn_last_interactionsnumber 类型默认 5trigger_contextnumber 类型默认 150。其get_info返回{id: simple, name: Simple QA, description: ...}。注意get_user_settings的键会与get_info()[id]拼接形成完整设置路径如reasoning.options.simple.llm这一点在下一节的get_pipeline中体现。六、构造流水线对象get_pipeline当 ktem 需要真正运行你的流水线时它会以完整的用户设置调用你的类方法get_pipeline并期望获得一个实例化好的流水线对象。你应在该方法内实现所有初始化逻辑class SoSimple(BaseComponent): ... # as above classmethod def get_pipeline(cls, setting): obj cls(arg1setting[reasoning.id.setting1]) return obj源码级范例get_pipeline 如何消费设置FullQAPipeline.get_pipeline(settings, states, retrievers)ktem/reasoning/simple.py是这一流程的完整示范以settings.get(reasoning.max_context_length, 32000)读取全局上下文长度用cls.get_info()[id]拼出前缀reasoning.options.simple读取llm、highlight_citation、n_last_interactions、system_prompt、qa_prompt等全部用户配置根据highlight_citation选择AnswerWithInlineCitationinline或AnswerWithContextPipelinehighlight/off并逐一注入llm、enable_citation、enable_mindmap、enable_citation_viz、use_multimodal等属性依据settings[reasoning.lang]经由SUPPORTED_LANGUAGE_MAP设置回答语言返回配置完毕的流水线实例。这说明get_pipeline是“用户设置 → 运行时实例”的映射器所有用户在设置页调整过的参数都会在这里被读取并写入子组件。子类FullDecomposeQAPipeline还示范了如何在prepare_pipeline_instance中依据自己的get_info()[id]complex读取额外设置如decompose_prompt实现设置项的按需扩展。七、推理流水线把输出流式推送到 UI为了获得流畅的用户体验可以将输出直接流式推送到 UI这样用户从 LLM 生成第一个 token 起就能看到输出而不必等待整个流水线跑完。流式输出需要做两件事把run函数改为async通过self.report_output把输出投递到专用队列。async def run(self, question: str, history: list, **kwargs) - Document: for char in This is a long messages: self.report_output({output: text.text})self.report_output的参数是一个字典可包含以下两个键中的任一个或全部output该字符串会被流式写入聊天消息evidence该字符串会被流式写入信息面板information panel。源码级印证输出队列机制底层实现见 component.pyset_output_queue(queue)会递归地把队列传递给所有子节点BaseComponent类型report_output(output)则执行self._queue.put_nowait(output)。应用端在 ktem/pages/chat/init.py 中创建asyncio.Queue后调用pipeline.set_output_queue(queue)随后通过pipeline.stream(...)消费输出并按channel渲染。内置的 react.py 与 rewoo.py 展示了实际用法self.report_output(Document(contentanswer.text, channelchat))推送答案、Document(contentstep_output, channelinfo)推送中间步骤到信息面板最后以self.report_output(None)通知结束。提示从仓库实现看report_output实际接收的是Document带channel或None而异步方式asyncrunreport_output与同步生成器方式streamyield Document(channel...)均可实现流式体验后者正是FullQAPipeline.streamsimple.py所采用的形式。八、访问应用级 LLM 与 Embedding自定义流水线可以直接访问用户配置好的 LLM 与 Embedding 模型集合from ktem.embeddings.manager import embeddings from ktem.llms.manager import llms llm llms.get_default() embedding_model embeddings.get_default()ktem.llms.manager.LLMManagerlibs/ktem/ktem/llms/manager.py维护一个模型池初始化时从flowsettings.KH_LLMS读取模型定义写入 SQLite 表get_default()返回默认模型无默认时随机挑选options()返回全部模型字典get(key, default)按名获取。这些模型在启动时通过deserialize(item.spec, safeFalse)从配置字典实例化。Embedding 管理器ktem.embeddings.manager结构类似。允许用户在设置中指定模型你也可以在get_user_settings中暴露模型选择项让用户为每条流水线单独指定 LLM 或 Embeddingclassmethod def get_user_settings(cls) - dict: from ktem.llms.manager import llms return { citation_llm: { name: LLM for citation, value: llms.get_default(), component: dropdown, choices: list(llms.options().keys()), }, ... }这正是FullQAPipeline的做法其llm设置项的choices来自llms.options().keys()而get_pipeline中用llms.get(llm_name, llms.get_default())完成回退解析——用户未指定时优雅地回退到应用默认模型。九、可选访问应用数据数据库与向量存储自定义流水线还可以访问用户的应用数据库与向量存储# get the database that contains the source files from ktem.db.models import Source, Index, Conversation, User # get the vector store会话、用户、设置等应用级数据表定义于 ktem/db/models.pyConversation、User、Settings、IssueReport其字段详见 ktem/db/base_models.py例如Conversation记录会话 id、名称、data_sourceJSON 字段存放消息与文件列表等。索引相关的Source表与Index表是在每个文件索引初始化时动态创建的见 ktem/index/file/index.pySource记录文件名称、路径、大小、所属用户与备注Index表则维护source_id → target_id的映射关系relation_type 区分 docstore 与 vectorstore 关联。检索器与索引流水线正是通过BaseFileIndexIndexing/BaseFileIndexRetriever基类注入的Source、Index、VS向量存储、DS文档存储、FSPath等Param来读写这些数据。向量存储与文档存储的全局配置见 flowsettings.py 的KH_DOCSTORE与KH_VECTORSTORE默认分别为 LanceDB 文档存储与 Chroma 向量存储可切换 Milvus、Qdrant 等。十、完整接入流程小结与实战检查清单把以上章节串起来一条自定义推理流水线从代码到上线的完整路径是定义类继承BaseReasoning或直接BaseComponent实现run/stream/ainvoke之一签名遵循“question、history、conv_id→ Document”的契约实现三个类方法get_infoid/name/description、get_user_settings设置字典、get_pipeline用设置构造实例可注入llms/embeddings与检索器在 flowsettings.py 的KH_REASONINGS列表中追加类路径启动python app.py应用会在 ktem/app.py 中自动注册该流水线在应用的 Settings 页中选择该流水线并调整其专属设置项。索引流水线的接入路径与之对称实现BaseFileIndexIndexing子类run处理file_paths与reindex在KH_INDEX_TYPES中声明并可选实现get_user_settings/get_pipeline以暴露索引参数如 chunk 大小、重排模型选择参考 ktem/index/file/pipelines.py 中FileIndexingPipeline与检索器的实现。调试建议仓库内置流水线在关键节点使用print输出如 “Retrievers ...”、“Got N retrieved documents”自定义流水线可沿用这一风格配合Document(channeldebug)输出进行排障同时善用Document(channelinfo)在信息面板展示中间步骤与证据这是内置 Agent 类流水线React/ReWoo的标准做法。参考阅读组件编写规范Creating a Component默认设置文件flowsettings.py推理流水线基类ktem/reasoning/base.py内置推理流水线示例ktem/reasoning/simple.py、ktem/reasoning/react.py、ktem/reasoning/rewoo.py索引流水线基类与实现ktem/index/file/base.py、ktem/index/file/pipelines.py、ktem/index/file/index.py数据模型ktem/db/models.py、ktem/db/base_models.py组件基类与 Document 定义libs/kotaemon/kotaemon/base/component.py、libs/kotaemon/kotaemon/base/schema.py【免费下载链接】kotaemonAn open-source RAG-based tool for chatting with your documents.项目地址: https://gitcode.com/GitHub_Trending/kot/kotaemon创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考