andre/crewAI
1
0
Fork 0
mirror of https://github.com/crewAIInc/crewAI.git synced 2026-07-08 08:25:09 +00:00
Code Issues Actions 6 Packages Projects Releases Wiki Activity

Implement user input handling in Flow class

Browse Source 
- Introduced `FlowInputRequestedEvent` and `FlowInputReceivedEvent` to manage user input requests and responses during flow execution.
- Added `InputProvider` protocol and `InputResponse` dataclass for customizable input handling.
- Enhanced `Flow` class with `ask()` method to request user input, including timeout handling and state checkpointing.
- Updated `FlowConfig` to support custom input providers.
- Created `input_provider.py` for default input provider implementations, including a console-based provider.
- Added comprehensive tests for `ask()` functionality, covering basic usage, timeout behavior, and integration with flow machinery.
This commit is contained in:
Joao Moura
2026-02-15 01:50:45 -08:00
parent 4aedd58829
commit 9ccbb1655c
7 changed files with 1665 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
lib/crewai/src/crewai/events/types/flow_events.py
View File

@@ -120,6 +120,52 @@ class FlowPlotEvent(FlowEvent):
type: str = "flow_plot"
class FlowInputRequestedEvent(FlowEvent):
"""Event emitted when a flow requests user input via ``Flow.ask()``.
This event is emitted before the flow suspends waiting for user input,
allowing UI frameworks and observability tools to know when a flow
needs user interaction.
Attributes:
flow_name: Name of the flow requesting input.
method_name: Name of the flow method that called ``ask()``.
message: The question or prompt being shown to the user.
metadata: Optional metadata sent with the question (e.g., user ID,
channel, session context).
"""
method_name: str
message: str
metadata: dict[str, Any] | None = None
type: str = "flow_input_requested"
class FlowInputReceivedEvent(FlowEvent):
"""Event emitted when user input is received after ``Flow.ask()``.
This event is emitted after the user provides input (or the request
times out), allowing UI frameworks and observability tools to track
input collection.
Attributes:
flow_name: Name of the flow that received input.
method_name: Name of the flow method that called ``ask()``.
message: The original question or prompt.
response: The user's response, or None if timed out / unavailable.
metadata: Optional metadata sent with the question.
response_metadata: Optional metadata from the provider about the
response (e.g., who responded, thread ID, timestamps).
"""
method_name: str
message: str
response: str | None = None
metadata: dict[str, Any] | None = None
response_metadata: dict[str, Any] | None = None
type: str = "flow_input_received"
class HumanFeedbackRequestedEvent(FlowEvent):
"""Event emitted when human feedback is requested.

3
lib/crewai/src/crewai/flow/__init__.py
View File

@@ -7,6 +7,7 @@ from crewai.flow.async_feedback import (
from crewai.flow.flow import Flow, and_, listen, or_, router, start
from crewai.flow.flow_config import flow_config
from crewai.flow.human_feedback import HumanFeedbackResult, human_feedback
from crewai.flow.input_provider import InputProvider, InputResponse
from crewai.flow.persistence import persist
from crewai.flow.visualization import (
FlowStructure,
@@ -22,6 +23,8 @@ __all__ = [
"HumanFeedbackPending",
"HumanFeedbackProvider",
"HumanFeedbackResult",
"InputProvider",
"InputResponse",
"PendingFeedbackContext",
"and_",
"build_flow_structure",

85
lib/crewai/src/crewai/flow/async_feedback/providers.py
View File

@@ -1,7 +1,8 @@
"""Default provider implementations for human feedback.
"""Default provider implementations for human feedback and user input.
This module provides the ConsoleProvider, which is the default synchronous
provider that collects feedback via console input.
provider that collects both feedback (for ``@human_feedback``) and user input
(for ``Flow.ask()``) via console.
"""
from __future__ import annotations
@@ -16,20 +17,23 @@ if TYPE_CHECKING:
class ConsoleProvider:
"""Default synchronous console-based feedback provider.
"""Default synchronous console-based provider for feedback and input.
This provider blocks execution and waits for console input from the user.
It displays the method output with formatting and prompts for feedback.
It serves two purposes:
- **Feedback** (``request_feedback``): Used by ``@human_feedback`` to
display method output and collect review feedback.
- **Input** (``request_input``): Used by ``Flow.ask()`` to prompt the
user with a question and collect a response.
This is the default provider used when no custom provider is specified
in the @human_feedback decorator.
in the ``@human_feedback`` decorator or on the Flow's ``input_provider``.
Example:
Example (feedback):
```python
from crewai.flow.async_feedback import ConsoleProvider
# Explicitly use console provider
@human_feedback(
message="Review this:",
provider=ConsoleProvider(),
@@ -37,9 +41,20 @@ class ConsoleProvider:
def my_method(self):
return "Content to review"
```
Example (input):
```python
from crewai.flow import Flow, start
class MyFlow(Flow):
@start()
def gather_info(self):
topic = self.ask("What topic should we research?")
return topic
```
"""
def __init__(self, verbose: bool = True):
def __init__(self, verbose: bool = True) -> None:
"""Initialize the console provider.
Args:
@@ -124,3 +139,55 @@ class ConsoleProvider:
finally:
# Resume live updates
formatter.resume_live_updates()
def request_input(
self,
message: str,
flow: Flow[Any],
metadata: dict[str, Any] | None = None,
) -> str | None:
"""Request user input via console (blocking).
Displays the prompt message with formatting and waits for the user
to type their response. Used by ``Flow.ask()``.
Unlike ``request_feedback``, this method does not display an
"OUTPUT FOR REVIEW" panel or emit feedback-specific events (those
are handled by ``ask()`` itself).
Args:
message: The question or prompt to display to the user.
flow: The Flow instance requesting input.
metadata: Optional metadata from the caller. Ignored by the
console provider (console has no concept of user routing).
Returns:
The user's input as a stripped string. Returns empty string
if user presses Enter without input. Never returns None
(console input is always available).
"""
from crewai.events.event_listener import event_listener
# Pause live updates during human input
formatter = event_listener.formatter
formatter.pause_live_updates()
try:
console = formatter.console
if self.verbose:
console.print()
console.print(message, style="yellow")
console.print()
response = input(">>> \n").strip()
else:
response = input(f"{message} ").strip()
# Add line break after input so formatter output starts clean
console.print()
return response
finally:
# Resume live updates
formatter.resume_live_updates()

