跳转到内容

知识

CrewAI 中的知识是一套强大的系统,允许 AI agents 在执行任务时访问并利用外部信息源。 你可以把它理解为给你的 agents 配备一个在工作中随时可查阅的参考资料库。

CrewAI 提供了一个与提供方无关的 RAG 客户端抽象,用于向量存储。默认提供方是 ChromaDB,同时也支持 Qdrant。你可以使用配置工具来切换提供方。

当前支持:

  • ChromaDB(默认)
  • Qdrant
from crewai.rag.config.utils import set_rag_config, get_rag_client, clear_rag_config
# ChromaDB (default)
from crewai.rag.chromadb.config import ChromaDBConfig
set_rag_config(ChromaDBConfig())
chromadb_client = get_rag_client()
# Qdrant
from crewai.rag.qdrant.config import QdrantConfig
set_rag_config(QdrantConfig())
qdrant_client = get_rag_client()
# Example operations (same API for any provider)
client = qdrant_client # or chromadb_client
client.create_collection(collection_name="docs")
client.add_documents(
collection_name="docs",
documents=[{"id": "1", "content": "CrewAI enables collaborative AI agents."}],
)
results = client.search(collection_name="docs", query="collaborative agents", limit=3)
clear_rag_config() # optional reset

这个 RAG 客户端与 Knowledge 的内置存储是分开的。当你需要直接控制向量存储或自定义检索管线时,请使用它。

from crewai import Agent, Task, Crew, Process, LLM
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Create a knowledge source
content = "Users name is John. He is 30 years old and lives in San Francisco."
string_source = StringKnowledgeSource(content=content)
# Create an LLM with a temperature of 0 to ensure deterministic outputs
llm = LLM(model="gpt-4o-mini", temperature=0)
# Create an agent with the knowledge store
agent = Agent(
role="About User",
goal="You know everything about the user.",
backstory="You are a master at understanding people and their preferences.",
verbose=True,
allow_delegation=False,
llm=llm,
)
task = Task(
description="Answer the following questions about the user: {question}",
expected_output="An answer to the question.",
agent=agent,
)
crew = Crew(
agents=[agent],
tasks=[task],
verbose=True,
process=Process.sequential,
knowledge_sources=[string_source], # Enable knowledge by adding the sources here
)
result = crew.kickoff(inputs={"question": "What city does John live in and how old is he?"})
from crewai import LLM, Agent, Crew, Process, Task
from crewai.knowledge.source.crew_docling_source import CrewDoclingSource
# Create a knowledge source from web content
content_source = CrewDoclingSource(
file_paths=[
"https://lilianweng.github.io/posts/2024-11-28-reward-hacking",
"https://lilianweng.github.io/posts/2024-07-07-hallucination",
],
)
# Create an LLM with a temperature of 0 to ensure deterministic outputs
llm = LLM(model="gpt-4o-mini", temperature=0)
# Create an agent with the knowledge store
agent = Agent(
role="About papers",
goal="You know everything about the papers.",
backstory="You are a master at understanding papers and their content.",
verbose=True,
allow_delegation=False,
llm=llm,
)
task = Task(
description="Answer the following questions about the papers: {question}",
expected_output="An answer to the question.",
agent=agent,
)
crew = Crew(
agents=[agent],
tasks=[task],
verbose=True,
process=Process.sequential,
knowledge_sources=[content_source],
)
result = crew.kickoff(
inputs={"question": "What is the reward hacking paper about? Be sure to provide sources."}
)

CrewAI 开箱即支持多种知识源:

文本源

  • 原始字符串
  • 文本文件(.txt)
  • PDF 文档

结构化数据

  • CSV 文件
  • Excel 电子表格
  • JSON 文档
from crewai.knowledge.source.text_file_knowledge_source import TextFileKnowledgeSource
text_source = TextFileKnowledgeSource(
file_paths=["document.txt", "another.txt"]
)
from crewai.knowledge.source.pdf_knowledge_source import PDFKnowledgeSource
pdf_source = PDFKnowledgeSource(
file_paths=["document.pdf", "another.pdf"]
)
from crewai.knowledge.source.csv_knowledge_source import CSVKnowledgeSource
csv_source = CSVKnowledgeSource(
file_paths=["data.csv"]
)
from crewai.knowledge.source.excel_knowledge_source import ExcelKnowledgeSource
excel_source = ExcelKnowledgeSource(
file_paths=["spreadsheet.xlsx"]
)
from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource
json_source = JSONKnowledgeSource(
file_paths=["data.json"]
)

