Merge branch 'main' into brandon/eng-266-conversation-crew-v1

src/crewai/agent.py

@@ -17,6 +17,7 @@ from crewai.memory.contextual.contextual_memory import ContextualMemory
 from crewai.task import Task
 from crewai.tools import BaseTool
 from crewai.tools.agent_tools.agent_tools import AgentTools
+from crewai.tools.base_tool import Tool
 from crewai.utilities import Converter, Prompts
 from crewai.utilities.constants import TRAINED_AGENTS_DATA_FILE, TRAINING_DATA_FILE
 from crewai.utilities.converter import generate_model_description
@@ -115,6 +116,10 @@ class Agent(BaseAgent):
         default=2,
         description="Maximum number of retries for an agent to execute a task when an error occurs.",
     )
+    multimodal: bool = Field(
+        default=False,
+        description="Whether the agent is multimodal.",
+    )
     code_execution_mode: Literal["safe", "unsafe"] = Field(
         default="safe",
         description="Mode for code execution: 'safe' (using Docker) or 'unsafe' (direct execution).",
@@ -329,6 +334,10 @@ class Agent(BaseAgent):
         tools = agent_tools.tools()
         return tools
 
+    def get_multimodal_tools(self) -> List[Tool]:
+        from crewai.tools.agent_tools.add_image_tool import AddImageTool
+        return [AddImageTool()]
+
     def get_code_execution_tools(self):
         try:
             from crewai_tools import CodeInterpreterTool

src/crewai/agents/crew_agent_executor.py

@@ -143,10 +143,20 @@ class CrewAgentExecutor(CrewAgentExecutorMixin):
                         tool_result = self._execute_tool_and_check_finality(
                             formatted_answer
                         )
-                        if self.step_callback:
-                            self.step_callback(tool_result)
 
-                        formatted_answer.text += f"\nObservation: {tool_result.result}"
+                        # Directly append the result to the messages if the
+                        # tool is "Add image to content" in case of multimodal
+                        # agents
+                        if formatted_answer.tool == self._i18n.tools("add_image")["name"]:
+                            self.messages.append(tool_result.result)
+                            continue
+
+                        else:
+                            if self.step_callback:
+                                self.step_callback(tool_result)
+
+                            formatted_answer.text += f"\nObservation: {tool_result.result}"
+
                         formatted_answer.result = tool_result.result
                         if tool_result.result_as_answer:
                             return AgentFinish(

src/crewai/crew.py

@@ -37,6 +37,7 @@ from crewai.tasks.task_output import TaskOutput
 from crewai.telemetry import Telemetry
 from crewai.tools.agent_tools.agent_tools import AgentTools
 from crewai.types.crew_chat import ChatInputs
+from crewai.tools.base_tool import Tool
 from crewai.types.usage_metrics import UsageMetrics
 from crewai.utilities import I18N, FileHandler, Logger, RPMController
 from crewai.utilities.constants import TRAINING_DATA_FILE
@@ -543,9 +544,6 @@ class Crew(BaseModel):
             if not agent.function_calling_llm:  # type: ignore # "BaseAgent" has no attribute "function_calling_llm"
                 agent.function_calling_llm = self.function_calling_llm  # type: ignore # "BaseAgent" has no attribute "function_calling_llm"
 
-            if agent.allow_code_execution:  # type: ignore # BaseAgent" has no attribute "allow_code_execution"
-                agent.tools += agent.get_code_execution_tools()  # type: ignore # "BaseAgent" has no attribute "get_code_execution_tools"; maybe "get_delegation_tools"?
-
             if not agent.step_callback:  # type: ignore # "BaseAgent" has no attribute "step_callback"
                 agent.step_callback = self.step_callback  # type: ignore # "BaseAgent" has no attribute "step_callback"
 
@@ -682,7 +680,6 @@ class Crew(BaseModel):
                 )
                 manager.tools = []
                 raise Exception("Manager agent should not have tools")
-            manager.tools = self.manager_agent.get_delegation_tools(self.agents)
         else:
             self.manager_llm = (
                 getattr(self.manager_llm, "model_name", None)
@@ -694,6 +691,7 @@ class Crew(BaseModel):
                 goal=i18n.retrieve("hierarchical_manager_agent", "goal"),
                 backstory=i18n.retrieve("hierarchical_manager_agent", "backstory"),
                 tools=AgentTools(agents=self.agents).tools(),
+                allow_delegation=True,
                 llm=self.manager_llm,
                 verbose=self.verbose,
             )
@@ -736,7 +734,10 @@ class Crew(BaseModel):
                     f"No agent available for task: {task.description}. Ensure that either the task has an assigned agent or a manager agent is provided."
                 )
 
-            self._prepare_agent_tools(task)
+            # Determine which tools to use - task tools take precedence over agent tools
+            tools_for_task = task.tools or agent_to_use.tools or []
+            tools_for_task = self._prepare_tools(agent_to_use, task, tools_for_task)
+
             self._log_task_start(task, agent_to_use.role)
 
             if isinstance(task, ConditionalTask):
@@ -753,7 +754,7 @@ class Crew(BaseModel):
                 future = task.execute_async(
                     agent=agent_to_use,
                     context=context,
-                    tools=agent_to_use.tools,
+                    tools=tools_for_task,
                 )
                 futures.append((task, future, task_index))
             else:
@@ -765,7 +766,7 @@ class Crew(BaseModel):
                 task_output = task.execute_sync(
                     agent=agent_to_use,
                     context=context,
-                    tools=agent_to_use.tools,
+                    tools=tools_for_task,
                 )
                 task_outputs = [task_output]
                 self._process_task_result(task, task_output)
@@ -802,45 +803,77 @@ class Crew(BaseModel):
             return skipped_task_output
         return None
 
-    def _prepare_agent_tools(self, task: Task):
-        if self.process == Process.hierarchical:
-            if self.manager_agent:
-                self._update_manager_tools(task)
-            else:
-                raise ValueError("Manager agent is required for hierarchical process.")
-        elif task.agent and task.agent.allow_delegation:
-            self._add_delegation_tools(task)
+    def _prepare_tools(
+        self, agent: BaseAgent, task: Task, tools: List[Tool]
+    ) -> List[Tool]:
+        # Add delegation tools if agent allows delegation
+        if agent.allow_delegation:
+            if self.process == Process.hierarchical:
+                if self.manager_agent:
+                    tools = self._update_manager_tools(task, tools)
+                else:
+                    raise ValueError(
+                        "Manager agent is required for hierarchical process."
+                    )
+
+            elif agent and agent.allow_delegation:
+                tools = self._add_delegation_tools(task, tools)
+
+        # Add code execution tools if agent allows code execution
+        if agent.allow_code_execution:
+            tools = self._add_code_execution_tools(agent, tools)
+
+        if agent and agent.multimodal:
+            tools = self._add_multimodal_tools(agent, tools)
+
+        return tools
 
     def _get_agent_to_use(self, task: Task) -> Optional[BaseAgent]:
         if self.process == Process.hierarchical:
             return self.manager_agent
         return task.agent
 
-    def _add_delegation_tools(self, task: Task):
+    def _merge_tools(
+        self, existing_tools: List[Tool], new_tools: List[Tool]
+    ) -> List[Tool]:
+        """Merge new tools into existing tools list, avoiding duplicates by tool name."""
+        if not new_tools:
+            return existing_tools
+
+        # Create mapping of tool names to new tools
+        new_tool_map = {tool.name: tool for tool in new_tools}
+
+        # Remove any existing tools that will be replaced
+        tools = [tool for tool in existing_tools if tool.name not in new_tool_map]
+
+        # Add all new tools
+        tools.extend(new_tools)
+
+        return tools
+
+    def _inject_delegation_tools(
+        self, tools: List[Tool], task_agent: BaseAgent, agents: List[BaseAgent]
+    ):
+        delegation_tools = task_agent.get_delegation_tools(agents)
+        return self._merge_tools(tools, delegation_tools)
+
+    def _add_multimodal_tools(self, agent: BaseAgent, tools: List[Tool]):
+        multimodal_tools = agent.get_multimodal_tools()
+        return self._merge_tools(tools, multimodal_tools)
+
+    def _add_code_execution_tools(self, agent: BaseAgent, tools: List[Tool]):
+        code_tools = agent.get_code_execution_tools()
+        return self._merge_tools(tools, code_tools)
+
+    def _add_delegation_tools(self, task: Task, tools: List[Tool]):
         agents_for_delegation = [agent for agent in self.agents if agent != task.agent]
         if len(self.agents) > 1 and len(agents_for_delegation) > 0 and task.agent:
-            delegation_tools = task.agent.get_delegation_tools(agents_for_delegation)
-
-            # Add tools if they are not already in task.tools
-            for new_tool in delegation_tools:
-                # Find the index of the tool with the same name
-                existing_tool_index = next(
-                    (
-                        index
-                        for index, tool in enumerate(task.tools or [])
-                        if tool.name == new_tool.name
-                    ),
-                    None,
-                )
-                if not task.tools:
-                    task.tools = []
-
-                if existing_tool_index is not None:
-                    # Replace the existing tool
-                    task.tools[existing_tool_index] = new_tool
-                else:
-                    # Add the new tool
-                    task.tools.append(new_tool)
+            if not tools:
+                tools = []
+            tools = self._inject_delegation_tools(
+                tools, task.agent, agents_for_delegation
+            )
+        return tools
 
     def _log_task_start(self, task: Task, role: str = "None"):
         if self.output_log_file:
@@ -848,14 +881,15 @@ class Crew(BaseModel):
                 task_name=task.name, task=task.description, agent=role, status="started"
             )
 
-    def _update_manager_tools(self, task: Task):
+    def _update_manager_tools(self, task: Task, tools: List[Tool]):
         if self.manager_agent:
             if task.agent:
-                self.manager_agent.tools = task.agent.get_delegation_tools([task.agent])
+                tools = self._inject_delegation_tools(tools, task.agent, [task.agent])
             else:
