Fix issue #2233: Add support for task-specific knowledge

Co-Authored-By: Joe Moura <joao@crewai.com>

src/crewai/agent.py

@@ -207,6 +207,17 @@ class Agent(BaseAgent):
             if memory.strip() != "":
                 task_prompt += self.i18n.slice("memory").format(memory=memory)
 
+        # Check if the task has knowledge first
+        if hasattr(task, 'knowledge') and task.knowledge:
+            task_knowledge_snippets = task.knowledge.query([task.prompt()])
+            if task_knowledge_snippets:
+                task_knowledge_context = extract_knowledge_context(
+                    task_knowledge_snippets
+                )
+                if task_knowledge_context:
+                    task_prompt += task_knowledge_context
+
+        # Then check agent's knowledge
         if self.knowledge:
             agent_knowledge_snippets = self.knowledge.query([task.prompt()])
             if agent_knowledge_snippets:
@@ -216,6 +227,7 @@ class Agent(BaseAgent):
                 if agent_knowledge_context:
                     task_prompt += agent_knowledge_context
 
+        # Finally check crew's knowledge
         if self.crew:
             knowledge_snippets = self.crew.query_knowledge([task.prompt()])
             if knowledge_snippets:

src/crewai/knowledge/__init__.py

@@ -0,0 +1,3 @@
+from crewai.knowledge.knowledge import Knowledge
+
+__all__ = ["Knowledge"]

src/crewai/task.py

@@ -32,6 +32,7 @@ from pydantic import (
 from pydantic_core import PydanticCustomError
 
 from crewai.agents.agent_builder.base_agent import BaseAgent
+from crewai.knowledge import Knowledge
 from crewai.tasks.guardrail_result import GuardrailResult
 from crewai.tasks.output_format import OutputFormat
 from crewai.tasks.task_output import TaskOutput
@@ -144,6 +145,10 @@ class Task(BaseModel):
     end_time: Optional[datetime.datetime] = Field(
         default=None, description="End time of the task execution"
     )
+    knowledge: Optional[Knowledge] = Field(
+        default=None,
+        description="Knowledge sources for the task. This knowledge will be used by the agent when executing the task.",
+    )
 
     @field_validator("guardrail")
     @classmethod

tests/agent_test.py

@@ -12,6 +12,7 @@ from crewai.agents.crew_agent_executor import AgentFinish, CrewAgentExecutor
 from crewai.agents.parser import AgentAction, CrewAgentParser, OutputParserException
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
 from crewai.knowledge.source.string_knowledge_source import StringKnowledgeSource
+from crewai.knowledge import Knowledge
 from crewai.llm import LLM
 from crewai.tools import tool
 from crewai.tools.tool_calling import InstructorToolCalling
@@ -1661,6 +1662,54 @@ def test_agent_with_knowledge_sources_works_with_copy():
             assert isinstance(agent_copy.llm, LLM)
 
 
+@pytest.mark.vcr(filter_headers=["authorization"])
+def test_agent_uses_task_knowledge():
+    """Test that an agent uses the knowledge provided in the task."""
+    # Create a knowledge source with specific content
+    content = "The capital of France is Paris. The Eiffel Tower is located in Paris."
+    
+    # Create a mock Knowledge object
+    with patch("crewai.knowledge.Knowledge", autospec=True) as MockKnowledge:
+        # Configure the mock
+        mock_knowledge = MockKnowledge.return_value
+        mock_knowledge.query.return_value = [{"content": content}]
+        
+        # Create an agent without knowledge sources
+        agent = Agent(
+            role="Geography Teacher",
+            goal="Provide accurate geographic information",
+            backstory="You are a geography expert who teaches students about world capitals.",
+            llm=LLM(model="gpt-4o-mini"),
+        )
+        
+        # Create a task with knowledge
+        task = Task(
+            description="What is the capital of France?",
+            expected_output="The capital of France.",
+            agent=agent,
+            knowledge=mock_knowledge,
+        )
+        
+        # Mock the agent's execute_task method to avoid actual LLM calls
+        with patch.object(agent.llm, "call") as mock_llm_call:
+            mock_llm_call.return_value = "The capital of France is Paris, where the Eiffel Tower is located."
+            
+            # Execute the task
+            result = agent.execute_task(task)
+            
+            # Assert that the agent provides the correct information
+            assert "paris" in result.lower()
+            assert "eiffel tower" in result.lower()
+            
+            # Verify that the task's knowledge was queried
+            mock_knowledge.query.assert_called_once()
+            
+            # The query should include the task prompt
+            query_arg = mock_knowledge.query.call_args[0][0]
+            assert isinstance(query_arg, list)
+            assert "capital of france" in query_arg[0].lower()
+
+
 @pytest.mark.vcr(filter_headers=["authorization"])
 def test_litellm_auth_error_handling():
     """Test that LiteLLM authentication errors are handled correctly and not retried."""