当你使用 knowledge 时,实际会发生以下事情:

from crewai import Agent, Task, Crew
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Agent with its own knowledge - NO crew knowledge needed
specialist_knowledge = StringKnowledgeSource(
content="Specialized technical information for this agent only"
)
specialist_agent = Agent(
role="Technical Specialist",
goal="Provide technical expertise",
backstory="Expert in specialized technical domains",
knowledge_sources=[specialist_knowledge] # Agent-specific knowledge
)
task = Task(
description="Answer technical questions",
agent=specialist_agent,
expected_output="Technical answer"
)
# No crew-level knowledge required
crew = Crew(
agents=[specialist_agent],
tasks=[task]
)
result = crew.kickoff() # Agent knowledge works independently

当你调用 crew.kickoff() 时,实际执行顺序如下:

# During kickoff
for agent in self.agents:
agent.crew = self # Agent gets reference to crew
agent.set_knowledge(crew_embedder=self.embedder) # Agent knowledge initialized
agent.create_agent_executor()

每个 knowledge 层级都使用独立的存储 collection:

# Agent knowledge storage
agent_collection_name = agent.role # e.g., "Technical Specialist"
# Crew knowledge storage
crew_collection_name = "crew"
# Both stored in same ChromaDB instance but different collections
# Path: ~/.local/share/CrewAI/{project}/knowledge/
# ├── crew/ # Crew knowledge collection
# ├── Technical Specialist/ # Agent knowledge collection
# └── Another Agent Role/ # Another agent's collection
from crewai import Agent, Task, Crew
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Agent-specific knowledge
agent_knowledge = StringKnowledgeSource(
content="Agent-specific information that only this agent needs"
)
agent = Agent(
role="Specialist",
goal="Use specialized knowledge",
backstory="Expert with specific knowledge",
knowledge_sources=[agent_knowledge],
embedder={ # Agent can have its own embedder
"provider": "openai",
"config": {"model": "text-embedding-3-small"}
}
)
task = Task(
description="Answer using your specialized knowledge",
agent=agent,
expected_output="Answer based on agent knowledge"
)
# No crew knowledge needed
crew = Crew(agents=[agent], tasks=[task])
result = crew.kickoff() # Works perfectly

示例 2:同时使用 Agent 和 Crew 知识

Section titled “示例 2:同时使用 Agent 和 Crew 知识”
# Crew-wide knowledge (shared by all agents)
crew_knowledge = StringKnowledgeSource(
content="Company policies and general information for all agents"
)
# Agent-specific knowledge
specialist_knowledge = StringKnowledgeSource(
content="Technical specifications only the specialist needs"
)
specialist = Agent(
role="Technical Specialist",
goal="Provide technical expertise",
backstory="Technical expert",
knowledge_sources=[specialist_knowledge] # Agent-specific
)
generalist = Agent(
role="General Assistant",
goal="Provide general assistance",
backstory="General helper"
# No agent-specific knowledge
)
crew = Crew(
agents=[specialist, generalist],
tasks=[...],
knowledge_sources=[crew_knowledge] # Crew-wide knowledge
)
# Result:
# - specialist gets: crew_knowledge + specialist_knowledge
# - generalist gets: crew_knowledge only

示例 3:多个 agents 使用不同知识

