Implementation:CrewAIInc CrewAI Crew Query Knowledge
Appearance
Metadata
| Field | Value |
|---|---|
| Implementation Name | Crew Query Knowledge |
| Workflow | Knowledge_RAG_Pipeline |
| Category | Programmatic Access |
| Repository | crewAIInc/crewAI |
| Implements | Principle:CrewAIInc_CrewAI_Direct_Knowledge_Querying |
Overview
Concrete method on Crew for direct knowledge base querying without agent execution provided by the CrewAI framework. The query_knowledge method exposes the crew's internal knowledge store for programmatic search, enabling debugging, testing, and custom application integration.
Source Reference
| File | Lines |
|---|---|
| src/crewai/crew.py | L1617-1625 |
Signature
class Crew(BaseModel):
# ... other fields and methods ...
def query_knowledge(
self,
query: list[str],
results_limit: int = 3,
score_threshold: float = 0.35,
) -> list[SearchResult] | None:
"""Query the crew's knowledge base directly.
Args:
query: List of query strings to search for.
results_limit: Maximum number of results to return.
score_threshold: Minimum similarity score for inclusion.
Returns:
List of SearchResult objects, or None if no knowledge is configured.
"""
...
Import
from crewai import Crew
I/O Contract
| Direction | Type | Description |
|---|---|---|
| Input | query: list[str] |
One or more query strings to search for in the knowledge base |
| Input | results_limit: int |
Maximum number of results to return (default: 3) |
| Input | score_threshold: float |
Minimum cosine similarity score for a result to be included (default: 0.35) |
| Output | None | List of matching search results, or None if no knowledge is configured |
Method Behavior
The query_knowledge method operates as follows:
- Check knowledge availability -- If the crew has no knowledge configured (no
knowledge_sourcesor no internal Knowledge object), returnsNone - Delegate to Knowledge.query() -- Calls the internal Knowledge object's
query()method with the provided parameters - Return results -- Returns the list of
SearchResultobjects from the vector store search
The method is a thin wrapper that provides a convenient API on the Crew object, delegating all actual search logic to the underlying Knowledge and KnowledgeStorage classes.
Default Parameter Differences
Note the different defaults compared to the internal automatic retrieval:
| Parameter | query_knowledge() Default | Automatic Retrieval Default | Reason |
|---|---|---|---|
results_limit |
3 | 5 | Direct queries typically need fewer, more focused results |
score_threshold |
0.35 | 0.6 | Lower threshold for exploratory/debugging use cases |
Code Examples
Basic Direct Query
from crewai import Crew, Agent, Task
from crewai.knowledge.source import PDFKnowledgeSource
# Configure crew with knowledge
crew = Crew(
agents=[
Agent(
role="Assistant",
goal="Help users",
backstory="Helpful assistant",
)
],
tasks=[
Task(
description="Placeholder task",
expected_output="Response",
)
],
knowledge_sources=[
PDFKnowledgeSource(file_paths=["docs/product_manual.pdf"])
],
embedder={
"provider": "openai",
"config": {"model": "text-embedding-3-small"},
},
)
# Query knowledge directly without running agents
results = crew.query_knowledge(["What is the return policy?"])
if results:
for result in results:
print(f"Score: {result.score:.3f}")
print(f"Content: {result.content}")
print("---")
else:
print("No knowledge configured or no results found.")
Testing Ingestion Quality
# Verify that specific topics were ingested correctly
test_queries = [
"password reset instructions",
"billing and payment methods",
"account deletion process",
]
for q in test_queries:
results = crew.query_knowledge(
query=[q],
results_limit=3,
score_threshold=0.3,
)
count = len(results) if results else 0
top_score = results[0].score if results else 0.0
print(f"Query: '{q}' -> {count} results (top score: {top_score:.3f})")
Custom Search with Adjusted Parameters
# Broader search with more results and lower threshold
results = crew.query_knowledge(
query=["network configuration"],
results_limit=10,
score_threshold=0.2,
)
# Stricter search with fewer, higher-quality results
results = crew.query_knowledge(
query=["exact error message: ConnectionRefused"],
results_limit=1,
score_threshold=0.7,
)
Integration with Application Code
# Using direct querying in a web API endpoint
from fastapi import FastAPI
app = FastAPI()
@app.get("/search")
def search_docs(q: str):
results = crew.query_knowledge(
query=[q],
results_limit=5,
score_threshold=0.35,
)
if not results:
return {"results": [], "message": "No matching documents found."}
return {
"results": [
{
"content": r.content,
"score": round(r.score, 3),
"metadata": r.metadata,
}
for r in results
]
}
Handling the None Return Case
# Safe handling when knowledge may not be configured
results = crew.query_knowledge(["some query"])
if results is None:
# No knowledge is configured on this crew
print("This crew has no knowledge sources configured.")
elif len(results) == 0:
# Knowledge exists but no results matched the query/threshold
print("No results matched the query above the score threshold.")
else:
# Results found
for r in results:
print(f"[{r.score:.3f}] {r.content[:150]}")
Related Pages
- Principle:CrewAIInc_CrewAI_Direct_Knowledge_Querying -- The principle this implements
- Implementation:CrewAIInc_CrewAI_Knowledge_Storage_Search -- Underlying storage search called by this method
- Implementation:CrewAIInc_CrewAI_Knowledge_Constructor -- Knowledge object that manages the query pipeline
- Implementation:CrewAIInc_CrewAI_Knowledge_Attachment_Config -- Configuration that enables knowledge on the crew
- Heuristic:CrewAIInc_CrewAI_RAG_Search_Defaults
Page Connections
Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment