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

Report bounded planning step outcomes

Browse Source
This commit is contained in:
lorenzejay
2026-07-22 15:31:25 -07:00
parent 40279e3152
commit 67c68b96dd
7 changed files with 326 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
lib/crewai/src/crewai/agents/step_executor.py
View File

@@ -43,7 +43,7 @@ from crewai.utilities.agent_utils import (
setup_native_tools,
)
from crewai.utilities.i18n import I18N_DEFAULT
from crewai.utilities.planning_types import TodoItem
from crewai.utilities.planning_types import StepOutcome, TodoItem
from crewai.utilities.step_execution_context import StepExecutionContext, StepResult
from crewai.utilities.string_utils import sanitize_tool_name
from crewai.utilities.tool_utils import execute_tool_and_check_finality
@@ -60,6 +60,21 @@ if TYPE_CHECKING:
from crewai.tools.structured_tool import CrewStructuredTool
class _StepExecutionTerminatedError(Exception):
"""Signal a bounded step termination while preserving partial results."""
def __init__(
self,
outcome: StepOutcome,
result: str,
reason: str,
) -> None:
super().__init__(reason)
self.outcome = outcome
self.result = result
self.reason = reason
class StepExecutor:
"""Executes a SINGLE todo item using direct-action execution.
@@ -175,11 +190,20 @@ class StepExecutor:
elapsed = time.monotonic() - start_time
return StepResult(
success=True,
outcome="completed",
result=result_text,
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
except _StepExecutionTerminatedError as termination:
elapsed = time.monotonic() - start_time
return StepResult(
outcome=termination.outcome,
result=termination.result,
termination_reason=termination.reason,
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
except Exception as e:
if self._use_native_tools and is_native_tool_calling_unsupported_error(e):
try:
@@ -213,19 +237,28 @@ class StepExecutor:
self._validate_expected_tool_usage(todo, tool_calls_made)
elapsed = time.monotonic() - start_time
return StepResult(
success=True,
outcome="completed",
result=result_text,
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
except _StepExecutionTerminatedError as termination:
elapsed = time.monotonic() - start_time
return StepResult(
outcome=termination.outcome,
result=termination.result,
termination_reason=termination.reason,
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
except Exception as fallback_error:
e = fallback_error
elapsed = time.monotonic() - start_time
return StepResult(
success=False,
outcome="failed",
result="",
error=str(e),
termination_reason=str(e),
tool_calls_made=tool_calls_made,
execution_time=elapsed,
)
@@ -337,7 +370,14 @@ class StepExecutor:
if step_timeout and start_time:
elapsed = time.monotonic() - start_time
if elapsed >= step_timeout:
return last_tool_result or f"Step timed out after {elapsed:.0f}s"
reason = (
f"Step timed out after {elapsed:.0f}s (limit: {step_timeout}s)."
)
raise _StepExecutionTerminatedError(
"timed_out",
last_tool_result,
reason,
)
answer = self.llm.call(
messages,
callbacks=self.callbacks,
@@ -364,7 +404,12 @@ class StepExecutor:
return answer_str
return last_tool_result
raise _StepExecutionTerminatedError(
"iteration_exhausted",
last_tool_result,
f"Step reached max_step_iterations ({max_step_iterations}) "
"before producing a final answer.",
)
def _execute_text_tool_with_events(
self, formatted: AgentAction, todo: TodoItem
@@ -547,10 +592,13 @@ class StepExecutor:
if step_timeout and start_time:
elapsed = time.monotonic() - start_time
if elapsed >= step_timeout:
return (
"\n\n".join(accumulated_results)
if accumulated_results
else f"Step timed out after {elapsed:.0f}s"
reason = (
f"Step timed out after {elapsed:.0f}s (limit: {step_timeout}s)."
)
raise _StepExecutionTerminatedError(
"timed_out",
"\n\n".join(accumulated_results),
reason,
)
answer = self.llm.call(
messages,
@@ -575,7 +623,12 @@ class StepExecutor:
return str(answer)
return "\n".join(filter(None, accumulated_results))
raise _StepExecutionTerminatedError(
"iteration_exhausted",
"\n".join(filter(None, accumulated_results)),
f"Step reached max_step_iterations ({max_step_iterations}) "
"before producing a final answer.",
)
def _execute_native_tool_calls(
self,

3
lib/crewai/src/crewai/events/types/observation_events.py
View File

@@ -8,6 +8,7 @@ continuation, refinement, or replanning.
from typing import Any, Literal
from crewai.events.base_events import BaseEvent
from crewai.utilities.planning_types import StepOutcome
class ObservationEvent(BaseEvent):
@@ -54,6 +55,8 @@ class PlanStepCompletedEvent(PlanStepEvent):
type: Literal["plan_step_completed"] = "plan_step_completed"
success: bool = True
outcome: StepOutcome = "completed"
termination_reason: str | None = None
result: str | None = None
error: str | None = None

55
lib/crewai/src/crewai/experimental/agent_executor.py
View File

@@ -103,6 +103,7 @@ from crewai.utilities.i18n import I18N_DEFAULT
from crewai.utilities.planning_types import (
PlanStep,
StepObservation,
StepOutcome,
TodoItem,
TodoList,
)
@@ -417,6 +418,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
todo: TodoItem,
*,
success: bool,
outcome: StepOutcome,
termination_reason: str | None = None,
result: str | None = None,
error: str | None = None,
) -> None:
@@ -429,6 +432,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
step_description=todo.description,
tool_to_use=todo.tool_to_use,
success=success,
outcome=outcome,
termination_reason=termination_reason,
result=result,
error=error,
from_task=self.task,
@@ -454,22 +459,44 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
self.state.todos.mark_completed(step_number, result=result)
todo = self.state.todos.get_by_step_number(step_number)
if todo and previous_status != "completed":
self._emit_plan_step_completed(todo, success=True, result=result)
self._emit_plan_step_completed(
todo,
success=True,
outcome="completed",
result=result,
)
def _mark_todo_failed(
self,
step_number: int,
result: str | None = None,
error: str | None = None,
outcome: StepOutcome | None = None,
termination_reason: str | None = None,
) -> None:
todo = self.state.todos.get_by_step_number(step_number)
previous_status = todo.status if todo else None
self.state.todos.mark_failed(step_number, result=result)
final_outcome = outcome or (
todo.outcome
if todo and todo.outcome not in (None, "completed")
else "failed"
)
final_reason = (
termination_reason or error or (todo.termination_reason if todo else None)
)
self.state.todos.mark_failed(
step_number,
result=result,
outcome=final_outcome,
termination_reason=final_reason,
)
todo = self.state.todos.get_by_step_number(step_number)
if todo and previous_status != "failed":
self._emit_plan_step_completed(
todo,
success=False,
outcome=final_outcome,
termination_reason=final_reason,
result=result,
error=error,
)
@@ -578,17 +605,23 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
"""Observe a completed step via LLM or a lightweight heuristic."""
from crewai.agents.planner_observer import PlannerObserver
if step_success is None:
step_success = self._step_success_from_log(completed_step.step_number)
if self._should_observe_steps():
observer = self._ensure_planner_observer()
return observer.observe(
observation = observer.observe(
completed_step=completed_step,
result=result,
all_completed=all_completed,
remaining_todos=remaining_todos,
)
if step_success is False and observation.step_completed_successfully:
return observation.model_copy(
update={"step_completed_successfully": False}
)
return observation
if step_success is None:
step_success = self._step_success_from_log(completed_step.step_number)
if step_success is None:
step_success = True
@@ -1129,12 +1162,16 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
# Do NOT mark completed here — observation logic decides
current.result = result.result
current.outcome = result.outcome
current.termination_reason = result.termination_reason
self.state.execution_log.append(
{
"type": "step_execution",
"step_number": current.step_number,
"success": result.success,
"outcome": result.outcome,
"termination_reason": result.termination_reason,
"result_preview": result.result[:200] if result.result else "",
"error": result.error,
"tool_calls": result.tool_calls_made,
@@ -1251,10 +1288,14 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
if isinstance(item, BaseException):
error_msg = f"Error: {item!s}"
todo.result = error_msg
todo.outcome = "failed"
todo.termination_reason = error_msg
self._mark_todo_failed(
todo.step_number,
result=error_msg,
error=error_msg,
outcome="failed",
termination_reason=error_msg,
)
if self.agent.verbose:
PRINTER.print(
@@ -1265,12 +1306,16 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor):
_returned_todo, result = item
step_result = cast(StepResult, result)
todo.result = step_result.result
todo.outcome = step_result.outcome
todo.termination_reason = step_result.termination_reason
self.state.execution_log.append(
{
"type": "step_execution",
"step_number": todo.step_number,
"success": step_result.success,
"outcome": step_result.outcome,
"termination_reason": step_result.termination_reason,
"result_preview": step_result.result[:200]
if step_result.result
else "",

11
lib/crewai/src/crewai/lite_agent_output.py
View File

@@ -7,7 +7,7 @@ from typing import Any
from pydantic import BaseModel, Field
from crewai.types.usage_metrics import UsageMetrics
from crewai.utilities.planning_types import TodoItem
from crewai.utilities.planning_types import StepOutcome, TodoItem
from crewai.utilities.types import LLMMessage
@@ -23,6 +23,13 @@ class TodoExecutionResult(BaseModel):
result: str | None = Field(
default=None, description="Result or error message from execution"
)
outcome: StepOutcome | None = Field(
default=None, description="How execution of the step terminated"
)
termination_reason: str | None = Field(
default=None,
description="Human-readable reason for non-completed step termination",
)
depends_on: list[int] = Field(
default_factory=list, description="Step numbers this depended on"
)
@@ -82,6 +89,8 @@ class LiteAgentOutput(BaseModel):
tool_used=item.tool_to_use,
status=item.status,
result=item.result,
outcome=item.outcome,
termination_reason=item.termination_reason,
depends_on=item.depends_on,
)
for item in todo_items

22
lib/crewai/src/crewai/utilities/planning_types.py
View File

@@ -9,6 +9,7 @@ from pydantic import BaseModel, Field, field_validator
TodoStatus = Literal["pending", "running", "completed", "failed"]
StepOutcome = Literal["completed", "timed_out", "iteration_exhausted", "failed"]
class PlanStep(BaseModel):
@@ -40,6 +41,14 @@ class TodoItem(BaseModel):
result: str | None = Field(
default=None, description="Result after completion, if any"
)
outcome: StepOutcome | None = Field(
default=None,
description="How execution of this step terminated, once available",
)
termination_reason: str | None = Field(
default=None,
description="Human-readable reason for non-completed step termination",
)
class TodoList(BaseModel):
@@ -98,14 +107,25 @@ class TodoList(BaseModel):
item = self.get_by_step_number(step_number)
if item:
item.status = "completed"
item.outcome = "completed"
item.termination_reason = None
if result is not None:
item.result = result
def mark_failed(self, step_number: int, result: str | None = None) -> None:
def mark_failed(
self,
step_number: int,
result: str | None = None,
*,
outcome: StepOutcome = "failed",
termination_reason: str | None = None,
) -> None:
"""Mark a todo as failed by step number."""
item = self.get_by_step_number(step_number)
if item:
item.status = "failed"
item.outcome = outcome
item.termination_reason = termination_reason
if result is not None:
item.result = result

20
lib/crewai/src/crewai/utilities/step_execution_context.py
View File

@@ -9,6 +9,8 @@ from __future__ import annotations
from dataclasses import dataclass, field
from crewai.utilities.planning_types import StepOutcome
@dataclass(frozen=True)
class StepExecutionContext:
@@ -50,15 +52,25 @@ class StepResult:
to subsequent steps or the Planner.
Attributes:
success: Whether the step completed successfully.
outcome: How execution of the step terminated.
result: The final output string from the step.
error: Error message if the step failed (None on success).
termination_reason: Human-readable reason for a non-completed outcome.
tool_calls_made: List of tool names invoked (for debugging/logging only).
execution_time: Wall-clock time in seconds for the step execution.
"""
success: bool
outcome: StepOutcome
result: str
error: str | None = None
termination_reason: str | None = None
tool_calls_made: list[str] = field(default_factory=list)
execution_time: float = 0.0
@property
def success(self) -> bool:
"""Whether the step reached the completed outcome."""
return self.outcome == "completed"
@property
def error(self) -> str | None:
"""Backward-compatible alias for the termination reason."""
return self.termination_reason

