问题背景

在使用 CrewAI 1.8.0 构建多智能体系统时，如果为 Agent 配置了知识源（Knowledge Sources），会遇到一个性能问题：每次调用 crew.kickoff() 时，所有知识源都会被重新嵌入到向量数据库中。

问题表现

# 典型的 Agent 配置
@agent
def my_agent(self) -> Agent:
    return Agent(
        config=self.agents_config["my_agent"],
        llm=self.llm,
        knowledge_sources=[
            JSONKnowledgeSource(file_paths=["data.json"])
        ]
    )

# 每次 kickoff 都会重新嵌入
crew = MyCrew().crew()
result = crew.kickoff(inputs={...})

性能影响：

  • 每次启动延迟增加 10-30 秒（取决于知识源大小）
  • 重复调用 Embedding API，产生额外费用
  • 浪费计算资源，特别是批处理场景

日志表现

即使第二次、第三次运行相同任务，日志中仍会反复出现：

INFO: 将 88 条数据逐条嵌入
INFO: 嵌入进度: 10/88
INFO: 嵌入进度: 20/88
...
INFO: 所有 88 条数据嵌入完成

问题根源分析

1. CrewAI 的知识源初始化机制

查看 CrewAI 源码可以发现，在 crew.kickoff() 执行时，会调用以下逻辑：

# crewai/crew.py (简化版)
class Crew:
    def kickoff(self, inputs):
        # 在 kickoff 时重新设置知识源
        for agent in self.agents:
            agent.set_knowledge(crew_embedder=self.embedder)

在 Agent.set_knowledge() 方法中：

# crewai/agent.py (简化版)
class Agent:
    def set_knowledge(self, crew_embedder):
        if self.knowledge_sources:
            # 重新创建 Knowledge 对象，触发重新嵌入
            self.knowledge = Knowledge(
                sources=self.knowledge_sources,
                embedder=crew_embedder,
                # 没有使用持久化的 collection_name
            )

关键问题：

  • knowledge_sources 参数传递的 KnowledgeSource 对象在每次 kickoff() 时都会被重新包装成新的 Knowledge 对象
  • 新的 Knowledge 对象会生成随机的 collection_name（如 knowledge_Agent_Name_12345）
  • ChromaDB 找不到已有的 collection，认为是新数据，触发重新嵌入

2. ChromaDB 存储结构

CrewAI 选用 ChromaDB 来充当向量数据库，它的存储结构如下：

knowledge_sources/
├── chroma.sqlite3                    # 元数据数据库
├── 00dce6e3-281b-4450-8f65-b3d8.../  # UUID 目录（实际向量数据）
├── 85d93e0c-3b79-4200-90ab-360b.../
└── ...

Collection 名称是存储在 chroma.sqlite3 里面的，而向量数据则存储在以 UUID 命名的目录中。要是 collection_name 每次都不一样，那就会去创建新的 collection。

———存储在 chroma.sqlite3 中，向量数据存储在 UUID 命名的目录中。如果 collection_name 每次都不同，就会

解决方案：使用 knowledge 参数替代 knowledge_sources

CrewAI 官方文档中提到：

“With the typical file structure provided by CrewAI, knowledge sources are embedded every time the kickoff is triggered… To resolve this, directly initialize the knowledge parameter instead of the knowledge_sources parameter.“

核心思路

直接创建 Knowledge 对象并指定固定的 collection_name，让 ChromaDB 能够找到已有的 collection，避免重复嵌入。

实现步骤

第 1 步：创建知识管理器模块

创建一个单例模块来管理 Knowledge 对象：

# tools/knowledge_manager.py
from crewai import Knowledge
from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource
from crewai.knowledge.storage.knowledge_storage import KnowledgeStorage
from pathlib import Path
import os

_KNOWLEDGE_OBJECTS = {}
_INITIALIZED = False

def _initialize_knowledge_objects():
    """初始化 Knowledge 对象（仅执行一次）"""
    global _KNOWLEDGE_OBJECTS, _INITIALIZED
    
    if _INITIALIZED:
        return
    
    # embedder 配置
    embedder_config = {
        "provider": "openai",
        "config": {
            "model": "text-embedding-3-small"
        }
    }
    
    project_root = Path(__file__).parent.parent
    original_cwd = Path.cwd()
    os.chdir(project_root)
    
    try:
        # 创建知识源
        data_source = JSONKnowledgeSource(
            file_paths=["knowledge/data.json"]
        )
        data_storage = KnowledgeStorage(
            embedder=embedder_config,
            collection_name="data_persistent"  # 固定名称
        )
        _KNOWLEDGE_OBJECTS['data'] = Knowledge(
            collection_name="data_persistent",
            sources=[data_source],
            embedder=embedder_config,
            storage=data_storage
        )
        
        _INITIALIZED = True
        print("Knowledge 对象初始化完成")
        
    finally:
        os.chdir(original_cwd)

def get_knowledge(name: str):
    """获取预初始化的 Knowledge 对象"""
    if not _INITIALIZED:
        _initialize_knowledge_objects()
    
    if name not in _KNOWLEDGE_OBJECTS:
        raise ValueError(f"未知的知识源: {name}")
    
    return _KNOWLEDGE_OBJECTS[name]

第 2 步：在 Crew 中使用 knowledge 参数

错误方式（会重复嵌入）：

from crewai import Agent, Crew
from crewai.project import CrewBase, agent, crew
from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource

@CrewBase
class MyCrew:
    @agent
    def my_agent(self) -> Agent:
        return Agent(
            config=self.agents_config["my_agent"],
            llm=self.llm,
            knowledge_sources=[JSONKnowledgeSource(file_paths=["data.json"])]
        )

正确方式（使用持久化）：

from crewai import Agent, Crew
from crewai.project import CrewBase, agent, crew
from tools.knowledge_manager import get_knowledge

@CrewBase
class MyCrew:
    data_knowledge = get_knowledge('data')
    
    @agent
    def my_agent(self) -> Agent:
        return Agent(
            config=self.agents_config["my_agent"],
            llm=self.llm,
            knowledge=self.data_knowledge
        )