核心是一张 PROVIDER_PATHS 表 + build_embedder（rag/embeddings/factory.py:90）：

# rag/embeddings/factory.py:90（节选）
PROVIDER_PATHS = {
    "openai":  "crewai.rag.embeddings.providers.openai.openai_provider.OpenAIProvider",
    "cohere":  "crewai.rag.embeddings.providers.cohere.cohere_provider.CohereProvider",
    "ollama":  "crewai.rag.embeddings.providers.ollama.ollama_provider.OllamaProvider",
    "google":  "crewai.rag.embeddings.providers.google.generative_ai.GenerativeAiProvider",
    "huggingface": "...huggingface_provider.HuggingFaceProvider",
    "voyageai": "...voyageai_provider.VoyageAIProvider", ...    # 共 ~18 种
}

# rag/embeddings/factory.py:223（节选）
def build_embedder_from_dict(spec):
    provider_name = spec["provider"]
    if provider_name not in PROVIDER_PATHS:
        raise ValueError(f"Unknown provider: {provider_name}. Available: {list(PROVIDER_PATHS.keys())}")
    provider_path = PROVIDER_PATHS[provider_name]
    provider_class = import_and_validate_definition(provider_path)   # ★动态 import
    provider_config = spec.get("config", {})
    provider = provider_class(**provider_config)          # 实例化 provider
    return build_embedder_from_provider(provider)

# rag/embeddings/factory.py:113
def build_embedder_from_provider(provider):
    return provider.embedding_callable(                   # 造出真正的嵌入函数
        **provider.model_dump(exclude={"embedding_callable"}))

看最常用的 OpenAIProvider（rag/embeddings/providers/openai/openai_provider.py:13）：

# rag/embeddings/providers/openai/openai_provider.py:13（节选）
class OpenAIProvider(BaseEmbeddingsProvider[OpenAIEmbeddingFunction]):
    @model_validator(mode="before")
    @classmethod
    def _normalize_model_alias(cls, data):
        if isinstance(data, dict) and "model" in data and "model_name" not in data:
            data = data.copy(); data["model_name"] = data["model"]   # model→model_name 兼容
        return data

    embedding_callable: type[OpenAIEmbeddingFunction] = Field(default=OpenAIEmbeddingFunction)
    api_key: str | None = Field(default=None,
        validation_alias=AliasChoices("EMBEDDINGS_OPENAI_API_KEY", "OPENAI_API_KEY"))  # ★从环境变量读
    model_name: str = Field(default="text-embedding-3-large",                          # ★默认大模型
        validation_alias=AliasChoices("EMBEDDINGS_OPENAI_MODEL_NAME", "model_name"))
    dimensions: int | None = Field(default=None, ...)

转到存储侧。LanceDBStorage 初始化时不知道向量多少维，靠自检（memory/storage/lancedb_storage.py:97）：

# memory/storage/lancedb_storage.py:97（节选）
try:
    self._table = self._db.open_table(self._table_name)         # 表已存在
    self._vector_dim = self._infer_dim_from_table(self._table)  # ★从表结构读维度
    with store_lock(self._lock_name):
        self._ensure_scope_index()                              # 确保 scope 索引在
    self._compact_if_needed()                                   # 顺手整理碎片
except Exception:
    self._table = None
    self._vector_dim = vector_dim or 0     # 0 = 还不知道，等第一次 save 再定

# memory/storage/lancedb_storage.py:116
@staticmethod
def _infer_dim_from_table(table) -> int:
    schema = table.schema
    for field in schema:
        if field.name == "vector":
            return int(field.type.list_size)   # 向量列的固定长度就是维度
    return DEFAULT_VECTOR_DIM   # 3072

save 存前严格校验维度（memory/storage/lancedb_storage.py:289）：

# memory/storage/lancedb_storage.py:289（节选）
def save(self, records: list[MemoryRecord]) -> None:
    if not records: return
    dim = None
    for r in records:
        if r.embedding and len(r.embedding) > 0:
            if dim is None: dim = len(r.embedding)
            elif len(r.embedding) != dim:
                raise EmbeddingDimensionMismatchError(dim, len(r.embedding))   # 批内不一致
    is_new_table = self._table is None
    if not is_new_table and dim and self._vector_dim and dim != self._vector_dim:
        raise EmbeddingDimensionMismatchError(self._vector_dim, dim)           # 与既有表不一致
    with store_lock(self._lock_name):
        self._ensure_table(vector_dim=dim)      # 首存时用真实维度建表
        rows = [self._record_to_row(rec) for rec in records]
        for row in rows:
            if row["vector"] is None or len(row["vector"]) != self._vector_dim:
                row["vector"] = [0.0] * self._vector_dim   # 无向量→零向量占位
        self._do_write("add", rows)

LanceDB 用乐观并发，高并发写会撞"提交冲突"，_do_write 自动重试（memory/storage/lancedb_storage.py:128）：

# memory/storage/lancedb_storage.py:128（节选）
_MAX_RETRIES = 5
_RETRY_BASE_DELAY = 0.2   # 秒，每次翻倍：0.2+0.4+0.8+1.6+3.2 ≈ 6.2s

def _do_write(self, op, *args, **kwargs):
    delay = _RETRY_BASE_DELAY
    for attempt in range(_MAX_RETRIES + 1):
        try:
            return getattr(self._table, op)(*args, **kwargs)   # add/delete/update
        except OSError as e:
            if "Commit conflict" not in str(e) or attempt >= _MAX_RETRIES:
                raise                                          # 不是冲突/次数用尽→抛
            try:
                self._table = self._db.open_table(self._table_name)  # ★重新打开（拿最新版本）
            except Exception:
                pass
            time.sleep(delay)                                  # 退避等待
            delay *= 2                                         # ★指数退避

每次 save 产生一个碎片文件，攒多了拖慢查询，所以定期后台整理（memory/storage/lancedb_storage.py:314）：

# memory/storage/lancedb_storage.py:314（save 尾部）
self._save_count += 1
if self._compact_every > 0 and self._save_count % self._compact_every == 0:  # 每 100 次
    self._compact_async()               # ★后台守护线程 table.optimize()

# lancedb_storage.py:183 —— scope 标量索引（加速前缀过滤）
def _ensure_scope_index(self):
    self._table.create_scalar_index("scope", index_type="BTREE", replace=False)

# lancedb_storage.py:403 —— 距离→相似分（和 D38 chromadb 同思路）
distance = row.get("_distance", 0.0)
score = 1.0 / (1.0 + float(distance)) if distance is not None else 1.0