-                self.manager_agent.tools = self.manager_agent.get_delegation_tools(
-                    self.agents
+                tools = self._inject_delegation_tools(
+                    tools, self.manager_agent, self.agents
                 )
+        return tools
 
     def _get_context(self, task: Task, task_outputs: List[TaskOutput]):
         context = (

src/crewai/flow/flow.py

@@ -30,7 +30,47 @@ from crewai.telemetry import Telemetry
 T = TypeVar("T", bound=Union[BaseModel, Dict[str, Any]])
 
 
-def start(condition=None):
+def start(condition: Optional[Union[str, dict, Callable]] = None) -> Callable:
+    """
+    Marks a method as a flow's starting point.
+
+    This decorator designates a method as an entry point for the flow execution.
+    It can optionally specify conditions that trigger the start based on other
+    method executions.
+
+    Parameters
+    ----------
+    condition : Optional[Union[str, dict, Callable]], optional
+        Defines when the start method should execute. Can be:
+        - str: Name of a method that triggers this start
+        - dict: Contains "type" ("AND"/"OR") and "methods" (list of triggers)
+        - Callable: A method reference that triggers this start
+        Default is None, meaning unconditional start.
+
+    Returns
+    -------
+    Callable
+        A decorator function that marks the method as a flow start point.
+
+    Raises
+    ------
+    ValueError
+        If the condition format is invalid.
+
+    Examples
+    --------
+    >>> @start()  # Unconditional start
+    >>> def begin_flow(self):
+    ...     pass
+
+    >>> @start("method_name")  # Start after specific method
+    >>> def conditional_start(self):
+    ...     pass
+
+    >>> @start(and_("method1", "method2"))  # Start after multiple methods
+    >>> def complex_start(self):
+    ...     pass
+    """
     def decorator(func):
         func.__is_start_method__ = True
         if condition is not None:
@@ -55,8 +95,42 @@ def start(condition=None):
 
     return decorator
 
+def listen(condition: Union[str, dict, Callable]) -> Callable:
+    """
+    Creates a listener that executes when specified conditions are met.
 
-def listen(condition):
+    This decorator sets up a method to execute in response to other method
+    executions in the flow. It supports both simple and complex triggering
+    conditions.
+
+    Parameters
+    ----------
+    condition : Union[str, dict, Callable]
+        Specifies when the listener should execute. Can be:
+        - str: Name of a method that triggers this listener
+        - dict: Contains "type" ("AND"/"OR") and "methods" (list of triggers)
+        - Callable: A method reference that triggers this listener
+
+    Returns
+    -------
+    Callable
+        A decorator function that sets up the method as a listener.
+
+    Raises
+    ------
+    ValueError
+        If the condition format is invalid.
+
+    Examples
+    --------
+    >>> @listen("process_data")  # Listen to single method
+    >>> def handle_processed_data(self):
+    ...     pass
+
+    >>> @listen(or_("success", "failure"))  # Listen to multiple methods
+    >>> def handle_completion(self):
+    ...     pass
+    """
     def decorator(func):
         if isinstance(condition, str):
             func.__trigger_methods__ = [condition]
@@ -80,16 +154,103 @@ def listen(condition):
     return decorator
 
 
-def router(method):
+def router(condition: Union[str, dict, Callable]) -> Callable:
+    """
+    Creates a routing method that directs flow execution based on conditions.
+
+    This decorator marks a method as a router, which can dynamically determine
+    the next steps in the flow based on its return value. Routers are triggered
+    by specified conditions and can return constants that determine which path
+    the flow should take.
+
+    Parameters
+    ----------
+    condition : Union[str, dict, Callable]
+        Specifies when the router should execute. Can be:
+        - str: Name of a method that triggers this router
+        - dict: Contains "type" ("AND"/"OR") and "methods" (list of triggers)
+        - Callable: A method reference that triggers this router
+
+    Returns
+    -------
+    Callable
+        A decorator function that sets up the method as a router.
+
+    Raises
+    ------
+    ValueError
+        If the condition format is invalid.
+
+    Examples
+    --------
+    >>> @router("check_status")
+    >>> def route_based_on_status(self):
+    ...     if self.state.status == "success":
+    ...         return SUCCESS
+    ...     return FAILURE
+
+    >>> @router(and_("validate", "process"))
+    >>> def complex_routing(self):
+    ...     if all([self.state.valid, self.state.processed]):
+    ...         return CONTINUE
+    ...     return STOP
+    """
     def decorator(func):
         func.__is_router__ = True
-        func.__router_for__ = method.__name__
+        if isinstance(condition, str):
+            func.__trigger_methods__ = [condition]
+            func.__condition_type__ = "OR"
+        elif (
+            isinstance(condition, dict)
+            and "type" in condition
+            and "methods" in condition
+        ):
+            func.__trigger_methods__ = condition["methods"]
+            func.__condition_type__ = condition["type"]
+        elif callable(condition) and hasattr(condition, "__name__"):
+            func.__trigger_methods__ = [condition.__name__]
+            func.__condition_type__ = "OR"
+        else:
+            raise ValueError(
+                "Condition must be a method, string, or a result of or_() or and_()"
+            )
         return func
 
     return decorator
 
+def or_(*conditions: Union[str, dict, Callable]) -> dict:
+    """
+    Combines multiple conditions with OR logic for flow control.
 
-def or_(*conditions):
+    Creates a condition that is satisfied when any of the specified conditions
+    are met. This is used with @start, @listen, or @router decorators to create
+    complex triggering conditions.
+
+    Parameters
+    ----------
+    *conditions : Union[str, dict, Callable]
+        Variable number of conditions that can be:
+        - str: Method names
+        - dict: Existing condition dictionaries
+        - Callable: Method references
+
+    Returns
+    -------
+    dict
+        A condition dictionary with format:
+        {"type": "OR", "methods": list_of_method_names}
+
+    Raises
+    ------
+    ValueError
+        If any condition is invalid.
+
+    Examples
+    --------
+    >>> @listen(or_("success", "timeout"))
+    >>> def handle_completion(self):
+    ...     pass
+    """
     methods = []
     for condition in conditions:
         if isinstance(condition, dict) and "methods" in condition:
@@ -103,7 +264,39 @@ def or_(*conditions):
     return {"type": "OR", "methods": methods}
 
 
-def and_(*conditions):
+def and_(*conditions: Union[str, dict, Callable]) -> dict:
+    """
+    Combines multiple conditions with AND logic for flow control.
+
+    Creates a condition that is satisfied only when all specified conditions
+    are met. This is used with @start, @listen, or @router decorators to create
+    complex triggering conditions.
+
+    Parameters
+    ----------
+    *conditions : Union[str, dict, Callable]
+        Variable number of conditions that can be:
+        - str: Method names
+        - dict: Existing condition dictionaries
+        - Callable: Method references
+
+    Returns
+    -------
+    dict
+        A condition dictionary with format:
+        {"type": "AND", "methods": list_of_method_names}
+
+    Raises
+    ------
+    ValueError
+        If any condition is invalid.
+
+    Examples
+    --------
+    >>> @listen(and_("validated", "processed"))
+    >>> def handle_complete_data(self):
+    ...     pass
+    """
     methods = []
     for condition in conditions:
         if isinstance(condition, dict) and "methods" in condition:
@@ -123,8 +316,8 @@ class FlowMeta(type):
 
         start_methods = []
         listeners = {}
-        routers = {}
         router_paths = {}
+        routers = set()
 
         for attr_name, attr_value in dct.items():
             if hasattr(attr_value, "__is_start_method__"):
@@ -137,18 +330,11 @@ class FlowMeta(type):
                 methods = attr_value.__trigger_methods__
                 condition_type = getattr(attr_value, "__condition_type__", "OR")
                 listeners[attr_name] = (condition_type, methods)
-
-            elif hasattr(attr_value, "__is_router__"):
-                routers[attr_value.__router_for__] = attr_name
-                possible_returns = get_possible_return_constants(attr_value)
-                if possible_returns:
-                    router_paths[attr_name] = possible_returns
-
-                # Register router as a listener to its triggering method
-                trigger_method_name = attr_value.__router_for__
-                methods = [trigger_method_name]
-                condition_type = "OR"
-                listeners[attr_name] = (condition_type, methods)
+                if hasattr(attr_value, "__is_router__") and attr_value.__is_router__:
+                    routers.add(attr_name)
+                    possible_returns = get_possible_return_constants(attr_value)
+                    if possible_returns:
+                        router_paths[attr_name] = possible_returns
 
         setattr(cls, "_start_methods", start_methods)
         setattr(cls, "_listeners", listeners)
@@ -163,7 +349,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
 
     _start_methods: List[str] = []
     _listeners: Dict[str, tuple[str, List[str]]] = {}
-    _routers: Dict[str, str] = {}
+    _routers: Set[str] = set()
     _router_paths: Dict[str, List[str]] = {}
     initial_state: Union[Type[T], T, None] = None
     event_emitter = Signal("event_emitter")
@@ -210,20 +396,10 @@ class Flow(Generic[T], metaclass=FlowMeta):
         return self._method_outputs
 
     def _initialize_state(self, inputs: Dict[str, Any]) -> None:
-        """
-        Initializes or updates the state with the provided inputs.
-
-        Args:
-            inputs: Dictionary of inputs to initialize or update the state.
-
-        Raises:
-            ValueError: If inputs do not match the structured state model.
-            TypeError: If state is neither a BaseModel instance nor a dictionary.
-        """
         if isinstance(self._state, BaseModel):
-            # Structured state management
+            # Structured state
             try:
-                # Define a function to create the dynamic class
+
                 def create_model_with_extra_forbid(
                     base_model: Type[BaseModel],
                 ) -> Type[BaseModel]:
@@ -233,34 +409,20 @@ class Flow(Generic[T], metaclass=FlowMeta):
 
                     return ModelWithExtraForbid
 
-                # Create the dynamic class
                 ModelWithExtraForbid = create_model_with_extra_forbid(
                     self._state.__class__
                 )
-
-                # Create a new instance using the combined state and inputs
                 self._state = cast(
                     T, ModelWithExtraForbid(**{**self._state.model_dump(), **inputs})
                 )
-
             except ValidationError as e:
                 raise ValueError(f"Invalid inputs for structured state: {e}") from e
         elif isinstance(self._state, dict):
-            # Unstructured state management
             self._state.update(inputs)
         else:
             raise TypeError("State must be a BaseModel instance or a dictionary.")
 
     def kickoff(self, inputs: Optional[Dict[str, Any]] = None) -> Any:
-        """
-        Starts the execution of the flow synchronously.
-
-        Args:
-            inputs: Optional dictionary of inputs to initialize or update the state.
-
-        Returns:
-            The final output from the flow execution.
-        """
         self.event_emitter.send(
             self,
             event=FlowStartedEvent(
@@ -274,15 +436,6 @@ class Flow(Generic[T], metaclass=FlowMeta):
         return asyncio.run(self.kickoff_async())
 
     async def kickoff_async(self, inputs: Optional[Dict[str, Any]] = None) -> Any:
-        """
-        Starts the execution of the flow asynchronously.
-
-        Args:
-            inputs: Optional dictionary of inputs to initialize or update the state.
-
-        Returns:
-            The final output from the flow execution.
-        """
         if not self._start_methods:
             raise ValueError("No start method defined")
 
@@ -290,16 +443,12 @@ class Flow(Generic[T], metaclass=FlowMeta):
             self.__class__.__name__, list(self._methods.keys())
         )
 
-        # Create tasks for all start methods
         tasks = [
             self._execute_start_method(start_method)
             for start_method in self._start_methods
         ]
-
-        # Run all start methods concurrently
         await asyncio.gather(*tasks)
 
-        # Determine the final output (from the last executed method)
         final_output = self._method_outputs[-1] if self._method_outputs else None
 
         self.event_emitter.send(
@@ -310,10 +459,26 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 result=final_output,
             ),
         )
-
         return final_output
 
     async def _execute_start_method(self, start_method_name: str) -> None:
+        """
+        Executes a flow's start method and its triggered listeners.
+
+        This internal method handles the execution of methods marked with @start
+        decorator and manages the subsequent chain of listener executions.
+
+        Parameters
+        ----------
+        start_method_name : str
+            The name of the start method to execute.
+
+        Notes
+        -----
+        - Executes the start method and captures its result
+        - Triggers execution of any listeners waiting on this start method
+        - Part of the flow's initialization sequence
+        """
         result = await self._execute_method(
             start_method_name, self._methods[start_method_name]
         )
@@ -327,51 +492,146 @@ class Flow(Generic[T], metaclass=FlowMeta):
             if asyncio.iscoroutinefunction(method)
             else method(*args, **kwargs)
         )
-        self._method_outputs.append(result)  # Store the output
-
-        # Track method execution counts
+        self._method_outputs.append(result)
         self._method_execution_counts[method_name] = (
             self._method_execution_counts.get(method_name, 0) + 1
         )
-
         return result
 
     async def _execute_listeners(self, trigger_method: str, result: Any) -> None:
-        listener_tasks = []
+        """
+        Executes all listeners and routers triggered by a method completion.
 
-        if trigger_method in self._routers:
-            router_method = self._methods[self._routers[trigger_method]]
-            path = await self._execute_method(
-                self._routers[trigger_method], router_method
+        This internal method manages the execution flow by:
+        1. First executing all triggered routers sequentially
+        2. Then executing all triggered listeners in parallel
+
+        Parameters
+        ----------
+        trigger_method : str
+            The name of the method that triggered these listeners.
+        result : Any
+            The result from the triggering method, passed to listeners
+            that accept parameters.
+
+        Notes
+        -----
+        - Routers are executed sequentially to maintain flow control
+        - Each router's result becomes the new trigger_method
+        - Normal listeners are executed in parallel for efficiency
+        - Listeners can receive the trigger method's result as a parameter
+        """
+        # First, handle routers repeatedly until no router triggers anymore
+        while True:
+            routers_triggered = self._find_triggered_methods(
+                trigger_method, router_only=True
             )
-            trigger_method = path
+            if not routers_triggered:
+                break
+            for router_name in routers_triggered:
+                await self._execute_single_listener(router_name, result)
+                # After executing router, the router's result is the path
+                # The last router executed sets the trigger_method
+                # The router result is the last element in self._method_outputs
+                trigger_method = self._method_outputs[-1]
 
+        # Now that no more routers are triggered by current trigger_method,
+        # execute normal listeners
+        listeners_triggered = self._find_triggered_methods(
+            trigger_method, router_only=False
+        )
+        if listeners_triggered:
+            tasks = [
+                self._execute_single_listener(listener_name, result)
+                for listener_name in listeners_triggered
+            ]
+            await asyncio.gather(*tasks)
+
+    def _find_triggered_methods(
+        self, trigger_method: str, router_only: bool
+    ) -> List[str]:
+        """
+        Finds all methods that should be triggered based on conditions.
+
+        This internal method evaluates both OR and AND conditions to determine
+        which methods should be executed next in the flow.
+
+        Parameters
+        ----------
+        trigger_method : str
+            The name of the method that just completed execution.
+        router_only : bool
+            If True, only consider router methods.
+            If False, only consider non-router methods.
+
+        Returns
+        -------
+        List[str]
+            Names of methods that should be triggered.
+
+        Notes
+        -----
+        - Handles both OR and AND conditions:
+          * OR: Triggers if any condition is met
+          * AND: Triggers only when all conditions are met
+        - Maintains state for AND conditions using _pending_and_listeners
+        - Separates router and normal listener evaluation
+        """
+        triggered = []
         for listener_name, (condition_type, methods) in self._listeners.items():
+            is_router = listener_name in self._routers
+
+            if router_only != is_router:
+                continue
+
             if condition_type == "OR":
+                # If the trigger_method matches any in methods, run this
                 if trigger_method in methods:
-                    # Schedule the listener without preventing re-execution
-                    listener_tasks.append(
-                        self._execute_single_listener(listener_name, result)
-                    )
+                    triggered.append(listener_name)
             elif condition_type == "AND":
                 # Initialize pending methods for this listener if not already done
                 if listener_name not in self._pending_and_listeners:
                     self._pending_and_listeners[listener_name] = set(methods)
                 # Remove the trigger method from pending methods
-                self._pending_and_listeners[listener_name].discard(trigger_method)
+                if trigger_method in self._pending_and_listeners[listener_name]:
+                    self._pending_and_listeners[listener_name].discard(trigger_method)
+
                 if not self._pending_and_listeners[listener_name]:
                     # All required methods have been executed
-                    listener_tasks.append(
-                        self._execute_single_listener(listener_name, result)
-                    )
+                    triggered.append(listener_name)
                     # Reset pending methods for this listener
                     self._pending_and_listeners.pop(listener_name, None)
 
-        # Run all listener tasks concurrently and wait for them to complete
-        if listener_tasks:
-            await asyncio.gather(*listener_tasks)
+        return triggered
 
     async def _execute_single_listener(self, listener_name: str, result: Any) -> None:
+        """
+        Executes a single listener method with proper event handling.
+
+        This internal method manages the execution of an individual listener,
+        including parameter inspection, event emission, and error handling.
+
+        Parameters
+        ----------
+        listener_name : str
+            The name of the listener method to execute.
+        result : Any
+            The result from the triggering method, which may be passed
+            to the listener if it accepts parameters.
+
+        Notes
+        -----
+        - Inspects method signature to determine if it accepts the trigger result
+        - Emits events for method execution start and finish
+        - Handles errors gracefully with detailed logging
+        - Recursively triggers listeners of this listener
+        - Supports both parameterized and parameter-less listeners
+
+        Error Handling
+        -------------
+        Catches and logs any exceptions during execution, preventing
+        individual listener failures from breaking the entire flow.
+        """
         try:
             method = self._methods[listener_name]
 
@@ -386,17 +646,13 @@ class Flow(Generic[T], metaclass=FlowMeta):
 
             sig = inspect.signature(method)
             params = list(sig.parameters.values())
-
-            # Exclude 'self' parameter
             method_params = [p for p in params if p.name != "self"]
 
             if method_params:
-                # If listener expects parameters, pass the result
                 listener_result = await self._execute_method(
                     listener_name, method, result
                 )
             else:
-                # If listener does not expect parameters, call without arguments
                 listener_result = await self._execute_method(listener_name, method)
 
             self.event_emitter.send(
@@ -408,8 +664,9 @@ class Flow(Generic[T], metaclass=FlowMeta):
                 ),
             )
 