202
lib/crewai/src/crewai/flow/flow.py
View File

@@ -738,6 +738,7 @@ class Flow(Generic[T], metaclass=FlowMeta):
tracing: bool | None = None
stream: bool = False
memory: Any = None # Memory | MemoryScope | MemorySlice | None; auto-created if not set
input_provider: Any = None # InputProvider | None; per-flow override for self.ask()
def __class_getitem__(cls: type[Flow[T]], item: type[T]) -> type[Flow[T]]:
class _FlowGeneric(cls): # type: ignore
@@ -784,6 +785,9 @@ class Flow(Generic[T], metaclass=FlowMeta):
self._pending_feedback_context: PendingFeedbackContext | None = None
self.suppress_flow_events: bool = suppress_flow_events
# User input history (for self.ask())
self._input_history: list[dict[str, Any]] = []
# Initialize state with initial values
self._state = self._create_initial_state()
self.tracing = tracing
@@ -2582,6 +2586,204 @@ class Flow(Generic[T], metaclass=FlowMeta):
logger.error(f"Error executing listener {listener_name}: {e}")
raise
# ── User Input (self.ask) ────────────────────────────────────────
def _resolve_input_provider(self) -> Any:
"""Resolve the input provider using the priority chain.
Resolution order:
1. ``self.input_provider`` (per-flow override)
2. ``flow_config.input_provider`` (global default)
3. ``ConsoleInputProvider()`` (built-in fallback)
Returns:
An object implementing the ``InputProvider`` protocol.
"""
from crewai.flow.async_feedback.providers import ConsoleProvider
from crewai.flow.flow_config import flow_config
if self.input_provider is not None:
return self.input_provider
if flow_config.input_provider is not None:
return flow_config.input_provider
return ConsoleProvider()
def _resolve_calling_method_name(self) -> str:
"""Resolve the name of the flow method that called ask().
Walks the call stack to find the first frame whose function name
matches a registered flow method on this instance.
Returns:
The flow method name, or ``"unknown"`` if not found.
"""
import inspect
for frame_info in inspect.stack():
fname = frame_info.function
if fname in self._methods:
return fname
return "unknown"
def _checkpoint_state_for_ask(self) -> None:
"""Auto-checkpoint flow state before waiting for user input.
If persistence is configured, saves the current state so that
``self.state`` is recoverable even if the process crashes while
waiting for input.
This is best-effort: if persistence is not configured, this is a no-op.
"""
if self._persistence is None:
return
try:
state_data = (
self._state
if isinstance(self._state, dict)
else self._state.model_dump()
)
self._persistence.save_state(
flow_uuid=self.flow_id,
method_name="_ask_checkpoint",
state_data=state_data,
)
except Exception:
logger.debug("Failed to checkpoint state before ask()", exc_info=True)
def ask(
self,
message: str,
timeout: float | None = None,
metadata: dict[str, Any] | None = None,
) -> str | None:
"""Request input from the user during flow execution.
Blocks the current thread until the user provides input or the
timeout expires. Works in both sync and async flow methods (the
flow framework runs sync methods in a thread pool via
``asyncio.to_thread``, so the event loop stays free).
Timeout ensures flows always terminate. When timeout expires,
``None`` is returned, enabling the pattern::
while (msg := self.ask("You: ", timeout=300)) is not None:
process(msg)
Before waiting for input, the current ``self.state`` is automatically
checkpointed to persistence (if configured) for durability.
Args:
message: The question or prompt to display to the user.
timeout: Maximum seconds to wait for input. ``None`` means
wait indefinitely. When timeout expires, returns ``None``.
metadata: Optional metadata to send to the input provider,
such as user ID, channel, session context. The provider
can use this to route the question to the right recipient.
Returns:
The user's input as a string, or ``None`` on timeout, disconnect,
or provider error. Empty string ``""`` means the user pressed
Enter without typing (intentional empty input).
Example:
```python
class MyFlow(Flow):
@start()
def gather_info(self):
topic = self.ask(
"What topic should we research?",
metadata={"user_id": "u123", "channel": "#research"},
)
if topic is None:
return "No input received"
return topic
```
"""
from concurrent.futures import ThreadPoolExecutor, TimeoutError as FuturesTimeoutError
from datetime import datetime
from crewai.events.types.flow_events import (
FlowInputReceivedEvent,
FlowInputRequestedEvent,
)
from crewai.flow.input_provider import InputResponse
method_name = self._resolve_calling_method_name()
# Emit input requested event
crewai_event_bus.emit(
self,
FlowInputRequestedEvent(
type="flow_input_requested",
flow_name=self.name or self.__class__.__name__,
method_name=method_name,
message=message,
metadata=metadata,
),
)
# Auto-checkpoint state before waiting
self._checkpoint_state_for_ask()
provider = self._resolve_input_provider()
raw: str | InputResponse | None = None
try:
if timeout is not None:
with ThreadPoolExecutor(max_workers=1) as executor:
future = executor.submit(
provider.request_input, message, self, metadata
)
try:
raw = future.result(timeout=timeout)
except FuturesTimeoutError:
raw = None
else:
raw = provider.request_input(message, self, metadata=metadata)
except KeyboardInterrupt:
raise
except Exception:
logger.debug("Input provider error in ask()", exc_info=True)
raw = None
# Normalize provider response: str, InputResponse, or None
response: str | None = None
response_metadata: dict[str, Any] | None = None
if isinstance(raw, InputResponse):
response = raw.text
response_metadata = raw.metadata
elif isinstance(raw, str):
response = raw
else:
response = None
# Record in history
self._input_history.append({
"message": message,
"response": response,
"method_name": method_name,
"timestamp": datetime.now(),
"metadata": metadata,
"response_metadata": response_metadata,
})
# Emit input received event
crewai_event_bus.emit(
self,
FlowInputReceivedEvent(
type="flow_input_received",
flow_name=self.name or self.__class__.__name__,
method_name=method_name,
message=message,
response=response,
metadata=metadata,
response_metadata=response_metadata,
),
)
return response
def _request_human_feedback(
self,
message: str,

32
lib/crewai/src/crewai/flow/flow_config.py
View File

@@ -11,6 +11,7 @@ from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from crewai.flow.async_feedback.types import HumanFeedbackProvider
from crewai.flow.input_provider import InputProvider
class FlowConfig:
@@ -20,10 +21,15 @@ class FlowConfig:
hitl_provider: The human-in-the-loop feedback provider.
Defaults to None (uses console input).
Can be overridden by deployments at startup.
input_provider: The input provider used by ``Flow.ask()``.
Defaults to None (uses ``ConsoleProvider``).
Can be overridden by
deployments at startup.
"""
def __init__(self) -> None:
self._hitl_provider: HumanFeedbackProvider | None = None
self._input_provider: InputProvider | None = None
@property
def hitl_provider(self) -> Any:
@@ -35,6 +41,32 @@ class FlowConfig:
"""Set the HITL provider."""
self._hitl_provider = provider
@property
def input_provider(self) -> Any:
"""Get the configured input provider for ``Flow.ask()``.
Returns:
The configured InputProvider instance, or None if not set
(in which case ``ConsoleInputProvider`` is used as default).
"""
return self._input_provider
@input_provider.setter
def input_provider(self, provider: Any) -> None:
"""Set the input provider for ``Flow.ask()``.
Args:
provider: An object implementing the ``InputProvider`` protocol.
Example:
```python
from crewai.flow import flow_config
flow_config.input_provider = WebSocketInputProvider(...)
```
"""
self._input_provider = provider
# Singleton instance
flow_config = FlowConfig()

151
lib/crewai/src/crewai/flow/input_provider.py Normal file
View File

@@ -0,0 +1,151 @@
"""Input provider protocol for Flow.ask().
This module provides the InputProvider protocol and InputResponse dataclass
used by Flow.ask() to request input from users during flow execution.
The default implementation is ``ConsoleProvider`` (from
``crewai.flow.async_feedback.providers``), which serves both feedback
and input collection via console.
Example (default console input):
```python
from crewai.flow import Flow, start
class MyFlow(Flow):
@start()
def gather_info(self):
topic = self.ask("What topic should we research?")
return topic
```
Example (custom provider with metadata):
```python
from crewai.flow import Flow, start
from crewai.flow.input_provider import InputProvider, InputResponse
class SlackProvider:
def request_input(self, message, flow, metadata=None):
channel = metadata.get("channel", "#general") if metadata else "#general"
thread = self.post_question(channel, message)
reply = self.wait_for_reply(thread)
return InputResponse(
text=reply.text,
metadata={"responded_by": reply.user_id, "thread_id": thread.id},
)
class MyFlow(Flow):
input_provider = SlackProvider()
@start()
def gather_info(self):
topic = self.ask("What topic?", metadata={"channel": "#research"})
return topic
```
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable
if TYPE_CHECKING:
from crewai.flow.flow import Flow
@dataclass
class InputResponse:
"""Response from an InputProvider, optionally carrying metadata.
Simple providers can just return a string from ``request_input()``.
Providers that need to send metadata back (e.g., who responded,
thread ID, external timestamps) return an ``InputResponse`` instead.
``ask()`` normalizes both cases -- callers always get ``str | None``.
The response metadata is stored in ``_input_history`` and emitted
in ``FlowInputReceivedEvent``.
Attributes:
text: The user's input text, or None if unavailable.
metadata: Optional metadata from the provider about the response
(e.g., who responded, thread ID, timestamps).
Example:
```python
class MyProvider:
def request_input(self, message, flow, metadata=None):
response = get_response_from_external_system(message)
return InputResponse(
text=response.text,
metadata={"responded_by": response.user_id},
)
```
"""
text: str | None
metadata: dict[str, Any] | None = field(default=None)
@runtime_checkable
class InputProvider(Protocol):
"""Protocol for user input collection strategies.
Implement this protocol to create custom input providers that integrate
with external systems like websockets, web UIs, Slack, or custom APIs.
The default provider is ``ConsoleProvider``, which blocks waiting for
console input via Python's built-in ``input()`` function.
Providers are always synchronous. The flow framework runs sync methods
in a thread pool (via ``asyncio.to_thread``), so ``ask()`` never blocks
the event loop even inside async flow methods.
Providers can return either:
- ``str | None`` for simple cases (no response metadata)
- ``InputResponse`` when they need to send metadata back with the answer
Example (simple):
```python
class SimpleProvider:
def request_input(self, message: str, flow: Flow) -> str | None:
return input(message)
```
Example (with metadata):
```python
class SlackProvider:
def request_input(self, message, flow, metadata=None):
channel = metadata.get("channel") if metadata else "#general"
reply = self.post_and_wait(channel, message)
return InputResponse(
text=reply.text,
metadata={"responded_by": reply.user_id},
)
```
"""
def request_input(
self,
message: str,
flow: Flow[Any],
metadata: dict[str, Any] | None = None,
) -> str | InputResponse | None:
"""Request input from the user.
Args:
message: The question or prompt to display to the user.
flow: The Flow instance requesting input. Can be used to
access flow state, name, or other context.
metadata: Optional metadata from the caller, such as user ID,
channel, session context, etc. Providers can use this to
route the question to the right recipient.
Returns:
The user's input as a string, an ``InputResponse`` with text
and optional response metadata, or None if input is unavailable
(e.g., user cancelled, connection dropped).
"""
...

1155
lib/crewai/tests/test_flow_ask.py Normal file
View File

File diff suppressed because it is too large Load Diff