Enhance input history management in Flow class

- Introduced a new `InputHistoryEntry` TypedDict to structure user input history for the `ask()` method, capturing details such as the question, user response, method name, timestamp, and associated metadata.
- Updated the `_input_history` attribute in the Flow class to utilize the new `InputHistoryEntry` type, improving type safety and clarity in input history management.
This commit is contained in:
Joao Moura
2026-02-16 10:55:32 -08:00
parent 162988134d
commit aef63341c2
2 changed files with 27 additions and 2 deletions

View File

@@ -77,7 +77,7 @@ from crewai.flow.flow_wrappers import (
StartMethod,
)
from crewai.flow.persistence.base import FlowPersistence
from crewai.flow.types import FlowExecutionData, FlowMethodName, PendingListenerKey
from crewai.flow.types import FlowExecutionData, FlowMethodName, InputHistoryEntry, PendingListenerKey
from crewai.flow.utils import (
_extract_all_methods,
_extract_all_methods_recursive,
@@ -786,7 +786,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
self.suppress_flow_events: bool = suppress_flow_events
# User input history (for self.ask())
self._input_history: list[dict[str, Any]] = []
self._input_history: list[InputHistoryEntry] = []
# Initialize state with initial values
self._state = self._create_initial_state()

View File

@@ -4,6 +4,7 @@ This module contains TypedDict definitions and type aliases used throughout
the Flow system.
"""
from datetime import datetime
from typing import (
Annotated,
Any,
@@ -101,6 +102,30 @@ class FlowData(TypedDict):
flow_methods_attributes: list[FlowMethodData]
class InputHistoryEntry(TypedDict):
"""A single entry in the flow's input history from ``self.ask()``.
Each call to ``Flow.ask()`` appends one entry recording the question,
the user's response, which method asked, and any metadata exchanged
between the caller and the input provider.
Attributes:
message: The question or prompt that was displayed to the user.
response: The user's response, or None on timeout/error.
method_name: The flow method that called ``ask()``.
timestamp: When the input was received.
metadata: Metadata sent with the question (caller to provider).
response_metadata: Metadata received with the answer (provider to caller).
"""
message: str
response: str | None
method_name: str
timestamp: datetime
metadata: dict[str, Any] | None
response_metadata: dict[str, Any] | None
class FlowExecutionData(TypedDict):
"""Flow execution data.