Knowledge（knowledge/knowledge.py:88）就是"一堆源 + 一个存储"的组合：

# knowledge/knowledge.py:88（节选）
class Knowledge(BaseModel):
    sources: Annotated[list[BaseKnowledgeSource], BeforeValidator(_resolve_knowledge_sources)] = ...
    storage: BaseKnowledgeStorage | None = Field(default=None)
    embedder: ... = None
    collection_name: str | None = None

    def __init__(self, collection_name, sources, embedder=None, storage=None, **data):
        super().__init__(**data)
        if storage is not None:
            self.storage = storage
        else:
            from crewai.knowledge.storage.factory import resolve_knowledge_storage
            custom = resolve_knowledge_storage(embedder, collection_name)   # 先问工厂
            self.storage = custom if custom is not None else KnowledgeStorage(
                embedder=embedder, collection_name=collection_name)         # 默认
        self.sources = sources

    def add_sources(self) -> None:
        for source in self.sources:
            source.storage = self.storage    # ★把存储注入每个源
            source.add()                     # ★每个源自己去切块+存

    def query(self, query, results_limit=5, score_threshold=0.6) -> list[SearchResult]:
        return self.storage.search(query, limit=results_limit, score_threshold=score_threshold)

sources 字段上挂了个 BeforeValidator，能把字典按 source_type 解析成对应类（knowledge/knowledge.py:23）：

# knowledge/knowledge.py:23
_KNOWN_SOURCES: dict[str, type[BaseKnowledgeSource]] = {
    "string": StringKnowledgeSource, "docling": CrewDoclingSource,
    "csv": CSVKnowledgeSource, "excel": ExcelKnowledgeSource,
    "json": JSONKnowledgeSource, "pdf": PDFKnowledgeSource,
    "text_file": TextFileKnowledgeSource,
}

# knowledge/knowledge.py:34
def _resolve_knowledge_sources(value: Any) -> Any:
    if not isinstance(value, list): return value
    resolved = []
    for idx, item in enumerate(value):
        if isinstance(item, dict):
            tag = item.get("source_type")
            cls = _KNOWN_SOURCES.get(tag)            # 按 source_type 找类
            if cls is None:
                raise ValueError(f"Unknown source_type={tag!r} at index {idx}: "
                                 f"expected one of {sorted(_KNOWN_SOURCES)}")
            resolved.append(cls.model_validate(item))   # 用对应类校验/构造
        else:
            resolved.append(item)                    # 已是实例，原样
    return resolved

所有源的基类（knowledge/source/base_knowledge_source.py:16），核心是切块和保存：

# knowledge/source/base_knowledge_source.py:16（节选）
class BaseKnowledgeSource(BaseModel, ABC):
    chunk_size: int = 4000                    # 每块最多 4000 字符
    chunk_overlap: int = 200                  # 相邻块重叠 200 字符
    chunks: list[str] = Field(default_factory=list)
    chunk_embeddings: list[np.ndarray] = Field(default_factory=list, exclude=True)
    storage: BaseKnowledgeStorage | None = Field(default=None)

    @abstractmethod
    def validate_content(self) -> Any: ...    # 各源自己实现：怎么读
    @abstractmethod
    def add(self) -> None: ...                # 各源自己实现：读→切→存

    def _chunk_text(self, text: str) -> list[str]:
        return [text[i : i + self.chunk_size]                     # ★滑动窗口切块
                for i in range(0, len(text), self.chunk_size - self.chunk_overlap)]

    def _save_documents(self) -> None:
        if self.storage is not None:
            self.storage.save(self.chunks)    # 交给 KnowledgeStorage
        else:
            raise ValueError("No storage found to save documents.")

最简单的 StringKnowledgeSource（knowledge/source/string_knowledge_source.py:8）：

# knowledge/source/string_knowledge_source.py:8（节选）
class StringKnowledgeSource(BaseKnowledgeSource):
    source_type: Literal["string"] = "string"
    content: str = Field(...)
    def validate_content(self) -> None:
        if not isinstance(self.content, str):
            raise ValueError("StringKnowledgeSource only accepts string content")
    def add(self) -> None:
        new_chunks = self._chunk_text(self.content)   # 切
        self.chunks.extend(new_chunks)
        self._save_documents()                        # 存

文件类源走 BaseFileKnowledgeSource（knowledge/source/base_file_knowledge_source.py:44）三步初始化：

# knowledge/source/base_file_knowledge_source.py:44
def model_post_init(self, _: Any) -> None:
    self.safe_file_paths = self._process_file_paths()   # ① 规整路径（相对→知识目录下）
    self.validate_content()                             # ② 校验文件存在/是文件
    self.content = self.load_content()                  # ③ 抽象：各源读各自格式

# text_file_knowledge_source.py:12 —— 一个具体实现
def load_content(self) -> dict[Path, str]:
    content = {}
    for path in self.safe_file_paths:
        path = self.convert_to_path(path)
        with open(path, "r", encoding="utf-8") as f:
            content[path] = f.read()
    return content

KnowledgeStorage（knowledge/storage/knowledge_storage.py:22）是"知识库"和"D38 RAG 客户端"之间的适配器：

# knowledge/storage/knowledge_storage.py:37（节选）
@model_validator(mode="after")
def _init_client(self) -> Self:
    if self.embedder:
        embedding_function = build_embedder(self.embedder)     # D40 的嵌入工厂
        config = ChromaDBConfig(embedding_function=cast(..., embedding_function))
        self._client = create_client(config)                   # 造一个 ChromaDBClient
    return self

def _get_client(self) -> BaseClient:
    return self._client if self._client else get_rag_client()  # 实例专属 or 全局共享

def save(self, documents: list[str]) -> None:
    if not documents: return
    client = self._get_client()
    collection_name = f"knowledge_{self.collection_name}" if self.collection_name else "knowledge"
    client.get_or_create_collection(collection_name=collection_name)
    rag_documents: list[BaseRecord] = [{"content": doc} for doc in documents]   # chunk→BaseRecord
    client.add_documents(collection_name=collection_name, documents=rag_documents)

crew 用 knowledge_sources 一键建知识库（crew.py:675）：

# crew.py:675（节选）
@model_validator(mode="after")
def create_crew_knowledge(self) -> Crew:
    if self.knowledge_sources:
        if isinstance(self.knowledge_sources, list) and all(
                isinstance(k, BaseKnowledgeSource) for k in self.knowledge_sources):
            self.knowledge = Knowledge(
                sources=self.knowledge_sources, embedder=self.embedder,
                collection_name="crew")
            self.knowledge.add_sources()      # ★建 crew 时就把资料灌进去
    return self

而 embedder 字段有个巧妙的序列化处理（knowledge/knowledge.py:69）：

# knowledge/knowledge.py:69
def _serialize_embedder_spec(value: Any) -> dict[str, Any] | None:
    if value is None: return None
    if isinstance(value, BaseEmbeddingsProvider): return value.model_dump(mode="json")
    if isinstance(value, dict): return value
    if isinstance(value, type) and issubclass(value, BaseEmbeddingsProvider):
        raise TypeError(f"Cannot checkpoint embedder class ...: "     # ★类不能序列化
            "build_embedder requires an instance or ProviderSpec dict, not a class.")
    raise TypeError(f"Cannot serialize embedder of type {type(value).__name__}: ...")