组件模型:一切皆可组合的 BaseComponent

30 秒导读: kotaemon 把 RAG 里的每一步(读文件、切分、向量检索、重排、问答、推理)都写成同一种东西——一个 BaseComponent。组件之间靠声明式的字段(Param/Node)拼成一棵树,靠统一的数据类型(Document 家族)互相传值。看懂这一章,后面所有 XxxPipeline 你都认识它的骨架了。

本章只讲抽象和数据模型,不讲具体的索引/检索/问答流程(那是 02–05 的事)。目标:让你以后读到任何 class XxxPipeline(BaseComponent),都知道它的 Param/Node、run/stream、Document 输出各自是什么,后面章节不再重复解释这些。

---

1. 这是什么(零基础也能懂)

一句话定义: BaseComponent 是 kotaemon 的最底层积木——一个"接受输入、产出输出"的可调用对象,而且它自己内部可以由更多同类积木拼成。

它要解决什么问题。 RAG(检索增强生成)是一条长流水线:文档进来 → 切块 → 存进向量库 → 检索 → 重排 → 塞进 prompt → 让 LLM 生成带引用的答案。如果每一步都用互不相同的类和函数各写各的,整条线就会又乱又难换件。kotaemon 的答案是:让每一步都长成同一个样子,于是它们能像乐高一样任意替换、嵌套、串联。

一句话直觉/类比。 把它想成 PyTorch 的 nn.Module:

nn.Module 世界	kotaemon 世界
每层是一个 Module	每步是一个 BaseComponent
子层写成 self.conv = nn.Conv2d(...)	子组件写成 Node(...) / Param(...) 字段
forward() 定义前向计算	run() 定义这步做什么
张量在层间流动	Document 在组件间流动

用起来什么样。 一个组件只需要继承 BaseComponent、声明几个子部件字段、实现 run,就能当函数一样直接调用:

# 示意,非源码。真实例子见 libs/kotaemon/kotaemon/llms/cot.py:13 的 Thought
class Thought(BaseComponent):
    prompt: str = Param()                       # 一个"参数"字段:配置值
    llm: LLM = Node(LCAzureChatOpenAI)          # 一个"节点"字段:子组件
    post_process: Function = Node()

    def run(self, **kwargs) -> Document:        # 这步做什么
        prompt = self.prompt_template(**kwargs).text
        response = self.llm(prompt).text        # 直接把子组件当函数调
        return Document(self.post_process(response))

thought = Thought(prompt="Word {word} in {language} is ", ...)
thought(word="hello", language="French")        # 像调函数一样调组件 → 触发 run

重点看两处:字段 llm = Node(...) 声明了"我内部含一个 LLM 子组件",而 self.llm(prompt) 直接把这个子组件当函数调——组件调用组件,这就是"可组合"。

---

2. 顶层全景(它大概怎么转)

kotaemon 的组件模型建在第三方库 theflow(声明式管线框架)之上。BaseComponent 本身很薄——它继承 theflow.Function,只加了几样 RAG 需要的东西:四种执行形态、inflow/flow 串联、以及一套流式输出机制。

真正的"声明式"魔法(Param/Node/lazy/.auto)全部来自 theflow,kotaemon 只是 from theflow import ... 再 re-export 出去(见 libs/kotaemon/kotaemon/base/component.py:4 与 :66 的 __all__)。

这套模型的三个支柱

                      ┌─────────────────────────────────┐
                      │         BaseComponent           │
                      │      (theflow.Function 的子类)   │
                      └─────────────────────────────────┘
                        │            │              │
          ┌─────────────┘            │              └──────────────┐
          ▼                          ▼                             ▼
 ┌──────────────────┐   ┌──────────────────────┐   ┌────────────────────────┐
 │ ① 结构:怎么拼    │   │ ② 执行:怎么跑        │   │ ③ 数据:传什么          │
 │                  │   │                      │   │                        │
 │ Param  = 配置    │   │ run     普通调用      │   │ Document   通用文档     │
 │ Node   = 子组件  │   │ invoke  单值        │   │  ├ RetrievedDocument   │
 │ .auto  = 惰性构建│   │ stream  同步流       │   │  ├ BaseMessage 家族     │
 │ .withx = 延迟实例│   │ astream 异步流       │   │  └ LLMInterface        │
 └──────────────────┘   └──────────────────────┘   └────────────────────────┘
      §3                       §4                          §5

