知识
CrewAI 中的知识是一套强大的系统,允许 AI agents 在执行任务时访问并利用外部信息源。 你可以把它理解为给你的 agents 配备一个在工作中随时可查阅的参考资料库。
快速开始示例
Section titled “快速开始示例”向量存储(RAG)客户端配置
Section titled “向量存储(RAG)客户端配置”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 ChromaDBConfigset_rag_config(ChromaDBConfig())chromadb_client = get_rag_client()
# Qdrantfrom crewai.rag.qdrant.config import QdrantConfigset_rag_config(QdrantConfig())qdrant_client = get_rag_client()
# Example operations (same API for any provider)client = qdrant_client # or chromadb_clientclient.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 的内置存储是分开的。当你需要直接控制向量存储或自定义检索管线时,请使用它。
基础字符串知识示例
Section titled “基础字符串知识示例”from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Create a knowledge sourcecontent = "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 outputsllm = LLM(model="gpt-4o-mini", temperature=0)
# Create an agent with the knowledge storeagent = 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?"})Web 内容知识示例
Section titled “Web 内容知识示例”from crewai import LLM, Agent, Crew, Process, Taskfrom crewai.knowledge.source.crew_docling_source import CrewDoclingSource
# Create a knowledge source from web contentcontent_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 outputsllm = LLM(model="gpt-4o-mini", temperature=0)
# Create an agent with the knowledge storeagent = 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."})支持的知识源
Section titled “支持的知识源”CrewAI 开箱即支持多种知识源:
文本源
- 原始字符串
- 文本文件(.txt)
- PDF 文档
结构化数据
- CSV 文件
- Excel 电子表格
- JSON 文档
文本文件知识源
Section titled “文本文件知识源”from crewai.knowledge.source.text_file_knowledge_source import TextFileKnowledgeSource
text_source = TextFileKnowledgeSource( file_paths=["document.txt", "another.txt"])PDF 知识源
Section titled “PDF 知识源”from crewai.knowledge.source.pdf_knowledge_source import PDFKnowledgeSource
pdf_source = PDFKnowledgeSource( file_paths=["document.pdf", "another.pdf"])CSV 知识源
Section titled “CSV 知识源”from crewai.knowledge.source.csv_knowledge_source import CSVKnowledgeSource
csv_source = CSVKnowledgeSource( file_paths=["data.csv"])Excel 知识源
Section titled “Excel 知识源”from crewai.knowledge.source.excel_knowledge_source import ExcelKnowledgeSource
excel_source = ExcelKnowledgeSource( file_paths=["spreadsheet.xlsx"])JSON 知识源
Section titled “JSON 知识源”from crewai.knowledge.source.json_knowledge_source import JSONKnowledgeSource
json_source = JSONKnowledgeSource( file_paths=["data.json"])Agent 与 Crew 知识:完整指南
Section titled “Agent 与 Crew 知识:完整指南”知识初始化的实际流程
Section titled “知识初始化的实际流程”当你使用 knowledge 时,实际会发生以下事情:
Agent 级知识(独立)
Section titled “Agent 级知识(独立)”from crewai import Agent, Task, Crewfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Agent with its own knowledge - NO crew knowledge neededspecialist_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 requiredcrew = Crew( agents=[specialist_agent], tasks=[task])
result = crew.kickoff() # Agent knowledge works independently在 crew.kickoff() 期间会发生什么
Section titled “在 crew.kickoff() 期间会发生什么”当你调用 crew.kickoff() 时,实际执行顺序如下:
# During kickofffor 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 storageagent_collection_name = agent.role # e.g., "Technical Specialist"
# Crew knowledge storagecrew_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完整可运行示例
Section titled “完整可运行示例”示例 1:仅 Agent 知识
Section titled “示例 1:仅 Agent 知识”from crewai import Agent, Task, Crewfrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Agent-specific knowledgeagent_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 neededcrew = 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 knowledgespecialist_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 agentssales_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 providersKnowledge 配置
Section titled “Knowledge 配置”你可以为 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)支持的 Knowledge 参数
Section titled “支持的 Knowledge 参数”sources List[BaseKnowledgeSource] required 提供内容以便存储和查询的 knowledge sources 列表。可以包含 PDF、CSV、Excel、JSON、文本文件,或字符串内容。
collection_name str 用于存储 knowledge 的 collection 名称。用于标识不同的 knowledge 集合。若未提供,默认值为 “knowledge”。
storage Optional[KnowledgeStorage] 用于管理 knowledge 存储与检索方式的自定义存储配置。如果未提供,会创建默认 storage。
Knowledge 存储透明性
Section titled “Knowledge 存储透明性”CrewAI 将 Knowledge 文件存在哪里
Section titled “CrewAI 将 Knowledge 文件存在哪里”默认情况下,CrewAI 使用与 memory 相同的存储系统,将 knowledge 存放在平台特定目录中:
按平台划分的默认存储位置
Section titled “按平台划分的默认存储位置”macOS:
~/Library/Application Support/CrewAI/{project_name}/└── knowledge/ # Knowledge ChromaDB files ├── chroma.sqlite3 # ChromaDB metadata ├── {collection_id}/ # Vector embeddings └── knowledge_{collection}/ # Named collectionsLinux:
~/.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}\查找你的 Knowledge 存储位置
Section titled “查找你的 Knowledge 存储位置”要精确查看 CrewAI 正在将 knowledge 文件存到哪里:
from crewai.utilities.paths import db_storage_pathimport os
# Get the knowledge storage pathknowledge_path = os.path.join(db_storage_path(), "knowledge")print(f"Knowledge storage location: {knowledge_path}")
# List knowledge collections and filesif 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.")控制 Knowledge 存储位置
Section titled “控制 Knowledge 存储位置”选项 1:环境变量(推荐)
Section titled “选项 1:环境变量(推荐)”import osfrom crewai import Crew
# Set custom storage location for all CrewAI dataos.environ["CREWAI_STORAGE_DIR"] = "./my_project_storage"
# All knowledge will now be stored in ./my_project_storage/knowledge/crew = Crew( agents=[...], tasks=[...], knowledge_sources=[...])选项 2:自定义 Knowledge Storage
Section titled “选项 2:自定义 Knowledge Storage”from crewai.knowledge.storage.knowledge_storage import KnowledgeStoragefrom crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Create custom storage with specific embeddercustom_storage = KnowledgeStorage( embedder={ "provider": "ollama", "config": {"model": "mxbai-embed-large"} }, collection_name="my_custom_knowledge")
# Use with knowledge sourcesknowledge_source = StringKnowledgeSource( content="Your knowledge content here")knowledge_source.storage = custom_storage选项 3:项目专属 Knowledge Storage
Section titled “选项 3:项目专属 Knowledge Storage”import osfrom pathlib import Path
# Store knowledge in project directoryproject_root = Path(__file__).parentknowledge_dir = project_root / "knowledge_storage"
os.environ["CREWAI_STORAGE_DIR"] = str(knowledge_dir)
# Now all knowledge will be stored in your project directory默认 Embedding Provider 行为
Section titled “默认 Embedding Provider 行为”默认行为说明
Section titled “默认行为说明”from crewai import Agent, Crew, LLMfrom 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 preferenceknowledge_source = StringKnowledgeSource(content="Research data...")
crew = Crew( agents=[agent], tasks=[...], knowledge_sources=[knowledge_source] # Default: Uses OpenAI embeddings even with Claude LLM)自定义 Knowledge Embedding Provider
Section titled “自定义 Knowledge Embedding Provider”# 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 customizationagent = 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
Section titled “配置 Azure OpenAI Embeddings”使用 Azure OpenAI embeddings 时:
- 请先确保你已在 Azure 平台中部署 embedding 模型
- 然后使用以下配置:
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 时,原始任务提示会自动转换为更有效的搜索查询。
查询重写如何工作
Section titled “查询重写如何工作”- 当 agent 执行一个可访问 knowledge sources 的任务时,会触发
_get_knowledge_search_query方法 - agent 的 LLM 会将原始任务提示转换为优化后的搜索查询
- 这个优化后的查询随后用于从 knowledge sources 中检索相关信息
查询重写的好处
Section titled “查询重写的好处”提升检索准确性
通过聚焦关键概念并移除无关内容,查询重写有助于检索到更相关的信息。
上下文感知
重写后的查询会更具体,也更能理解上下文,适合向量数据库检索。
# Original task prompttask_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?"重写后的查询更聚焦于核心信息需求,并移除了与输出格式相关的无关指令。
Knowledge 事件
Section titled “Knowledge 事件”CrewAI 会在 knowledge 检索过程中发出事件,你可以使用事件系统监听这些事件。这些事件可让你监控、调试并分析 knowledge 是如何被检索和使用的。
可用的 Knowledge 事件
Section titled “可用的 Knowledge 事件”- KnowledgeRetrievalStartedEvent:当 agent 开始从 sources 中检索 knowledge 时触发
- KnowledgeRetrievalCompletedEvent:当 knowledge 检索完成时触发,包括所用查询和检索到的内容
- KnowledgeQueryStartedEvent:当对 knowledge sources 的查询开始时触发
- KnowledgeQueryCompletedEvent:当查询成功完成时触发
- KnowledgeQueryFailedEvent:当对 knowledge sources 的查询失败时触发
- KnowledgeSearchQueryFailedEvent:当搜索查询失败时触发
示例:监控 Knowledge 检索
Section titled “示例:监控 Knowledge 检索”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 listenerknowledge_monitor = KnowledgeMonitorListener()有关事件使用的更多信息,请参阅 Event Listeners 文档。
自定义 Knowledge Sources
Section titled “自定义 Knowledge Sources”CrewAI 允许你通过扩展 BaseKnowledgeSource 类来为任何类型的数据创建自定义 knowledge source。下面我们创建一个实际示例,用于获取并处理太空新闻文章。
太空新闻 Knowledge Source 示例
Section titled “太空新闻 Knowledge Source 示例”from crewai import Agent, Task, Crew, Process, LLMfrom crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSourceimport requestsfrom datetime import datetimefrom typing import Dict, Anyfrom 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 sourcerecent_news = SpaceNewsKnowledgeSource( api_endpoint="https://api.spaceflightnewsapi.net/v4/articles", limit=10,)
# Create specialized agentspace_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 questionsanalysis_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 crewcrew = Crew( agents=[space_analyst], tasks=[analysis_task], verbose=True, process=Process.sequential)
# Example usageresult = 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/)调试与故障排查
Section titled “调试与故障排查”调试 Knowledge 问题
Section titled “调试 Knowledge 问题”检查 Agent Knowledge 初始化
Section titled “检查 Agent Knowledge 初始化”from crewai import Agent, Crew, Taskfrom 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 initializedprint(f"Before kickoff - Agent knowledge: {getattr(agent, 'knowledge', None)}")
crew.kickoff()
# After kickoff - knowledge initializedprint(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)}")验证 Knowledge 存储位置
Section titled “验证 Knowledge 存储位置”import osfrom crewai.utilities.paths import db_storage_path
# Check storage structurestorage_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}")测试 Knowledge 检索
Section titled “测试 Knowledge 检索”# Test agent knowledge retrievalif 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")检查 Knowledge Collections
Section titled “检查 Knowledge Collections”import chromadbfrom crewai.utilities.paths import db_storage_pathimport os
# Connect to CrewAI's knowledge ChromaDBknowledge_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")检查 Knowledge 处理
Section titled “检查 Knowledge 处理”from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
# Create a test knowledge sourcetest_source = StringKnowledgeSource( content="Test knowledge content for debugging", chunk_size=100, # Small chunks for testing chunk_overlap=20)
# Check chunking behaviorprint(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 chunkstest_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]}...")常见的 Knowledge 存储问题
Section titled “常见的 Knowledge 存储问题”“File not found” 错误:
# Ensure files are in the correct locationfrom crewai.utilities.constants import KNOWLEDGE_DIRECTORYimport 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 embeddingscrew.reset_memories(command_type='knowledge')
# Or use consistent embedding providerscrew = Crew( agents=[...], tasks=[...], knowledge_sources=[...], embedder={"provider": "openai", "config": {"model": "text-embedding-3-small"}})“ChromaDB permission denied” 错误:
# Fix storage permissionschmod -R 755 ~/.local/share/CrewAI/Knowledge 在运行之间没有持久化:
# Verify storage location consistencyimport osfrom 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"))Knowledge 重置命令
Section titled “Knowledge 重置命令”# Reset only agent-specific knowledgecrew.reset_memories(command_type='agent_knowledge')
# Reset both crew and agent knowledgecrew.reset_memories(command_type='knowledge')
# CLI commands# crewai reset-memories --agent-knowledge # Agent knowledge only# crewai reset-memories --knowledge # All knowledge清除 Knowledge
Section titled “清除 Knowledge”如果你需要清除 CrewAI 中存储的 knowledge,可以使用带有 --knowledge 选项的 crewai reset-memories 命令。
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 和敏感配置