Jump to content

Connect Leeroopedia MCP: Equip your AI agents to search best practices, build plans, verify code, diagnose failures, and look up hyperparameter defaults.

Implementation:CrewAIInc CrewAI Crew Query Knowledge

From Leeroopedia

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:

  1. Check knowledge availability -- If the crew has no knowledge configured (no knowledge_sources or no internal Knowledge object), returns None
  2. Delegate to Knowledge.query() -- Calls the internal Knowledge object's query() method with the provided parameters
  3. Return results -- Returns the list of SearchResult objects 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

Page Connections

Double-click a node to navigate. Hold to expand connections.
Principle
Implementation
Heuristic
Environment