-            # Execute listeners of this listener
+            # Execute listeners (and possibly routers) of this listener
             await self._execute_listeners(listener_name, listener_result)
+
         except Exception as e:
             print(
                 f"[Flow._execute_single_listener] Error in method {listener_name}: {e}"
@@ -422,5 +679,4 @@ class Flow(Generic[T], metaclass=FlowMeta):
         self._telemetry.flow_plotting_span(
             self.__class__.__name__, list(self._methods.keys())
         )
-
         plot_flow(self, filename)

src/crewai/flow/flow_visualizer.py

@@ -1,12 +1,14 @@
 # flow_visualizer.py
 
 import os
+from pathlib import Path
 
 from pyvis.network import Network
 
 from crewai.flow.config import COLORS, NODE_STYLES
 from crewai.flow.html_template_handler import HTMLTemplateHandler
 from crewai.flow.legend_generator import generate_legend_items_html, get_legend_items
+from crewai.flow.path_utils import safe_path_join, validate_path_exists
 from crewai.flow.utils import calculate_node_levels
 from crewai.flow.visualization_utils import (
     add_edges,
@@ -16,89 +18,209 @@ from crewai.flow.visualization_utils import (
 
 
 class FlowPlot:
+    """Handles the creation and rendering of flow visualization diagrams."""
+
     def __init__(self, flow):
+        """
+        Initialize FlowPlot with a flow object.
+
+        Parameters
+        ----------
+        flow : Flow
+            A Flow instance to visualize.
+
+        Raises
+        ------
+        ValueError
+            If flow object is invalid or missing required attributes.
+        """
+        if not hasattr(flow, '_methods'):
+            raise ValueError("Invalid flow object: missing '_methods' attribute")
+        if not hasattr(flow, '_listeners'):
+            raise ValueError("Invalid flow object: missing '_listeners' attribute")
+        if not hasattr(flow, '_start_methods'):
+            raise ValueError("Invalid flow object: missing '_start_methods' attribute")
+            
         self.flow = flow
         self.colors = COLORS
         self.node_styles = NODE_STYLES
 
     def plot(self, filename):
-        net = Network(
-            directed=True,
-            height="750px",
-            width="100%",
-            bgcolor=self.colors["bg"],
-            layout=None,
-        )
-
-        # Set options to disable physics
-        net.set_options(
-            """
-            var options = {
-                "nodes": {
-                    "font": {
-                        "multi": "html"
-                    }
-                },
-                "physics": {
-                    "enabled": false
-                }
-            }
         """
-        )
+        Generate and save an HTML visualization of the flow.
 
-        # Calculate levels for nodes
-        node_levels = calculate_node_levels(self.flow)
+        Parameters
+        ----------
+        filename : str
+            Name of the output file (without extension).
 
-        # Compute positions
-        node_positions = compute_positions(self.flow, node_levels)
+        Raises
+        ------
+        ValueError
+            If filename is invalid or network generation fails.
+        IOError
+            If file operations fail or visualization cannot be generated.
+        RuntimeError
+            If network visualization generation fails.
+        """
+        if not filename or not isinstance(filename, str):
+            raise ValueError("Filename must be a non-empty string")
+            
+        try:
+            # Initialize network
+            net = Network(
+                directed=True,
+                height="750px",
+                width="100%",
+                bgcolor=self.colors["bg"],
+                layout=None,
+            )
 
-        # Add nodes to the network
-        add_nodes_to_network(net, self.flow, node_positions, self.node_styles)
+            # Set options to disable physics
+            net.set_options(
+                """
+                var options = {
+                    "nodes": {
+                        "font": {
+                            "multi": "html"
+                        }
+                    },
+                    "physics": {
+                        "enabled": false
+                    }
+                }
+            """
+            )
 
-        # Add edges to the network
-        add_edges(net, self.flow, node_positions, self.colors)
+            # Calculate levels for nodes
+            try:
+                node_levels = calculate_node_levels(self.flow)
+            except Exception as e:
+                raise ValueError(f"Failed to calculate node levels: {str(e)}")
 
-        network_html = net.generate_html()
-        final_html_content = self._generate_final_html(network_html)
+            # Compute positions
+            try:
+                node_positions = compute_positions(self.flow, node_levels)
+            except Exception as e:
+                raise ValueError(f"Failed to compute node positions: {str(e)}")
 
-        # Save the final HTML content to the file
-        with open(f"{filename}.html", "w", encoding="utf-8") as f:
-            f.write(final_html_content)
-        print(f"Plot saved as {filename}.html")
+            # Add nodes to the network
+            try:
+                add_nodes_to_network(net, self.flow, node_positions, self.node_styles)
+            except Exception as e:
+                raise RuntimeError(f"Failed to add nodes to network: {str(e)}")
 
-        self._cleanup_pyvis_lib()
+            # Add edges to the network
+            try:
+                add_edges(net, self.flow, node_positions, self.colors)
+            except Exception as e:
+                raise RuntimeError(f"Failed to add edges to network: {str(e)}")
+
+            # Generate HTML
+            try:
+                network_html = net.generate_html()
+                final_html_content = self._generate_final_html(network_html)
+            except Exception as e:
+                raise RuntimeError(f"Failed to generate network visualization: {str(e)}")
+
+            # Save the final HTML content to the file
+            try:
+                with open(f"{filename}.html", "w", encoding="utf-8") as f:
+                    f.write(final_html_content)
+                print(f"Plot saved as {filename}.html")
+            except IOError as e:
+                raise IOError(f"Failed to save flow visualization to {filename}.html: {str(e)}")
+
+        except (ValueError, RuntimeError, IOError) as e:
+            raise e
+        except Exception as e:
+            raise RuntimeError(f"Unexpected error during flow visualization: {str(e)}")
+        finally:
+            self._cleanup_pyvis_lib()
 
     def _generate_final_html(self, network_html):
-        # Extract just the body content from the generated HTML
-        current_dir = os.path.dirname(__file__)
-        template_path = os.path.join(
-            current_dir, "assets", "crewai_flow_visual_template.html"
-        )
-        logo_path = os.path.join(current_dir, "assets", "crewai_logo.svg")
+        """
+        Generate the final HTML content with network visualization and legend.
 
-        html_handler = HTMLTemplateHandler(template_path, logo_path)
-        network_body = html_handler.extract_body_content(network_html)
+        Parameters
+        ----------
+        network_html : str
+            HTML content generated by pyvis Network.
 
-        # Generate the legend items HTML
-        legend_items = get_legend_items(self.colors)
-        legend_items_html = generate_legend_items_html(legend_items)
-        final_html_content = html_handler.generate_final_html(
-            network_body, legend_items_html
-        )
-        return final_html_content
+        Returns
+        -------
+        str
+            Complete HTML content with styling and legend.
+
+        Raises
+        ------
+        IOError
+            If template or logo files cannot be accessed.
+        ValueError
+            If network_html is invalid.
+        """
+        if not network_html:
+            raise ValueError("Invalid network HTML content")
+
+        try:
+            # Extract just the body content from the generated HTML
+            current_dir = os.path.dirname(__file__)
+            template_path = safe_path_join("assets", "crewai_flow_visual_template.html", root=current_dir)
+            logo_path = safe_path_join("assets", "crewai_logo.svg", root=current_dir)
+
+            if not os.path.exists(template_path):
+                raise IOError(f"Template file not found: {template_path}")
+            if not os.path.exists(logo_path):
+                raise IOError(f"Logo file not found: {logo_path}")
+
+            html_handler = HTMLTemplateHandler(template_path, logo_path)
+            network_body = html_handler.extract_body_content(network_html)
+
+            # Generate the legend items HTML
+            legend_items = get_legend_items(self.colors)
+            legend_items_html = generate_legend_items_html(legend_items)
+            final_html_content = html_handler.generate_final_html(
+                network_body, legend_items_html
+            )
+            return final_html_content
+        except Exception as e:
+            raise IOError(f"Failed to generate visualization HTML: {str(e)}")
 
     def _cleanup_pyvis_lib(self):
-        # Clean up the generated lib folder
-        lib_folder = os.path.join(os.getcwd(), "lib")
+        """
+        Clean up the generated lib folder from pyvis.
+
+        This method safely removes the temporary lib directory created by pyvis
+        during network visualization generation.
+        """
         try:
+            lib_folder = safe_path_join("lib", root=os.getcwd())
             if os.path.exists(lib_folder) and os.path.isdir(lib_folder):
                 import shutil
-
                 shutil.rmtree(lib_folder)
+        except ValueError as e:
+            print(f"Error validating lib folder path: {e}")
         except Exception as e:
-            print(f"Error cleaning up {lib_folder}: {e}")
+            print(f"Error cleaning up lib folder: {e}")
 
 
 def plot_flow(flow, filename="flow_plot"):
+    """
+    Convenience function to create and save a flow visualization.
+
+    Parameters
+    ----------
+    flow : Flow
+        Flow instance to visualize.
+    filename : str, optional
+        Output filename without extension, by default "flow_plot".
+
+    Raises
+    ------
+    ValueError
+        If flow object or filename is invalid.
+    IOError
+        If file operations fail.
+    """
     visualizer = FlowPlot(flow)
     visualizer.plot(filename)

src/crewai/flow/html_template_handler.py

@@ -1,26 +1,53 @@
 import base64
 import re
+from pathlib import Path
+
+from crewai.flow.path_utils import safe_path_join, validate_path_exists
 
 
 class HTMLTemplateHandler:
+    """Handles HTML template processing and generation for flow visualization diagrams."""
+
     def __init__(self, template_path, logo_path):
-        self.template_path = template_path
-        self.logo_path = logo_path
+        """
+        Initialize HTMLTemplateHandler with validated template and logo paths.
+
+        Parameters
+        ----------
+        template_path : str
+            Path to the HTML template file.
+        logo_path : str
+            Path to the logo image file.
+
+        Raises
+        ------
+        ValueError
+            If template or logo paths are invalid or files don't exist.
+        """
+        try:
+            self.template_path = validate_path_exists(template_path, "file")
+            self.logo_path = validate_path_exists(logo_path, "file")
+        except ValueError as e:
+            raise ValueError(f"Invalid template or logo path: {e}")
 
     def read_template(self):
+        """Read and return the HTML template file contents."""
         with open(self.template_path, "r", encoding="utf-8") as f:
             return f.read()
 
     def encode_logo(self):
+        """Convert the logo SVG file to base64 encoded string."""
         with open(self.logo_path, "rb") as logo_file:
             logo_svg_data = logo_file.read()
             return base64.b64encode(logo_svg_data).decode("utf-8")
 
     def extract_body_content(self, html):
+        """Extract and return content between body tags from HTML string."""
         match = re.search("<body.*?>(.*?)</body>", html, re.DOTALL)
         return match.group(1) if match else ""
 
     def generate_legend_items_html(self, legend_items):
+        """Generate HTML markup for the legend items."""
         legend_items_html = ""
         for item in legend_items:
             if "border" in item:
@@ -48,6 +75,7 @@ class HTMLTemplateHandler:
         return legend_items_html
 
     def generate_final_html(self, network_body, legend_items_html, title="Flow Plot"):
+        """Combine all components into final HTML document with network visualization."""
         html_template = self.read_template()
         logo_svg_base64 = self.encode_logo()
 

src/crewai/flow/legend_generator.py

@@ -1,3 +1,4 @@
+
 def get_legend_items(colors):
     return [
         {"label": "Start Method", "color": colors["start"]},

src/crewai/flow/path_utils.py

@@ -0,0 +1,135 @@
+"""
+Path utilities for secure file operations in CrewAI flow module.
+
+This module provides utilities for secure path handling to prevent directory
+traversal attacks and ensure paths remain within allowed boundaries.
+"""
+
+import os
+from pathlib import Path
+from typing import List, Union
+
+
+def safe_path_join(*parts: str, root: Union[str, Path, None] = None) -> str:
+    """
+    Safely join path components and ensure the result is within allowed boundaries.
+
+    Parameters
+    ----------
+    *parts : str
+        Variable number of path components to join.
+    root : Union[str, Path, None], optional
+        Root directory to use as base. If None, uses current working directory.
+
+    Returns
+    -------
+    str
+        String representation of the resolved path.
+
+    Raises
+    ------
+    ValueError
+        If the resulting path would be outside the root directory
+        or if any path component is invalid.
+    """
+    if not parts:
+        raise ValueError("No path components provided")
+
+    try:
+        # Convert all parts to strings and clean them
+        clean_parts = [str(part).strip() for part in parts if part]
+        if not clean_parts:
+            raise ValueError("No valid path components provided")
+
+        # Establish root directory
+        root_path = Path(root).resolve() if root else Path.cwd()
+        
+        # Join and resolve the full path
+        full_path = Path(root_path, *clean_parts).resolve()
+        
+        # Check if the resolved path is within root
+        if not str(full_path).startswith(str(root_path)):
+            raise ValueError(
+                f"Invalid path: Potential directory traversal. Path must be within {root_path}"
+            )
+            
+        return str(full_path)
+        
+    except Exception as e:
+        if isinstance(e, ValueError):
+            raise
+        raise ValueError(f"Invalid path components: {str(e)}")
+
+
+def validate_path_exists(path: Union[str, Path], file_type: str = "file") -> str:
+    """
+    Validate that a path exists and is of the expected type.
+
+    Parameters
+    ----------
+    path : Union[str, Path]
+        Path to validate.
+    file_type : str, optional
+        Expected type ('file' or 'directory'), by default 'file'.
+
+    Returns
+    -------
+    str
+        Validated path as string.
+
+    Raises
+    ------
+    ValueError
+        If path doesn't exist or is not of expected type.
+    """
+    try:
+        path_obj = Path(path).resolve()
+        
+        if not path_obj.exists():
+            raise ValueError(f"Path does not exist: {path}")
+            
+        if file_type == "file" and not path_obj.is_file():
+            raise ValueError(f"Path is not a file: {path}")
+        elif file_type == "directory" and not path_obj.is_dir():
+            raise ValueError(f"Path is not a directory: {path}")
+            
+        return str(path_obj)
+        
+    except Exception as e:
+        if isinstance(e, ValueError):
+            raise
+        raise ValueError(f"Invalid path: {str(e)}")
+
+
+def list_files(directory: Union[str, Path], pattern: str = "*") -> List[str]:
+    """
+    Safely list files in a directory matching a pattern.
+
+    Parameters
+    ----------
+    directory : Union[str, Path]
+        Directory to search in.
+    pattern : str, optional
+        Glob pattern to match files against, by default "*".
+
+    Returns
+    -------
+    List[str]
+        List of matching file paths.
+
+    Raises
+    ------
+    ValueError
+        If directory is invalid or inaccessible.
+    """
+    try:
+        dir_path = Path(directory).resolve()
+        if not dir_path.is_dir():
+            raise ValueError(f"Not a directory: {directory}")
+            
+        return [str(p) for p in dir_path.glob(pattern) if p.is_file()]
+        
+    except Exception as e:
+        if isinstance(e, ValueError):
+            raise
+        raise ValueError(f"Error listing files: {str(e)}")

src/crewai/flow/utils.py

@@ -1,9 +1,25 @@
+"""
+Utility functions for flow visualization and dependency analysis.
+
+This module provides core functionality for analyzing and manipulating flow structures,
+including node level calculation, ancestor tracking, and return value analysis.
+Functions in this module are primarily used by the visualization system to create
+accurate and informative flow diagrams.
+
+Example
+-------
+>>> flow = Flow()
+>>> node_levels = calculate_node_levels(flow)
+>>> ancestors = build_ancestor_dict(flow)
+"""
+
 import ast
 import inspect
 import textwrap
+from typing import Any, Dict, List, Optional, Set, Union
 
 
-def get_possible_return_constants(function):
+def get_possible_return_constants(function: Any) -> Optional[List[str]]:
     try:
         source = inspect.getsource(function)
     except OSError:
@@ -31,23 +47,80 @@ def get_possible_return_constants(function):
         print(f"Source code:\n{source}")
         return None
 
-    return_values = []
+    return_values = set()
+    dict_definitions = {}
+
+    class DictionaryAssignmentVisitor(ast.NodeVisitor):
+        def visit_Assign(self, node):
+            # Check if this assignment is assigning a dictionary literal to a variable
+            if isinstance(node.value, ast.Dict) and len(node.targets) == 1:
+                target = node.targets[0]
+                if isinstance(target, ast.Name):
+                    var_name = target.id
+                    dict_values = []
+                    # Extract string values from the dictionary
+                    for val in node.value.values:
+                        if isinstance(val, ast.Constant) and isinstance(val.value, str):
+                            dict_values.append(val.value)
+                        # If non-string, skip or just ignore
+                    if dict_values:
+                        dict_definitions[var_name] = dict_values
+            self.generic_visit(node)
 
     class ReturnVisitor(ast.NodeVisitor):
         def visit_Return(self, node):
-            # Check if the return value is a constant (Python 3.8+)
-            if isinstance(node.value, ast.Constant):
-                return_values.append(node.value.value)
+            # Direct string return
+            if isinstance(node.value, ast.Constant) and isinstance(
+                node.value.value, str
+            ):
+                return_values.add(node.value.value)
+            # Dictionary-based return, like return paths[result]
+            elif isinstance(node.value, ast.Subscript):
+                # Check if we're subscripting a known dictionary variable
+                if isinstance(node.value.value, ast.Name):
+                    var_name = node.value.value.id
+                    if var_name in dict_definitions:
+                        # Add all possible dictionary values
+                        for v in dict_definitions[var_name]:
+                            return_values.add(v)
+            self.generic_visit(node)
 
+    # First pass: identify dictionary assignments
+    DictionaryAssignmentVisitor().visit(code_ast)
+    # Second pass: identify returns
     ReturnVisitor().visit(code_ast)
-    return return_values
+
+    return list(return_values) if return_values else None
 
 
-def calculate_node_levels(flow):
-    levels = {}
-    queue = []
-    visited = set()
-    pending_and_listeners = {}
+def calculate_node_levels(flow: Any) -> Dict[str, int]:
+    """
+    Calculate the hierarchical level of each node in the flow.
+
+    Performs a breadth-first traversal of the flow graph to assign levels
+    to nodes, starting with start methods at level 0.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance containing methods, listeners, and router configurations.
+
+    Returns
+    -------
+    Dict[str, int]
+        Dictionary mapping method names to their hierarchical levels.
+
+    Notes
+    -----
+    - Start methods are assigned level 0
+    - Each subsequent connected node is assigned level = parent_level + 1
+    - Handles both OR and AND conditions for listeners
+    - Processes router paths separately
+    """
+    levels: Dict[str, int] = {}
+    queue: List[str] = []
+    visited: Set[str] = set()
+    pending_and_listeners: Dict[str, Set[str]] = {}
 
     # Make all start methods at level 0
     for method_name, method in flow._methods.items():
@@ -61,10 +134,7 @@ def calculate_node_levels(flow):
         current_level = levels[current]
         visited.add(current)
 
-        for listener_name, (
-            condition_type,
-            trigger_methods,
-        ) in flow._listeners.items():
+        for listener_name, (condition_type, trigger_methods) in flow._listeners.items():
             if condition_type == "OR":
                 if current in trigger_methods:
                     if (
@@ -89,7 +159,7 @@ def calculate_node_levels(flow):
                             queue.append(listener_name)
 
         # Handle router connections
-        if current in flow._routers.values():
+        if current in flow._routers:
             router_method_name = current
             paths = flow._router_paths.get(router_method_name, [])
             for path in paths:
@@ -105,10 +175,24 @@ def calculate_node_levels(flow):
                             levels[listener_name] = current_level + 1
                             if listener_name not in visited:
                                 queue.append(listener_name)
+
     return levels
 
 
-def count_outgoing_edges(flow):
+def count_outgoing_edges(flow: Any) -> Dict[str, int]:
+    """
+    Count the number of outgoing edges for each method in the flow.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to analyze.
+
+    Returns
+    -------
+    Dict[str, int]
+        Dictionary mapping method names to their outgoing edge count.
+    """
     counts = {}
     for method_name in flow._methods:
         counts[method_name] = 0
@@ -120,16 +204,53 @@ def count_outgoing_edges(flow):
     return counts
 
 
-def build_ancestor_dict(flow):
-    ancestors = {node: set() for node in flow._methods}
-    visited = set()
+def build_ancestor_dict(flow: Any) -> Dict[str, Set[str]]:
+    """
+    Build a dictionary mapping each node to its ancestor nodes.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to analyze.
+
+    Returns
+    -------
+    Dict[str, Set[str]]
+        Dictionary mapping each node to a set of its ancestor nodes.
+    """
+    ancestors: Dict[str, Set[str]] = {node: set() for node in flow._methods}
+    visited: Set[str] = set()
     for node in flow._methods:
         if node not in visited:
             dfs_ancestors(node, ancestors, visited, flow)
     return ancestors
 
 
-def dfs_ancestors(node, ancestors, visited, flow):
+def dfs_ancestors(
+    node: str,
+    ancestors: Dict[str, Set[str]],
+    visited: Set[str],
+    flow: Any
+) -> None:
+    """
+    Perform depth-first search to build ancestor relationships.
+
+    Parameters
+    ----------
+    node : str
+        Current node being processed.
+    ancestors : Dict[str, Set[str]]
+        Dictionary tracking ancestor relationships.
+    visited : Set[str]
+        Set of already visited nodes.
+    flow : Any
+        The flow instance being analyzed.
+
+    Notes
+    -----
+    This function modifies the ancestors dictionary in-place to build
+    the complete ancestor graph.
+    """
     if node in visited:
         return
     visited.add(node)
@@ -142,7 +263,7 @@ def dfs_ancestors(node, ancestors, visited, flow):
             dfs_ancestors(listener_name, ancestors, visited, flow)
 
     # Handle router methods separately
-    if node in flow._routers.values():
+    if node in flow._routers:
         router_method_name = node
         paths = flow._router_paths.get(router_method_name, [])
         for path in paths:
@@ -153,12 +274,48 @@ def dfs_ancestors(node, ancestors, visited, flow):
                     dfs_ancestors(listener_name, ancestors, visited, flow)
 
 
-def is_ancestor(node, ancestor_candidate, ancestors):
+def is_ancestor(node: str, ancestor_candidate: str, ancestors: Dict[str, Set[str]]) -> bool:
+    """
+    Check if one node is an ancestor of another.
+
+    Parameters
+    ----------
+    node : str
+        The node to check ancestors for.
+    ancestor_candidate : str
+        The potential ancestor node.
+    ancestors : Dict[str, Set[str]]
+        Dictionary containing ancestor relationships.
+
+    Returns
+    -------
+    bool
+        True if ancestor_candidate is an ancestor of node, False otherwise.
+    """
     return ancestor_candidate in ancestors.get(node, set())
 
 
-def build_parent_children_dict(flow):
-    parent_children = {}
+def build_parent_children_dict(flow: Any) -> Dict[str, List[str]]:
+    """
+    Build a dictionary mapping parent nodes to their children.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to analyze.
+
+    Returns
+    -------
+    Dict[str, List[str]]
+        Dictionary mapping parent method names to lists of their child method names.
+
+    Notes
+    -----
+    - Maps listeners to their trigger methods
+    - Maps router methods to their paths and listeners
+    - Children lists are sorted for consistent ordering
+    """
+    parent_children: Dict[str, List[str]] = {}
 
     # Map listeners to their trigger methods
     for listener_name, (_, trigger_methods) in flow._listeners.items():
@@ -182,7 +339,24 @@ def build_parent_children_dict(flow):
     return parent_children
 
 
-def get_child_index(parent, child, parent_children):
+def get_child_index(parent: str, child: str, parent_children: Dict[str, List[str]]) -> int:
+    """
+    Get the index of a child node in its parent's sorted children list.
+
+    Parameters
+    ----------
+    parent : str
+        The parent node name.
+    child : str
+        The child node name to find the index for.
+    parent_children : Dict[str, List[str]]
+        Dictionary mapping parents to their children lists.
+
+    Returns
+    -------
+    int
+        Zero-based index of the child in its parent's sorted children list.
+    """
     children = parent_children.get(parent, [])
     children.sort()
     return children.index(child)

src/crewai/flow/visualization_utils.py

@@ -1,5 +1,23 @@
+"""
+Utilities for creating visual representations of flow structures.
+
+This module provides functions for generating network visualizations of flows,
+including node placement, edge creation, and visual styling. It handles the
+conversion of flow structures into visual network graphs with appropriate
+styling and layout.
+
+Example
+-------
+>>> flow = Flow()
+>>> net = Network(directed=True)
+>>> node_positions = compute_positions(flow, node_levels)
+>>> add_nodes_to_network(net, flow, node_positions, node_styles)
+>>> add_edges(net, flow, node_positions, colors)
+"""
+
 import ast
 import inspect
+from typing import Any, Dict, List, Optional, Tuple, Union
 
 from .utils import (
     build_ancestor_dict,
@@ -9,8 +27,25 @@ from .utils import (
 )
 
 
-def method_calls_crew(method):
-    """Check if the method calls `.crew()`."""
+def method_calls_crew(method: Any) -> bool:
+    """
+    Check if the method contains a call to `.crew()`.
+
+    Parameters
+    ----------
+    method : Any
+        The method to analyze for crew() calls.
+
+    Returns
+    -------
+    bool
+        True if the method calls .crew(), False otherwise.
+
+    Notes
+    -----
+    Uses AST analysis to detect method calls, specifically looking for
+    attribute access of 'crew'.
+    """
     try:
         source = inspect.getsource(method)
         source = inspect.cleandoc(source)
@@ -20,6 +55,7 @@ def method_calls_crew(method):
         return False
 
     class CrewCallVisitor(ast.NodeVisitor):
+        """AST visitor to detect .crew() method calls."""
         def __init__(self):
             self.found = False
 
@@ -34,7 +70,34 @@ def method_calls_crew(method):
     return visitor.found
 
 
-def add_nodes_to_network(net, flow, node_positions, node_styles):
+def add_nodes_to_network(
+    net: Any,
+    flow: Any,
+    node_positions: Dict[str, Tuple[float, float]],
+    node_styles: Dict[str, Dict[str, Any]]
+) -> None:
+    """
+    Add nodes to the network visualization with appropriate styling.
+
+    Parameters
+    ----------
+    net : Any
+        The pyvis Network instance to add nodes to.
+    flow : Any
+        The flow instance containing method information.
+    node_positions : Dict[str, Tuple[float, float]]
+        Dictionary mapping node names to their (x, y) positions.
+    node_styles : Dict[str, Dict[str, Any]]
+        Dictionary containing style configurations for different node types.
+
+    Notes
+    -----
+    Node types include:
+    - Start methods
+    - Router methods
+    - Crew methods
+    - Regular methods
+    """
     def human_friendly_label(method_name):
         return method_name.replace("_", " ").title()
 
@@ -73,9 +136,33 @@ def add_nodes_to_network(net, flow, node_positions, node_styles):
         )
 
 
-def compute_positions(flow, node_levels, y_spacing=150, x_spacing=150):
-    level_nodes = {}
-    node_positions = {}
+def compute_positions(
+    flow: Any,
+    node_levels: Dict[str, int],
+    y_spacing: float = 150,
+    x_spacing: float = 150
+) -> Dict[str, Tuple[float, float]]:
+    """
+    Compute the (x, y) positions for each node in the flow graph.
+
+    Parameters
+    ----------
+    flow : Any
+        The flow instance to compute positions for.
+    node_levels : Dict[str, int]
+        Dictionary mapping node names to their hierarchical levels.
+    y_spacing : float, optional
+        Vertical spacing between levels, by default 150.
+    x_spacing : float, optional
+        Horizontal spacing between nodes, by default 150.
+
+    Returns
+    -------
+    Dict[str, Tuple[float, float]]
+        Dictionary mapping node names to their (x, y) coordinates.
+    """
+    level_nodes: Dict[int, List[str]] = {}
+    node_positions: Dict[str, Tuple[float, float]] = {}
 
     for method_name, level in node_levels.items():
         level_nodes.setdefault(level, []).append(method_name)
@@ -90,16 +177,44 @@ def compute_positions(flow, node_levels, y_spacing=150, x_spacing=150):
     return node_positions
 
 
-def add_edges(net, flow, node_positions, colors):
+def add_edges(
+    net: Any,
+    flow: Any,
+    node_positions: Dict[str, Tuple[float, float]],
+    colors: Dict[str, str]
+) -> None:
+    edge_smooth: Dict[str, Union[str, float]] = {"type": "continuous"}  # Default value
+    """
+    Add edges to the network visualization with appropriate styling.
+
+    Parameters
+    ----------
+    net : Any
+        The pyvis Network instance to add edges to.
+    flow : Any
+        The flow instance containing edge information.
+    node_positions : Dict[str, Tuple[float, float]]
+        Dictionary mapping node names to their positions.
+    colors : Dict[str, str]
+        Dictionary mapping edge types to their colors.
+
+    Notes
+    -----
+    - Handles both normal listener edges and router edges
+    - Applies appropriate styling (color, dashes) based on edge type
+    - Adds curvature to edges when needed (cycles or multiple children)
+    """
     ancestors = build_ancestor_dict(flow)
     parent_children = build_parent_children_dict(flow)
 
+    # Edges for normal listeners
     for method_name in flow._listeners:
         condition_type, trigger_methods = flow._listeners[method_name]
         is_and_condition = condition_type == "AND"
 
         for trigger in trigger_methods:
-            if trigger in flow._methods or trigger in flow._routers.values():
+            # Check if nodes exist before adding edges
+            if trigger in node_positions and method_name in node_positions:
                 is_router_edge = any(
                     trigger in paths for paths in flow._router_paths.values()
                 )
@@ -124,7 +239,7 @@ def add_edges(net, flow, node_positions, colors):
                     else:
                         edge_smooth = {"type": "cubicBezier"}
                 else:
-                    edge_smooth = False
+                    edge_smooth.update({"type": "continuous"})
 
                 edge_style = {
                     "color": edge_color,
@@ -135,7 +250,22 @@ def add_edges(net, flow, node_positions, colors):
                 }
 
                 net.add_edge(trigger, method_name, **edge_style)
+            else:
+                # Nodes not found in node_positions. Check if it's a known router outcome and a known method.
+                is_router_edge = any(
+                    trigger in paths for paths in flow._router_paths.values()
+                )
+                # Check if method_name is a known method
+                method_known = method_name in flow._methods
 
+                # If it's a known router edge and the method is known, don't warn.
+                # This means the path is legitimate, just not reflected as nodes here.
+                if not (is_router_edge and method_known):
+                    print(
+                        f"Warning: No node found for '{trigger}' or '{method_name}'. Skipping edge."
+                    )
+
+    # Edges for router return paths
     for router_method_name, paths in flow._router_paths.items():
         for path in paths:
             for listener_name, (
@@ -143,36 +273,49 @@ def add_edges(net, flow, node_positions, colors):
                 trigger_methods,
             ) in flow._listeners.items():
                 if path in trigger_methods:
-                    is_cycle_edge = is_ancestor(trigger, method_name, ancestors)
-                    parent_has_multiple_children = (
-                        len(parent_children.get(router_method_name, [])) > 1
-                    )
-                    needs_curvature = is_cycle_edge or parent_has_multiple_children
+                    if (
+                        router_method_name in node_positions
+                        and listener_name in node_positions
+                    ):
+                        is_cycle_edge = is_ancestor(
+                            router_method_name, listener_name, ancestors
+                        )
+                        parent_has_multiple_children = (
+                            len(parent_children.get(router_method_name, [])) > 1
+                        )
+                        needs_curvature = is_cycle_edge or parent_has_multiple_children
 
-                    if needs_curvature:
-                        source_pos = node_positions.get(router_method_name)
-                        target_pos = node_positions.get(listener_name)
+                        if needs_curvature:
+                            source_pos = node_positions.get(router_method_name)
+                            target_pos = node_positions.get(listener_name)
 
-                        if source_pos and target_pos:
-                            dx = target_pos[0] - source_pos[0]
-                            smooth_type = "curvedCCW" if dx <= 0 else "curvedCW"
-                            index = get_child_index(
-                                router_method_name, listener_name, parent_children
-                            )
-                            edge_smooth = {
-                                "type": smooth_type,
-                                "roundness": 0.2 + (0.1 * index),
-                            }
+                            if source_pos and target_pos:
+                                dx = target_pos[0] - source_pos[0]
+                                smooth_type = "curvedCCW" if dx <= 0 else "curvedCW"
+                                index = get_child_index(
+                                    router_method_name, listener_name, parent_children
+                                )
+                                edge_smooth = {
+                                    "type": smooth_type,
+                                    "roundness": 0.2 + (0.1 * index),
+                                }
+                            else:
+                                edge_smooth = {"type": "cubicBezier"}
                         else:
-                            edge_smooth = {"type": "cubicBezier"}
-                    else:
-                        edge_smooth = False
+                            edge_smooth.update({"type": "continuous"})
 
-                    edge_style = {
-                        "color": colors["router_edge"],
-                        "width": 2,
-                        "arrows": "to",
-                        "dashes": True,
-                        "smooth": edge_smooth,
-                    }
-                    net.add_edge(router_method_name, listener_name, **edge_style)
+                        edge_style = {
+                            "color": colors["router_edge"],
+                            "width": 2,
+                            "arrows": "to",
+                            "dashes": True,
+                            "smooth": edge_smooth,
+                        }
+                        net.add_edge(router_method_name, listener_name, **edge_style)
+                    else:
+                        # Same check here: known router edge and known method?
+                        method_known = listener_name in flow._methods
+                        if not method_known:
+                            print(
+                                f"Warning: No node found for '{router_method_name}' or '{listener_name}'. Skipping edge."
+                            )

src/crewai/knowledge/knowledge.py

@@ -14,13 +14,13 @@ class Knowledge(BaseModel):
     Knowledge is a collection of sources and setup for the vector store to save and query relevant context.
     Args:
         sources: List[BaseKnowledgeSource] = Field(default_factory=list)
-        storage: KnowledgeStorage = Field(default_factory=KnowledgeStorage)
+        storage: Optional[KnowledgeStorage] = Field(default=None)
         embedder_config: Optional[Dict[str, Any]] = None
     """
 
     sources: List[BaseKnowledgeSource] = Field(default_factory=list)
     model_config = ConfigDict(arbitrary_types_allowed=True)
-    storage: KnowledgeStorage = Field(default_factory=KnowledgeStorage)
+    storage: Optional[KnowledgeStorage] = Field(default=None)
     embedder_config: Optional[Dict[str, Any]] = None
     collection_name: Optional[str] = None
 
@@ -49,8 +49,13 @@ class Knowledge(BaseModel):
         """
         Query across all knowledge sources to find the most relevant information.
         Returns the top_k most relevant chunks.
+        
+        Raises:
+            ValueError: If storage is not initialized.
         """
-
+        if self.storage is None:
+            raise ValueError("Storage is not initialized.")
+            
         results = self.storage.search(
             query,
             limit,

src/crewai/knowledge/source/base_file_knowledge_source.py

@@ -22,13 +22,14 @@ class BaseFileKnowledgeSource(BaseKnowledgeSource, ABC):
         default_factory=list, description="The path to the file"
     )
     content: Dict[Path, str] = Field(init=False, default_factory=dict)
-    storage: KnowledgeStorage = Field(default_factory=KnowledgeStorage)
+    storage: Optional[KnowledgeStorage] = Field(default=None)
     safe_file_paths: List[Path] = Field(default_factory=list)
 
     @field_validator("file_path", "file_paths", mode="before")
-    def validate_file_path(cls, v, values):
+    def validate_file_path(cls, v, info):
         """Validate that at least one of file_path or file_paths is provided."""
-        if v is None and ("file_path" not in values or values.get("file_path") is None):
+        # Single check if both are None, O(1) instead of nested conditions
+        if v is None and info.data.get("file_path" if info.field_name == "file_paths" else "file_paths") is None:
             raise ValueError("Either file_path or file_paths must be provided")
         return v
 
@@ -62,7 +63,10 @@ class BaseFileKnowledgeSource(BaseKnowledgeSource, ABC):
 
     def _save_documents(self):
         """Save the documents to the storage."""
-        self.storage.save(self.chunks)
+        if self.storage:
+            self.storage.save(self.chunks)
+        else:
+            raise ValueError("No storage found to save documents.")
 
     def convert_to_path(self, path: Union[Path, str]) -> Path:
         """Convert a path to a Path object."""
@@ -71,28 +75,29 @@ class BaseFileKnowledgeSource(BaseKnowledgeSource, ABC):
     def _process_file_paths(self) -> List[Path]:
         """Convert file_path to a list of Path objects."""
 
-        # Check if old file_path is being used
         if hasattr(self, "file_path") and self.file_path is not None:
             self._logger.log(
                 "warning",
                 "The 'file_path' attribute is deprecated and will be removed in a future version. Please use 'file_paths' instead.",
                 color="yellow",
             )
-            paths = (
-                [self.file_path]
-                if isinstance(self.file_path, (str, Path))
-                else self.file_path
-            )
-        else:
-            if self.file_paths is None:
-                raise ValueError("Your source must be provided with a file_paths: []")
-            elif isinstance(self.file_paths, list) and len(self.file_paths) == 0:
-                raise ValueError("Empty file_paths are not allowed")
-            else:
-                paths = (
-                    [self.file_paths]
-                    if isinstance(self.file_paths, (str, Path))
-                    else self.file_paths
-                )
+            self.file_paths = self.file_path
 
-        return [self.convert_to_path(path) for path in paths]
+        if self.file_paths is None:
+            raise ValueError("Your source must be provided with a file_paths: []")
+
+        # Convert single path to list
+        path_list: List[Union[Path, str]] = (
+            [self.file_paths]
+            if isinstance(self.file_paths, (str, Path))
+            else list(self.file_paths)
+            if isinstance(self.file_paths, list)
+            else []
+        )
+
+        if not path_list:
+            raise ValueError(
+                "file_path/file_paths must be a Path, str, or a list of these types"
+            )
+
+        return [self.convert_to_path(path) for path in path_list]

src/crewai/knowledge/source/base_knowledge_source.py

@@ -16,7 +16,7 @@ class BaseKnowledgeSource(BaseModel, ABC):
     chunk_embeddings: List[np.ndarray] = Field(default_factory=list)
 
     model_config = ConfigDict(arbitrary_types_allowed=True)
-    storage: KnowledgeStorage = Field(default_factory=KnowledgeStorage)
+    storage: Optional[KnowledgeStorage] = Field(default=None)
     metadata: Dict[str, Any] = Field(default_factory=dict)  # Currently unused
     collection_name: Optional[str] = Field(default=None)
 
@@ -46,4 +46,7 @@ class BaseKnowledgeSource(BaseModel, ABC):
         Save the documents to the storage.
         This method should be called after the chunks and embeddings are generated.
         """
-        self.storage.save(self.chunks)
+        if self.storage:
+            self.storage.save(self.chunks)
+        else:
+            raise ValueError("No storage found to save documents.")

src/crewai/llm.py

@@ -7,11 +7,15 @@ import warnings
 from contextlib import contextmanager
 from typing import Any, Dict, List, Optional, Union, cast
 
-# Load environment variables from .env file
-import litellm
 from dotenv import load_dotenv
-from litellm import Choices, get_supported_openai_params
-from litellm.types.utils import ModelResponse
+
+with warnings.catch_warnings():
+    warnings.simplefilter("ignore", UserWarning)
+    import litellm
+    from litellm import get_supported_openai_params
+    from litellm import Choices, get_supported_openai_params
+    from litellm.types.utils import ModelResponse
+
 
 from crewai.utilities.exceptions.context_window_exceeding_exception import (
     LLMContextLengthExceededException,
@@ -71,6 +75,8 @@ LLM_CONTEXT_WINDOW_SIZES = {
     "llama3-70b-8192": 8192,
     "llama3-8b-8192": 8192,
     "mixtral-8x7b-32768": 32768,
+    "llama-3.3-70b-versatile": 128000,
+    "llama-3.3-70b-instruct": 128000,
 }
 
 DEFAULT_CONTEXT_WINDOW_SIZE = 8192

src/crewai/project/annotations.py

@@ -4,18 +4,23 @@ from typing import Callable
 from crewai import Crew
 from crewai.project.utils import memoize
 
+"""Decorators for defining crew components and their behaviors."""
+
 
 def before_kickoff(func):
+    """Marks a method to execute before crew kickoff."""
     func.is_before_kickoff = True
     return func
 
 
 def after_kickoff(func):
+    """Marks a method to execute after crew kickoff."""
     func.is_after_kickoff = True
     return func
 
 
 def task(func):
+    """Marks a method as a crew task."""
     func.is_task = True
 
     @wraps(func)
@@ -29,43 +34,51 @@ def task(func):
 
 
 def agent(func):
+    """Marks a method as a crew agent."""
     func.is_agent = True
     func = memoize(func)
     return func
 
 
 def llm(func):
+    """Marks a method as an LLM provider."""
     func.is_llm = True
     func = memoize(func)
     return func
 
 
 def output_json(cls):
+    """Marks a class as JSON output format."""
     cls.is_output_json = True
     return cls
 
 
 def output_pydantic(cls):
+    """Marks a class as Pydantic output format."""
     cls.is_output_pydantic = True
     return cls
 
 
 def tool(func):
+    """Marks a method as a crew tool."""
     func.is_tool = True
     return memoize(func)
 
 
 def callback(func):
+    """Marks a method as a crew callback."""
     func.is_callback = True
     return memoize(func)
 
 
 def cache_handler(func):
+    """Marks a method as a cache handler."""
     func.is_cache_handler = True
     return memoize(func)
 
 
 def crew(func) -> Callable[..., Crew]:
+    """Marks a method as the main crew execution point."""
 
     @wraps(func)
     def wrapper(self, *args, **kwargs) -> Crew:

src/crewai/project/crew_base.py

@@ -9,8 +9,10 @@ load_dotenv()
 
 T = TypeVar("T", bound=type)
 
+"""Base decorator for creating crew classes with configuration and function management."""
 
 def CrewBase(cls: T) -> T:
+    """Wraps a class with crew functionality and configuration management."""
     class WrappedClass(cls):  # type: ignore
         is_crew_class: bool = True  # type: ignore
 

src/crewai/task.py

@@ -180,6 +180,7 @@ class Task(BaseModel):
     _execution_span: Optional[Span] = PrivateAttr(default=None)
     _original_description: Optional[str] = PrivateAttr(default=None)
     _original_expected_output: Optional[str] = PrivateAttr(default=None)
+    _original_output_file: Optional[str] = PrivateAttr(default=None)
     _thread: Optional[threading.Thread] = PrivateAttr(default=None)
     _execution_time: Optional[float] = PrivateAttr(default=None)
 
@@ -214,8 +215,46 @@ class Task(BaseModel):
 
     @field_validator("output_file")
     @classmethod
-    def output_file_validation(cls, value: str) -> str:
-        """Validate the output file path by removing the / from the beginning of the path."""
+    def output_file_validation(cls, value: Optional[str]) -> Optional[str]:
+        """Validate the output file path.
+        
+        Args:
+            value: The output file path to validate. Can be None or a string.
+                  If the path contains template variables (e.g. {var}), leading slashes are preserved.
+                  For regular paths, leading slashes are stripped.
+        
+        Returns:
+            The validated and potentially modified path, or None if no path was provided.
+            
+        Raises:
+            ValueError: If the path contains invalid characters, path traversal attempts,
+                      or other security concerns.
+        """
+        if value is None:
+            return None
+
+        # Basic security checks
+        if ".." in value:
+            raise ValueError("Path traversal attempts are not allowed in output_file paths")
+        
+        # Check for shell expansion first
+        if value.startswith('~') or value.startswith('$'):
+            raise ValueError("Shell expansion characters are not allowed in output_file paths")
+            
+        # Then check other shell special characters
+        if any(char in value for char in ['|', '>', '<', '&', ';']):
+            raise ValueError("Shell special characters are not allowed in output_file paths")
+
+        # Don't strip leading slash if it's a template path with variables
+        if "{" in value or "}" in value: 
+            # Validate template variable format
+            template_vars = [part.split("}")[0] for part in value.split("{")[1:]]
+            for var in template_vars:
+                if not var.isidentifier():
+                    raise ValueError(f"Invalid template variable name: {var}")
+            return value
+
+        # Strip leading slash for regular paths
         if value.startswith("/"):
             return value[1:]
         return value
@@ -396,21 +435,50 @@ class Task(BaseModel):
         tasks_slices = [self.description, output]
         return "\n".join(tasks_slices)
 
-    def interpolate_inputs_and_add_conversation_history(
-        self, inputs: Dict[str, Any]
-    ) -> None:
-        """Interpolate inputs into the task description and expected output."""
+
+    def interpolate_inputs_and_add_conversation_history(self, inputs: Dict[str, Union[str, int, float]]) -> None:
+        """Interpolate inputs into the task description, expected output, and output file path.
+           Add conversation history if present.
+        
+        Args:
+            inputs: Dictionary mapping template variables to their values.
+                   Supported value types are strings, integers, and floats.
+        
+        Raises:
+            ValueError: If a required template variable is missing from inputs.
+        """
         if self._original_description is None:
             self._original_description = self.description
         if self._original_expected_output is None:
             self._original_expected_output = self.expected_output
+        if self.output_file is not None and self._original_output_file is None:
+            self._original_output_file = self.output_file
 
-        if inputs:
+        if not inputs:
+            return
+
+        try:
             self.description = self._original_description.format(**inputs)
+        except KeyError as e:
+            raise ValueError(f"Missing required template variable '{e.args[0]}' in description") from e
+        except ValueError as e:
+            raise ValueError(f"Error interpolating description: {str(e)}") from e
+
+        try:
             self.expected_output = self.interpolate_only(
                 input_string=self._original_expected_output, inputs=inputs
             )
+        except (KeyError, ValueError) as e:
+            raise ValueError(f"Error interpolating expected_output: {str(e)}") from e
 
+        if self.output_file is not None:
+            try:
+                self.output_file = self.interpolate_only(
+                    input_string=self._original_output_file, inputs=inputs
+                )
+            except (KeyError, ValueError) as e:
+                raise ValueError(f"Error interpolating output_file path: {str(e)}") from e
+                
         if "crew_chat_messages" in inputs and inputs["crew_chat_messages"]:
             # Fetch the conversation history instruction using self.i18n.slice
             conversation_instruction = self.i18n.slice(
@@ -439,14 +507,47 @@ class Task(BaseModel):
                 f"\n\n{conversation_instruction}\n\n{conversation_history}"
             )
 
-    def interpolate_only(self, input_string: str, inputs: Dict[str, Any]) -> str:
-        """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched."""
-        escaped_string = input_string.replace("{", "{{").replace("}", "}}")
+    def interpolate_only(self, input_string: Optional[str], inputs: Dict[str, Union[str, int, float]]) -> str:
+        """Interpolate placeholders (e.g., {key}) in a string while leaving JSON untouched.
+        
+        Args:
+            input_string: The string containing template variables to interpolate.
+                         Can be None or empty, in which case an empty string is returned.
+            inputs: Dictionary mapping template variables to their values.
+                   Supported value types are strings, integers, and floats.
+                   If input_string is empty or has no placeholders, inputs can be empty.
+        
+        Returns:
+            The interpolated string with all template variables replaced with their values.
+            Empty string if input_string is None or empty.
+            
+        Raises:
+            ValueError: If a required template variable is missing from inputs.
+            KeyError: If a template variable is not found in the inputs dictionary.
+        """
+        if input_string is None or not input_string:
+            return ""
+        if "{" not in input_string and "}" not in input_string:
+            return input_string
+        if not inputs:
+            raise ValueError("Inputs dictionary cannot be empty when interpolating variables")
 
-        for key in inputs.keys():
-            escaped_string = escaped_string.replace(f"{{{{{key}}}}}", f"{{{key}}}")
+        try:
+            # Validate input types
+            for key, value in inputs.items():
+                if not isinstance(value, (str, int, float)):
+                    raise ValueError(f"Value for key '{key}' must be a string, integer, or float, got {type(value).__name__}")
 
-        return escaped_string.format(**inputs)
+            escaped_string = input_string.replace("{", "{{").replace("}", "}}")
+
+            for key in inputs.keys():
+                escaped_string = escaped_string.replace(f"{{{{{key}}}}}", f"{{{key}}}")
+
+            return escaped_string.format(**inputs)
+        except KeyError as e:
+            raise KeyError(f"Template variable '{e.args[0]}' not found in inputs dictionary") from e
+        except ValueError as e:
+            raise ValueError(f"Error during string interpolation: {str(e)}") from edssfasf
 
     def increment_tools_errors(self) -> None:
         """Increment the tools errors counter."""

src/crewai/tools/agent_tools/add_image_tool.py

@@ -0,0 +1,45 @@
+from typing import Dict, Optional, Union
+
+from pydantic import BaseModel, Field
+
+from crewai.tools.base_tool import BaseTool
+from crewai.utilities import I18N
+
+i18n = I18N()
+
+class AddImageToolSchema(BaseModel):
+    image_url: str = Field(..., description="The URL or path of the image to add")
+    action: Optional[str] = Field(
+        default=None,
+        description="Optional context or question about the image"
+    )
+
+
+class AddImageTool(BaseTool):
+    """Tool for adding images to the content"""
+
+    name: str = Field(default_factory=lambda: i18n.tools("add_image")["name"])  # type: ignore
+    description: str = Field(default_factory=lambda: i18n.tools("add_image")["description"])  # type: ignore
+    args_schema: type[BaseModel] = AddImageToolSchema
+
+    def _run(
+        self,
+        image_url: str,
+        action: Optional[str] = None,
+        **kwargs,
+    ) -> dict:
+        action = action or i18n.tools("add_image")["default_action"]  # type: ignore
+        content = [
+            {"type": "text", "text": action},
+            {
+                "type": "image_url",
+                "image_url": {
+                    "url": image_url,
+                },
+            }
+        ]
+
+        return {
+            "role": "user",
+            "content": content
+        }

src/crewai/tools/agent_tools/agent_tools.py

@@ -20,13 +20,13 @@ class AgentTools:
         delegate_tool = DelegateWorkTool(
             agents=self.agents,
             i18n=self.i18n,
-            description=self.i18n.tools("delegate_work").format(coworkers=coworkers),
+            description=self.i18n.tools("delegate_work").format(coworkers=coworkers),  # type: ignore
         )
 
         ask_tool = AskQuestionTool(
             agents=self.agents,
             i18n=self.i18n,
-            description=self.i18n.tools("ask_question").format(coworkers=coworkers),
+            description=self.i18n.tools("ask_question").format(coworkers=coworkers),  # type: ignore
         )
 
         return [delegate_tool, ask_tool]

src/crewai/tools/agent_tools/base_agent_tools.py

@@ -1,3 +1,4 @@
+import logging
 from typing import Optional, Union
 
 from pydantic import Field
@@ -7,6 +8,8 @@ from crewai.task import Task
 from crewai.tools.base_tool import BaseTool
 from crewai.utilities import I18N
 
+logger = logging.getLogger(__name__)
+
 
 class BaseAgentTool(BaseTool):
     """Base class for agent-related tools"""
@@ -16,6 +19,25 @@ class BaseAgentTool(BaseTool):
         default_factory=I18N, description="Internationalization settings"
     )
 
+    def sanitize_agent_name(self, name: str) -> str:
+        """
+        Sanitize agent role name by normalizing whitespace and setting to lowercase.
+        Converts all whitespace (including newlines) to single spaces and removes quotes.
+
+        Args:
+            name (str): The agent role name to sanitize
+
+        Returns:
+            str: The sanitized agent role name, with whitespace normalized,
+                 converted to lowercase, and quotes removed
+        """
+        if not name:
+            return ""
+        # Normalize all whitespace (including newlines) to single spaces
+        normalized = " ".join(name.split())
+        # Remove quotes and convert to lowercase
+        return normalized.replace('"', "").casefold()
+
     def _get_coworker(self, coworker: Optional[str], **kwargs) -> Optional[str]:
         coworker = coworker or kwargs.get("co_worker") or kwargs.get("coworker")
         if coworker:
@@ -25,11 +47,27 @@ class BaseAgentTool(BaseTool):
         return coworker
 
     def _execute(
-        self, agent_name: Union[str, None], task: str, context: Union[str, None]
+        self,
+        agent_name: Optional[str],
+        task: str,
+        context: Optional[str] = None
     ) -> str:
+        """
+        Execute delegation to an agent with case-insensitive and whitespace-tolerant matching.
+        
+        Args:
+            agent_name: Name/role of the agent to delegate to (case-insensitive)
+            task: The specific question or task to delegate
+            context: Optional additional context for the task execution
+            
+        Returns:
+            str: The execution result from the delegated agent or an error message
+                 if the agent cannot be found
+        """
         try:
             if agent_name is None:
                 agent_name = ""
+                logger.debug("No agent name provided, using empty string")
 
             # It is important to remove the quotes from the agent name.
             # The reason we have to do this is because less-powerful LLM's
@@ -38,31 +76,49 @@ class BaseAgentTool(BaseTool):
             # {"task": "....", "coworker": "....
             # when it should look like this:
             # {"task": "....", "coworker": "...."}
-            agent_name = agent_name.casefold().replace('"', "").replace("\n", "")
+            sanitized_name = self.sanitize_agent_name(agent_name)
+            logger.debug(f"Sanitized agent name from '{agent_name}' to '{sanitized_name}'")
+
+            available_agents = [agent.role for agent in self.agents]
+            logger.debug(f"Available agents: {available_agents}")
+
             agent = [  # type: ignore # Incompatible types in assignment (expression has type "list[BaseAgent]", variable has type "str | None")
                 available_agent
                 for available_agent in self.agents
-                if available_agent.role.casefold().replace("\n", "") == agent_name
+                if self.sanitize_agent_name(available_agent.role) == sanitized_name
             ]
-        except Exception as _:
+            logger.debug(f"Found {len(agent)} matching agents for role '{sanitized_name}'")
+        except (AttributeError, ValueError) as e:
+            # Handle specific exceptions that might occur during role name processing
             return self.i18n.errors("agent_tool_unexisting_coworker").format(
                 coworkers="\n".join(
-                    [f"- {agent.role.casefold()}" for agent in self.agents]
-                )
+                    [f"- {self.sanitize_agent_name(agent.role)}" for agent in self.agents]
+                ),
+                error=str(e)
             )
 
         if not agent:
+            # No matching agent found after sanitization
             return self.i18n.errors("agent_tool_unexisting_coworker").format(
                 coworkers="\n".join(
-                    [f"- {agent.role.casefold()}" for agent in self.agents]
-                )
+                    [f"- {self.sanitize_agent_name(agent.role)}" for agent in self.agents]
+                ),
+                error=f"No agent found with role '{sanitized_name}'"
             )
 
         agent = agent[0]
-        task_with_assigned_agent = Task(  # type: ignore # Incompatible types in assignment (expression has type "Task", variable has type "str")
-            description=task,
-            agent=agent,
-            expected_output=agent.i18n.slice("manager_request"),
-            i18n=agent.i18n,
-        )
-        return agent.execute_task(task_with_assigned_agent, context)
+        try:
+            task_with_assigned_agent = Task(
+                description=task,
+                agent=agent,
+                expected_output=agent.i18n.slice("manager_request"),
+                i18n=agent.i18n,
+            )
+            logger.debug(f"Created task for agent '{self.sanitize_agent_name(agent.role)}': {task}")
+            return agent.execute_task(task_with_assigned_agent, context)
+        except Exception as e:
+            # Handle task creation or execution errors
+            return self.i18n.errors("agent_tool_execution_error").format(
+                agent_role=self.sanitize_agent_name(agent.role),
+                error=str(e)
+            )

src/crewai/tools/tool_usage.py

@@ -10,6 +10,7 @@ from crewai.agents.tools_handler import ToolsHandler
 from crewai.task import Task
 from crewai.telemetry import Telemetry
 from crewai.tools import BaseTool
+from crewai.tools.structured_tool import CrewStructuredTool
 from crewai.tools.tool_calling import InstructorToolCalling, ToolCalling
 from crewai.tools.tool_usage_events import ToolUsageError, ToolUsageFinished
 from crewai.utilities import I18N, Converter, ConverterError, Printer
@@ -18,8 +19,7 @@ try:
     import agentops  # type: ignore
 except ImportError:
     agentops = None
-
-OPENAI_BIGGER_MODELS = ["gpt-4", "gpt-4o", "o1-preview", "o1-mini"]
+OPENAI_BIGGER_MODELS = ["gpt-4", "gpt-4o", "o1-preview", "o1-mini", "o1", "o3", "o3-mini"]
 
 
 class ToolUsageErrorException(Exception):
@@ -103,6 +103,19 @@ class ToolUsage:
             if self.agent.verbose:
                 self._printer.print(content=f"\n\n{error}\n", color="red")
             return error
+
+        if isinstance(tool, CrewStructuredTool) and tool.name == self._i18n.tools("add_image")["name"]:  # type: ignore
+            try:
+                result = self._use(tool_string=tool_string, tool=tool, calling=calling)
+                return result
+
+            except Exception as e:
+                error = getattr(e, "message", str(e))
+                self.task.increment_tools_errors()
+                if self.agent.verbose:
+                    self._printer.print(content=f"\n\n{error}\n", color="red")
+                return error
+
         return f"{self._use(tool_string=tool_string, tool=tool, calling=calling)}"  # type: ignore # BUG?: "_use" of "ToolUsage" does not return a value (it only ever returns None)
 
     def _use(

src/crewai/translations/en.json

@@ -34,10 +34,16 @@
     "tool_usage_error": "I encountered an error: {error}",
     "tool_arguments_error": "Error: the Action Input is not a valid key, value dictionary.",
     "wrong_tool_name": "You tried to use the tool {tool}, but it doesn't exist. You must use one of the following tools, use one at time: {tools}.",
-    "tool_usage_exception": "I encountered an error while trying to use the tool. This was the error: {error}.\n Tool {tool} accepts these inputs: {tool_inputs}"
+    "tool_usage_exception": "I encountered an error while trying to use the tool. This was the error: {error}.\n Tool {tool} accepts these inputs: {tool_inputs}",
+    "agent_tool_execution_error": "Error executing task with agent '{agent_role}'. Error: {error}"
   },
   "tools": {
     "delegate_work": "Delegate a specific task to one of the following coworkers: {coworkers}\nThe input to this tool should be the coworker, the task you want them to do, and ALL necessary context to execute the task, they know nothing about the task, so share absolute everything you know, don't reference things but instead explain them.",
-    "ask_question": "Ask a specific question to one of the following coworkers: {coworkers}\nThe input to this tool should be the coworker, the question you have for them, and ALL necessary context to ask the question properly, they know nothing about the question, so share absolute everything you know, don't reference things but instead explain them."
+    "ask_question": "Ask a specific question to one of the following coworkers: {coworkers}\nThe input to this tool should be the coworker, the question you have for them, and ALL necessary context to ask the question properly, they know nothing about the question, so share absolute everything you know, don't reference things but instead explain them.",
+    "add_image": {
+      "name": "Add image to content",
+      "description": "See image to understand it's content, you can optionally ask a question about the image",
+      "default_action": "Please provide a detailed description of this image, including all visual elements, context, and any notable details you can observe."
+    }
   }
 }

src/crewai/utilities/crew_json_encoder.py

@@ -1,3 +1,5 @@
+"""JSON encoder for handling CrewAI specific types."""
+
 import json
 from datetime import date, datetime
 from decimal import Decimal
@@ -8,6 +10,7 @@ from pydantic import BaseModel
 
 
 class CrewJSONEncoder(json.JSONEncoder):
+    """Custom JSON encoder for CrewAI objects and special types."""
     def default(self, obj):
         if isinstance(obj, BaseModel):
             return self._handle_pydantic_model(obj)

src/crewai/utilities/crew_pydantic_output_parser.py

@@ -6,9 +6,10 @@ from pydantic import BaseModel, ValidationError
 
 from crewai.agents.parser import OutputParserException
 
+"""Parser for converting text outputs into Pydantic models."""
 
 class CrewPydanticOutputParser:
-    """Parses the text into pydantic models"""
+    """Parses text outputs into specified Pydantic models."""
 
     pydantic_object: Type[BaseModel]
 

src/crewai/utilities/i18n.py

@@ -1,11 +1,13 @@
 import json
 import os
-from typing import Dict, Optional
+from typing import Dict, Optional, Union
 
 from pydantic import BaseModel, Field, PrivateAttr, model_validator
 
+"""Internationalization support for CrewAI prompts and messages."""
 
 class I18N(BaseModel):
+    """Handles loading and retrieving internationalized prompts."""
     _prompts: Dict[str, Dict[str, str]] = PrivateAttr()
     prompt_file: Optional[str] = Field(
         default=None,
@@ -41,8 +43,8 @@ class I18N(BaseModel):
     def errors(self, error: str) -> str:
         return self.retrieve("errors", error)
 
-    def tools(self, error: str) -> str:
-        return self.retrieve("tools", error)
+    def tools(self, tool: str) -> Union[str, Dict[str, str]]:
+        return self.retrieve("tools", tool)
 
     def retrieve(self, kind, key) -> str:
         try:

src/crewai/utilities/internal_instructor.py

@@ -1,3 +1,4 @@
+import warnings
 from typing import Any, Optional, Type
 
 
@@ -25,9 +26,10 @@ class InternalInstructor:
         if self.agent and not self.llm:
             self.llm = self.agent.function_calling_llm or self.agent.llm
 
-        # Lazy import
-        import instructor
-        from litellm import completion
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", UserWarning)
+            import instructor
+            from litellm import completion
 
         self._client = instructor.from_litellm(
             completion,

src/crewai/utilities/paths.py

@@ -3,8 +3,10 @@ from pathlib import Path
 
 import appdirs
 
+"""Path management utilities for CrewAI storage and configuration."""
 
 def db_storage_path():
+    """Returns the path for database storage."""
     app_name = get_project_directory_name()
     app_author = "CrewAI"
 
@@ -14,6 +16,7 @@ def db_storage_path():
 
 
 def get_project_directory_name():
+    """Returns the current project directory name."""
     project_directory_name = os.environ.get("CREWAI_STORAGE_DIR")
 
     if project_directory_name:

src/crewai/utilities/planning_handler.py

@@ -1,3 +1,5 @@
+import json
+import logging
 from typing import Any, List, Optional
 
 from pydantic import BaseModel, Field
@@ -5,8 +7,11 @@ from pydantic import BaseModel, Field
 from crewai.agent import Agent
 from crewai.task import Task
 
+"""Handles planning and coordination of crew tasks."""
+logger = logging.getLogger(__name__)
 
 class PlanPerTask(BaseModel):
+    """Represents a plan for a specific task."""
     task: str = Field(..., description="The task for which the plan is created")
     plan: str = Field(
         ...,
@@ -15,6 +20,7 @@ class PlanPerTask(BaseModel):
 
 
 class PlannerTaskPydanticOutput(BaseModel):
+    """Output format for task planning results."""
     list_of_plans_per_task: List[PlanPerTask] = Field(
         ...,
         description="Step by step plan on how the agents can execute their tasks using the available tools with mastery",
@@ -22,6 +28,7 @@ class PlannerTaskPydanticOutput(BaseModel):
 
 
 class CrewPlanner:
+    """Plans and coordinates the execution of crew tasks."""
     def __init__(self, tasks: List[Task], planning_agent_llm: Optional[Any] = None):
         self.tasks = tasks
 
@@ -68,19 +75,39 @@ class CrewPlanner:
             output_pydantic=PlannerTaskPydanticOutput,
         )
 
+    def _get_agent_knowledge(self, task: Task) -> List[str]:
+        """
+        Safely retrieve knowledge source content from the task's agent.
+        
+        Args:
+            task: The task containing an agent with potential knowledge sources
+            
+        Returns:
+            List[str]: A list of knowledge source strings
+        """
+        try:
+            if task.agent and task.agent.knowledge_sources:
+                return [source.content for source in task.agent.knowledge_sources]
+        except AttributeError:
+            logger.warning("Error accessing agent knowledge sources")
+        return []
+
     def _create_tasks_summary(self) -> str:
         """Creates a summary of all tasks."""
         tasks_summary = []
         for idx, task in enumerate(self.tasks):
-            tasks_summary.append(
-                f"""
+            knowledge_list = self._get_agent_knowledge(task)
+            task_summary = f"""
                 Task Number {idx + 1} - {task.description}
                 "task_description": {task.description}
                 "task_expected_output": {task.expected_output}
                 "agent": {task.agent.role if task.agent else "None"}
                 "agent_goal": {task.agent.goal if task.agent else "None"}
                 "task_tools": {task.tools}
-                "agent_tools": {task.agent.tools if task.agent else "None"}
-                """
-            )
+                "agent_tools": %s%s""" % (
+                    f"[{', '.join(str(tool) for tool in task.agent.tools)}]" if task.agent and task.agent.tools else '"agent has no tools"',
+                    f',\n                "agent_knowledge": "[\\"{knowledge_list[0]}\\"]"' if knowledge_list and str(knowledge_list) != "None" else ""
+                )
+            
+            tasks_summary.append(task_summary)
         return " ".join(tasks_summary)

src/crewai/utilities/printer.py

@@ -1,7 +1,11 @@
+"""Utility for colored console output."""
+
 from typing import Optional
 
 
 class Printer:
+    """Handles colored console output formatting."""
+
     def print(self, content: str, color: Optional[str] = None):
         if color == "purple":
             self._print_purple(content)

src/crewai/utilities/rpm_controller.py

@@ -6,8 +6,10 @@ from pydantic import BaseModel, Field, PrivateAttr, model_validator
 
 from crewai.utilities.logger import Logger
 
+"""Controls request rate limiting for API calls."""
 
 class RPMController(BaseModel):
+    """Manages requests per minute limiting."""
     max_rpm: Optional[int] = Field(default=None)
     logger: Logger = Field(default_factory=lambda: Logger(verbose=False))
     _current_rpm: int = PrivateAttr(default=0)

src/crewai/utilities/task_output_storage_handler.py

@@ -8,8 +8,10 @@ from crewai.memory.storage.kickoff_task_outputs_storage import (
 )
 from crewai.task import Task
 
+"""Handles storage and retrieval of task execution outputs."""
 
 class ExecutionLog(BaseModel):
+    """Represents a log entry for task execution."""
     task_id: str
     expected_output: Optional[str] = None
     output: Dict[str, Any]
@@ -22,6 +24,8 @@ class ExecutionLog(BaseModel):
         return getattr(self, key)
 
 
+"""Manages storage and retrieval of task outputs."""
+
 class TaskOutputStorageHandler:
     def __init__(self) -> None:
         self.storage = KickoffTaskOutputsSQLiteStorage()

src/crewai/utilities/token_counter_callback.py

@@ -1,3 +1,5 @@
+import warnings
+
 from litellm.integrations.custom_logger import CustomLogger
 from litellm.types.utils import Usage
 
@@ -12,11 +14,13 @@ class TokenCalcHandler(CustomLogger):
         if self.token_cost_process is None:
             return
 
-        usage: Usage = response_obj["usage"]
-        self.token_cost_process.sum_successful_requests(1)
-        self.token_cost_process.sum_prompt_tokens(usage.prompt_tokens)
-        self.token_cost_process.sum_completion_tokens(usage.completion_tokens)
-        if usage.prompt_tokens_details:
-            self.token_cost_process.sum_cached_prompt_tokens(
-                usage.prompt_tokens_details.cached_tokens
-            )
+        with warnings.catch_warnings():
+            warnings.simplefilter("ignore", UserWarning)
+            usage: Usage = response_obj["usage"]
+            self.token_cost_process.sum_successful_requests(1)
+            self.token_cost_process.sum_prompt_tokens(usage.prompt_tokens)
+            self.token_cost_process.sum_completion_tokens(usage.completion_tokens)
+            if usage.prompt_tokens_details:
+                self.token_cost_process.sum_cached_prompt_tokens(
+                    usage.prompt_tokens_details.cached_tokens
+                )