Section titled “示例 3:多个 agents 使用不同知识”
# Different knowledge for different agents
sales_knowledge = StringKnowledgeSource(content="Sales procedures and pricing")
tech_knowledge = StringKnowledgeSource(content="Technical documentation")
support_knowledge = StringKnowledgeSource(content="Support procedures")
sales_agent = Agent(
role="Sales Representative",
knowledge_sources=[sales_knowledge],
embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}}
)
tech_agent = Agent(
role="Technical Expert",
knowledge_sources=[tech_knowledge],
embedder={"provider": "ollama", "config": {"model": "mxbai-embed-large"}}
)
support_agent = Agent(
role="Support Specialist",
knowledge_sources=[support_knowledge]
# Will use crew embedder as fallback
)
crew = Crew(
agents=[sales_agent, tech_agent, support_agent],
tasks=[...],
embedder={ # Fallback embedder for agents without their own
"provider": "google-generativeai",
"config": {"model_name": "gemini-embedding-001"}
}
)
# Each agent gets only their specific knowledge
# Each can use different embedding providers

你可以为 crew 或 agent 配置 knowledge。

from crewai.knowledge.knowledge_config import KnowledgeConfig
knowledge_config = KnowledgeConfig(results_limit=10, score_threshold=0.5)
agent = Agent(
...
knowledge_config=knowledge_config
)
sources List[BaseKnowledgeSource] required

提供内容以便存储和查询的 knowledge sources 列表。可以包含 PDF、CSV、Excel、JSON、文本文件,或字符串内容。

collection_name str

用于存储 knowledge 的 collection 名称。用于标识不同的 knowledge 集合。若未提供,默认值为 “knowledge”。

storage Optional[KnowledgeStorage]

用于管理 knowledge 存储与检索方式的自定义存储配置。如果未提供,会创建默认 storage。

默认情况下,CrewAI 使用与 memory 相同的存储系统,将 knowledge 存放在平台特定目录中:

macOS:

~/Library/Application Support/CrewAI/{project_name}/
└── knowledge/ # Knowledge ChromaDB files
├── chroma.sqlite3 # ChromaDB metadata
├── {collection_id}/ # Vector embeddings
└── knowledge_{collection}/ # Named collections

Linux:

~/.local/share/CrewAI/{project_name}/
└── knowledge/
├── chroma.sqlite3
├── {collection_id}/
└── knowledge_{collection}/

Windows:

C:\Users\{username}\AppData\Local\CrewAI\{project_name}\
└── knowledge\
├── chroma.sqlite3
├── {collection_id}\
└── knowledge_{collection}\

要精确查看 CrewAI 正在将 knowledge 文件存到哪里:

from crewai.utilities.paths import db_storage_path
import os
# Get the knowledge storage path
knowledge_path = os.path.join(db_storage_path(), "knowledge")
print(f"Knowledge storage location: {knowledge_path}")
# List knowledge collections and files
if os.path.exists(knowledge_path):
print("\nKnowledge storage contents:")
for item in os.listdir(knowledge_path):
item_path = os.path.join(knowledge_path, item)
if os.path.isdir(item_path):
print(f"📁 Collection: {item}/")
# Show collection contents
try:
for subitem in os.listdir(item_path):
print(f" └── {subitem}")
except PermissionError:
print(f" └── (permission denied)")
else:
print(f"📄 {item}")
else:
print("No knowledge storage found yet.")
import os
from crewai import Crew
# Set custom storage location for all CrewAI data
os.environ["CREWAI_STORAGE_DIR"] = "./my_project_storage"
# All knowledge will now be stored in ./my_project_storage/knowledge/
crew = Crew(
agents=[...],
tasks=[...],
knowledge_sources=[...]
)
from crewai.knowledge.storage.knowledge_storage import KnowledgeStorage
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Create custom storage with specific embedder
custom_storage = KnowledgeStorage(
embedder={
"provider": "ollama",
"config": {"model": "mxbai-embed-large"}
},
collection_name="my_custom_knowledge"
)
# Use with knowledge sources
knowledge_source = StringKnowledgeSource(
content="Your knowledge content here"
)
knowledge_source.storage = custom_storage
import os
from pathlib import Path
# Store knowledge in project directory
project_root = Path(__file__).parent
knowledge_dir = project_root / "knowledge_storage"
os.environ["CREWAI_STORAGE_DIR"] = str(knowledge_dir)
# Now all knowledge will be stored in your project directory
from crewai import Agent, Crew, LLM
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# When using Claude as your LLM...
agent = Agent(
role="Researcher",
goal="Research topics",
backstory="Expert researcher",
llm=LLM(provider="anthropic", model="claude-3-sonnet") # Using Claude
)
# CrewAI will still use OpenAI embeddings by default for knowledge
# This ensures consistency but may not match your LLM provider preference
knowledge_source = StringKnowledgeSource(content="Research data...")
crew = Crew(
agents=[agent],
tasks=[...],
knowledge_sources=[knowledge_source]
# Default: Uses OpenAI embeddings even with Claude LLM
)
# Option 1: Use Voyage AI (recommended by Anthropic for Claude users)
crew = Crew(
agents=[agent],
tasks=[...],
knowledge_sources=[knowledge_source],
embedder={
"provider": "voyageai", # Recommended for Claude users
"config": {
"api_key": "your-voyage-api-key",
"model": "voyage-3" # or "voyage-3-large" for best quality
}
}
)
# Option 2: Use local embeddings (no external API calls)
crew = Crew(
agents=[agent],
tasks=[...],
knowledge_sources=[knowledge_source],
embedder={
"provider": "ollama",
"config": {
"model": "mxbai-embed-large",
"url": "http://localhost:11434/api/embeddings"
}
}
)
# Option 3: Agent-level embedding customization
agent = Agent(
role="Researcher",
goal="Research topics",
backstory="Expert researcher",
knowledge_sources=[knowledge_source],
embedder={
"provider": "google-generativeai",
"config": {
"model_name": "gemini-embedding-001",
"api_key": "your-google-key"
}
}
)