三个支柱各管一件事,后面三节各讲一个:

支柱	管什么	关键符号	定义位置
① 结构	组件如何声明子部件、如何拼成树	Param Node .auto .withx	来自 theflow,component.py:4 re-export
② 执行	一个组件能以哪几种方式被跑	run invoke stream astream	component.py:46-63
③ 数据	组件之间流动的是什么类型	Document 及其子类	schema.py:21-157

一条主线走一遍(高层)。 用户调 component(x) → theflow 的 Function.__call__ 最终派发到你写的 run(x) → run 里调用 self.某子组件(...)(子组件也是 BaseComponent,递归重复)→ 每层产出 Document 往上返回。整个 RAG 流水线就是这样一棵 BaseComponent 树,一次调用自顶向下递归执行。

---

3. 支柱①:结构——用 Param / Node 声明式地拼组件

它要解决的小问题: 一个组件"内部有哪些配置、有哪些子组件",怎么写才能既清楚、又能被框架看见(好做缓存、日志、可视化)?

思路/直觉: 不用 __init__ 里手写一堆 self.x = ...,而是像 Pydantic / dataclass 那样,把每个部件声明成类字段,并用两个标记区分它是"配置"还是"子组件":

• Param —— 一个参数/配置值(数字、字符串、开关、回调函数)。例:top_k: int = 5、reader_mode: str = Param("default", ...)。
• Node —— 一个子组件(本身也是可运行的 BaseComponent)。例:llm: LLM = Node(LCAzureChatOpenAI)。

真实例子,来自 chain-of-thought 的 Thought:

# libs/kotaemon/kotaemon/llms/cot.py:69-83
prompt: str = Param(help="The prompt template string. ...")
llm: LLM = Node(LCAzureChatOpenAI, help="The LLM model to execute the input prompt")
post_process: Function = Node(help="The function post-processor ...")

prompt 是配置(一串模板文字),llm 和 post_process 是子组件——框架据此知道这个组件的"零件清单"。

3.1 .withx():先声明、晚实例化

问题: 类字段的默认值在类定义时就要写出来。但有时你想给一个字段一个"已配好参数、但还没真正构造完"的子组件当默认值——直接 LLMReranking(...) 可能过早触发昂贵的初始化。

.withx() 就是"带着这些参数,但先别真造出来"的延迟实例化(lazy instantiation):

# libs/ktem/ktem/index/file/pipelines.py:99
llm_scorer: LLMReranking | None = LLMReranking.withx()

它返回的是一个"待构造"的占位,等这个字段真正被用到时框架才把它实例化。.withx 由 theflow 的 lazy 机制提供(lazy 同样从 theflow re-export,见 component.py:4)。

3.2 Node.auto / Param.auto(depends_on=...):按依赖惰性构建

问题: 有些子组件/参数不能写成静态默认值,因为它得由别的字段算出来。比如"用哪个文档读取器"取决于用户选的 reader_mode;"向量检索器"取决于 embedding 和向量库对象。

.auto 装饰器把一个方法变成"惰性计算的字段":它平时不跑,等被第一次访问时才执行方法体、把返回值当作该字段的值;depends_on= 声明它依赖哪些别的字段,那些字段变了就重算。

例一:Node.auto 惰性构建子组件。 DocumentRetrievalPipeline 的向量检索器依赖 embedding 和向量库/文档库:

# libs/ktem/ktem/index/file/pipelines.py:105-113
@Node.auto(depends_on=["embedding", "VS", "DS"])
def vector_retrieval(self) -> VectorRetrieval:
    return VectorRetrieval(
        embedding=self.embedding,
        vector_store=self.VS,
        doc_store=self.DS,
        retrieval_mode=self.retrieval_mode,
        rerankers=self.rerankers,
    )

