Merge branch 'main' into main

src/crewai/agents/agent_builder/utilities/base_output_converter.py

@@ -25,7 +25,7 @@ class OutputConverter(BaseModel, ABC):
     llm: Any = Field(description="The language model to be used to convert the text.")
     model: Any = Field(description="The model to be used to convert the text.")
     instructions: str = Field(description="Conversion instructions to the LLM.")
-    max_attempts: Optional[int] = Field(
+    max_attempts: int = Field(
         description="Max number of attempts to try to get the output formatted.",
         default=3,
     )

src/crewai/agents/agent_builder/utilities/base_token_process.py

@@ -2,11 +2,12 @@ from crewai.types.usage_metrics import UsageMetrics
 
 
 class TokenProcess:
-    total_tokens: int = 0
-    prompt_tokens: int = 0
-    cached_prompt_tokens: int = 0
-    completion_tokens: int = 0
-    successful_requests: int = 0
+    def __init__(self):
+        self.total_tokens: int = 0
+        self.prompt_tokens: int = 0
+        self.cached_prompt_tokens: int = 0
+        self.completion_tokens: int = 0
+        self.successful_requests: int = 0
 
     def sum_prompt_tokens(self, tokens: int):
         self.prompt_tokens = self.prompt_tokens + tokens

src/crewai/flow/flow.py

@@ -1,5 +1,6 @@
 import asyncio
 import inspect
+import uuid
 from typing import (
     Any,
     Callable,
@@ -12,6 +13,7 @@ from typing import (
     TypeVar,
     Union,
     cast,
+    overload,
 )
 from uuid import uuid4
 
@@ -25,6 +27,8 @@ from crewai.flow.flow_events import (
     MethodExecutionStartedEvent,
 )
 from crewai.flow.flow_visualizer import plot_flow
+from crewai.flow.persistence import FlowPersistence
+from crewai.flow.persistence.base import FlowPersistence
 from crewai.flow.utils import get_possible_return_constants
 from crewai.telemetry import Telemetry
 
@@ -33,7 +37,46 @@ class FlowState(BaseModel):
     """Base model for all flow states, ensuring each state has a unique ID."""
     id: str = Field(default_factory=lambda: str(uuid4()), description="Unique identifier for the flow state")
 
-T = TypeVar("T", bound=Union[FlowState, Dict[str, Any]])
+# Type variables with explicit bounds
+T = TypeVar("T", bound=Union[Dict[str, Any], BaseModel])  # Generic flow state type parameter
+StateT = TypeVar("StateT", bound=Union[Dict[str, Any], BaseModel])  # State validation type parameter
+
+def ensure_state_type(state: Any, expected_type: Type[StateT]) -> StateT:
+    """Ensure state matches expected type with proper validation.
+    
+    Args:
+        state: State instance to validate
+        expected_type: Expected type for the state
+        
+    Returns:
+        Validated state instance
+        
+    Raises:
+        TypeError: If state doesn't match expected type
+        ValueError: If state validation fails
+    """
+    """Ensure state matches expected type with proper validation.
+    
+    Args:
+        state: State instance to validate
+        expected_type: Expected type for the state
+        
+    Returns:
+        Validated state instance
+        
+    Raises:
+        TypeError: If state doesn't match expected type
+        ValueError: If state validation fails
+    """
+    if expected_type == dict:
+        if not isinstance(state, dict):
+            raise TypeError(f"Expected dict, got {type(state).__name__}")
+        return cast(StateT, state)
+    if isinstance(expected_type, type) and issubclass(expected_type, BaseModel):
+        if not isinstance(state, expected_type):
+            raise TypeError(f"Expected {expected_type.__name__}, got {type(state).__name__}")
+        return cast(StateT, state)
+    raise TypeError(f"Invalid expected_type: {expected_type}")
 
 
 def start(condition: Optional[Union[str, dict, Callable]] = None) -> Callable:
@@ -326,21 +369,27 @@ class FlowMeta(type):
         routers = set()
 
         for attr_name, attr_value in dct.items():
-            if hasattr(attr_value, "__is_start_method__"):
-                start_methods.append(attr_name)
+            # Check for any flow-related attributes
+            if (hasattr(attr_value, "__is_flow_method__") or
+                hasattr(attr_value, "__is_start_method__") or
+                hasattr(attr_value, "__trigger_methods__") or
+                hasattr(attr_value, "__is_router__")):
+                
+                # Register start methods
+                if hasattr(attr_value, "__is_start_method__"):
+                    start_methods.append(attr_name)
+                
+                # Register listeners and routers
                 if hasattr(attr_value, "__trigger_methods__"):
                     methods = attr_value.__trigger_methods__
                     condition_type = getattr(attr_value, "__condition_type__", "OR")
                     listeners[attr_name] = (condition_type, methods)
-            elif hasattr(attr_value, "__trigger_methods__"):
-                methods = attr_value.__trigger_methods__
-                condition_type = getattr(attr_value, "__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
+                    
+                    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)
@@ -351,6 +400,9 @@ class FlowMeta(type):
 
 
 class Flow(Generic[T], metaclass=FlowMeta):
+    """Base class for all flows.
+    
+    Type parameter T must be either Dict[str, Any] or a subclass of BaseModel."""
     _telemetry = Telemetry()
 
     _start_methods: List[str] = []
@@ -367,53 +419,220 @@ class Flow(Generic[T], metaclass=FlowMeta):
         _FlowGeneric.__name__ = f"{cls.__name__}[{item.__name__}]"
         return _FlowGeneric
 
-    def __init__(self) -> None:
+    def __init__(
+        self,
+        persistence: Optional[FlowPersistence] = None,
+        restore_uuid: Optional[str] = None,
+        **kwargs: Any,
+    ) -> None:
+        """Initialize a new Flow instance.
+        
+        Args:
+            persistence: Optional persistence backend for storing flow states
+            restore_uuid: Optional UUID to restore state from persistence
+            **kwargs: Additional state values to initialize or override
+        """
+        # Initialize basic instance attributes
         self._methods: Dict[str, Callable] = {}
-        self._state: T = self._create_initial_state()
         self._method_execution_counts: Dict[str, int] = {}
         self._pending_and_listeners: Dict[str, Set[str]] = {}
         self._method_outputs: List[Any] = []  # List to store all method outputs
+        self._persistence: Optional[FlowPersistence] = persistence
+        
+        # Validate state model before initialization
+        if isinstance(self.initial_state, type):
+            if issubclass(self.initial_state, BaseModel) and not issubclass(self.initial_state, FlowState):
+                # Check if model has id field
+                model_fields = getattr(self.initial_state, "model_fields", None)
+                if not model_fields or "id" not in model_fields:
+                    raise ValueError("Flow state model must have an 'id' field")
+        
+        # Handle persistence and potential ID conflicts
+        stored_state = None
+        if self._persistence is not None:
+            if restore_uuid and kwargs and "id" in kwargs and restore_uuid != kwargs["id"]:
+                raise ValueError(
+                    f"Conflicting IDs provided: restore_uuid='{restore_uuid}' "
+                    f"vs kwargs['id']='{kwargs['id']}'. Use only one ID for restoration."
+                )
+            
+            # Attempt to load state, prioritizing restore_uuid
+            if restore_uuid:
+                stored_state = self._persistence.load_state(restore_uuid)
+                if not stored_state:
+                    raise ValueError(f"No state found for restore_uuid='{restore_uuid}'")
+            elif kwargs and "id" in kwargs:
+                stored_state = self._persistence.load_state(kwargs["id"])
+                if not stored_state:
+                    # For kwargs["id"], we allow creating new state if not found
+                    self._state = self._create_initial_state()
+                    if kwargs:
+                        self._initialize_state(kwargs)
+                    return
+        
+        # Initialize state based on persistence and kwargs
+        if stored_state:
+            # Create initial state and restore from persistence
+            self._state = self._create_initial_state()
+            self._restore_state(stored_state)
+            # Apply any additional kwargs to override specific fields
+            if kwargs:
+                filtered_kwargs = {k: v for k, v in kwargs.items() if k != "id"}
+                if filtered_kwargs:
+                    self._initialize_state(filtered_kwargs)
+        else:
+            # No stored state, create new state with initial values
+            self._state = self._create_initial_state()
+            # Apply any additional kwargs
+            if kwargs:
+                self._initialize_state(kwargs)
 
         self._telemetry.flow_creation_span(self.__class__.__name__)
 
+        # Register all flow-related methods
         for method_name in dir(self):
-            if callable(getattr(self, method_name)) and not method_name.startswith(
-                "__"
-            ):
-                self._methods[method_name] = getattr(self, method_name)
+            if not method_name.startswith("_"):
+                method = getattr(self, method_name)
+                # Check for any flow-related attributes
+                if (hasattr(method, "__is_flow_method__") or
+                    hasattr(method, "__is_start_method__") or
+                    hasattr(method, "__trigger_methods__") or
+                    hasattr(method, "__is_router__")):
+                    # Ensure method is bound to this instance
+                    if not hasattr(method, "__self__"):
+                        method = method.__get__(self, self.__class__)
+                    self._methods[method_name] = method
 
+
+    
     def _create_initial_state(self) -> T:
+        """Create and initialize flow state with UUID and default values.
+        
+        Returns:
+            New state instance with UUID and default values initialized
+            
+        Raises:
+            ValueError: If structured state model lacks 'id' field
+            TypeError: If state is neither BaseModel nor dictionary
+        """
         # Handle case where initial_state is None but we have a type parameter
         if self.initial_state is None and hasattr(self, "_initial_state_T"):
             state_type = getattr(self, "_initial_state_T")
             if isinstance(state_type, type):
                 if issubclass(state_type, FlowState):
-                    return state_type()  # type: ignore
+                    # Create instance without id, then set it
+                    instance = state_type()
+                    if not hasattr(instance, 'id'):
+                        setattr(instance, 'id', str(uuid4()))
+                    return cast(T, instance)
                 elif issubclass(state_type, BaseModel):
                     # Create a new type that includes the ID field
                     class StateWithId(state_type, FlowState):  # type: ignore
                         pass
-                    return StateWithId()  # type: ignore
+                    instance = StateWithId()
+                    if not hasattr(instance, 'id'):
+                        setattr(instance, 'id', str(uuid4()))
+                    return cast(T, instance)
+                elif state_type == dict:
+                    return cast(T, {"id": str(uuid4())})  # Minimal dict state
+        
+        # Handle case where no initial state is provided
+        if self.initial_state is None:
+            return cast(T, {"id": str(uuid4())})
+        
+        # Handle case where initial_state is a type (class)
+        if isinstance(self.initial_state, type):
+            if issubclass(self.initial_state, FlowState):
+                return cast(T, self.initial_state())  # Uses model defaults
+            elif issubclass(self.initial_state, BaseModel):
+                # Validate that the model has an id field
+                model_fields = getattr(self.initial_state, "model_fields", None)
+                if not model_fields or "id" not in model_fields:
+                    raise ValueError("Flow state model must have an 'id' field")
+                return cast(T, self.initial_state())  # Uses model defaults
+            elif self.initial_state == dict:
+                return cast(T, {"id": str(uuid4())})
+        
+        # Handle dictionary instance case
+        if isinstance(self.initial_state, dict):
+            new_state = dict(self.initial_state)  # Copy to avoid mutations
+            if "id" not in new_state:
+                new_state["id"] = str(uuid4())
+            return cast(T, new_state)
+        
+        # Handle BaseModel instance case
+        if isinstance(self.initial_state, BaseModel):
+            model = cast(BaseModel, self.initial_state)
+            if not hasattr(model, "id"):
+                raise ValueError("Flow state model must have an 'id' field")
+            
+            # Create new instance with same values to avoid mutations
+            if hasattr(model, "model_dump"):
+                # Pydantic v2
+                state_dict = model.model_dump()
+            elif hasattr(model, "dict"):
+                # Pydantic v1
+                state_dict = model.dict()
+            else:
+                # Fallback for other BaseModel implementations
+                state_dict = {
+                    k: v for k, v in model.__dict__.items()
+                    if not k.startswith("_")
+                }
+            
+            # Create new instance of the same class
+            model_class = type(model)
+            return cast(T, model_class(**state_dict))
+            
+        raise TypeError(
+            f"Initial state must be dict or BaseModel, got {type(self.initial_state)}"
+        )
+        # Handle case where initial_state is None but we have a type parameter
+        if self.initial_state is None and hasattr(self, "_initial_state_T"):
+            state_type = getattr(self, "_initial_state_T")
+            if isinstance(state_type, type):
+                if issubclass(state_type, FlowState):
+                    return cast(T, state_type())
+                elif issubclass(state_type, BaseModel):
+                    # Create a new type that includes the ID field
+                    class StateWithId(state_type, FlowState):  # type: ignore
+                        pass
+                    return cast(T, StateWithId())
+                elif state_type == dict:
+                    return cast(T, {"id": str(uuid4())})
 
         # Handle case where no initial state is provided
         if self.initial_state is None:
-            return {"id": str(uuid4())}  # type: ignore
+            return cast(T, {"id": str(uuid4())})
 
         # Handle case where initial_state is a type (class)
         if isinstance(self.initial_state, type):
             if issubclass(self.initial_state, FlowState):
-                return self.initial_state()  # type: ignore
+                return cast(T, self.initial_state())
             elif issubclass(self.initial_state, BaseModel):
-                # Create a new type that includes the ID field
-                class StateWithId(self.initial_state, FlowState):  # type: ignore
-                    pass
-                return StateWithId()  # type: ignore
+                # Validate that the model has an id field
+                model_fields = getattr(self.initial_state, "model_fields", None)
+                if not model_fields or "id" not in model_fields:
+                    raise ValueError("Flow state model must have an 'id' field")
+                return cast(T, self.initial_state())
+            elif self.initial_state == dict:
+                return cast(T, {"id": str(uuid4())})
 
-        # Handle dictionary case
-        if isinstance(self.initial_state, dict) and "id" not in self.initial_state:
-            self.initial_state["id"] = str(uuid4())
+        # Handle dictionary instance case
+        if isinstance(self.initial_state, dict):
+            if "id" not in self.initial_state:
+                self.initial_state["id"] = str(uuid4())
+            return cast(T, dict(self.initial_state))  # Create new dict to avoid mutations
 
-        return self.initial_state  # type: ignore
+        # Handle BaseModel instance case
+        if isinstance(self.initial_state, BaseModel):
+            if not hasattr(self.initial_state, "id"):
+                raise ValueError("Flow state model must have an 'id' field")
+            return cast(T, self.initial_state)
+            
+        raise TypeError(
+            f"Initial state must be dict or BaseModel, got {type(self.initial_state)}"
+        )
 
     @property
     def state(self) -> T:
@@ -425,50 +644,95 @@ class Flow(Generic[T], metaclass=FlowMeta):
         return self._method_outputs
 
     def _initialize_state(self, inputs: Dict[str, Any]) -> None:
+        """Initialize or update flow state with new inputs.
+        
+        Args:
+            inputs: Dictionary of state values to set/update
+            
+        Raises:
+            ValueError: If validation fails for structured state
+            TypeError: If state is neither BaseModel nor dictionary
+        """
         if isinstance(self._state, dict):
-            # Preserve the ID when updating unstructured state
+            # For dict states, preserve existing fields unless overridden
             current_id = self._state.get("id")
-            self._state.update(inputs)
+            # Only update specified fields
+            for k, v in inputs.items():
+                self._state[k] = v
+            # Ensure ID is preserved or generated
             if current_id:
                 self._state["id"] = current_id
             elif "id" not in self._state:
                 self._state["id"] = str(uuid4())
         elif isinstance(self._state, BaseModel):
-            # Structured state
+            # For BaseModel states, preserve existing fields unless overridden
             try:
-                def create_model_with_extra_forbid(
-                    base_model: Type[BaseModel],
-                ) -> Type[BaseModel]:
-                    class ModelWithExtraForbid(base_model):  # type: ignore
-                        model_config = base_model.model_config.copy()
-                        model_config["extra"] = "forbid"
-
-                    return ModelWithExtraForbid
-
-                # Get current state as dict, preserving the ID if it exists
-                state_model = cast(BaseModel, self._state)
-                current_state = (
-                    state_model.model_dump()
-                    if hasattr(state_model, "model_dump")
-                    else state_model.dict()
-                    if hasattr(state_model, "dict")
-                    else {
-                        k: v
-                        for k, v in state_model.__dict__.items()
+                model = cast(BaseModel, self._state)
+                # Get current state as dict
+                if hasattr(model, "model_dump"):
+                    current_state = model.model_dump()
+                elif hasattr(model, "dict"):
+                    current_state = model.dict()
+                else:
+                    current_state = {
+                        k: v for k, v in model.__dict__.items()
                         if not k.startswith("_")
                     }
-                )
-
-                ModelWithExtraForbid = create_model_with_extra_forbid(
-                    self._state.__class__
-                )
-                self._state = cast(
-                    T, ModelWithExtraForbid(**{**current_state, **inputs})
-                )
+                
+                # Create new state with preserved fields and updates
+                new_state = {**current_state, **inputs}
+                
+                # Create new instance with merged state
+                model_class = type(model)
+                if hasattr(model_class, "model_validate"):
+                    # Pydantic v2
+                    self._state = cast(T, model_class.model_validate(new_state))
+                elif hasattr(model_class, "parse_obj"):
+                    # Pydantic v1
+                    self._state = cast(T, model_class.parse_obj(new_state))
+                else:
+                    # Fallback for other BaseModel implementations
+                    self._state = cast(T, model_class(**new_state))
             except ValidationError as e:
                 raise ValueError(f"Invalid inputs for structured state: {e}") from e
         else:
             raise TypeError("State must be a BaseModel instance or a dictionary.")
+            
+    def _restore_state(self, stored_state: Dict[str, Any]) -> None:
+        """Restore flow state from persistence.
+        
+        Args:
+            stored_state: Previously stored state to restore
+            
+        Raises:
+            ValueError: If validation fails for structured state
+            TypeError: If state is neither BaseModel nor dictionary
+        """
+        # When restoring from persistence, use the stored ID
+        stored_id = stored_state.get("id")
+        if not stored_id:
+            raise ValueError("Stored state must have an 'id' field")
+        
+        if isinstance(self._state, dict):
+            # For dict states, update all fields from stored state
+            self._state.clear()
+            self._state.update(stored_state)
+        elif isinstance(self._state, BaseModel):
+            # For BaseModel states, create new instance with stored values
+            model = cast(BaseModel, self._state)
+            if hasattr(model, "model_validate"):
+                # Pydantic v2
+                self._state = cast(T, type(model).model_validate(stored_state))
+            elif hasattr(model, "parse_obj"):
+                # Pydantic v1
+                self._state = cast(T, type(model).parse_obj(stored_state))
+            else:
+                # Fallback for other BaseModel implementations
+                self._state = cast(T, type(model)(**stored_state))
+        else:
+            raise TypeError(
+                f"State must be dict or BaseModel, got {type(self._state)}"
+            )
 
     def kickoff(self, inputs: Optional[Dict[str, Any]] = None) -> Any:
         self.event_emitter.send(

src/crewai/flow/persistence/__init__.py

@@ -0,0 +1,18 @@
+"""
+CrewAI Flow Persistence.
+
+This module provides interfaces and implementations for persisting flow states.
+"""
+
+from typing import Any, Dict, TypeVar, Union
+
+from pydantic import BaseModel
+
+from crewai.flow.persistence.base import FlowPersistence
+from crewai.flow.persistence.decorators import persist
+from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+
+__all__ = ["FlowPersistence", "persist", "SQLiteFlowPersistence"]
+
+StateType = TypeVar('StateType', bound=Union[Dict[str, Any], BaseModel])
+DictStateType = Dict[str, Any]

src/crewai/flow/persistence/base.py

@@ -0,0 +1,53 @@
+"""Base class for flow state persistence."""
+
+import abc
+from typing import Any, Dict, Optional, Union
+
+from pydantic import BaseModel
+
+
+class FlowPersistence(abc.ABC):
+    """Abstract base class for flow state persistence.
+    
+    This class defines the interface that all persistence implementations must follow.
+    It supports both structured (Pydantic BaseModel) and unstructured (dict) states.
+    """
+    
+    @abc.abstractmethod
+    def init_db(self) -> None:
+        """Initialize the persistence backend.
+        
+        This method should handle any necessary setup, such as:
+        - Creating tables
+        - Establishing connections
+        - Setting up indexes
+        """
+        pass
+    
+    @abc.abstractmethod
+    def save_state(
+        self,
+        flow_uuid: str,
+        method_name: str,
+        state_data: Union[Dict[str, Any], BaseModel]
+    ) -> None:
+        """Persist the flow state after method completion.
+        
+        Args:
+            flow_uuid: Unique identifier for the flow instance
+            method_name: Name of the method that just completed
+            state_data: Current state data (either dict or Pydantic model)
+        """
+        pass
+    
+    @abc.abstractmethod
+    def load_state(self, flow_uuid: str) -> Optional[Dict[str, Any]]:
+        """Load the most recent state for a given flow UUID.
+        
+        Args:
+            flow_uuid: Unique identifier for the flow instance
+            
+        Returns:
+            The most recent state as a dictionary, or None if no state exists
+        """
+        pass

src/crewai/flow/persistence/decorators.py

@@ -0,0 +1,177 @@
+"""
+Decorators for flow state persistence.
+
+Example:
+    ```python
+    from crewai.flow.flow import Flow, start
+    from crewai.flow.persistence import persist, SQLiteFlowPersistence
+    
+    class MyFlow(Flow):
+        @start()
+        @persist(SQLiteFlowPersistence())
+        def sync_method(self):
+            # Synchronous method implementation
+            pass
+            
+        @start()
+        @persist(SQLiteFlowPersistence())
+        async def async_method(self):
+            # Asynchronous method implementation
+            await some_async_operation()
+    ```
+"""
+
+import asyncio
+import functools
+import inspect
+import logging
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Optional,
+    Type,
+    TypeVar,
+    Union,
+    cast,
+    get_type_hints,
+)
+
+from pydantic import BaseModel
+
+from crewai.flow.persistence.base import FlowPersistence
+from crewai.flow.persistence.sqlite import SQLiteFlowPersistence
+
+logger = logging.getLogger(__name__)
+T = TypeVar("T")
+
+
+def persist(persistence: Optional[FlowPersistence] = None):
+    """Decorator to persist flow state.
+    
+    This decorator can be applied at either the class level or method level.
+    When applied at the class level, it automatically persists all flow method
+    states. When applied at the method level, it persists only that method's
+    state.
+    
+    Args:
+        persistence: Optional FlowPersistence implementation to use.
+                    If not provided, uses SQLiteFlowPersistence.
+    
+    Returns:
+        A decorator that can be applied to either a class or method
+    
+    Raises:
+        ValueError: If the flow state doesn't have an 'id' field
+        RuntimeError: If state persistence fails
+        
+    Example:
+        @persist  # Class-level persistence with default SQLite
+        class MyFlow(Flow[MyState]):
+            @start()
+            def begin(self):
+                pass
+    """
+    def _persist_state(flow_instance: Any, method_name: str, persistence_instance: FlowPersistence) -> None:
+        """Helper to persist state with error handling."""
+        try:
+            # Get flow UUID from state
+            state = getattr(flow_instance, 'state', None)
+            if state is None:
+                raise ValueError("Flow instance has no state")
+                
+            flow_uuid: Optional[str] = None
+            if isinstance(state, dict):
+                flow_uuid = state.get('id')
+            elif isinstance(state, BaseModel):
+                flow_uuid = getattr(state, 'id', None)
+                
+            if not flow_uuid:
+                raise ValueError(
+                    "Flow state must have an 'id' field for persistence"
+                )
+                
+            # Persist the state
+            persistence_instance.save_state(
+                flow_uuid=flow_uuid,
+                method_name=method_name,
+                state_data=state,
+            )
+        except Exception as e:
+            logger.error(
+                f"Failed to persist state for method {method_name}: {str(e)}"
+            )
+            raise RuntimeError(f"State persistence failed: {str(e)}") from e
+    
+    def decorator(target: Union[Type, Callable[..., T]]) -> Union[Type, Callable[..., T]]:
+        """Decorator that handles both class and method decoration."""
+        actual_persistence = persistence or SQLiteFlowPersistence()
+        
+        if isinstance(target, type):
+            # Class decoration
+            class_methods = {}
+            for name, method in target.__dict__.items():
+                if callable(method) and hasattr(method, "__is_flow_method__"):
+                    # Wrap each flow method with persistence
+                    if asyncio.iscoroutinefunction(method):
+                        @functools.wraps(method)
+                        async def class_async_wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
+                            method_coro = method(self, *args, **kwargs)
+                            if asyncio.iscoroutine(method_coro):
+                                result = await method_coro
+                            else:
+                                result = method_coro
+                            _persist_state(self, method.__name__, actual_persistence)
+                            return result
+                        class_methods[name] = class_async_wrapper
+                    else:
+                        @functools.wraps(method)
+                        def class_sync_wrapper(self: Any, *args: Any, **kwargs: Any) -> Any:
+                            result = method(self, *args, **kwargs)
+                            _persist_state(self, method.__name__, actual_persistence)
+                            return result
+                        class_methods[name] = class_sync_wrapper
+                    
+                    # Preserve flow-specific attributes
+                    for attr in ["__is_start_method__", "__trigger_methods__", "__condition_type__", "__is_router__"]:
+                        if hasattr(method, attr):
+                            setattr(class_methods[name], attr, getattr(method, attr))
+                    setattr(class_methods[name], "__is_flow_method__", True)
+            
+            # Update class with wrapped methods
+            for name, method in class_methods.items():
+                setattr(target, name, method)
+            return target
+        else:
+            # Method decoration
+            method = target
+            setattr(method, "__is_flow_method__", True)
+            
+            if asyncio.iscoroutinefunction(method):
+                @functools.wraps(method)
+                async def method_async_wrapper(flow_instance: Any, *args: Any, **kwargs: Any) -> T:
+                    method_coro = method(flow_instance, *args, **kwargs)
+                    if asyncio.iscoroutine(method_coro):
+                        result = await method_coro
+                    else:
+                        result = method_coro
+                    _persist_state(flow_instance, method.__name__, actual_persistence)
+                    return result
+                for attr in ["__is_start_method__", "__trigger_methods__", "__condition_type__", "__is_router__"]:
+                    if hasattr(method, attr):
+                        setattr(method_async_wrapper, attr, getattr(method, attr))
+                setattr(method_async_wrapper, "__is_flow_method__", True)
+                return cast(Callable[..., T], method_async_wrapper)
+            else:
+                @functools.wraps(method)
+                def method_sync_wrapper(flow_instance: Any, *args: Any, **kwargs: Any) -> T:
+                    result = method(flow_instance, *args, **kwargs)
+                    _persist_state(flow_instance, method.__name__, actual_persistence)
+                    return result
+                for attr in ["__is_start_method__", "__trigger_methods__", "__condition_type__", "__is_router__"]:
+                    if hasattr(method, attr):
+                        setattr(method_sync_wrapper, attr, getattr(method, attr))
+                setattr(method_sync_wrapper, "__is_flow_method__", True)
+                return cast(Callable[..., T], method_sync_wrapper)
+            
+    return decorator

src/crewai/flow/persistence/sqlite.py

@@ -0,0 +1,124 @@
+"""
+SQLite-based implementation of flow state persistence.
+"""
+
+import json
+import os
+import sqlite3
+import tempfile
+from datetime import datetime
+from typing import Any, Dict, Optional, Union
+
+from pydantic import BaseModel
+
+from crewai.flow.persistence.base import FlowPersistence
+
+
+class SQLiteFlowPersistence(FlowPersistence):
+    """SQLite-based implementation of flow state persistence.
+    
+    This class provides a simple, file-based persistence implementation using SQLite.
+    It's suitable for development and testing, or for production use cases with
+    moderate performance requirements.
+    """
+    
+    db_path: str  # Type annotation for instance variable
+    
+    def __init__(self, db_path: Optional[str] = None):
+        """Initialize SQLite persistence.
+        
+        Args:
+            db_path: Path to the SQLite database file. If not provided, uses
+                    db_storage_path() from utilities.paths.
+                    
+        Raises:
+            ValueError: If db_path is invalid
+        """
+        from crewai.utilities.paths import db_storage_path
+        # Get path from argument or default location
+        path = db_path or db_storage_path()
+        
+        if not path:
+            raise ValueError("Database path must be provided")
+            
+        self.db_path = path  # Now mypy knows this is str
+        self.init_db()
+    
+    def init_db(self) -> None:
+        """Create the necessary tables if they don't exist."""
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute("""
+            CREATE TABLE IF NOT EXISTS flow_states (
+                id INTEGER PRIMARY KEY AUTOINCREMENT,
+                flow_uuid TEXT NOT NULL,
+                method_name TEXT NOT NULL,
+                timestamp DATETIME NOT NULL,
+                state_json TEXT NOT NULL
+            )
+            """)
+            # Add index for faster UUID lookups
+            conn.execute("""
+            CREATE INDEX IF NOT EXISTS idx_flow_states_uuid 
+            ON flow_states(flow_uuid)
+            """)
+    
+    def save_state(
+        self,
+        flow_uuid: str,
+        method_name: str,
+        state_data: Union[Dict[str, Any], BaseModel],
+    ) -> None:
+        """Save the current flow state to SQLite.
+        
+        Args:
+            flow_uuid: Unique identifier for the flow instance
+            method_name: Name of the method that just completed
+            state_data: Current state data (either dict or Pydantic model)
+        """
+        # Convert state_data to dict, handling both Pydantic and dict cases
+        if isinstance(state_data, BaseModel):
+            state_dict = dict(state_data)  # Use dict() for better type compatibility
+        elif isinstance(state_data, dict):
+            state_dict = state_data
+        else:
+            raise ValueError(
+                f"state_data must be either a Pydantic BaseModel or dict, got {type(state_data)}"
+            )
+            
+        with sqlite3.connect(self.db_path) as conn:
+            conn.execute("""
+            INSERT INTO flow_states (
+                flow_uuid,
+                method_name,
+                timestamp,
+                state_json
+            ) VALUES (?, ?, ?, ?)
+            """, (
+                flow_uuid,
+                method_name,
+                datetime.utcnow().isoformat(),
+                json.dumps(state_dict),
+            ))
+    
+    def load_state(self, flow_uuid: str) -> Optional[Dict[str, Any]]:
+        """Load the most recent state for a given flow UUID.
+        
+        Args:
+            flow_uuid: Unique identifier for the flow instance
+            
+        Returns:
+            The most recent state as a dictionary, or None if no state exists
+        """
+        with sqlite3.connect(self.db_path) as conn:
+            cursor = conn.execute("""
+            SELECT state_json
+            FROM flow_states
+            WHERE flow_uuid = ?
+            ORDER BY id DESC
+            LIMIT 1
+            """, (flow_uuid,))
+            row = cursor.fetchone()
+            
+        if row:
+            return json.loads(row[0])
+        return None

src/crewai/knowledge/source/crew_docling_source.py

@@ -8,6 +8,7 @@ try:
     from docling.exceptions import ConversionError
     from docling_core.transforms.chunker.hierarchical_chunker import HierarchicalChunker
     from docling_core.types.doc.document import DoclingDocument
+
     DOCLING_AVAILABLE = True
 except ImportError:
     DOCLING_AVAILABLE = False
@@ -38,8 +39,8 @@ class CrewDoclingSource(BaseKnowledgeSource):
     file_paths: List[Union[Path, str]] = Field(default_factory=list)
     chunks: List[str] = Field(default_factory=list)
     safe_file_paths: List[Union[Path, str]] = Field(default_factory=list)
-    content: List[DoclingDocument] = Field(default_factory=list)
-    document_converter: DocumentConverter = Field(
+    content: List["DoclingDocument"] = Field(default_factory=list)
+    document_converter: "DocumentConverter" = Field(
         default_factory=lambda: DocumentConverter(
             allowed_formats=[
                 InputFormat.MD,
@@ -65,7 +66,7 @@ class CrewDoclingSource(BaseKnowledgeSource):
         self.safe_file_paths = self.validate_content()
         self.content = self._load_content()
 
-    def _load_content(self) -> List[DoclingDocument]:
+    def _load_content(self) -> List["DoclingDocument"]:
         try:
             return self._convert_source_to_docling_documents()
         except ConversionError as e:
@@ -87,11 +88,11 @@ class CrewDoclingSource(BaseKnowledgeSource):
             self.chunks.extend(list(new_chunks_iterable))
         self._save_documents()
 
-    def _convert_source_to_docling_documents(self) -> List[DoclingDocument]:
+    def _convert_source_to_docling_documents(self) -> List["DoclingDocument"]:
         conv_results_iter = self.document_converter.convert_all(self.safe_file_paths)
         return [result.document for result in conv_results_iter]
 
-    def _chunk_doc(self, doc: DoclingDocument) -> Iterator[str]:
+    def _chunk_doc(self, doc: "DoclingDocument") -> Iterator[str]:
         chunker = HierarchicalChunker()
         for chunk in chunker.chunk(doc):
             yield chunk.text

src/crewai/llm.py

@@ -222,6 +222,19 @@ class LLM:
                 ].message
                 text_response = response_message.content or ""
                 tool_calls = getattr(response_message, "tool_calls", [])
+                
+                # Ensure callbacks get the full response object with usage info
+                if callbacks and len(callbacks) > 0:
+                    for callback in callbacks:
+                        if hasattr(callback, "log_success_event"):
+                            usage_info = getattr(response, "usage", None)
+                            if usage_info:
+                                callback.log_success_event(
+                                    kwargs=params,
+                                    response_obj={"usage": usage_info},
+                                    start_time=0,
+                                    end_time=0,
+                                )
 
                 # --- 2) If no tool calls, return the text response
                 if not tool_calls or not available_functions:

src/crewai/memory/storage/kickoff_task_outputs_storage.py

@@ -1,12 +1,17 @@
 import json
+import logging
 import sqlite3
+from pathlib import Path
 from typing import Any, Dict, List, Optional
 
 from crewai.task import Task
 from crewai.utilities import Printer
 from crewai.utilities.crew_json_encoder import CrewJSONEncoder
+from crewai.utilities.errors import DatabaseError, DatabaseOperationError
 from crewai.utilities.paths import db_storage_path
 
+logger = logging.getLogger(__name__)
+
 
 class KickoffTaskOutputsSQLiteStorage:
     """
@@ -14,15 +19,24 @@ class KickoffTaskOutputsSQLiteStorage:
     """
 
     def __init__(
-        self, db_path: str = f"{db_storage_path()}/latest_kickoff_task_outputs.db"
+        self, db_path: Optional[str] = None
     ) -> None:
+        if db_path is None:
+            # Get the parent directory of the default db path and create our db file there
+            db_path = str(Path(db_storage_path()).parent / "latest_kickoff_task_outputs.db")
         self.db_path = db_path
         self._printer: Printer = Printer()
         self._initialize_db()
 
-    def _initialize_db(self):
-        """
-        Initializes the SQLite database and creates LTM table
+    def _initialize_db(self) -> None:
+        """Initialize the SQLite database and create the latest_kickoff_task_outputs table.
+
+        This method sets up the database schema for storing task outputs. It creates
+        a table with columns for task_id, expected_output, output (as JSON),
+        task_index, inputs (as JSON), was_replayed flag, and timestamp.
+
+        Raises:
+            DatabaseOperationError: If database initialization fails due to SQLite errors.
         """
         try:
             with sqlite3.connect(self.db_path) as conn:
@@ -43,10 +57,9 @@ class KickoffTaskOutputsSQLiteStorage:
 
                 conn.commit()
         except sqlite3.Error as e:
-            self._printer.print(
-                content=f"SAVING KICKOFF TASK OUTPUTS ERROR: An error occurred during database initialization: {e}",
-                color="red",
-            )
+            error_msg = DatabaseError.format_error(DatabaseError.INIT_ERROR, e)
+            logger.error(error_msg)
+            raise DatabaseOperationError(error_msg, e)
 
     def add(
         self,
@@ -55,9 +68,22 @@ class KickoffTaskOutputsSQLiteStorage:
         task_index: int,
         was_replayed: bool = False,
         inputs: Dict[str, Any] = {},
-    ):
+    ) -> None:
+        """Add a new task output record to the database.
+
+        Args:
+            task: The Task object containing task details.
+            output: Dictionary containing the task's output data.
+            task_index: Integer index of the task in the sequence.
+            was_replayed: Boolean indicating if this was a replay execution.
+            inputs: Dictionary of input parameters used for the task.
+
+        Raises:
+            DatabaseOperationError: If saving the task output fails due to SQLite errors.
+        """
         try:
             with sqlite3.connect(self.db_path) as conn:
+                conn.execute("BEGIN TRANSACTION")
                 cursor = conn.cursor()
                 cursor.execute(
                     """
@@ -76,21 +102,31 @@ class KickoffTaskOutputsSQLiteStorage:
                 )
                 conn.commit()
         except sqlite3.Error as e:
-            self._printer.print(
-                content=f"SAVING KICKOFF TASK OUTPUTS ERROR: An error occurred during database initialization: {e}",
-                color="red",
-            )
+            error_msg = DatabaseError.format_error(DatabaseError.SAVE_ERROR, e)
+            logger.error(error_msg)
+            raise DatabaseOperationError(error_msg, e)
 
     def update(
         self,
         task_index: int,
-        **kwargs,
-    ):
-        """
-        Updates an existing row in the latest_kickoff_task_outputs table based on task_index.
+        **kwargs: Any,
+    ) -> None:
+        """Update an existing task output record in the database.
+
+        Updates fields of a task output record identified by task_index. The fields
+        to update are provided as keyword arguments.
+
+        Args:
+            task_index: Integer index of the task to update.
+            **kwargs: Arbitrary keyword arguments representing fields to update.
+                     Values that are dictionaries will be JSON encoded.
+
+        Raises:
+            DatabaseOperationError: If updating the task output fails due to SQLite errors.
         """
         try:
             with sqlite3.connect(self.db_path) as conn:
+                conn.execute("BEGIN TRANSACTION")
                 cursor = conn.cursor()
 
                 fields = []
@@ -110,14 +146,23 @@ class KickoffTaskOutputsSQLiteStorage:
                 conn.commit()
 
                 if cursor.rowcount == 0:
-                    self._printer.print(
-                        f"No row found with task_index {task_index}. No update performed.",
-                        color="red",
-                    )
+                    logger.warning(f"No row found with task_index {task_index}. No update performed.")
         except sqlite3.Error as e:
-            self._printer.print(f"UPDATE KICKOFF TASK OUTPUTS ERROR: {e}", color="red")
+            error_msg = DatabaseError.format_error(DatabaseError.UPDATE_ERROR, e)
+            logger.error(error_msg)
+            raise DatabaseOperationError(error_msg, e)
 
-    def load(self) -> Optional[List[Dict[str, Any]]]:
+    def load(self) -> List[Dict[str, Any]]:
+        """Load all task output records from the database.
+
+        Returns:
+            List of dictionaries containing task output records, ordered by task_index.
+            Each dictionary contains: task_id, expected_output, output, task_index,
+            inputs, was_replayed, and timestamp.
+
+        Raises:
+            DatabaseOperationError: If loading task outputs fails due to SQLite errors.
+        """
         try:
             with sqlite3.connect(self.db_path) as conn:
                 cursor = conn.cursor()
@@ -144,23 +189,26 @@ class KickoffTaskOutputsSQLiteStorage:
                 return results
 
         except sqlite3.Error as e:
-            self._printer.print(
-                content=f"LOADING KICKOFF TASK OUTPUTS ERROR: An error occurred while querying kickoff task outputs: {e}",
-                color="red",
-            )
-            return None
+            error_msg = DatabaseError.format_error(DatabaseError.LOAD_ERROR, e)
+            logger.error(error_msg)
+            raise DatabaseOperationError(error_msg, e)
 
-    def delete_all(self):
-        """
-        Deletes all rows from the latest_kickoff_task_outputs table.
+    def delete_all(self) -> None:
+        """Delete all task output records from the database.
+
+        This method removes all records from the latest_kickoff_task_outputs table.
+        Use with caution as this operation cannot be undone.
+
+        Raises:
+            DatabaseOperationError: If deleting task outputs fails due to SQLite errors.
         """
         try:
             with sqlite3.connect(self.db_path) as conn:
+                conn.execute("BEGIN TRANSACTION")
                 cursor = conn.cursor()
                 cursor.execute("DELETE FROM latest_kickoff_task_outputs")
                 conn.commit()
         except sqlite3.Error as e:
-            self._printer.print(
-                content=f"ERROR: Failed to delete all kickoff task outputs: {e}",
-                color="red",
-            )
+            error_msg = DatabaseError.format_error(DatabaseError.DELETE_ERROR, e)
+            logger.error(error_msg)
+            raise DatabaseOperationError(error_msg, e)

src/crewai/memory/storage/ltm_sqlite_storage.py

@@ -1,5 +1,6 @@
 import json
 import sqlite3
+from pathlib import Path
 from typing import Any, Dict, List, Optional, Union
 
 from crewai.utilities import Printer
@@ -12,10 +13,15 @@ class LTMSQLiteStorage:
     """
 
     def __init__(
-        self, db_path: str = f"{db_storage_path()}/long_term_memory_storage.db"
+        self, db_path: Optional[str] = None
     ) -> None:
+        if db_path is None:
+            # Get the parent directory of the default db path and create our db file there
+            db_path = str(Path(db_storage_path()).parent / "long_term_memory_storage.db")
         self.db_path = db_path
         self._printer: Printer = Printer()
+        # Ensure parent directory exists
+        Path(self.db_path).parent.mkdir(parents=True, exist_ok=True)
         self._initialize_db()
 
     def _initialize_db(self):

src/crewai/utilities/converter.py

@@ -26,17 +26,24 @@ class Converter(OutputConverter):
             if self.llm.supports_function_calling():
                 return self._create_instructor().to_pydantic()
             else:
-                return self.llm.call(
+                response = self.llm.call(
                     [
                         {"role": "system", "content": self.instructions},
                         {"role": "user", "content": self.text},
                     ]
                 )
+                return self.model.model_validate_json(response)
+        except ValidationError as e:
+            if current_attempt < self.max_attempts:
+                return self.to_pydantic(current_attempt + 1)
+            raise ConverterError(
+                f"Failed to convert text into a Pydantic model due to the following validation error: {e}"
+            )
         except Exception as e:
             if current_attempt < self.max_attempts:
                 return self.to_pydantic(current_attempt + 1)
-            return ConverterError(
-                f"Failed to convert text into a pydantic model due to the following error: {e}"
+            raise ConverterError(
+                f"Failed to convert text into a Pydantic model due to the following error: {e}"
             )
 
     def to_json(self, current_attempt=1):
@@ -66,7 +73,6 @@ class Converter(OutputConverter):
             llm=self.llm,
             model=self.model,
             content=self.text,
-            instructions=self.instructions,
         )
         return inst
 
@@ -187,10 +193,15 @@ def convert_with_instructions(
 
 
 def get_conversion_instructions(model: Type[BaseModel], llm: Any) -> str:
-    instructions = "I'm gonna convert this raw text into valid JSON."
+    instructions = "Please convert the following text into valid JSON."
     if llm.supports_function_calling():
         model_schema = PydanticSchemaParser(model=model).get_schema()
-        instructions = f"{instructions}\n\nThe json should have the following structure, with the following keys:\n{model_schema}"
+        instructions += (
+            f"\n\nThe JSON should follow this schema:\n```json\n{model_schema}\n```"
+        )
+    else:
+        model_description = generate_model_description(model)
+        instructions += f"\n\nThe JSON should follow this format:\n{model_description}"
     return instructions
 
 

src/crewai/utilities/errors.py

@@ -0,0 +1,39 @@
+"""Error message definitions for CrewAI database operations."""
+from typing import Optional
+
+
+class DatabaseOperationError(Exception):
+    """Base exception class for database operation errors."""
+
+    def __init__(self, message: str, original_error: Optional[Exception] = None):
+        """Initialize the database operation error.
+
+        Args:
+            message: The error message to display
+            original_error: The original exception that caused this error, if any
+        """
+        super().__init__(message)
+        self.original_error = original_error
+
+
+class DatabaseError:
+    """Standardized error message templates for database operations."""
+
+    INIT_ERROR: str = "Database initialization error: {}"
+    SAVE_ERROR: str = "Error saving task outputs: {}"
+    UPDATE_ERROR: str = "Error updating task outputs: {}"
+    LOAD_ERROR: str = "Error loading task outputs: {}"
+    DELETE_ERROR: str = "Error deleting task outputs: {}"
+
+    @classmethod
+    def format_error(cls, template: str, error: Exception) -> str:
+        """Format an error message with the given template and error.
+
+        Args:
+            template: The error message template to use
+            error: The exception to format into the template
+
+        Returns:
+            The formatted error message
+        """
+        return template.format(str(error))

src/crewai/utilities/internal_instructor.py

@@ -11,12 +11,10 @@ class InternalInstructor:
         model: Type,
         agent: Optional[Any] = None,
         llm: Optional[str] = None,
-        instructions: Optional[str] = None,
     ):
         self.content = content
         self.agent = agent
         self.llm = llm
-        self.instructions = instructions
         self.model = model
         self._client = None
         self.set_instructor()
@@ -31,10 +29,7 @@ class InternalInstructor:
             import instructor
             from litellm import completion
 
-            self._client = instructor.from_litellm(
-                completion,
-                mode=instructor.Mode.TOOLS,
-            )
+            self._client = instructor.from_litellm(completion)
 
     def to_json(self):
         model = self.to_pydantic()
@@ -42,8 +37,6 @@ class InternalInstructor:
 
     def to_pydantic(self):
         messages = [{"role": "user", "content": self.content}]
-        if self.instructions:
-            messages.append({"role": "system", "content": self.instructions})
         model = self._client.chat.completions.create(
             model=self.llm.model, response_model=self.model, messages=messages
         )

src/crewai/utilities/paths.py

@@ -5,14 +5,18 @@ import appdirs
 
 """Path management utilities for CrewAI storage and configuration."""
 
-def db_storage_path():
-    """Returns the path for database storage."""
+def db_storage_path() -> str:
+    """Returns the path for SQLite database storage.
+    
+    Returns:
+        str: Full path to the SQLite database file
+    """
     app_name = get_project_directory_name()
     app_author = "CrewAI"
 
     data_dir = Path(appdirs.user_data_dir(app_name, app_author))
     data_dir.mkdir(parents=True, exist_ok=True)
-    return data_dir
+    return str(data_dir / "crewai_flows.db")
 
 
 def get_project_directory_name():

src/crewai/utilities/pydantic_schema_parser.py

@@ -1,4 +1,4 @@
-from typing import Type, Union, get_args, get_origin
+from typing import Dict, List, Type, Union, get_args, get_origin
 
 from pydantic import BaseModel
 
@@ -10,40 +10,83 @@ class PydanticSchemaParser(BaseModel):
         """
         Public method to get the schema of a Pydantic model.
 
-        :param model: The Pydantic model class to generate schema for.
         :return: String representation of the model schema.
         """
-        return self._get_model_schema(self.model)
+        return "{\n" + self._get_model_schema(self.model) + "\n}"
 
-    def _get_model_schema(self, model, depth=0) -> str:
-        indent = "    " * depth
-        lines = [f"{indent}{{"]
-        for field_name, field in model.model_fields.items():
-            field_type_str = self._get_field_type(field, depth + 1)
-            lines.append(f"{indent}    {field_name}: {field_type_str},")
-        lines[-1] = lines[-1].rstrip(",")  # Remove trailing comma from last item
-        lines.append(f"{indent}}}")
-        return "\n".join(lines)
+    def _get_model_schema(self, model: Type[BaseModel], depth: int = 0) -> str:
+        indent = " " * 4 * depth
+        lines = [
+            f"{indent}    {field_name}: {self._get_field_type(field, depth + 1)}"
+            for field_name, field in model.model_fields.items()
+        ]
+        return ",\n".join(lines)
 
-    def _get_field_type(self, field, depth) -> str:
+    def _get_field_type(self, field, depth: int) -> str:
         field_type = field.annotation
-        if get_origin(field_type) is list:
+        origin = get_origin(field_type)
+
+        if origin in {list, List}:
             list_item_type = get_args(field_type)[0]
-            if isinstance(list_item_type, type) and issubclass(
-                list_item_type, BaseModel
-            ):
-                nested_schema = self._get_model_schema(list_item_type, depth + 1)
-                return f"List[\n{nested_schema}\n{' ' * 4 * depth}]"
+            return self._format_list_type(list_item_type, depth)
+
+        if origin in {dict, Dict}:
+            key_type, value_type = get_args(field_type)
+            return f"Dict[{key_type.__name__}, {value_type.__name__}]"
+
+        if origin is Union:
+            return self._format_union_type(field_type, depth)
+
+        if isinstance(field_type, type) and issubclass(field_type, BaseModel):
+            nested_schema = self._get_model_schema(field_type, depth)
+            nested_indent = " " * 4 * depth
+            return f"{field_type.__name__}\n{nested_indent}{{\n{nested_schema}\n{nested_indent}}}"
+
+        return field_type.__name__
+
+    def _format_list_type(self, list_item_type, depth: int) -> str:
+        if isinstance(list_item_type, type) and issubclass(list_item_type, BaseModel):
+            nested_schema = self._get_model_schema(list_item_type, depth + 1)
+            nested_indent = " " * 4 * (depth)
+            return f"List[\n{nested_indent}{{\n{nested_schema}\n{nested_indent}}}\n{nested_indent}]"
+        return f"List[{list_item_type.__name__}]"
+
+    def _format_union_type(self, field_type, depth: int) -> str:
+        args = get_args(field_type)
+        if type(None) in args:
+            # It's an Optional type
+            non_none_args = [arg for arg in args if arg is not type(None)]
+            if len(non_none_args) == 1:
+                inner_type = self._get_field_type_for_annotation(
+                    non_none_args[0], depth
+                )
+                return f"Optional[{inner_type}]"
             else:
-                return f"List[{list_item_type.__name__}]"
-        elif get_origin(field_type) is Union:
-            union_args = get_args(field_type)
-            if type(None) in union_args:
-                non_none_type = next(arg for arg in union_args if arg is not type(None))
-                return f"Optional[{self._get_field_type(field.__class__(annotation=non_none_type), depth)}]"
-            else:
-                return f"Union[{', '.join(arg.__name__ for arg in union_args)}]"
-        elif isinstance(field_type, type) and issubclass(field_type, BaseModel):
-            return self._get_model_schema(field_type, depth)
+                # Union with None and multiple other types
+                inner_types = ", ".join(
+                    self._get_field_type_for_annotation(arg, depth)
+                    for arg in non_none_args
+                )
+                return f"Optional[Union[{inner_types}]]"
         else:
-            return getattr(field_type, "__name__", str(field_type))
+            # General Union type
+            inner_types = ", ".join(
+                self._get_field_type_for_annotation(arg, depth) for arg in args
+            )
+            return f"Union[{inner_types}]"
+
+    def _get_field_type_for_annotation(self, annotation, depth: int) -> str:
+        origin = get_origin(annotation)
+        if origin in {list, List}:
+            list_item_type = get_args(annotation)[0]
+            return self._format_list_type(list_item_type, depth)
+        if origin in {dict, Dict}:
+            key_type, value_type = get_args(annotation)
+            return f"Dict[{key_type.__name__}, {value_type.__name__}]"
+        if origin is Union:
+            return self._format_union_type(annotation, depth)
+        if isinstance(annotation, type) and issubclass(annotation, BaseModel):
+            nested_schema = self._get_model_schema(annotation, depth)
+            nested_indent = " " * 4 * depth
+            return f"{annotation.__name__}\n{nested_indent}{{\n{nested_schema}\n{nested_indent}}}"
+        return annotation.__name__

src/crewai/utilities/token_counter_callback.py

@@ -23,11 +23,15 @@ class TokenCalcHandler(CustomLogger):
 
         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
-                )
+            if isinstance(response_obj, dict) and "usage" in response_obj:
+                usage: Usage = response_obj["usage"]
+                if usage:
+                    self.token_cost_process.sum_successful_requests(1)
+                    if hasattr(usage, "prompt_tokens"):
+                        self.token_cost_process.sum_prompt_tokens(usage.prompt_tokens)
+                    if hasattr(usage, "completion_tokens"):
+                        self.token_cost_process.sum_completion_tokens(usage.completion_tokens)
+                    if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details:
+                        self.token_cost_process.sum_cached_prompt_tokens(
+                            usage.prompt_tokens_details.cached_tokens
+                        )