使用 Azure OpenAI embeddings 时:

  1. 请先确保你已在 Azure 平台中部署 embedding 模型
  2. 然后使用以下配置:
agent = Agent(
role="Researcher",
goal="Research topics",
backstory="Expert researcher",
knowledge_sources=[knowledge_source],
embedder={
"provider": "azure",
"config": {
"api_key": "your-azure-api-key",
"model": "text-embedding-ada-002", # change to the model you are using and is deployed in Azure
"api_base": "https://your-azure-endpoint.openai.azure.com/",
"api_version": "2024-02-01"
}
}
)

CrewAI 实现了智能的查询重写机制,以优化 knowledge 检索。当 agent 需要搜索 knowledge sources 时,原始任务提示会自动转换为更有效的搜索查询。

  1. 当 agent 执行一个可访问 knowledge sources 的任务时,会触发 _get_knowledge_search_query 方法
  2. agent 的 LLM 会将原始任务提示转换为优化后的搜索查询
  3. 这个优化后的查询随后用于从 knowledge sources 中检索相关信息

提升检索准确性

通过聚焦关键概念并移除无关内容,查询重写有助于检索到更相关的信息。

上下文感知

重写后的查询会更具体,也更能理解上下文,适合向量数据库检索。

# Original task prompt
task_prompt = "Answer the following questions about the user's favorite movies: What movie did John watch last week? Format your answer in JSON."
# Behind the scenes, this might be rewritten as:
rewritten_query = "What movies did John watch last week?"

重写后的查询更聚焦于核心信息需求,并移除了与输出格式相关的无关指令。

CrewAI 会在 knowledge 检索过程中发出事件,你可以使用事件系统监听这些事件。这些事件可让你监控、调试并分析 knowledge 是如何被检索和使用的。

  • KnowledgeRetrievalStartedEvent:当 agent 开始从 sources 中检索 knowledge 时触发
  • KnowledgeRetrievalCompletedEvent:当 knowledge 检索完成时触发,包括所用查询和检索到的内容
  • KnowledgeQueryStartedEvent:当对 knowledge sources 的查询开始时触发
  • KnowledgeQueryCompletedEvent:当查询成功完成时触发
  • KnowledgeQueryFailedEvent:当对 knowledge sources 的查询失败时触发
  • KnowledgeSearchQueryFailedEvent:当搜索查询失败时触发