之后代码里直接 self.vector_retrieval(text=text, top_k=...)(同文件 :177)当子组件调——它是被 .auto 现造出来的一个 Node。

例二:Param.auto 惰性计算配置值。 IndexDocumentPipeline 的"读取器表"随 reader_mode 变化:

# libs/ktem/ktem/index/file/pipelines.py:675-711(节选)
@Param.auto(depends_on="reader_mode")
def readers(self):
    readers = deepcopy(KH_DEFAULT_FILE_EXTRACTORS)
    if self.reader_mode == "adobe":
        readers[".pdf"] = adobe_reader
    elif self.reader_mode == "azure-di":
        readers[".pdf"] = azure_reader
    # ... 其它模式
    return readers

reader_mode 一改,readers 就按新模式重算——配置随依赖自动派生,不用手动同步。

一句话记住 .auto: Node.auto 造出来的是子组件,Param.auto 算出来的是配置值;二者都惰性、都能 depends_on 别的字段。

---

4. 支柱②:执行——一个组件的四种"跑法"

它要解决的小问题: 同一个组件,有时你要它一次算完返回结果,有时要它边算边吐(流式,让 UI 逐字显示),还要区分同步/异步。怎么用一套接口覆盖这四种?

四种执行形态。 BaseComponent 声明了四个方法,子类按需实现(libs/kotaemon/kotaemon/base/component.py:46-63):

方法	同步/异步	返回	用途
run	同步	单值 / 列表 / 迭代器 / Any	唯一抽象方法,必须实现;组件的核心逻辑
invoke	同步	Document / list[Document] / None	单值执行形态
ainvoke	异步	同上	invoke 的 async 版
stream	同步	Iterator[Document]	流式:一边算一边 yield
astream	异步	AsyncGenerator[Document]	流式的 async 版

注意:只有 run 带 @abstractmethod(component.py:58),是必须实现的那一个;其余四个是空实现(...),子类用哪种就覆盖哪种。你调 component(x) 时,theflow 的 Function.__call__ 会把调用派发到 run。

流式的真实样子。 带引用问答组件覆盖 stream,把 LLM 的增量输出一段段 yield 成 Document:

# libs/kotaemon/kotaemon/indices/qa/citation_qa.py:262-272(节选)
try:
    for out_msg in self.llm.stream(messages):        # 子组件也支持 stream
        output += out_msg.text
        yield Document(channel="chat", content=out_msg.text)   # 逐段吐出
except NotImplementedError:
    output = self.llm(messages).text                 # 不支持流式就退回一次算完
    yield Document(channel="chat", content=output)

重点看:stream 里调用子组件的 stream,把子组件吐出的片段再包成 Document 往上 yield——流式沿组件树逐层透传。

4.1 inflow / flow:把上一环的输出喂给下一环

BaseComponent 还带一对轻量的串联原语(component.py:22-33):把某个组件的 inflow 设成上游组件,调 .flow() 时它会先跑上游的 flow()、再把结果喂给自己:

# libs/kotaemon/kotaemon/base/component.py:24-33(节选)
def flow(self):
    if self.inflow is None:
        raise ValueError("No inflow provided.")
    if not isinstance(self.inflow, BaseComponent):
        raise ValueError(f"inflow must be a BaseComponent, found {type(self.inflow)}")
    return self.__call__(self.inflow.flow())         # 先跑上游,再喂给自己

这是一条"链式管道"的简单写法,与 Node/Param 的树状组合互补——前者是"A 接 B 接 C"的线性流,后者是"组件内部含子组件"的嵌套树。

4.2 set_output_queue / report_output:旁路流式通道

问题: stream 只能吐"主输出"(逐字答案)。但一个推理 agent 还想在生成答案的同时往 UI 的不同面板推送中间过程(思考步骤、检索到的证据),这些不是主返回值,怎么送出去?

