跳到主要内容

组件契约：@component 到底做了什么

本章讲一个普通 Python 类怎么变成 Haystack 组件，以及 connect() 怎么靠类型把两个组件接上、组件怎么存盘恢复。

1. 组件的契约（白话）​

一个组件类必须满足（见 component.py 顶部模块 docstring，它自称「the source of truth for components contract」，component.py:10）：

  • 用 @component 装饰；
  • 有一个 run(self, ...) 方法（必须），返回一个 dict；
  • 用 @component.output_types(...) 声明输出口的名字和类型；
  • __init__ 要轻量（构造管道时会频繁实例化），重活放 warm_up()；
  • 输入口由 run() 的参数签名自动推出。

一个最小组件（示意，非源码，但就是真实写法）：

from haystack import component

@component
class DoubleIt:
    @component.output_types(result=int)   # 声明：输出口叫 result，类型 int
    def run(self, value: int):            # 输入口 value 从签名推出
        return {"result": value * 2}      # 返回 dict，键 = 输出口名

2. 核心机制：@component 装饰器做了什么​

2.1 思路​

装饰器要做三件事：(1) 把类登记进一个全局 registry（反序列化时按类路径找回类）；(2) 让类用 Haystack 的元类 ComponentMeta，好在实例化时自动建 socket；(3) 装上统一的 __repr__。

2.2 真实实现​

_Component._component（component.py:572）：

  • 先 if not hasattr(cls, "run") 就报错（component.py:579）；
  • 用 typing.new_class(...) 重建这个类，让它带上 ComponentMeta 元类（component.py:597）——注释解释为何重建而不是直接改：是为了让类型检查器仍认得类的类型；
  • 把 {module}.{ClassName} 存进 self.registry（component.py:600-610）。

@component 既能带括号也能不带，靠 __call__ 里的 wrap 兼容两种写法（component.py:630-641）。

2.3 socket 什么时候建？​

不在装饰时，而在每次实例化时，由元类 ComponentMeta.__call__（component.py:294）触发：

  • _parse_and_set_input_sockets（component.py:231）：读 run 的签名，每个参数（除 self、*args、**kwargs）变成一个 InputSocket，带名字、类型、默认值（有默认值 = 非必填）；
  • _parse_and_set_output_sockets（component.py:207）：从 run 上被 @output_types 缓存的 _output_types_cache 取输出 socket；
  • 若组件同时有 run 和 run_async，会强制校验两者签名一致（component.py:288-292），不一致直接报错。

@output_types 本身（component.py:534）只是把类型字典塞到 run 方法的 _output_types_cache 属性上，留给元类在实例化时读取（component.py:550-568）。

2.4 动态 socket​

如果 run(self, **kwargs) 用了 kwargs，组件可在 __init__ 里调 component.set_input_types(self, ...)（component.py:449）/ set_output_types（component.py:499）手动声明端口。Agent 就是这么干的——它的输入口随 state_schema 动态变（见第 4 章）。

3. 核心机制：connect() 靠类型配对​

3.1 思路​

connect("a", "b.documents") 要在 a 的输出 socket 和 b 的 documents 输入 socket 之间连一条边——前提是类型兼容。

3.2 真实实现​

PipelineBase.connect（base.py:441）：

  1. 解析 "component.socket" 字符串，分出组件名和 socket 名（parse_connect_string）；
  2. 没指定 socket 名时，把两边所有 socket 当候选；
  3. 对所有 (输出 socket × 输入 socket) 组合，调 _types_are_compatible（core/type_utils.py:233）判断能否连、要不要类型转换（base.py:529-534）；
  4. 多个可连时，严格匹配优先于「需转换」匹配（base.py:538-545）——这是为了向后兼容（老版本不允许类型转换）；
  5. 找不到任何可连组合就报 PipelineConnectError，错误信息里列出两边 socket 和类型（base.py:555-560）。

连上后，边上记着 from_socket / to_socket / conn_type，同时把 sender 名写进接收 socket 的 senders 列表——这正是第 1 章调度时判断「上游有没有送值」的依据。

4. 序列化:组件存盘与恢复​

Haystack 的卖点之一是管道可存成 YAML/JSON。机制：

  • 存：default_to_dict(obj, **init_params)（serialization.py:177）产出 {"type": 全限定类名, "init_parameters": {...}}。全限定名由 generate_qualified_class_name（serialization.py:127）算出。组件可重写 to_dict 处理特殊字段（如把回调函数序列化成 import 路径）。
  • 读：default_from_dict(cls, data)（serialization.py:250）校验 data["type"] 等于该类全限定名，再用 init_parameters 重建。整条管道的 from_dict（base.py:177）按每个组件的 type 去 component.registry 里查类（base.py:232）——这就是 §2.2 那个 registry 的用途。

约束：init_parameters 里的值必须 JSON 可序列化（component.py:31-35 的契约）。要存类/可调用对象，约定存成 "module.symbol" 字符串，from_dict 里用 importlib 还原。

下一章：把引擎和契约串起来，看一条真实 RAG 请求怎么流。见 03-rag-path.md。