Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 21 additions & 6 deletions lib/crewai/src/crewai/utilities/reasoning_handler.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import json
import logging
import re
from typing import TYPE_CHECKING, Any, Final, Literal, cast

from pydantic import BaseModel, Field
Expand Down Expand Up @@ -409,7 +410,7 @@ def _create_reasoning_plan(
return (
response_str,
[],
"READY: I am ready to execute the task." in response_str,
self._is_ready(response_str),
)

except Exception as e:
Expand All @@ -433,7 +434,7 @@ def _create_reasoning_plan(
return (
fallback_str,
[],
"READY: I am ready to execute the task." in fallback_str,
self._is_ready(fallback_str),
)
except Exception as inner_e:
self.logger.error(f"Error during fallback text parsing: {inner_e!s}")
Expand Down Expand Up @@ -579,6 +580,23 @@ def _create_refine_prompt(self, current_plan: str) -> str:
current_plan=current_plan,
)

@staticmethod
def _is_ready(response: str) -> bool:
"""Check whether a text response indicates the agent is ready.

The prompt templates instruct models to conclude with "READY" or
"NOT READY". Older prompts used the full phrase
"READY: I am ready to execute the task." Both forms are accepted
so that models following either convention work correctly.
"""
upper = response.upper()
if "NOT READY" in upper:
return False
return bool(
"READY: I AM READY TO EXECUTE THE TASK." in upper
or re.search(r"\bREADY\b", upper)
)

@staticmethod
def _parse_planning_response(response: str) -> tuple[str, bool]:
"""Parses the planning response to extract the plan and readiness.
Expand All @@ -592,10 +610,7 @@ def _parse_planning_response(response: str) -> tuple[str, bool]:
if not response:
return "No plan was generated.", False

plan = response
ready = "READY: I am ready to execute the task." in response

return plan, ready
return response, AgentReasoning._is_ready(response)


AgentPlanning = AgentReasoning
Expand Down
61 changes: 61 additions & 0 deletions lib/crewai/tests/agents/test_agent_reasoning.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,67 @@

from crewai import Agent, PlanningConfig, Task
from crewai.llm import LLM
from crewai.utilities.reasoning_handler import AgentReasoning


class TestIsReady:
"""Regression tests for AgentReasoning._is_ready.
The prompt templates tell models to conclude with "READY" or "NOT READY".
The detection must handle both the short form from current prompts and the
legacy long form, and must never treat "NOT READY" as ready.
"""

def test_legacy_full_phrase(self):
assert AgentReasoning._is_ready("READY: I am ready to execute the task.")

def test_legacy_full_phrase_case_insensitive(self):
assert AgentReasoning._is_ready("ready: i am ready to execute the task.")

def test_short_form_uppercase(self):
assert AgentReasoning._is_ready("Here is my plan.\n\nREADY")

def test_short_form_mixed_case(self):
assert AgentReasoning._is_ready("My plan is complete.\n\nReady")

def test_not_ready_returns_false(self):
assert not AgentReasoning._is_ready("NOT READY")

def test_not_ready_inline_returns_false(self):
assert not AgentReasoning._is_ready("My plan needs more work. NOT READY")

def test_empty_string_returns_false(self):
assert not AgentReasoning._is_ready("")

def test_no_keyword_returns_false(self):
assert not AgentReasoning._is_ready("Here is my plan. I need more information.")

def test_ready_as_substring_returns_false(self):
assert not AgentReasoning._is_ready("I already need more info.")

def test_unready_substring_returns_false(self):
assert not AgentReasoning._is_ready("The system is unready for deployment.")


class TestParsePlanningResponse:
def test_empty_response(self):
plan, ready = AgentReasoning._parse_planning_response("")
assert plan == "No plan was generated."
assert ready is False

def test_short_ready(self):
_, ready = AgentReasoning._parse_planning_response("Step 1: do X.\n\nREADY")
assert ready is True

def test_not_ready(self):
_, ready = AgentReasoning._parse_planning_response("Step 1: do X.\n\nNOT READY")
assert ready is False

def test_legacy_phrase(self):
_, ready = AgentReasoning._parse_planning_response(
"Step 1: do X.\nREADY: I am ready to execute the task."
)
assert ready is True



Expand Down