机制:一条共享队列。 UI 层调 pipeline.set_output_queue(queue) 把一个队列注入组件;set_output_queue 会递归把队列装进所有子组件(component.py:35-40,遍历 self._ff_nodes)。之后组件内部任意深处调 self.report_output(doc),就把一个 Document 塞进队列;UI 那头从队列里取,推给对应面板。传 None 表示"流结束"。

# libs/kotaemon/kotaemon/base/component.py:35-44
def set_output_queue(self, queue):
    self._queue = queue
    for name in self._ff_nodes:                      # 递归下发到每个子组件
        node = getattr(self, name)
        if isinstance(node, BaseComponent):
            node.set_output_queue(queue)

def report_output(self, output: Optional[Document]):
    if self._queue is not None:
        self._queue.put_nowait(output)               # 无阻塞塞进队列

真实用法在 ReAct 推理里——主答案发到 chat 频道,每个中间步骤发到 info 频道,最后 report_output(None) 收尾:

# libs/ktem/ktem/reasoning/react.py:220-226
self.report_output(Document(content=answer.text, channel="chat"))
for _, step_output in answer.intermediate_steps:
    self.report_output(Document(content=step_output, channel="info"))
self.report_output(None)                             # None = 流结束

UI 侧则是 pipeline.set_output_queue(queue) 后消费这条队列(libs/ktem/ktem/pages/chat/__init__.py:1341)。channel 字段决定这段内容显示在哪个面板——见下节 Document 的 channel。

---

5. 支柱③:数据契约——所有组件之间流动的是 Document 家族

它要解决的小问题: 上百个组件互相传值,如果各传各的类型,接口就对不上。kotaemon 的做法是:统一都传 Document(或其子类)。这就是 component.py:16-19 注释说的"容忍多种输入类型,但强制单一输出类型"。

Document 是家族之根(libs/kotaemon/kotaemon/base/schema.py:21),继承自 llama-index 的 Document。它的巧妙处是构造函数接受一个任意类型的位置参数 content——文本、embedding、甚至另一个 Document 都能塞进去(schema.py:43-62)。几个关键字段:

字段	作用	位置
content: Any	原始内容,任意类型	schema.py:39
source: Optional[str]	来源 id	schema.py:40
channel	显示频道:chat/info/index/debug/plot	schema.py:41

channel 正是上一节 report_output 用来决定"这段内容送到 UI 哪个面板"的开关。

家族全景

Document  (schema.py:21) ── 万物之根,content 可为任意类型
  │
  ├─ DocumentWithEmbedding (:87) ── 强制带 embedding 向量的文档
  │
  ├─ RetrievedDocument (:121) ── 检索结果:多带 score + retrieval_metadata
  │
  ├─ ExtractorOutput (:151) ── 抽取器输出:多带 matches 列表
  │
  └─ BaseMessage (:98) ── 对话消息之根(带 to_openai_format)
       ├─ SystemMessage (:106)   role=system
       ├─ HumanMessage  (:116)   role=user
       └─ AIMessage     (:111)   role=assistant
            └─ LLMInterface (:135) ── LLM 调用的完整结果契约
                 └─ StructuredOutputLLMInterface (:146)

三个你在后续章节会反复见到的成员:

• RetrievedDocument(schema.py:121-132):检索/重排环节的产物,在 Document 基础上加了 score: float 和 retrieval_metadata: dict——retrieval_metadata 专门用来让检索管线里不同组件"互相传话"(见 03 混合检索)。

• BaseMessage 家族(schema.py:98-118):把对话消息也建模成 Document 的子类,并各自实现 to_openai_format() 转成 OpenAI 的 {"role": ..., "content": ...}。于是"一段文档"和"一条聊天消息"共用同一套基类。

• LLMInterface(schema.py:135-143):所有 LLM 组件的统一返回契约。它是 AIMessage 的子类,除了文本还带 candidates、completion_tokens/total_tokens/prompt_tokens、total_cost、logprobs 等——不管底层接的是哪家模型,上层拿到的都是同一个 LLMInterface。