161
lib/crewai/tests/agents/test_agent_executor.py
View File

@@ -812,9 +812,106 @@ class TestStepExecutorCriticalFixes:
result = step_executor.execute(todo, context)
assert result.success is False
assert result.outcome == "failed"
assert result.error is not None
assert "Expected tool 'count_words' was not called" in result.error
def test_step_executor_marks_text_iteration_exhaustion_as_failure(
self, step_executor
):
"""A tool result at the iteration limit is partial, not successful."""
todo = TodoItem(
step_number=1,
description="Count words in input text.",
tool_to_use="count_words",
)
context = StepExecutionContext(task_description="task", task_goal="goal")
action = AgentAction(
thought="Need a tool",
tool="count_words",
tool_input='{"text":"hello world"}',
text="Action: count_words",
)
with (
patch(
"crewai.agents.step_executor.process_llm_response",
return_value=action,
),
patch.object(
step_executor,
"_execute_text_tool_with_events",
return_value="2",
),
):
step_executor.llm.call.return_value = action.text
result = step_executor.execute(
todo,
context,
max_step_iterations=1,
)
assert result.success is False
assert result.outcome == "iteration_exhausted"
assert result.result == "2"
assert result.termination_reason is not None
assert "max_step_iterations (1)" in result.termination_reason
def test_step_executor_marks_native_iteration_exhaustion_as_failure(
self, step_executor
):
"""Native tool output at the iteration limit remains a partial result."""
step_executor._use_native_tools = True
step_executor._openai_tools = [
{"type": "function", "function": {"name": "count_words"}}
]
todo = TodoItem(step_number=1, description="Count words")
context = StepExecutionContext(task_description="task", task_goal="goal")
tool_calls = [object()]
step_executor.llm.call.return_value = tool_calls
with (
patch("crewai.agents.step_executor.is_tool_call_list", return_value=True),
patch.object(
step_executor,
"_execute_native_tool_calls",
return_value="2",
),
):
result = step_executor.execute(
todo,
context,
max_step_iterations=1,
)
assert result.success is False
assert result.outcome == "iteration_exhausted"
assert result.result == "2"
assert result.termination_reason is not None
assert "max_step_iterations (1)" in result.termination_reason
def test_step_executor_marks_timeout_as_failure(self, step_executor):
"""Reaching the wall-clock limit returns an explicit timeout outcome."""
step_executor._use_native_tools = False
todo = TodoItem(step_number=1, description="Count words")
context = StepExecutionContext(task_description="task", task_goal="goal")
with patch(
"crewai.agents.step_executor.time.monotonic",
side_effect=[0.5, 2.0, 2.0],
):
result = step_executor.execute(
todo,
context,
step_timeout=1,
)
assert result.success is False
assert result.outcome == "timed_out"
assert result.result == ""
assert result.termination_reason == "Step timed out after 2s (limit: 1s)."
step_executor.llm.call.assert_not_called()
def test_step_executor_text_tool_emits_usage_events(self, step_executor):
"""Text-parsed tool execution should emit started and finished events."""
started_events: list[ToolUsageStartedEvent] = []
@@ -929,6 +1026,8 @@ class TestStepExecutorCriticalFixes:
assert started_events[0].tool_to_use == "search"
assert len(completed_events) == 1
assert completed_events[0].success is True
assert completed_events[0].outcome == "completed"
assert completed_events[0].termination_reason is None
assert completed_events[0].step_number == 1
assert completed_events[0].result == "Found release"
@@ -953,16 +1052,44 @@ class TestStepExecutorCriticalFixes:
1,
result="Error: no result",
error="No result",
outcome="timed_out",
termination_reason="Step timed out after 30s.",
)
crewai_event_bus.flush()
assert todo.status == "failed"
assert len(completed_events) == 1
assert completed_events[0].success is False
assert completed_events[0].outcome == "timed_out"
assert completed_events[0].termination_reason == "Step timed out after 30s."
assert completed_events[0].step_number == 1
assert completed_events[0].result == "Error: no result"
assert completed_events[0].error == "No result"
def test_lite_agent_output_includes_step_termination_details(self):
"""Standalone agent output exposes how each planned step terminated."""
from crewai.lite_agent_output import LiteAgentOutput
todos = LiteAgentOutput.from_todo_items(
[
TodoItem(
step_number=1,
description="Run bounded work",
status="failed",
result="partial result",
outcome="iteration_exhausted",
termination_reason=(
"Step reached max_step_iterations (1) before producing "
"a final answer."
),
)
]
)
assert todos[0].outcome == "iteration_exhausted"
assert todos[0].termination_reason is not None
assert "max_step_iterations (1)" in todos[0].termination_reason
@patch("crewai.experimental.agent_executor.handle_output_parser_exception")
def test_recover_from_parser_error(
self, mock_handle_exception, mock_dependencies
@@ -1849,6 +1976,40 @@ class TestReasoningEffort:
executor._ensure_planner_observer.assert_not_called()
assert observation.step_completed_successfully is True
def test_observer_cannot_override_failed_step_execution(self):
"""An optimistic observer cannot turn an execution failure into success."""
from crewai.agent.planning_config import PlanningConfig
from crewai.experimental.agent_executor import AgentExecutor
from crewai.utilities.planning_types import StepObservation, TodoItem
executor = Mock(spec=AgentExecutor)
executor.agent = Mock()
executor.agent.planning_config = PlanningConfig(reasoning_effort="medium")
executor._should_observe_steps = (
AgentExecutor._should_observe_steps.__get__(executor)
)
executor._observe_completed_step = (
AgentExecutor._observe_completed_step.__get__(executor)
)
executor._ensure_planner_observer.return_value.observe.return_value = (
StepObservation(
step_completed_successfully=True,
key_information_learned="Partial data is usable",
remaining_plan_still_valid=True,
)
)
todo = TodoItem(step_number=1, description="Run bounded work")
observation = executor._observe_completed_step(
completed_step=todo,
result="partial result",
all_completed=[],
remaining_todos=[],
step_success=False,
)
assert observation.step_completed_successfully is False
@pytest.mark.vcr()
def test_reasoning_effort_low_skips_decide_and_replan(self):
"""Low effort: heuristic observe, no decide/replan/refine LLM pipeline.