from crewai.events import (
KnowledgeRetrievalStartedEvent,
KnowledgeRetrievalCompletedEvent,
BaseEventListener,
)
class KnowledgeMonitorListener(BaseEventListener):
def setup_listeners(self, crewai_event_bus):
@crewai_event_bus.on(KnowledgeRetrievalStartedEvent)
def on_knowledge_retrieval_started(source, event):
print(f"Agent '{event.agent.role}' started retrieving knowledge")
@crewai_event_bus.on(KnowledgeRetrievalCompletedEvent)
def on_knowledge_retrieval_completed(source, event):
print(f"Agent '{event.agent.role}' completed knowledge retrieval")
print(f"Query: {event.query}")
print(f"Retrieved {len(event.retrieved_knowledge)} knowledge chunks")
# Create an instance of your listener
knowledge_monitor = KnowledgeMonitorListener()

有关事件使用的更多信息,请参阅 Event Listeners 文档。

CrewAI 允许你通过扩展 BaseKnowledgeSource 类来为任何类型的数据创建自定义 knowledge source。下面我们创建一个实际示例,用于获取并处理太空新闻文章。

Code
from crewai import Agent, Task, Crew, Process, LLM
from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
import requests
from datetime import datetime
from typing import Dict, Any
from pydantic import BaseModel, Field
class SpaceNewsKnowledgeSource(BaseKnowledgeSource):
"""Knowledge source that fetches data from Space News API."""
api_endpoint: str = Field(description="API endpoint URL")
limit: int = Field(default=10, description="Number of articles to fetch")
def load_content(self) -> Dict[Any, str]:
"""Fetch and format space news articles."""
try:
response = requests.get(
f"{self.api_endpoint}?limit={self.limit}"
)
response.raise_for_status()
data = response.json()
articles = data.get('results', [])
formatted_data = self.validate_content(articles)
return {self.api_endpoint: formatted_data}
except Exception as e:
raise ValueError(f"Failed to fetch space news: {str(e)}")
def validate_content(self, articles: list) -> str:
"""Format articles into readable text."""
formatted = "Space News Articles:\n\n"
for article in articles:
formatted += f"""
Title: {article['title']}
Published: {article['published_at']}
Summary: {article['summary']}
News Site: {article['news_site']}
URL: {article['url']}
-------------------"""
return formatted
def add(self) -> None:
"""Process and store the articles."""
content = self.load_content()
for _, text in content.items():
chunks = self._chunk_text(text)
self.chunks.extend(chunks)
self._save_documents()
# Create knowledge source
recent_news = SpaceNewsKnowledgeSource(
api_endpoint="https://api.spaceflightnewsapi.net/v4/articles",
limit=10,
)
# Create specialized agent
space_analyst = Agent(
role="Space News Analyst",
goal="Answer questions about space news accurately and comprehensively",
backstory="""You are a space industry analyst with expertise in space exploration,
satellite technology, and space industry trends. You excel at answering questions
about space news and providing detailed, accurate information.""",
knowledge_sources=[recent_news],
llm=LLM(model="gpt-4", temperature=0.0)
)
# Create task that handles user questions
analysis_task = Task(
description="Answer this question about space news: {user_question}",
expected_output="A detailed answer based on the recent space news articles",
agent=space_analyst
)
# Create and run the crew
crew = Crew(
agents=[space_analyst],
tasks=[analysis_task],
verbose=True,
process=Process.sequential
)
# Example usage
result = crew.kickoff(
inputs={"user_question": "What are the latest developments in space exploration?"}
)
# Agent: Space News Analyst
## Task: Answer this question about space news: What are the latest developments in space exploration?
# Agent: Space News Analyst
## Final Answer:
The latest developments in space exploration, based on recent space news articles, include the following:
1. SpaceX has received the final regulatory approvals to proceed with the second integrated Starship/Super Heavy launch, scheduled for as soon as the morning of Nov. 17, 2023. This is a significant step in SpaceX's ambitious plans for space exploration and colonization. [Source: SpaceNews](https://spacenews.com/starship-cleared-for-nov-17-launch/)
2. SpaceX has also informed the US Federal Communications Commission (FCC) that it plans to begin launching its first next-generation Starlink Gen2 satellites. This represents a major upgrade to the Starlink satellite internet service, which aims to provide high-speed internet access worldwide. [Source: Teslarati](https://www.teslarati.com/spacex-first-starlink-gen2-satellite-launch-2022/)
3. AI startup Synthetaic has raised $15 million in Series B funding. The company uses artificial intelligence to analyze data from space and air sensors, which could have significant applications in space exploration and satellite technology. [Source: SpaceNews](https://spacenews.com/ai-startup-synthetaic-raises-15-million-in-series-b-funding/)
4. The Space Force has formally established a unit within the U.S. Indo-Pacific Command, marking a permanent presence in the Indo-Pacific region. This could have significant implications for space security and geopolitics. [Source: SpaceNews](https://spacenews.com/space-force-establishes-permanent-presence-in-indo-pacific-region/)
5. Slingshot Aerospace, a space tracking and data analytics company, is expanding its network of ground-based optical telescopes to increase coverage of low Earth orbit. This could improve our ability to track and analyze objects in low Earth orbit, including satellites and space debris. [Source: SpaceNews](https://spacenews.com/slingshots-space-tracking-network-to-extend-coverage-of-low-earth-orbit/)
6. The National Natural Science Foundation of China has outlined a five-year project for researchers to study the assembly of ultra-large spacecraft. This could lead to significant advancements in spacecraft technology and space exploration capabilities. [Source: SpaceNews](https://spacenews.com/china-researching-challenges-of-kilometer-scale-ultra-large-spacecraft/)
7. The Center for AEroSpace Autonomy Research (CAESAR) at Stanford University is focusing on spacecraft autonomy. The center held a kickoff event on May 22, 2024, to highlight the industry, academia, and government collaboration it seeks to foster. This could lead to significant advancements in autonomous spacecraft technology. [Source: SpaceNews](https://spacenews.com/stanford-center-focuses-on-spacecraft-autonomy/)
from crewai import Agent, Crew, Task
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
knowledge_source = StringKnowledgeSource(content="Test knowledge")
agent = Agent(
role="Test Agent",
goal="Test knowledge",
backstory="Testing",
knowledge_sources=[knowledge_source]
)
crew = Crew(agents=[agent], tasks=[Task(...)])
# Before kickoff - knowledge not initialized
print(f"Before kickoff - Agent knowledge: {getattr(agent, 'knowledge', None)}")
crew.kickoff()
# After kickoff - knowledge initialized
print(f"After kickoff - Agent knowledge: {agent.knowledge}")
print(f"Agent knowledge collection: {agent.knowledge.storage.collection_name}")
print(f"Number of sources: {len(agent.knowledge.sources)}")
import os
from crewai.utilities.paths import db_storage_path
# Check storage structure
storage_path = db_storage_path()
knowledge_path = os.path.join(storage_path, "knowledge")
if os.path.exists(knowledge_path):
print("Knowledge collections found:")
for collection in os.listdir(knowledge_path):
collection_path = os.path.join(knowledge_path, collection)
if os.path.isdir(collection_path):
print(f" - {collection}/")
# Show collection contents
for item in os.listdir(collection_path):
print(f" └── {item}")
# Test agent knowledge retrieval
if hasattr(agent, 'knowledge') and agent.knowledge:
test_query = ["test query"]
results = agent.knowledge.query(test_query)
print(f"Agent knowledge results: {len(results)} documents found")
# Test crew knowledge retrieval (if exists)
if hasattr(crew, 'knowledge') and crew.knowledge:
crew_results = crew.query_knowledge(test_query)
print(f"Crew knowledge results: {len(crew_results)} documents found")
import chromadb
from crewai.utilities.paths import db_storage_path
import os
# Connect to CrewAI's knowledge ChromaDB
knowledge_path = os.path.join(db_storage_path(), "knowledge")
if os.path.exists(knowledge_path):
client = chromadb.PersistentClient(path=knowledge_path)
collections = client.list_collections()
print("Knowledge Collections:")
for collection in collections:
print(f" - {collection.name}: {collection.count()} documents")
# Sample a few documents to verify content
if collection.count() > 0:
sample = collection.peek(limit=2)
print(f" Sample content: {sample['documents'][0][:100]}...")
else:
print("No knowledge storage found")
from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Create a test knowledge source
test_source = StringKnowledgeSource(
content="Test knowledge content for debugging",
chunk_size=100, # Small chunks for testing
chunk_overlap=20
)
# Check chunking behavior
print(f"Original content length: {len(test_source.content)}")
print(f"Chunk size: {test_source.chunk_size}")
print(f"Chunk overlap: {test_source.chunk_overlap}")
# Process and inspect chunks
test_source.add()
print(f"Number of chunks created: {len(test_source.chunks)}")
for i, chunk in enumerate(test_source.chunks[:3]): # Show first 3 chunks
print(f"Chunk {i+1}: {chunk[:50]}...")

“File not found” 错误:

# Ensure files are in the correct location
from crewai.utilities.constants import KNOWLEDGE_DIRECTORY
import os
knowledge_dir = KNOWLEDGE_DIRECTORY # Usually "knowledge"
file_path = os.path.join(knowledge_dir, "your_file.pdf")
if not os.path.exists(file_path):
print(f"File not found: {file_path}")
print(f"Current working directory: {os.getcwd()}")
print(f"Expected knowledge directory: {os.path.abspath(knowledge_dir)}")

“Embedding dimension mismatch” 错误:

# This happens when switching embedding providers
# Reset knowledge storage to clear old embeddings
crew.reset_memories(command_type='knowledge')
# Or use consistent embedding providers
crew = Crew(
agents=[...],
tasks=[...],
knowledge_sources=[...],
embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}}
)

“ChromaDB permission denied” 错误:

Terminal window
# Fix storage permissions
chmod -R 755 ~/.local/share/CrewAI/

Knowledge 在运行之间没有持久化:

# Verify storage location consistency
import os
from crewai.utilities.paths import db_storage_path
print("CREWAI_STORAGE_DIR:", os.getenv("CREWAI_STORAGE_DIR"))
print("Computed storage path:", db_storage_path())
print("Knowledge path:", os.path.join(db_storage_path(), "knowledge"))
# Reset only agent-specific knowledge
crew.reset_memories(command_type='agent_knowledge')
# Reset both crew and agent knowledge
crew.reset_memories(command_type='knowledge')
# CLI commands
# crewai reset-memories --agent-knowledge # Agent knowledge only
# crewai reset-memories --knowledge # All knowledge

如果你需要清除 CrewAI 中存储的 knowledge,可以使用带有 --knowledge 选项的 crewai reset-memories 命令。

Terminal window
crewai reset-memories --knowledge

当你更新了 knowledge sources,并希望确保 agents 使用最新信息时,这非常有用。

内容组织
  • 为你的内容类型选择合适的 chunk 大小
  • 考虑通过重叠内容保留上下文
  • 将相关信息组织到不同的 knowledge sources 中
性能建议
  • 根据内容复杂度调整 chunk 大小
  • 配置合适的 embedding 模型
  • 考虑使用本地 embedding provider 以提升处理速度
一次性知识
  • 使用 CrewAI 提供的典型文件结构时,每次触发 kickoff 时,knowledge sources 都会被重新嵌入。
  • 如果 knowledge sources 很大,这会导致效率低下并增加延迟,因为每次都会重复嵌入相同数据。
  • 为了解决这个问题,请直接初始化 knowledge 参数,而不是 knowledge_sources 参数。
  • 要获取完整思路,可查看对应问题 Github Issue
Knowledge 管理
  • 为角色专属信息使用 agent 级 knowledge
  • 为所有 agents 共享所需的信息使用 crew 级 knowledge
  • 如果你需要不同的 embedding 策略,请在 agent 级设置 embedder
  • 通过保持 agent role 描述清晰来使用一致的 collection 命名
  • 通过在 kickoff 后检查 agent.knowledge 来测试 knowledge 初始化
  • 监控存储位置,以了解 knowledge 存储在哪里
  • 使用正确的命令类型适当地重置 knowledge
生产环境最佳实践
  • 在生产环境中将 CREWAI_STORAGE_DIR 设置为已知位置
  • 选择明确的 embedding provider,以匹配你的 LLM 设置并避免 API key 冲突
  • 随着文档增加,持续监控 knowledge 存储大小
  • 使用 collection 名称按领域或用途组织 knowledge sources
  • 将 knowledge 目录纳入备份和部署策略
  • 为 knowledge 文件和存储目录设置合适的文件权限
  • 使用环境变量存放 API keys 和敏感配置