为什么这很关键。 因为输出类型统一,任意两个组件才能接上:检索组件吐 list[RetrievedDocument],问答组件就能直接消费;LLM 组件吐 LLMInterface,谁调都拿到一样的字段。这正是"一切皆可组合"能成立的前提——结构统一(§3)+ 执行统一(§4)+ 数据统一(§5),缺一不可。

---

6. 边界与局限(诚实)

• BaseComponent 很薄,重活在 theflow。 缓存、日志、可视化、Param/Node/lazy/.auto 的真实实现都在第三方库 theflow(kotaemon 只 re-export,component.py:4、:66)。本仓库源码里看不到这些机制的内部实现;要深挖 .auto 的求值时机、缓存策略需读 theflow 本身(本克隆内未包含)。

• invoke/ainvoke/stream/astream 是"可选契约"。 它们在基类里是空实现(...),不保证每个组件都实现。调 stream 前不能假定它一定支持——citation_qa.py:269 就用 try/except NotImplementedError 兜底退回一次算完的路径。

• report_output 依赖外部先注入队列。 若没人调 set_output_queue,self._queue 为 None,report_output 会静默丢弃(component.py:43 的 if self._queue is not None)——旁路流式在没有 UI 宿主时是"哑"的。

---

7. 横向对比

同为 agent/RAG 框架,对"最小可组合单元"的取舍不同:

框架	组合单元	声明子部件的方式	数据契约
kotaemon	BaseComponent(theflow.Function)	Param / Node 类字段 + .auto 惰性	统一 Document 家族
LangChain (LCEL)	Runnable	| 管道操作符串联	松散(dict / str / Message)
LlamaIndex	QueryPipeline 模块	显式 add_link 连边	Node/Response 等多种

kotaemon 的特点是把"配置"和"子组件"在字段层面分成 Param/Node 两类,并用 .auto(depends_on=...) 表达派生依赖——比纯管道符更接近 nn.Module 的"含子模块"心智模型。想看这套抽象怎么落到真实 RAG 流程,继续读 02 索引与切分 → 03 混合检索与重排 → 04 带引证问答 → 05 可插拔推理。

---

8. 代码地图(导航索引)

主题	文件路径	符号名
组件基类(四种执行形态 + inflow/flow)	libs/kotaemon/kotaemon/base/component.py	BaseComponent、run、invoke、stream、astream、flow
旁路流式输出机制	libs/kotaemon/kotaemon/base/component.py	set_output_queue、report_output
Param/Node/lazy 的 re-export	libs/kotaemon/kotaemon/base/component.py	__all__(Param、Node、lazy)
统一数据契约:文档之根	libs/kotaemon/kotaemon/base/schema.py	Document(content/source/channel)
检索结果契约	libs/kotaemon/kotaemon/base/schema.py	RetrievedDocument(score、retrieval_metadata)
对话消息家族	libs/kotaemon/kotaemon/base/schema.py	BaseMessage、SystemMessage、HumanMessage、AIMessage、to_openai_format
LLM 返回契约	libs/kotaemon/kotaemon/base/schema.py	LLMInterface、StructuredOutputLLMInterface
Node.auto 惰性构建子组件示例	libs/ktem/ktem/index/file/pipelines.py	DocumentRetrievalPipeline.vector_retrieval
Param.auto(depends_on=...) 派生配置示例	libs/ktem/ktem/index/file/pipelines.py	IndexDocumentPipeline.readers、reader_mode
Param/Node/Node.auto/.withx 综合示例	libs/kotaemon/kotaemon/llms/cot.py	Thought(prompt、llm、prompt_template)
stream 覆盖 + 流式透传示例	libs/kotaemon/kotaemon/indices/qa/citation_qa.py	AnswerWithContextPipeline.stream
report_output 真实用法(多频道)	libs/ktem/ktem/reasoning/react.py	report_output、channel="chat"/"info"
set_output_queue 宿主注入点	libs/ktem/ktem/pages/chat/__init__.py	pipeline.set_output_queue(queue)