目录结构（lib/crewai/src/crewai/rag/）分得很清楚：

rag/
├── core/
│   ├── base_client.py            # ★BaseClient 协议（所有后端的契约）
│   ├── base_embeddings_callable.py  # EmbeddingFunction 协议
│   └── base_embeddings_provider.py  # 嵌入 provider 基类（D40）
├── types.py                      # BaseRecord / SearchResult / EmbeddingFunction
├── chromadb/                     # ★默认后端 ChromaDB 实现
│   ├── client.py                 # ChromaDBClient(BaseClient)
│   ├── config.py / factory.py    # 配置与工厂
│   └── utils.py                  # 文档准备 / distance→score / 名称清洗
├── qdrant/                       # 备选后端 Qdrant 实现
├── embeddings/                   # 嵌入 provider 工厂 + 十几个 provider（D40）
│   ├── factory.py
│   └── providers/{openai,cohere,ollama,...}
├── config/                       # 全局 RAG 配置（选后端、可选依赖）
└── factory.py                    # create_client：按 config 造出对应后端 client

核心协议 BaseClient（rag/core/base_client.py:66），方法都用 Unpack[TypedDict] 收关键字参数：

# rag/core/base_client.py:66（节选）
@runtime_checkable
class BaseClient(Protocol):
    client: Any                           # 底层真实的向量库 client
    embedding_function: EmbeddingFunction # 文字→向量

    @classmethod
    def __get_pydantic_core_schema__(cls, _src, _handler) -> CoreSchema:
        return core_schema.any_schema()   # ★让 Protocol 能塞进 pydantic 模型字段

    @abstractmethod
    def create_collection(self, **kwargs: Unpack[BaseCollectionParams]) -> None: ...
    @abstractmethod
    def get_or_create_collection(self, **kwargs: Unpack[BaseCollectionParams]) -> Any: ...
    @abstractmethod
    def add_documents(self, **kwargs: Unpack[BaseCollectionAddParams]) -> None: ...
    @abstractmethod
    def search(self, **kwargs: Unpack[BaseCollectionSearchParams]) -> list[SearchResult]: ...
    @abstractmethod
    def delete_collection(self, **kwargs: Unpack[BaseCollectionParams]) -> None: ...
    @abstractmethod
    def reset(self) -> None: ...
    # 每个都有 a* 异步版：acreate_collection / aadd_documents / asearch ...

入库前，文档要被"准备"成向量库能吃的三元组，关键是生成稳定的 doc_id（rag/chromadb/utils.py:56）：

# rag/chromadb/utils.py:72（节选）
for doc in documents:
    if "doc_id" in doc:
        doc_id = str(doc["doc_id"])                      # ① 显式给了就用
    else:
        metadata = doc.get("metadata")
        if metadata and isinstance(metadata, dict) and "doc_id" in metadata:
            doc_id = str(metadata["doc_id"])             # ② metadata 里有也用
        else:
            content_for_hash = doc["content"]
            if metadata:
                metadata_str = json.dumps(metadata, sort_keys=True)   # 稳定序列化
                content_for_hash = f"{content_for_hash}|{metadata_str}"
            doc_id = hashlib.sha256(content_for_hash.encode()).hexdigest()  # ③ 内容哈希

ChromaDB 的入库实现（rag/chromadb/client.py:309）：

# rag/chromadb/client.py:309（节选）
def add_documents(self, **kwargs: Unpack[BaseCollectionAddParams]) -> None:
    if not _is_sync_client(self.client):                 # ★类型守卫（L07）
        raise TypeError("Synchronous method add_documents() requires a ClientAPI. "
                        "Use aadd_documents() for AsyncClientAPI.")
    collection_name = kwargs["collection_name"]
    documents = kwargs["documents"]
    batch_size = kwargs.get("batch_size", self.default_batch_size)   # 默认 100
    if not documents:
        raise ValueError("Documents list cannot be empty")

    with self._locked():                                 # 跨进程锁
        collection = self.client.get_or_create_collection(
            name=_sanitize_collection_name(collection_name),
            embedding_function=self.embedding_function)
        prepared = _prepare_documents_for_chromadb(documents)   # 生成 ids/texts/metadatas
        for i in range(0, len(prepared.ids), batch_size):       # ★分批
            batch_ids, batch_texts, batch_metadatas = _create_batch_slice(
                prepared=prepared, start_index=i, batch_size=batch_size)
            collection.upsert(ids=batch_ids, documents=batch_texts,   # ★upsert 而非 add
                              metadatas=batch_metadatas)

搜索先查 collection，拿到的是"距离"，再翻译成"分数"（rag/chromadb/client.py:410）：

# rag/chromadb/client.py:410（节选）
def search(self, **kwargs) -> list[SearchResult]:
    if "limit" not in kwargs: kwargs["limit"] = self.default_limit           # 默认 5
    if "score_threshold" not in kwargs: kwargs["score_threshold"] = self.default_score_threshold  # 0.6
    params = _extract_search_params(kwargs)
    collection = self.client.get_or_create_collection(
        name=_sanitize_collection_name(params.collection_name),
        embedding_function=self.embedding_function)
    where = params.where if params.where is not None else params.metadata_filter
    results = collection.query(query_texts=[params.query], n_results=params.limit,
                               where=where, ...)          # chromadb 返回 distances
    return _process_query_results(collection=collection, ...)

翻译公式在 _convert_distance_to_score（rag/chromadb/utils.py:167）：

# rag/chromadb/utils.py:167
def _convert_distance_to_score(distance, distance_metric) -> float:
    """Convert ChromaDB distance to similarity score. [0,1], 1=most similar."""
    if distance_metric == "cosine":
        score = 1.0 - 0.5 * distance          # 余弦距离 [0,2] → 分 [0,1]
        return max(0.0, min(1.0, score))
    if distance_metric == "l2":
        score = 1.0 / (1.0 + distance)        # L2 距离 [0,∞) → 分 (0,1]
        return max(0.0, min(1.0, score))
    raise ValueError(f"Unsupported distance metric: {distance_metric}")

ChromaDB 有同步 ClientAPI 和异步 AsyncClientAPI 两种，ChromaDBClient 都支持，但每个方法开头都守卫（rag/chromadb/client.py:328）：

# rag/chromadb/client.py:328（同步方法开头）
if not _is_sync_client(self.client):
    raise TypeError("Synchronous method add_documents() requires a ClientAPI. "
                    "Use aadd_documents() for AsyncClientAPI.")
# rag/chromadb/client.py:379（异步方法开头）
if not _is_async_client(self.client):
    raise TypeError("Asynchronous method aadd_documents() requires an AsyncClientAPI. "
                    "Use add_documents() for ClientAPI.")

并发保护用跨进程锁（rag/chromadb/client.py:78）：

# rag/chromadb/client.py:78
def _locked(self) -> AbstractContextManager[None]:
    return store_lock(self._lock_name) if self._lock_name else nullcontext()

@asynccontextmanager
async def _alocked(self):                     # 异步版：在 executor 里拿/放同步锁
    if not self._lock_name: yield; return
    lock_cm = store_lock(self._lock_name)
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, lock_cm.__enter__)
    try: yield
    finally: await loop.run_in_executor(None, lock_cm.__exit__, None, None, None)