diff --git a/lib/crewai/src/crewai/agent/core.py b/lib/crewai/src/crewai/agent/core.py index 0ac21743a..5f5c5a027 100644 --- a/lib/crewai/src/crewai/agent/core.py +++ b/lib/crewai/src/crewai/agent/core.py @@ -74,6 +74,7 @@ from crewai.events.types.memory_events import ( ) from crewai.events.types.skill_events import SkillUsedEvent from crewai.experimental.agent_executor import AgentExecutor +from crewai.hooks.dispatch import HookAborted from crewai.knowledge.knowledge import Knowledge from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource from crewai.lite_agent_output import LiteAgentOutput @@ -139,7 +140,10 @@ if TYPE_CHECKING: # Deliberate stops, not transient errors: never swallowed into the # max_retry_limit loop. -_passthrough_exceptions: tuple[type[Exception], ...] = (ToolExecutionFailedError,) +_passthrough_exceptions: tuple[type[Exception], ...] = ( + ToolExecutionFailedError, + HookAborted, +) _EXECUTOR_CLASS_MAP: dict[str, type] = { "CrewAgentExecutor": CrewAgentExecutor, @@ -711,6 +715,9 @@ class Agent(BaseAgent): error=str(e), ), ) + # a deny aborts the task; any other failure degrades to no memory + if isinstance(e, HookAborted): + raise return task_prompt @@ -1438,6 +1445,18 @@ class Agent(BaseAgent): ), ) return rewritten_query + except HookAborted as e: + # A deny still owes the started event above its terminal event; only + # the fallback to no query is skipped. + crewai_event_bus.emit( + self, + event=KnowledgeQueryFailedEvent( + error=str(e), + from_task=task, + from_agent=self, + ), + ) + raise except Exception as e: crewai_event_bus.emit( self, @@ -1634,6 +1653,9 @@ class Agent(BaseAgent): error=str(e), ), ) + # a deny aborts the kickoff; any other failure degrades to no memory + if isinstance(e, HookAborted): + raise inputs: dict[str, Any] = { "input": formatted_messages, @@ -1815,6 +1837,8 @@ class Agent(BaseAgent): extracted = agent_memory.extract_memories(raw) if extracted: agent_memory.remember_many(extracted) + except HookAborted: + raise except Exception as e: self._logger.log("error", f"Failed to save kickoff result to memory: {e}") diff --git a/lib/crewai/src/crewai/agent/utils.py b/lib/crewai/src/crewai/agent/utils.py index 47701c8e4..20eb06b47 100644 --- a/lib/crewai/src/crewai/agent/utils.py +++ b/lib/crewai/src/crewai/agent/utils.py @@ -15,6 +15,7 @@ from crewai.events.types.knowledge_events import ( KnowledgeRetrievalStartedEvent, KnowledgeSearchQueryFailedEvent, ) +from crewai.hooks.dispatch import HookAborted from crewai.knowledge.utils.knowledge_utils import extract_knowledge_context from crewai.utilities.pydantic_schema_utils import generate_model_description from crewai.utilities.types import LLMMessage @@ -53,6 +54,8 @@ def handle_reasoning(agent: Agent, task: Task) -> None: planning_handler.handle_agent_reasoning() ) task.description += f"\n\nPlanning:\n{planning_output.plan.plan}" + except HookAborted: + raise except Exception as e: agent._logger.log("error", f"Error during planning: {e!s}") @@ -195,6 +198,9 @@ def handle_knowledge_retrieval( from_agent=agent, ), ) + # a deny aborts the task; any other failure degrades to no knowledge + if isinstance(e, HookAborted): + raise return task_prompt @@ -391,4 +397,7 @@ async def ahandle_knowledge_retrieval( from_agent=agent, ), ) + # a deny aborts the task; any other failure degrades to no knowledge + if isinstance(e, HookAborted): + raise return task_prompt diff --git a/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py b/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py index d251b1d36..39ac025a5 100644 --- a/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py +++ b/lib/crewai/src/crewai/agents/agent_builder/base_agent_executor.py @@ -30,6 +30,8 @@ class BaseAgentExecutor(BaseModel): def _save_to_memory(self, output: AgentFinish) -> None: """Save task result to unified memory (memory or crew._memory).""" + from crewai.hooks.dispatch import HookAborted + if self.agent is None: return memory = getattr(self.agent, "memory", None) or ( @@ -61,5 +63,7 @@ class BaseAgentExecutor(BaseModel): ) else: memory.remember_many(extracted, agent_role=self.agent.role) + except HookAborted: + raise except Exception as e: self.agent._logger.log("error", f"Failed to save to memory: {e}") diff --git a/lib/crewai/src/crewai/agents/planner_observer.py b/lib/crewai/src/crewai/agents/planner_observer.py index 2882ac780..ce08e6416 100644 --- a/lib/crewai/src/crewai/agents/planner_observer.py +++ b/lib/crewai/src/crewai/agents/planner_observer.py @@ -131,7 +131,13 @@ class PlannerObserver: StepObservation with the Planner's analysis. Any suggested refinements are structured StepRefinement objects ready for direct application — no second LLM call needed. + + Raises: + HookAborted: A `pre_model_call` hook denied the observation call. + Every other failure degrades to a conservative observation. """ + from crewai.hooks.dispatch import HookAborted + agent_role = self.agent.role crewai_event_bus.emit( @@ -188,6 +194,21 @@ class PlannerObserver: return observation + except HookAborted as e: + # A deny still owes the started event above its terminal event; only + # the conservative-replan fallback is skipped. + crewai_event_bus.emit( + self.agent, + event=StepObservationFailedEvent( + agent_role=agent_role, + step_number=completed_step.step_number, + step_description=completed_step.description, + error=str(e), + from_task=self.task, + from_agent=self.agent, + ), + ) + raise except Exception as e: logger.warning( f"Observation LLM call failed: {e}. Defaulting to conservative replan." diff --git a/lib/crewai/src/crewai/agents/step_executor.py b/lib/crewai/src/crewai/agents/step_executor.py index 415586bb9..a25db4e2f 100644 --- a/lib/crewai/src/crewai/agents/step_executor.py +++ b/lib/crewai/src/crewai/agents/step_executor.py @@ -28,6 +28,7 @@ from crewai.events.types.tool_usage_events import ( ToolUsageFinishedEvent, ToolUsageStartedEvent, ) +from crewai.hooks.dispatch import HookAborted from crewai.tools.tool_failure import ToolExecutionFailedError from crewai.utilities.agent_utils import ( build_text_tool_calling_fallback_message, @@ -181,7 +182,7 @@ class StepExecutor: tool_calls_made=tool_calls_made, execution_time=elapsed, ) - except ToolExecutionFailedError: + except (ToolExecutionFailedError, HookAborted): # A deliberate stop: StepResult(success=False) would let the plan # carry on. raise @@ -224,7 +225,7 @@ class StepExecutor: tool_calls_made=tool_calls_made, execution_time=elapsed, ) - except ToolExecutionFailedError: + except (ToolExecutionFailedError, HookAborted): # Same as the outer handler, reached via the text-tooling # fallback. raise diff --git a/lib/crewai/src/crewai/experimental/agent_executor.py b/lib/crewai/src/crewai/experimental/agent_executor.py index 45fffa1e4..d51e7aaca 100644 --- a/lib/crewai/src/crewai/experimental/agent_executor.py +++ b/lib/crewai/src/crewai/experimental/agent_executor.py @@ -57,6 +57,7 @@ from crewai.events.types.tool_usage_events import ( from crewai.flow.flow import Flow, listen, or_, router, start from crewai.flow.flow_context import current_flow_id from crewai.flow.types import FlowMethodName +from crewai.hooks.dispatch import HookAborted from crewai.hooks.llm_hooks import ( get_after_llm_call_hooks, get_before_llm_call_hooks, @@ -420,6 +421,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): # Do NOT mutate task.description — it's a shared object that # accumulates plan text on re-invoke. + except HookAborted: + raise except Exception as e: if hasattr(self.agent, "_logger"): self.agent._logger.log("error", f"Error during planning: {e!s}") @@ -1295,6 +1298,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): # todo ↔ result (or exception) mapping. step_results: list[tuple[TodoItem, StepResult]] = [] for todo, item in zip(ready, gathered, strict=True): + if isinstance(item, HookAborted): + raise item if isinstance(item, BaseException): error_msg = f"Error: {item!s}" todo.result = error_msg @@ -2552,6 +2557,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): ) return + except HookAborted: + raise except Exception as e: if self.agent and self.agent.verbose: PRINTER.print( @@ -2704,6 +2711,8 @@ class AgentExecutor(Flow[AgentExecutorState], BaseAgentExecutor): color="green", ) + except HookAborted: + raise except Exception as e: if hasattr(self.agent, "_logger"): self.agent._logger.log("error", f"Error during replanning: {e!s}") diff --git a/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py b/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py index 3f2314a36..6d462a007 100644 --- a/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py +++ b/lib/crewai/src/crewai/experimental/evaluation/evaluation_display.py @@ -311,6 +311,8 @@ class EvaluationDisplayFormatter: scores: list[float | None], strategy: AggregationStrategy, ) -> str: + from crewai.hooks.dispatch import HookAborted + if len(feedbacks) <= 2 and all(len(fb) < 200 for fb in feedbacks): return "\n\n".join( [f"Feedback {i + 1}: {fb}" for i, fb in enumerate(feedbacks)] @@ -372,6 +374,8 @@ class EvaluationDisplayFormatter: raise ValueError("LLM must be initialized") return llm.call(prompt) + except HookAborted: + raise except Exception: return "Synthesized from multiple tasks: " + "\n\n".join( [f"- {fb[:500]}..." for fb in feedbacks] diff --git a/lib/crewai/src/crewai/flow/human_feedback.py b/lib/crewai/src/crewai/flow/human_feedback.py index a1fabd658..0e0785f99 100644 --- a/lib/crewai/src/crewai/flow/human_feedback.py +++ b/lib/crewai/src/crewai/flow/human_feedback.py @@ -253,6 +253,8 @@ def _pre_review_with_lessons( learn_source: str, learn_strict: bool, ) -> Any: + from crewai.hooks.dispatch import HookAborted + try: mem = flow_instance.memory if mem is None: @@ -282,6 +284,8 @@ def _pre_review_with_lessons( return PreReviewResult.model_validate(response).improved_output reviewed = llm_inst.call(messages) return reviewed if isinstance(reviewed, str) else str(reviewed) + except HookAborted: + raise except Exception: if learn_strict: logger.warning( @@ -308,6 +312,8 @@ def _distill_and_store_lessons( learn_source: str, learn_strict: bool, ) -> None: + from crewai.hooks.dispatch import HookAborted + try: mem = flow_instance.memory if mem is None: @@ -344,6 +350,8 @@ def _distill_and_store_lessons( if lessons: mem.remember_many(lessons, source=learn_source) # type: ignore[union-attr] + except HookAborted: + raise except Exception: if learn_strict: logger.warning( diff --git a/lib/crewai/src/crewai/flow/runtime/__init__.py b/lib/crewai/src/crewai/flow/runtime/__init__.py index 01f645e2e..8c756a39d 100644 --- a/lib/crewai/src/crewai/flow/runtime/__init__.py +++ b/lib/crewai/src/crewai/flow/runtime/__init__.py @@ -3847,6 +3847,7 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): from pydantic import BaseModel, Field + from crewai.hooks.dispatch import HookAborted from crewai.llm import LLM from crewai.llms.base_llm import BaseLLM as BaseLLMClass from crewai.utilities.i18n import I18N_DEFAULT @@ -3902,6 +3903,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): logger.warning(f"Unexpected response type: {type(response)}") return outcomes[0] + except HookAborted: + raise except Exception as e: logger.warning( f"Structured output failed, falling back to simple prompting: {e}" @@ -3933,6 +3936,8 @@ class Flow(BaseModel, Generic[T], metaclass=FlowMeta): ) return outcomes[0] + except HookAborted: + raise except Exception as fallback_err: logger.warning( f"Simple prompting also failed: {fallback_err}. " diff --git a/lib/crewai/src/crewai/hooks/dispatch.py b/lib/crewai/src/crewai/hooks/dispatch.py index fd692c900..92826a71f 100644 --- a/lib/crewai/src/crewai/hooks/dispatch.py +++ b/lib/crewai/src/crewai/hooks/dispatch.py @@ -209,7 +209,7 @@ def _resolve_hooks(point: InterceptionPoint) -> list[HookFn]: return global_hooks -def _source_name(source: Any) -> str | None: +def source_name(source: Any) -> str | None: """Best-effort readable name for a hook source.""" if source is None: return None @@ -341,7 +341,7 @@ def run_hooks( except HookAborted as aborted: outcome = "aborted" abort_reason = aborted.reason - abort_source = _source_name(aborted.source) + abort_source = source_name(aborted.source) raise finally: _emit_telemetry( diff --git a/lib/crewai/src/crewai/hooks/llm_hooks.py b/lib/crewai/src/crewai/hooks/llm_hooks.py index 55563e84e..7658bf514 100644 --- a/lib/crewai/src/crewai/hooks/llm_hooks.py +++ b/lib/crewai/src/crewai/hooks/llm_hooks.py @@ -1,5 +1,8 @@ from __future__ import annotations +from collections.abc import Iterator +from contextlib import contextmanager +from contextvars import ContextVar from typing import TYPE_CHECKING, Any, cast from crewai_core.printer import PRINTER @@ -166,14 +169,50 @@ _after_llm_call_hooks: list[AfterLLMCallHookType | AfterLLMCallHookCallable] = ( ) +class LegacyHookBlocked(HookAborted): + """A ``before_llm_call`` hook blocked the call by returning ``False``. + + Distinguishes the boolean convention, which the LLM layer keeps surfacing as + the documented ``ValueError``, from a hook that raised :class:`HookAborted` + itself and must reach the caller as the deny it is. Raised by the reducer and + consumed inside the LLM layer, which re-raises it as + :class:`~crewai.llms.base_llm.LLMCallBlockedError`. + """ + + +_model_call_hooks_dispatched: ContextVar[bool] = ContextVar( + "model_call_hooks_dispatched", default=False +) + + +@contextmanager +def model_call_hooks_dispatched() -> Iterator[None]: + """Mark the window where the model-call hooks already ran for a pending call. + + The executor dispatches with its own richer context (executor, task, crew) + and only then reaches the LLM. Without this marker the LLM layer would + dispatch a second time for the same call. + """ + token = _model_call_hooks_dispatched.set(True) + try: + yield + finally: + _model_call_hooks_dispatched.reset(token) + + +def model_call_hooks_already_dispatched() -> bool: + """Whether an enclosing caller already dispatched hooks for the current call.""" + return _model_call_hooks_dispatched.get() + + def before_llm_call_reducer(context: LLMCallHookContext, result: object) -> bool: """Legacy calling convention for ``pre_model_call`` hooks. - A ``False`` return aborts the call (mapped to :class:`HookAborted`); messages - are modified in place, so no payload replacement occurs here. + A ``False`` return aborts the call (mapped to :class:`LegacyHookBlocked`); + messages are modified in place, so no payload replacement occurs here. """ if result is False: - raise HookAborted(reason="before_llm_call hook returned False") + raise LegacyHookBlocked(reason="before_llm_call hook returned False") return False diff --git a/lib/crewai/src/crewai/lite_agent.py b/lib/crewai/src/crewai/lite_agent.py index 373b774e4..48a130956 100644 --- a/lib/crewai/src/crewai/lite_agent.py +++ b/lib/crewai/src/crewai/lite_agent.py @@ -598,6 +598,8 @@ class LiteAgent(FlowTrackable, BaseModel): def _inject_memory_context(self) -> None: """Recall relevant memories and append to the system message. No-op if _memory is None.""" + from crewai.hooks.dispatch import HookAborted + if self._memory is None: return query = self._get_last_user_content() @@ -641,9 +643,14 @@ class LiteAgent(FlowTrackable, BaseModel): error=str(e), ), ) + # a deny aborts the run; any other failure degrades to no memory + if isinstance(e, HookAborted): + raise def _save_to_memory(self, output_text: str) -> None: """Extract discrete memories from the run and remember each. No-op if _memory is None or read-only.""" + from crewai.hooks.dispatch import HookAborted + if self._memory is None or self._memory.read_only: return input_str = self._get_last_user_content() or "User request" @@ -652,6 +659,8 @@ class LiteAgent(FlowTrackable, BaseModel): extracted = self._memory.extract_memories(raw) if extracted: self._memory.remember_many(extracted, agent_role=self.role) + except HookAborted: + raise except Exception as e: if self.verbose: PRINTER.print( diff --git a/lib/crewai/src/crewai/llm.py b/lib/crewai/src/crewai/llm.py index 371aadf87..1eb52255f 100644 --- a/lib/crewai/src/crewai/llm.py +++ b/lib/crewai/src/crewai/llm.py @@ -31,10 +31,12 @@ from crewai.events.types.tool_usage_events import ( ToolUsageFinishedEvent, ToolUsageStartedEvent, ) +from crewai.hooks.dispatch import HookAborted from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.base_llm import ( BaseLLM, JsonResponseFormat, + LLMCallBlockedError, get_current_call_id, llm_call_context, ) @@ -1039,12 +1041,15 @@ class LLM(BaseLLM): if not tool_calls or not available_functions: if response_model and self.is_litellm: + from crewai.hooks.llm_hooks import model_call_hooks_dispatched + instructor_instance = InternalInstructor( content=full_response, model=response_model, llm=self, ) - result = instructor_instance.to_pydantic() + with model_call_hooks_dispatched(): + result = instructor_instance.to_pydantic() structured_response = result.model_dump_json() usage_dict = self._usage_to_dict(usage_info) self._handle_emit_call_events( @@ -1241,6 +1246,7 @@ class LLM(BaseLLM): str: The response text """ if response_model and self.is_litellm: + from crewai.hooks.llm_hooks import model_call_hooks_dispatched from crewai.utilities.internal_instructor import InternalInstructor messages = params.get("messages", []) @@ -1256,7 +1262,8 @@ class LLM(BaseLLM): model=response_model, llm=self, ) - result = instructor_instance.to_pydantic() + with model_call_hooks_dispatched(): + result = instructor_instance.to_pydantic() structured_response = result.model_dump_json() self._handle_emit_call_events( response=structured_response, @@ -1396,6 +1403,7 @@ class LLM(BaseLLM): str: The response text """ if response_model and self.is_litellm: + from crewai.hooks.llm_hooks import model_call_hooks_dispatched from crewai.utilities.internal_instructor import InternalInstructor messages = params.get("messages", []) @@ -1411,7 +1419,8 @@ class LLM(BaseLLM): model=response_model, llm=self, ) - result = instructor_instance.to_pydantic() + with model_call_hooks_dispatched(): + result = instructor_instance.to_pydantic() structured_response = result.model_dump_json() self._handle_emit_call_events( response=structured_response, @@ -1874,8 +1883,11 @@ class LLM(BaseLLM): msg_role: Literal["assistant"] = "assistant" message["role"] = msg_role - if not self._invoke_before_llm_call_hooks(messages, from_agent): - raise ValueError("LLM call blocked by before_llm_call hook") + try: + self._invoke_before_llm_call_hooks(messages, from_agent) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise with suppress_warnings(): if callbacks and len(callbacks) > 0: @@ -2016,6 +2028,12 @@ class LLM(BaseLLM): msg_role: Literal["assistant"] = "assistant" message["role"] = msg_role + try: + self._invoke_before_llm_call_hooks(messages, from_agent) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise + with suppress_warnings(): if callbacks and len(callbacks) > 0: self.set_callbacks(callbacks) diff --git a/lib/crewai/src/crewai/llms/base_llm.py b/lib/crewai/src/crewai/llms/base_llm.py index 3b40601bc..9bd964a31 100644 --- a/lib/crewai/src/crewai/llms/base_llm.py +++ b/lib/crewai/src/crewai/llms/base_llm.py @@ -72,6 +72,15 @@ class JsonResponseFormat(TypedDict): type: Literal["json_object"] +class LLMCallBlockedError(ValueError): + """A ``before_llm_call`` hook blocked the call by returning ``False``. + + A ``ValueError`` so the fail-open handlers around internal model calls keep + absorbing it, and its own type so a provider can report it as the decision + it is instead of letting it read as a provider outage. + """ + + DEFAULT_CONTEXT_WINDOW_SIZE: Final[int] = 4096 DEFAULT_SUPPORTS_STOP_WORDS: Final[bool] = True _JSON_EXTRACTION_PATTERN: Final[re.Pattern[str]] = re.compile(r"\{.*}", re.DOTALL) @@ -660,6 +669,28 @@ class BaseLLM(BaseModel, ABC): ), ) + def _emit_call_denied_event( + self, + denial: Exception, + from_task: Task | None = None, + from_agent: BaseAgent | None = None, + ) -> None: + """Report a hook deny as a deny rather than as a provider failure. + + The call still owes its started event a terminal one, so the failed event + is emitted — with a message that names the decision instead of blaming + the provider for an outage that never happened. + """ + from crewai.hooks.dispatch import source_name + + source = source_name(getattr(denial, "source", None)) + reason = getattr(denial, "reason", str(denial)) + message = f"LLM call denied by {source or 'hook'}: {reason}" + logging.warning(message) + self._emit_call_failed_event( + error=message, from_task=from_task, from_agent=from_agent + ) + def _emit_stream_chunk_event( self, chunk: str, @@ -990,43 +1021,51 @@ class BaseLLM(BaseModel, ABC): messages: list[LLMMessage], from_agent: BaseAgent | None = None, ) -> bool: - """Invoke before_llm_call hooks for direct LLM calls (no agent context). + """Invoke before_llm_call hooks for an LLM call reaching the provider. This method should be called by native provider implementations before - making the actual LLM call when from_agent is None (direct calls). + making the actual LLM call. It no-ops when an enclosing caller — the + executor — already dispatched the hooks for this same call. Args: messages: The messages being sent to the LLM - from_agent: The agent making the call (None for direct calls) + from_agent: The agent making the call, when there is one Returns: - True if LLM call should proceed, False if blocked by hook + True, so a provider may still guard the call with ``if not ...``. + A block is raised, never returned. + + Raises: + HookAborted: If a hook raised it. The deny reaches the caller intact + instead of being flattened into a provider-style error. + LLMCallBlockedError: If a legacy hook blocked the call by returning + ``False``. A ``ValueError``, so the fail-open handlers around + internal model calls keep absorbing it. Example: >>> # In a native provider's call() method: - >>> if from_agent is None and not self._invoke_before_llm_call_hooks( - ... messages, from_agent - ... ): - ... raise ValueError("LLM call blocked by hook") + >>> self._invoke_before_llm_call_hooks(messages, from_agent) """ - if from_agent is not None: - return True - from crewai_core.printer import PRINTER - from crewai.hooks.dispatch import HookAborted, InterceptionPoint, dispatch + from crewai.hooks.dispatch import InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, + LegacyHookBlocked, before_llm_call_reducer, + model_call_hooks_already_dispatched, ) + if model_call_hooks_already_dispatched(): + return True + # No early global-list guard: dispatch resolves global + execution-scoped # hooks and has its own no-op fast path, so scoped hooks still run here. hook_context = LLMCallHookContext( executor=None, messages=messages, llm=self, - agent=None, + agent=from_agent, task=None, crew=None, ) @@ -1037,12 +1076,14 @@ class BaseLLM(BaseModel, ABC): hook_context, reducer=before_llm_call_reducer, ) - except HookAborted: + except LegacyHookBlocked as blocked: PRINTER.print( content="LLM call blocked by before_llm_call hook", color="yellow", ) - return False + raise LLMCallBlockedError( + "LLM call blocked by before_llm_call hook" + ) from blocked return True @@ -1052,42 +1093,44 @@ class BaseLLM(BaseModel, ABC): response: str, from_agent: BaseAgent | None = None, ) -> str: - """Invoke after_llm_call hooks for direct LLM calls (no agent context). + """Invoke after_llm_call hooks for an LLM call that reached the provider. This method should be called by native provider implementations after - receiving the LLM response when from_agent is None (direct calls). + receiving the LLM response. It no-ops when an enclosing caller — the + executor — already dispatched the hooks for this same call. Args: messages: The messages that were sent to the LLM response: The response from the LLM - from_agent: The agent that made the call (None for direct calls) + from_agent: The agent that made the call, when there is one Returns: The potentially modified response string Example: >>> # In a native provider's call() method: - >>> if from_agent is None and isinstance(result, str): + >>> if isinstance(result, str): ... result = self._invoke_after_llm_call_hooks( ... messages, result, from_agent ... ) """ - if from_agent is not None or not isinstance(response, str): - return response - from crewai.hooks.dispatch import InterceptionPoint, dispatch from crewai.hooks.llm_hooks import ( LLMCallHookContext, after_llm_call_reducer, + model_call_hooks_already_dispatched, ) + if model_call_hooks_already_dispatched() or not isinstance(response, str): + return response + # No early global-list guard: dispatch resolves global + execution-scoped # hooks and has its own no-op fast path, so scoped hooks still run here. hook_context = LLMCallHookContext( executor=None, messages=messages, llm=self, - agent=None, + agent=from_agent, task=None, crew=None, response=response, diff --git a/lib/crewai/src/crewai/llms/providers/anthropic/completion.py b/lib/crewai/src/crewai/llms/providers/anthropic/completion.py index e6bc4255f..71ce23a58 100644 --- a/lib/crewai/src/crewai/llms/providers/anthropic/completion.py +++ b/lib/crewai/src/crewai/llms/providers/anthropic/completion.py @@ -8,7 +8,13 @@ from typing import Any, Final, Literal, Protocol, TypeGuard, TypedDict, cast from pydantic import BaseModel, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType -from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context +from crewai.hooks.dispatch import HookAborted +from crewai.llms.base_llm import ( + BaseLLM, + JsonResponseFormat, + LLMCallBlockedError, + llm_call_context, +) from crewai.llms.hooks.base import BaseInterceptor from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport from crewai.llms.providers.utils.common import safe_tool_conversion @@ -401,10 +407,7 @@ class AnthropicCompletion(BaseLLM): self._format_messages_for_anthropic(messages) ) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) completion_params = self._prepare_completion_params( formatted_messages, system_message, tools, available_functions @@ -429,6 +432,9 @@ class AnthropicCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"Anthropic API call failed: {e!s}" logging.error(error_msg) @@ -476,6 +482,8 @@ class AnthropicCompletion(BaseLLM): self._format_messages_for_anthropic(messages) ) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + completion_params = self._prepare_completion_params( formatted_messages, system_message, tools, available_functions ) @@ -499,6 +507,9 @@ class AnthropicCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"Anthropic API call failed: {e!s}" logging.error(error_msg) diff --git a/lib/crewai/src/crewai/llms/providers/azure/completion.py b/lib/crewai/src/crewai/llms/providers/azure/completion.py index 0f7cbd4fc..8cdb49cca 100644 --- a/lib/crewai/src/crewai/llms/providers/azure/completion.py +++ b/lib/crewai/src/crewai/llms/providers/azure/completion.py @@ -9,6 +9,7 @@ from urllib.parse import urlparse from pydantic import BaseModel, PrivateAttr, model_validator from typing_extensions import Self +from crewai.hooks.dispatch import HookAborted from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id from crewai.llms.hooks.base import BaseInterceptor from crewai.utilities.agent_utils import is_context_length_exceeded @@ -42,7 +43,12 @@ try: ) from crewai.events.types.llm_events import LLMCallType - from crewai.llms.base_llm import BaseLLM, call_stream_override, llm_call_context + from crewai.llms.base_llm import ( + BaseLLM, + LLMCallBlockedError, + call_stream_override, + llm_call_context, + ) except ImportError: raise ImportError( @@ -521,10 +527,7 @@ class AzureCompletion(BaseLLM): formatted_messages = self._format_messages_for_azure(messages) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) completion_params = self._prepare_completion_params( formatted_messages, tools, effective_response_model @@ -547,6 +550,9 @@ class AzureCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: return self._handle_api_error(e, from_task, from_agent) # type: ignore[func-returns-value] @@ -603,6 +609,8 @@ class AzureCompletion(BaseLLM): formatted_messages = self._format_messages_for_azure(messages) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + completion_params = self._prepare_completion_params( formatted_messages, tools, effective_response_model ) @@ -624,6 +632,9 @@ class AzureCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: self._handle_api_error(e, from_task, from_agent) diff --git a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py index d10fc2ba8..ab05c55da 100644 --- a/lib/crewai/src/crewai/llms/providers/bedrock/completion.py +++ b/lib/crewai/src/crewai/llms/providers/bedrock/completion.py @@ -11,7 +11,8 @@ from pydantic import BaseModel, PrivateAttr, model_validator from typing_extensions import Required from crewai.events.types.llm_events import LLMCallType -from crewai.llms.base_llm import BaseLLM, llm_call_context +from crewai.hooks.dispatch import HookAborted +from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, llm_call_context from crewai.llms.providers.utils.common import safe_tool_conversion from crewai.utilities.agent_utils import is_context_length_exceeded from crewai.utilities.exceptions.context_window_exceeding_exception import ( @@ -377,10 +378,7 @@ class BedrockCompletion(BaseLLM): messages ) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) body: BedrockConverseRequestBody = { "inferenceConfig": self._get_inference_config(), @@ -447,6 +445,9 @@ class BedrockCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: if is_context_length_exceeded(e): logging.error(f"Context window exceeded: {e}") @@ -510,6 +511,8 @@ class BedrockCompletion(BaseLLM): messages ) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + body: BedrockConverseRequestBody = { "inferenceConfig": self._get_inference_config(), } @@ -575,6 +578,9 @@ class BedrockCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: if is_context_length_exceeded(e): logging.error(f"Context window exceeded: {e}") diff --git a/lib/crewai/src/crewai/llms/providers/gemini/completion.py b/lib/crewai/src/crewai/llms/providers/gemini/completion.py index b099fe237..0bdf8ec0e 100644 --- a/lib/crewai/src/crewai/llms/providers/gemini/completion.py +++ b/lib/crewai/src/crewai/llms/providers/gemini/completion.py @@ -10,7 +10,8 @@ from typing import Any, Literal, cast from pydantic import BaseModel, Field, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType -from crewai.llms.base_llm import BaseLLM, llm_call_context +from crewai.hooks.dispatch import HookAborted +from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, llm_call_context from crewai.llms.hooks.base import BaseInterceptor from crewai.utilities.agent_utils import is_context_length_exceeded from crewai.utilities.exceptions.context_window_exceeding_exception import ( @@ -313,10 +314,7 @@ class GeminiCompletion(BaseLLM): messages_for_hooks = self._convert_contents_to_dict(formatted_content) - if not self._invoke_before_llm_call_hooks( - messages_for_hooks, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(messages_for_hooks, from_agent) config = self._prepare_generation_config( system_instruction, tools, effective_response_model @@ -341,6 +339,9 @@ class GeminiCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except APIError as e: error_msg = f"Google Gemini API error: {e.code} - {e.message}" logging.error(error_msg) @@ -397,6 +398,10 @@ class GeminiCompletion(BaseLLM): self._format_messages_for_gemini(messages) ) + messages_for_hooks = self._convert_contents_to_dict(formatted_content) + + self._invoke_before_llm_call_hooks(messages_for_hooks, from_agent) + config = self._prepare_generation_config( system_instruction, tools, effective_response_model ) @@ -420,6 +425,9 @@ class GeminiCompletion(BaseLLM): effective_response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except APIError as e: error_msg = f"Google Gemini API error: {e.code} - {e.message}" logging.error(error_msg) diff --git a/lib/crewai/src/crewai/llms/providers/openai/completion.py b/lib/crewai/src/crewai/llms/providers/openai/completion.py index 2b667e168..20234e501 100644 --- a/lib/crewai/src/crewai/llms/providers/openai/completion.py +++ b/lib/crewai/src/crewai/llms/providers/openai/completion.py @@ -36,8 +36,14 @@ from openai.types.responses import ( from pydantic import BaseModel, PrivateAttr, model_validator from crewai.events.types.llm_events import LLMCallType +from crewai.hooks.dispatch import HookAborted from crewai.llms._finish_reason_utils import extract_choices_finish_reason_and_id -from crewai.llms.base_llm import BaseLLM, JsonResponseFormat, llm_call_context +from crewai.llms.base_llm import ( + BaseLLM, + JsonResponseFormat, + LLMCallBlockedError, + llm_call_context, +) from crewai.llms.hooks.base import BaseInterceptor from crewai.llms.hooks.transport import AsyncHTTPTransport, HTTPTransport from crewai.llms.providers.utils.common import safe_tool_conversion @@ -462,10 +468,7 @@ class OpenAICompletion(BaseLLM): formatted_messages = self._format_messages(messages) - if not self._invoke_before_llm_call_hooks( - formatted_messages, from_agent - ): - raise ValueError("LLM call blocked by before_llm_call hook") + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) if self._effective_api() == "responses": return self._call_responses( @@ -486,6 +489,9 @@ class OpenAICompletion(BaseLLM): response_model=response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"OpenAI API call failed: {e!s}" logging.error(error_msg) @@ -600,6 +606,8 @@ class OpenAICompletion(BaseLLM): formatted_messages = self._format_messages(messages) + self._invoke_before_llm_call_hooks(formatted_messages, from_agent) + if self._effective_api() == "responses": return await self._acall_responses( messages=formatted_messages, @@ -619,6 +627,9 @@ class OpenAICompletion(BaseLLM): response_model=response_model, ) + except (HookAborted, LLMCallBlockedError) as e: + self._emit_call_denied_event(e, from_task, from_agent) + raise except Exception as e: error_msg = f"OpenAI API call failed: {e!s}" logging.error(error_msg) diff --git a/lib/crewai/src/crewai/memory/analyze.py b/lib/crewai/src/crewai/memory/analyze.py index 65d671d0d..3bb3a92a4 100644 --- a/lib/crewai/src/crewai/memory/analyze.py +++ b/lib/crewai/src/crewai/memory/analyze.py @@ -168,6 +168,8 @@ def extract_memories_from_content(content: str, llm: Any) -> list[str]: Returns: List of short, self-contained memory statements (or [content] on failure). """ + from crewai.hooks.dispatch import HookAborted + if not (content or "").strip(): return [] user = _get_prompt("extract_memories_user").format(content=content) @@ -188,6 +190,8 @@ def extract_memories_from_content(content: str, llm: Any) -> list[str]: data = json.loads(response) return ExtractedMemories.model_validate(data).memories return ExtractedMemories.model_validate(response).memories + except HookAborted: + raise except Exception as e: _logger.warning( "Memory extraction failed, storing full content as single memory: %s", @@ -216,6 +220,8 @@ def analyze_query( Returns: QueryAnalysis with keywords, suggested_scopes, complexity, recall_queries, time_filter. """ + from crewai.hooks.dispatch import HookAborted + scope_desc = "" if scope_info: scope_desc = f"Current scope has {scope_info.record_count} records, categories: {scope_info.categories}" @@ -241,6 +247,8 @@ def analyze_query( data = json.loads(response) return QueryAnalysis.model_validate(data) return QueryAnalysis.model_validate(response) + except HookAborted: + raise except Exception as e: _logger.warning( "Query analysis failed, using defaults (complexity=simple): %s", @@ -284,6 +292,8 @@ def analyze_for_save( Returns: MemoryAnalysis with suggested_scope, categories, importance, extracted_metadata. """ + from crewai.hooks.dispatch import HookAborted + user = _get_prompt("save_user").format( content=content, existing_scopes=existing_scopes or ["/"], @@ -306,6 +316,8 @@ def analyze_for_save( data = json.loads(response) return MemoryAnalysis.model_validate(data) return MemoryAnalysis.model_validate(response) + except HookAborted: + raise except Exception as e: _logger.warning( "Memory save analysis failed, using defaults: %s", @@ -336,6 +348,8 @@ def analyze_for_consolidation( Returns: ConsolidationPlan with actions per record and whether to insert the new content. """ + from crewai.hooks.dispatch import HookAborted + if not existing_records: return ConsolidationPlan(actions=[], insert_new=True) records_lines: list[str] = [] @@ -366,6 +380,8 @@ def analyze_for_consolidation( data = json.loads(response) return ConsolidationPlan.model_validate(data) return ConsolidationPlan.model_validate(response) + except HookAborted: + raise except Exception as e: _logger.warning( "Consolidation analysis failed, defaulting to insert: %s", diff --git a/lib/crewai/src/crewai/memory/recall_flow.py b/lib/crewai/src/crewai/memory/recall_flow.py index 8b2d64651..fbeb5c5c6 100644 --- a/lib/crewai/src/crewai/memory/recall_flow.py +++ b/lib/crewai/src/crewai/memory/recall_flow.py @@ -296,6 +296,8 @@ class RecallFlow(Flow[RecallState]): Decrements the exploration budget so the loop terminates. """ + from crewai.hooks.dispatch import HookAborted + self.state.exploration_budget -= 1 enhanced = [] @@ -321,6 +323,8 @@ class RecallFlow(Flow[RecallState]): "results": finding["results"], } ) + except HookAborted: + raise except Exception: enhanced.append( { diff --git a/lib/crewai/src/crewai/tasks/llm_guardrail.py b/lib/crewai/src/crewai/tasks/llm_guardrail.py index 754596ab7..da3498cc0 100644 --- a/lib/crewai/src/crewai/tasks/llm_guardrail.py +++ b/lib/crewai/src/crewai/tasks/llm_guardrail.py @@ -105,7 +105,11 @@ class LLMGuardrail: Tuple[bool, Any]: A tuple containing: - bool: True if validation passed, False otherwise - Any: The validation result or error message + + Raises: + HookAborted: A `pre_model_call` hook denied the validation call. """ + from crewai.hooks.dispatch import HookAborted try: result = self._validate_output(task_output) @@ -115,5 +119,7 @@ class LLMGuardrail: if result.pydantic.valid: return True, task_output.raw return False, result.pydantic.feedback + except HookAborted: + raise except Exception as e: return False, f"Error while validating the task output: {e!s}" diff --git a/lib/crewai/src/crewai/utilities/agent_utils.py b/lib/crewai/src/crewai/utilities/agent_utils.py index 6c0ea6bdc..b8f664c11 100644 --- a/lib/crewai/src/crewai/utilities/agent_utils.py +++ b/lib/crewai/src/crewai/utilities/agent_utils.py @@ -24,7 +24,7 @@ from crewai.agents.parser import ( OutputParserError, parse, ) -from crewai.llms.base_llm import BaseLLM, call_stop_override +from crewai.llms.base_llm import BaseLLM, LLMCallBlockedError, call_stop_override from crewai.tools import BaseTool as CrewAITool from crewai.tools.base_tool import BaseTool from crewai.tools.structured_tool import ( @@ -481,31 +481,41 @@ def enforce_rpm_limit( request_within_rpm_limit() +@contextlib.contextmanager def _prepare_llm_call( executor_context: CrewAgentExecutor | AgentExecutor | LiteAgent | None, messages: list[LLMMessage], printer: Printer, verbose: bool = True, -) -> list[LLMMessage]: +) -> Iterator[list[LLMMessage]]: """Shared pre-call logic: run before hooks and resolve messages. + Yields for the duration of the LLM call so the LLM layer knows the hooks + already ran with this executor's context and does not dispatch them twice. + Args: executor_context: Optional executor context for hook invocation. messages: The messages to send to the LLM. printer: Printer instance for output. verbose: Whether to print output. - Returns: + Yields: The resolved messages list (may come from executor_context). Raises: ValueError: If a before hook blocks the call. """ - if executor_context is not None: - if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose): - raise ValueError("LLM call blocked by before_llm_call hook") - messages = executor_context.messages - return messages + from crewai.hooks.llm_hooks import model_call_hooks_dispatched + + if executor_context is None: + yield messages + return + + if not _setup_before_llm_call_hooks(executor_context, printer, verbose=verbose): + raise LLMCallBlockedError("LLM call blocked by before_llm_call hook") + + with model_call_hooks_dispatched(): + yield executor_context.messages def _validate_and_finalize_llm_response( @@ -577,17 +587,18 @@ def get_llm_response( Exception: If an error occurs. ValueError: If the response is None or empty. """ - messages = _prepare_llm_call(executor_context, messages, printer, verbose=verbose) - - answer = llm.call( - messages, - tools=tools, - callbacks=callbacks, - available_functions=available_functions, - from_task=from_task, - from_agent=from_agent, - response_model=response_model, - ) + with _prepare_llm_call( + executor_context, messages, printer, verbose=verbose + ) as prepared_messages: + answer = llm.call( + prepared_messages, + tools=tools, + callbacks=callbacks, + available_functions=available_functions, + from_task=from_task, + from_agent=from_agent, + response_model=response_model, + ) return _validate_and_finalize_llm_response( answer, executor_context, printer, verbose=verbose @@ -630,17 +641,18 @@ async def aget_llm_response( Exception: If an error occurs. ValueError: If the response is None or empty. """ - messages = _prepare_llm_call(executor_context, messages, printer, verbose=verbose) - - answer = await llm.acall( - messages, - tools=tools, - callbacks=callbacks, - available_functions=available_functions, - from_task=from_task, - from_agent=from_agent, - response_model=response_model, - ) + with _prepare_llm_call( + executor_context, messages, printer, verbose=verbose + ) as prepared_messages: + answer = await llm.acall( + prepared_messages, + tools=tools, + callbacks=callbacks, + available_functions=available_functions, + from_task=from_task, + from_agent=from_agent, + response_model=response_model, + ) return _validate_and_finalize_llm_response( answer, executor_context, printer, verbose=verbose @@ -1953,16 +1965,23 @@ def _setup_before_llm_call_hooks( verbose: Whether to print output. Returns: - True if LLM execution should proceed, False if blocked by a hook. + True if LLM execution should proceed, False if a hook blocked it by + returning ``False``. + + Raises: + HookAborted: If a hook raised it, so the deny reaches the caller intact. """ if executor_context: from crewai.hooks.dispatch import ( - HookAborted, InterceptionPoint, get_scoped_hooks, run_hooks, ) - from crewai.hooks.llm_hooks import LLMCallHookContext, before_llm_call_reducer + from crewai.hooks.llm_hooks import ( + LLMCallHookContext, + LegacyHookBlocked, + before_llm_call_reducer, + ) # Executor snapshot first, then execution-scoped hooks — the same # ordering dispatch() applies to global vs scoped hooks. @@ -1984,7 +2003,7 @@ def _setup_before_llm_call_hooks( reducer=before_llm_call_reducer, verbose=verbose, ) - except HookAborted: + except LegacyHookBlocked: if verbose: printer.print( content="LLM call blocked by before_llm_call hook", diff --git a/lib/crewai/src/crewai/utilities/converter.py b/lib/crewai/src/crewai/utilities/converter.py index d31b76f48..f55888589 100644 --- a/lib/crewai/src/crewai/utilities/converter.py +++ b/lib/crewai/src/crewai/utilities/converter.py @@ -93,6 +93,8 @@ class Converter(OutputConverter): Raises: ConverterError: If conversion fails after maximum attempts. """ + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): response = self.llm.call( @@ -108,6 +110,8 @@ class Converter(OutputConverter): raise ConverterError( f"Failed to convert text into a Pydantic model due to validation error: {e}" ) from e + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return self.to_pydantic(current_attempt + 1) @@ -117,6 +121,8 @@ class Converter(OutputConverter): async def ato_pydantic(self, current_attempt: int = 1) -> BaseModel: """Async equivalent of ``to_pydantic`` — uses ``acall`` so the event loop is not blocked.""" + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): response = await self.llm.acall( @@ -132,6 +138,8 @@ class Converter(OutputConverter): raise ConverterError( f"Failed to convert text into a Pydantic model due to validation error: {e}" ) from e + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return await self.ato_pydantic(current_attempt + 1) @@ -152,10 +160,14 @@ class Converter(OutputConverter): ConverterError: If conversion fails after maximum attempts. """ + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): return self._create_instructor().to_json() return json.dumps(self.llm.call(self._build_messages())) + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return self.to_json(current_attempt + 1) @@ -168,10 +180,14 @@ class Converter(OutputConverter): sync-only); we run it via ``asyncio.to_thread`` so the event loop stays free. """ + from crewai.hooks.dispatch import HookAborted + try: if self.llm.supports_function_calling(): return await asyncio.to_thread(self._create_instructor().to_json) return json.dumps(await self.llm.acall(self._build_messages())) + except HookAborted: + raise except Exception as e: if current_attempt < self.max_attempts: return await self.ato_json(current_attempt + 1) diff --git a/lib/crewai/src/crewai/utilities/crew_chat.py b/lib/crewai/src/crewai/utilities/crew_chat.py index 0ae40ea5b..8fa5a685c 100644 --- a/lib/crewai/src/crewai/utilities/crew_chat.py +++ b/lib/crewai/src/crewai/utilities/crew_chat.py @@ -417,6 +417,8 @@ def generate_input_description_with_ai( Returns: A concise description of the input. """ + from crewai.hooks.dispatch import HookAborted + context_texts = [] placeholder_pattern = re.compile(r"\{(.+?)}") @@ -460,6 +462,8 @@ def generate_input_description_with_ai( ) try: response = chat_llm.call(messages=[{"role": "user", "content": prompt}]) + except HookAborted: + raise except Exception as exc: click.secho( f"Warning: failed to generate input description for '{input_name}' " @@ -480,6 +484,8 @@ def generate_crew_description_with_ai(crew: Crew, chat_llm: LLM | BaseLLM) -> st Returns: A concise description of the crew's purpose (15 words or less). """ + from crewai.hooks.dispatch import HookAborted + context_texts = [] placeholder_pattern = re.compile(r"\{(.+?)}") @@ -514,6 +520,8 @@ def generate_crew_description_with_ai(crew: Crew, chat_llm: LLM | BaseLLM) -> st ) try: response = chat_llm.call(messages=[{"role": "user", "content": prompt}]) + except HookAborted: + raise except Exception as exc: click.secho( f"Warning: failed to generate crew description ({exc}); using default.", diff --git a/lib/crewai/src/crewai/utilities/guardrail.py b/lib/crewai/src/crewai/utilities/guardrail.py index faf27fa9f..8454ffd92 100644 --- a/lib/crewai/src/crewai/utilities/guardrail.py +++ b/lib/crewai/src/crewai/utilities/guardrail.py @@ -144,6 +144,7 @@ def process_guardrail( Raises: TypeError: If output is not a TaskOutput or LiteAgentOutput ValueError: If guardrail is None + HookAborted: A `pre_model_call` hook denied an LLM-backed guardrail. """ from crewai.lite_agent_output import LiteAgentOutput from crewai.tasks.task_output import TaskOutput @@ -158,6 +159,7 @@ def process_guardrail( LLMGuardrailCompletedEvent, LLMGuardrailStartedEvent, ) + from crewai.hooks.dispatch import HookAborted started_event = LLMGuardrailStartedEvent( guardrail=guardrail, @@ -167,7 +169,26 @@ def process_guardrail( ) crewai_event_bus.emit(event_source, started_event) - result = guardrail(output) + try: + result = guardrail(output) + except HookAborted as e: + # a deny ends the validation, so the started event above still needs a + # terminal one before it leaves + crewai_event_bus.emit( + event_source, + LLMGuardrailCompletedEvent( + success=False, + result=None, + error=str(e), + retry_count=retry_count, + guardrail_type=started_event.guardrail_type, + guardrail_name=started_event.guardrail_name, + from_agent=from_agent, + from_task=from_task, + ), + ) + raise + guardrail_result = GuardrailResult.from_tuple(result) crewai_event_bus.emit( diff --git a/lib/crewai/src/crewai/utilities/internal_instructor.py b/lib/crewai/src/crewai/utilities/internal_instructor.py index 074f3a824..21020275a 100644 --- a/lib/crewai/src/crewai/utilities/internal_instructor.py +++ b/lib/crewai/src/crewai/utilities/internal_instructor.py @@ -143,8 +143,12 @@ class InternalInstructor(Generic[T]): Instance of the specified Pydantic model with structured data Raises: - ValueError: If LLM is not provided or invalid + ValueError: If LLM is not provided or invalid, or if a hook blocked + the call by returning ``False``. + HookAborted: If a hook denied the call. """ + from crewai.llms.base_llm import BaseLLM + messages: list[LLMMessage] = [{"role": "user", "content": self.content}] if not _is_valid_llm(self.llm): @@ -157,6 +161,11 @@ class InternalInstructor(Generic[T]): else: model_name = self.llm.model + if isinstance(self.llm, BaseLLM): + # This reaches the provider client directly, so the hooks the LLM + # layer would have dispatched are dispatched here instead. + self.llm._invoke_before_llm_call_hooks(messages, self.agent) + return self._client.chat.completions.create( # type: ignore[no-any-return] model=model_name, response_model=self.model, messages=messages ) diff --git a/lib/crewai/src/crewai/utilities/reasoning_handler.py b/lib/crewai/src/crewai/utilities/reasoning_handler.py index 1028a3f3d..21c0a1e1d 100644 --- a/lib/crewai/src/crewai/utilities/reasoning_handler.py +++ b/lib/crewai/src/crewai/utilities/reasoning_handler.py @@ -362,6 +362,8 @@ class AgentReasoning: """ self.logger.debug(f"Using function calling for {plan_type} planning") + from crewai.hooks.dispatch import HookAborted + try: system_prompt = self._get_system_prompt() @@ -412,6 +414,8 @@ class AgentReasoning: "READY: I am ready to execute the task." in response_str, ) + except HookAborted: + raise except Exception as e: self.logger.warning( f"Error during function calling: {e!s}. Falling back to text parsing." @@ -435,6 +439,8 @@ class AgentReasoning: [], "READY: I am ready to execute the task." in fallback_str, ) + except HookAborted: + raise except Exception as inner_e: self.logger.error(f"Error during fallback text parsing: {inner_e!s}") return ( diff --git a/lib/crewai/tests/hooks/test_deny_reaches_the_caller.py b/lib/crewai/tests/hooks/test_deny_reaches_the_caller.py new file mode 100644 index 000000000..303b69ee7 --- /dev/null +++ b/lib/crewai/tests/hooks/test_deny_reaches_the_caller.py @@ -0,0 +1,309 @@ +"""A deny raised inside a run has to reach whoever started the run. + +The sibling propagation tests all call the frame that makes the model call +directly, so they prove a deny escapes *that* function and nothing about what +its callers do with it. Every regression in this area has lived one or more +frames up, in a broad ``except Exception`` that turned the deny into a degraded +result. These tests drive the public entry points instead, and count model calls +so a deny that gets retried reads as a failure rather than as a pass. +""" + +from __future__ import annotations + +from typing import Any + +from crewai.agent import Agent +from crewai.agents.step_executor import StepExecutor +from crewai.crew import Crew +from crewai.experimental.agent_executor import AgentExecutor +from crewai.hooks.dispatch import HookAborted, InterceptionPoint, clear_all, on +from crewai.lite_agent import LiteAgent +from crewai.llms.base_llm import BaseLLM +from crewai.task import Task +from crewai.utilities.planning_types import TodoItem +from crewai.utilities.step_execution_context import StepExecutionContext +from crewai.utilities.types import LLMMessage +import pytest + +from ..utils import wait_for_event_handlers + + +class StubProviderLLM(BaseLLM): + """Answers without a network, dispatching the before hooks like a provider.""" + + def __init__(self, fail_first_call: bool = False) -> None: + super().__init__(model="stub") + self.fail_first_call = fail_first_call + self.answered = 0 + + def call( + self, + messages: str | list[LLMMessage], + tools: list[dict[str, Any]] | None = None, + callbacks: list[Any] | None = None, + available_functions: dict[str, Any] | None = None, + from_task: Any | None = None, + from_agent: Any | None = None, + response_model: Any | None = None, + **kwargs: Any, + ) -> str: + formatted: list[LLMMessage] = ( + messages + if isinstance(messages, list) + else [{"role": "user", "content": messages}] + ) + self._invoke_before_llm_call_hooks(formatted, from_agent) + self.answered += 1 + if self.fail_first_call and self.answered == 1: + raise RuntimeError("the provider blipped") + return "Thought: done\nFinal Answer: ok" + + def supports_function_calling(self) -> bool: + return False + + +class DenyingMemory: + """Stands in for the memory whose own model call a hook denied.""" + + read_only = False + root_scope = None + + def __init__(self, deny_on: str, error: Exception | None = None) -> None: + self.deny_on = deny_on + self.error = error or HookAborted( + reason="memory is off limits", source="policy" + ) + self.touched: list[str] = [] + + def _step(self, name: str) -> None: + self.touched.append(name) + if name == self.deny_on: + raise self.error + + def drain_writes(self) -> None: + pass + + def recall(self, *args: Any, **kwargs: Any) -> list[Any]: + self._step("recall") + return [] + + def extract_memories(self, *args: Any, **kwargs: Any) -> list[str]: + self._step("extract_memories") + return ["a memory"] + + def remember_many(self, *args: Any, **kwargs: Any) -> None: + self._step("remember_many") + + def search(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + +class StubKnowledge: + def query(self, *args: Any, **kwargs: Any) -> list[Any]: + return [] + + +class DenyingStepExecutor: + """Stands in for the per-step executor whose own model call was denied.""" + + def __init__(self) -> None: + self.executed = 0 + + def execute(self, *args: Any, **kwargs: Any) -> Any: + self.executed += 1 + raise HookAborted(reason="no model calls allowed", source="policy") + + +@pytest.fixture(autouse=True) +def _clean_hooks(): + clear_all() + yield + # A kickoff emits events whose handlers run on a pool; draining them here + # keeps a straggler from firing inside an unrelated test. + wait_for_event_handlers() + clear_all() + + +def deny_nth_model_call(n: int) -> list[str]: + """Deny the nth model call of the run, returning the log of attempts.""" + attempts: list[str] = [] + + @on(InterceptionPoint.PRE_MODEL_CALL) + def gate(_ctx: Any) -> None: + attempts.append("attempt") + if len(attempts) == n: + raise HookAborted(reason="no model calls allowed", source="policy") + + return attempts + + +def build_agent(**kwargs: Any) -> Agent: + return Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(), + **kwargs, + ) + + +def build_crew(agent: Agent, **task_kwargs: Any) -> Crew: + task = Task( + description="Say ok", expected_output="ok", agent=agent, **task_kwargs + ) + return Crew(agents=[agent], tasks=[task]) + + +def run_agent_kickoff() -> Any: + return build_agent().kickoff("say ok") + + +def run_agent_kickoff_with_planning() -> Any: + return build_agent(planning=True).kickoff("say ok") + + +def run_crew_kickoff() -> Any: + return build_crew(build_agent()).kickoff() + + +def run_crew_kickoff_with_planning() -> Any: + return build_crew(build_agent(planning=True)).kickoff() + + +def run_crew_kickoff_with_knowledge() -> Any: + agent = build_agent() + agent.knowledge = StubKnowledge() + return build_crew(agent).kickoff() + + +def run_lite_agent_kickoff() -> Any: + return LiteAgent( + role="Worker", goal="Answer", backstory="You answer.", llm=StubProviderLLM() + ).kickoff("say ok") + + +@pytest.mark.parametrize( + "entry_point", + [ + run_agent_kickoff, + run_agent_kickoff_with_planning, + run_crew_kickoff, + run_crew_kickoff_with_planning, + run_crew_kickoff_with_knowledge, + run_lite_agent_kickoff, + ], + ids=[ + "agent.kickoff", + "agent.kickoff-planning", + "crew.kickoff", + "crew.kickoff-planning", + "crew.kickoff-knowledge", + "lite_agent.kickoff", + ], +) +def test_a_denied_model_call_reaches_the_caller(entry_point): + attempts = deny_nth_model_call(1) + + with pytest.raises(HookAborted): + entry_point() + + assert len(attempts) == 1 + + +def test_a_denied_guardrail_stops_the_crew_instead_of_retrying_the_task(): + attempts = deny_nth_model_call(2) + crew = build_crew(build_agent(), guardrail="The answer must be polite") + + with pytest.raises(HookAborted): + crew.kickoff() + + # the answer, then the denied validation, and nothing after it + assert len(attempts) == 2 + + +@pytest.mark.parametrize("deny_on", ["recall", "extract_memories"]) +def test_a_denied_memory_step_reaches_the_caller(deny_on): + memory = DenyingMemory(deny_on) + agent = build_agent() + agent.memory = memory + + with pytest.raises(HookAborted): + agent.kickoff("say ok") + + assert memory.touched.count(deny_on) == 1 + + +def test_a_denied_memory_save_stops_the_crew_instead_of_retrying_the_task(): + memory = DenyingMemory("extract_memories") + agent = build_agent() + agent.memory = memory + + with pytest.raises(HookAborted): + build_crew(agent).kickoff() + + assert memory.touched.count("extract_memories") == 1 + + +def build_step_executor(llm: StubProviderLLM) -> StepExecutor: + return StepExecutor(llm=llm, tools=[], agent=build_agent()) + + +def a_step() -> tuple[TodoItem, StepExecutionContext]: + return ( + TodoItem(step_number=1, description="Say ok"), + StepExecutionContext(task_description="Say ok", task_goal="ok"), + ) + + +def test_a_denied_step_stops_the_plan_instead_of_reporting_a_failed_step(): + attempts = deny_nth_model_call(1) + todo, context = a_step() + + with pytest.raises(HookAborted): + build_step_executor(StubProviderLLM()).execute(todo, context) + + assert len(attempts) == 1 + + +def test_an_ordinary_step_failure_still_reports_a_failed_step(): + todo, context = a_step() + + result = build_step_executor(StubProviderLLM(fail_first_call=True)).execute( + todo, context + ) + + assert result.success is False + + +@pytest.mark.asyncio +async def test_a_denied_parallel_step_reaches_the_caller(): + agent = build_agent() + executor = AgentExecutor(agent=agent, llm=agent.llm, task=None) + executor.state.todos.items = [ + TodoItem(step_number=1, description="first"), + TodoItem(step_number=2, description="second"), + ] + step_executor = DenyingStepExecutor() + object.__setattr__(executor, "_ensure_step_executor", lambda: step_executor) + + with pytest.raises(HookAborted): + await executor.execute_todos_parallel() + + +def test_an_ordinary_model_failure_is_still_retried(): + agent = Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(fail_first_call=True), + ) + + assert "ok" in str(build_crew(agent).kickoff()) + + +def test_an_ordinary_memory_failure_still_degrades(): + memory = DenyingMemory("extract_memories", error=RuntimeError("storage is down")) + agent = build_agent() + agent.memory = memory + + assert str(agent.kickoff("say ok")) == "ok" diff --git a/lib/crewai/tests/hooks/test_hook_abort_propagation.py b/lib/crewai/tests/hooks/test_hook_abort_propagation.py new file mode 100644 index 000000000..e498eaa04 --- /dev/null +++ b/lib/crewai/tests/hooks/test_hook_abort_propagation.py @@ -0,0 +1,522 @@ +"""A model-call deny must reach the caller as a deny. + +Two layers used to erase it. The LLM layer caught ``HookAborted`` and returned +``False``, which every provider turned into ``ValueError("LLM call blocked...")`` +— losing the reason, the source, and any way to tell a policy decision from a +provider outage. Downstream, every internal model call is wrapped in +``except Exception`` so a provider hiccup degrades instead of failing the run, +and those handlers then absorbed the flattened deny, often retrying the very +call that was just denied. + +A deny also owes the started event a terminal one, and owes it an honest label: +it is a decision, not an outage the provider caused. + +These use a real ``LLM`` with a real ``pre_model_call`` hook: the deny fires +inside ``dispatch`` before the provider is reached, so nothing here needs a +network or a cassette. +""" + +from __future__ import annotations + +from typing import Any + +from crewai.agent import Agent +from crewai.agent.planning_config import PlanningConfig +from crewai.agent.utils import ( + ahandle_knowledge_retrieval, + handle_knowledge_retrieval, + handle_reasoning, +) +from crewai.agents.planner_observer import PlannerObserver +from crewai.events.event_bus import crewai_event_bus +from crewai.events.types.knowledge_events import ( + KnowledgeQueryFailedEvent, + KnowledgeQueryStartedEvent, +) +from crewai.events.types.llm_events import LLMCallFailedEvent +from crewai.events.types.llm_guardrail_events import ( + LLMGuardrailCompletedEvent, + LLMGuardrailStartedEvent, +) +from crewai.events.types.observation_events import ( + StepObservationFailedEvent, + StepObservationStartedEvent, +) +from crewai.experimental.agent_executor import AgentExecutor +from crewai.hooks.dispatch import ( + HookAborted, + InterceptionPoint, + clear_all, + on, +) +from crewai.hooks.llm_hooks import register_before_llm_call_hook +from crewai.llm import LLM +from crewai.memory.analyze import ( + analyze_for_consolidation, + analyze_for_save, + analyze_query, + extract_memories_from_content, +) +from crewai.memory.types import MemoryRecord +from crewai.task import Task +from crewai.tasks.llm_guardrail import LLMGuardrail +from crewai.tasks.task_output import TaskOutput +from crewai.utilities.converter import Converter +from crewai.utilities.guardrail import process_guardrail +from crewai.utilities.planning_types import TodoItem +import pytest +from pydantic import BaseModel + +from ..utils import wait_for_event_handlers + + +@pytest.fixture(autouse=True) +def _clean_hooks(): + clear_all() + yield + # These calls emit LLM events whose handlers run on a pool; draining them + # here keeps a straggler from firing inside an unrelated test. + wait_for_event_handlers() + clear_all() + + +@pytest.fixture +def denied_calls() -> list[Any]: + return [] + + +@pytest.fixture +def denying_llm(denied_calls: list[Any]) -> LLM: + @on(InterceptionPoint.PRE_MODEL_CALL) + def deny(ctx: Any) -> None: + denied_calls.append(ctx) + raise HookAborted(reason="no model calls allowed", source="policy") + + return LLM(model="gpt-4o-mini") + + +class _Person(BaseModel): + name: str + + +def test_a_raised_deny_keeps_its_reason_out_of_the_llm_layer(denying_llm): + with pytest.raises(HookAborted) as exc: + denying_llm.call([{"role": "user", "content": "hi"}]) + + assert exc.value.reason == "no model calls allowed" + assert exc.value.source == "policy" + + +def test_the_boolean_convention_still_blocks_with_the_documented_error(): + register_before_llm_call_hook(lambda _ctx: False) + + with pytest.raises(ValueError, match="LLM call blocked by before_llm_call hook"): + LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}]) + + +def test_the_boolean_convention_is_still_absorbed_by_a_fail_open_handler(): + # Unlike a raised abort, the boolean deny must keep degrading rather than + # failing the run — otherwise adopting this fix breaks existing hooks. + register_before_llm_call_hook(lambda _ctx: False) + + analysis = analyze_query("a query", ["/"], None, LLM(model="gpt-4o-mini")) + + assert analysis.recall_queries == ["a query"] + + +@pytest.mark.parametrize( + "denied_helper", + [ + lambda llm: extract_memories_from_content("some content", llm), + lambda llm: analyze_query("a query", ["/"], None, llm), + lambda llm: analyze_for_save("some content", ["/"], [], llm), + lambda llm: analyze_for_consolidation( + "new content", + [MemoryRecord(id="1", content="old content", scope="/")], + llm, + ), + ], + ids=["extract", "query", "save", "consolidate"], +) +def test_memory_analysis_surfaces_a_deny_instead_of_a_safe_default( + denied_helper, denying_llm, denied_calls +): + with pytest.raises(HookAborted): + denied_helper(denying_llm) + + assert denied_calls, "the helper never reached the model" + + +def test_a_denied_conversion_is_not_retried(denying_llm, denied_calls): + converter = Converter( + text="Name: Ada", + llm=denying_llm, + model=_Person, + instructions="Extract the person", + max_attempts=3, + ) + + with pytest.raises(HookAborted): + converter.to_pydantic() + + assert len(denied_calls) == 1 + + +def test_a_denied_knowledge_query_still_reports_the_failure_it_started(denying_llm): + agent = Agent( + role="Researcher", + goal="Answer questions", + backstory="You look things up.", + llm=denying_llm, + ) + task = Task( + description="What is the capital of France?", + expected_output="A city name.", + agent=agent, + ) + started: list[KnowledgeQueryStartedEvent] = [] + failed: list[KnowledgeQueryFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(KnowledgeQueryStartedEvent) + def _on_started(_source: Any, event: KnowledgeQueryStartedEvent) -> None: + started.append(event) + + @crewai_event_bus.on(KnowledgeQueryFailedEvent) + def _on_failed(_source: Any, event: KnowledgeQueryFailedEvent) -> None: + failed.append(event) + + with pytest.raises(HookAborted): + agent._get_knowledge_search_query(task.description, task) + + wait_for_event_handlers() + + assert len(started) == 1 + assert len(failed) == 1 + assert failed[0].error == "no model calls allowed" + + +def test_a_denied_step_observation_still_reports_the_failure_it_started(denying_llm): + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + ) + observer = PlannerObserver(agent=agent, task=None) + step = TodoItem(step_number=1, description="do the thing", result="done") + started: list[StepObservationStartedEvent] = [] + failed: list[StepObservationFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(StepObservationStartedEvent) + def _on_started(_source: Any, event: StepObservationStartedEvent) -> None: + started.append(event) + + @crewai_event_bus.on(StepObservationFailedEvent) + def _on_failed(_source: Any, event: StepObservationFailedEvent) -> None: + failed.append(event) + + with pytest.raises(HookAborted): + observer.observe(step, "done", [], []) + + wait_for_event_handlers() + + assert len(started) == 1 + assert len(failed) == 1 + assert failed[0].error == "no model calls allowed" + assert failed[0].step_number == 1 + + +def test_a_denied_knowledge_retrieval_does_not_fall_back_to_the_plain_prompt( + denying_llm, +): + # the retrieval helper wraps the query rewrite in its own except Exception, + # so guarding the rewrite alone still let the task run without knowledge + agent = Agent( + role="Researcher", + goal="Answer questions", + backstory="You look things up.", + llm=denying_llm, + ) + # only has to be truthy: the deny fires before any knowledge is queried + agent.knowledge = object() + task = Task( + description="What is the capital of France?", + expected_output="A city name.", + agent=agent, + ) + + with pytest.raises(HookAborted): + handle_knowledge_retrieval( + agent, + task, + "the task prompt", + {}, + lambda *_args, **_kwargs: [], + lambda *_args, **_kwargs: [], + ) + + +@pytest.mark.asyncio +async def test_a_denied_async_knowledge_retrieval_reaches_the_caller(denying_llm): + agent = Agent( + role="Researcher", + goal="Answer questions", + backstory="You look things up.", + llm=denying_llm, + ) + agent.knowledge = object() + task = Task( + description="What is the capital of France?", + expected_output="A city name.", + agent=agent, + ) + + with pytest.raises(HookAborted): + await ahandle_knowledge_retrieval(agent, task, "the task prompt", {}) + + +def test_a_denied_plan_stops_the_legacy_planning_path(denying_llm): + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + planning=True, + ) + task = Task(description="Do the thing", expected_output="A result.", agent=agent) + + with pytest.raises(HookAborted): + handle_reasoning(agent, task) + + +def test_a_denied_replan_does_not_keep_executing_the_stale_plan(denying_llm): + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + planning_config=PlanningConfig( + reasoning_effort="low", + max_attempts=1, + max_steps=2, + max_replans=1, + max_step_iterations=2, + ), + ) + executor = AgentExecutor(agent=agent, llm=denying_llm, task=None) + executor._kickoff_input = "do the thing" + + with pytest.raises(HookAborted): + executor._trigger_replan("the first plan failed") + + +class _DenyingMemory: + """Stands in for unified memory whose model call was denied upstream.""" + + read_only = False + root_scope = None + + def __init__(self, error: Exception): + self._error = error + + def drain_writes(self) -> None: + pass + + def recall(self, *_args: Any, **_kwargs: Any): + raise self._error + + def extract_memories(self, *_args: Any, **_kwargs: Any): + raise self._error + + def remember_many(self, *_args: Any, **_kwargs: Any) -> None: + pass + + +def test_a_denied_memory_recall_does_not_run_the_task_without_memory(): + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(HookAborted(reason="no", source="policy")) + task = Task(description="Do the thing", expected_output="A result.", agent=agent) + + with pytest.raises(HookAborted): + agent._retrieve_memory_context(task, "the task prompt") + + +def test_an_ordinary_memory_recall_failure_still_degrades(): + # the guard must single out a deny: a broken store has always been allowed + # to degrade to no memory, and turning that into a failed run is a break + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(ValueError("vector store is down")) + task = Task(description="Do the thing", expected_output="A result.", agent=agent) + + assert agent._retrieve_memory_context(task, "the prompt") == "the prompt" + + +def test_a_denied_memory_save_does_not_report_a_clean_kickoff(): + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(HookAborted(reason="no", source="policy")) + + with pytest.raises(HookAborted): + agent._save_kickoff_to_memory("the input", "the output") + + +def test_an_ordinary_memory_save_failure_still_degrades(): + agent = Agent(role="Doer", goal="Do", backstory="You do.") + agent.memory = _DenyingMemory(ValueError("vector store is down")) + + agent._save_kickoff_to_memory("the input", "the output") + + +def test_a_denied_guardrail_does_not_read_as_a_failed_validation(denying_llm): + # returning (False, "Error while validating...") would feed the agent a + # retry prompt built from a call the policy refused + guardrail = LLMGuardrail(description="Must be polite", llm=denying_llm) + task_output = TaskOutput( + description="Say hi", raw="hi", agent="Doer", expected_output="A greeting." + ) + + with pytest.raises(HookAborted): + guardrail(task_output) + + +def test_a_denied_guardrail_still_reports_the_validation_it_started(denying_llm): + guardrail = LLMGuardrail(description="Must be polite", llm=denying_llm) + task_output = TaskOutput( + description="Say hi", raw="hi", agent="Doer", expected_output="A greeting." + ) + started: list[LLMGuardrailStartedEvent] = [] + completed: list[LLMGuardrailCompletedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMGuardrailStartedEvent) + def _on_started(_source: Any, event: LLMGuardrailStartedEvent) -> None: + started.append(event) + + @crewai_event_bus.on(LLMGuardrailCompletedEvent) + def _on_completed(_source: Any, event: LLMGuardrailCompletedEvent) -> None: + completed.append(event) + + with pytest.raises(HookAborted): + process_guardrail(output=task_output, guardrail=guardrail, retry_count=0) + + wait_for_event_handlers() + + assert len(started) == 1 + assert len(completed) == 1 + assert completed[0].success is False + assert "no model calls allowed" in (completed[0].error or "") + + +def test_a_denied_plan_stops_the_executor_instead_of_running_unplanned(denying_llm): + # the executor wraps planning in its own except Exception, so guarding the + # reasoning handler alone still left the agent proceeding with no plan + agent = Agent( + role="Planner", + goal="Plan work", + backstory="You plan.", + llm=denying_llm, + planning_config=PlanningConfig( + reasoning_effort="low", + max_attempts=1, + max_steps=2, + max_replans=0, + max_step_iterations=2, + ), + ) + executor = AgentExecutor(agent=agent, llm=denying_llm, task=None) + executor._kickoff_input = "do the thing" + + with pytest.raises(HookAborted): + executor.generate_plan() + + +@pytest.mark.asyncio +async def test_an_async_call_is_denied_like_a_sync_one(denying_llm): + with pytest.raises(HookAborted) as exc: + await denying_llm.acall([{"role": "user", "content": "hi"}]) + + assert exc.value.reason == "no model calls allowed" + assert exc.value.source == "policy" + + +def test_a_denied_structured_conversion_reaches_the_caller(denying_llm): + # The function-calling path goes through Instructor, which reaches the + # provider client without passing through ``llm.call``. + assert denying_llm.supports_function_calling() + converter = Converter( + text="Name: Ada", + llm=denying_llm, + model=_Person, + instructions="Extract the person", + max_attempts=1, + ) + + with pytest.raises(HookAborted): + converter.to_json() + + +def test_a_deny_is_not_reported_as_a_provider_failure(denying_llm): + failures: list[LLMCallFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMCallFailedEvent) + def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None: + failures.append(event) + + with pytest.raises(HookAborted): + denying_llm.call([{"role": "user", "content": "hi"}]) + + wait_for_event_handlers() + + assert len(failures) == 1 + assert failures[0].error == "LLM call denied by policy: no model calls allowed" + + +def test_a_deny_names_the_hook_that_raised_it_rather_than_its_repr(): + def gate_on_approved_models(_ctx: Any) -> None: + raise HookAborted(reason="model not approved", source=gate_on_approved_models) + + on(InterceptionPoint.PRE_MODEL_CALL)(gate_on_approved_models) + failures: list[LLMCallFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMCallFailedEvent) + def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None: + failures.append(event) + + with pytest.raises(HookAborted): + LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}]) + + wait_for_event_handlers() + + assert len(failures) == 1 + assert failures[0].error == ( + "LLM call denied by gate_on_approved_models: model not approved" + ) + + +def test_the_boolean_convention_is_not_reported_as_a_provider_failure(): + register_before_llm_call_hook(lambda _ctx: False) + failures: list[LLMCallFailedEvent] = [] + + with crewai_event_bus.scoped_handlers(): + + @crewai_event_bus.on(LLMCallFailedEvent) + def _on_failed(_source: Any, event: LLMCallFailedEvent) -> None: + failures.append(event) + + with pytest.raises(ValueError): + LLM(model="gpt-4o-mini").call([{"role": "user", "content": "hi"}]) + + wait_for_event_handlers() + + assert len(failures) == 1 + assert failures[0].error == ( + "LLM call denied by hook: LLM call blocked by before_llm_call hook" + ) diff --git a/lib/crewai/tests/hooks/test_llm_hooks.py b/lib/crewai/tests/hooks/test_llm_hooks.py index 0c917712d..42c7c9ef4 100644 --- a/lib/crewai/tests/hooks/test_llm_hooks.py +++ b/lib/crewai/tests/hooks/test_llm_hooks.py @@ -646,11 +646,27 @@ class TestDirectLLMScopedHooks: with scoped_hooks(): register_scoped(InterceptionPoint.PRE_MODEL_CALL, block) - proceed = llm._invoke_before_llm_call_hooks( - [{"role": "user", "content": "hi"}], from_agent=None - ) + with pytest.raises(HookAborted, match="blocked by scoped hook"): + llm._invoke_before_llm_call_hooks( + [{"role": "user", "content": "hi"}], from_agent=None + ) - assert proceed is False + def test_a_scoped_hook_returning_false_blocks_the_call(self): + from crewai.hooks import InterceptionPoint + from crewai.hooks.dispatch import register_scoped, scoped_hooks + from crewai.llms.base_llm import LLMCallBlockedError + + llm = self._stub_llm() + + with scoped_hooks(): + register_scoped(InterceptionPoint.PRE_MODEL_CALL, lambda _ctx: False) + with pytest.raises(LLMCallBlockedError) as blocked: + llm._invoke_before_llm_call_hooks( + [{"role": "user", "content": "hi"}], from_agent=None + ) + + # a ValueError so the fail-open handlers around internal calls still absorb it + assert isinstance(blocked.value, ValueError) def test_scoped_after_hook_modifies_direct_response(self): from crewai.hooks import InterceptionPoint diff --git a/lib/crewai/tests/hooks/test_model_call_hook_reach.py b/lib/crewai/tests/hooks/test_model_call_hook_reach.py new file mode 100644 index 000000000..ad89d9239 --- /dev/null +++ b/lib/crewai/tests/hooks/test_model_call_hook_reach.py @@ -0,0 +1,149 @@ +"""Which model calls the ``pre_model_call`` hooks can see, and how many times. + +The LLM layer used to skip the hooks whenever a call carried an agent, assuming +the executor had already dispatched them. That holds inside the executor loop +and left every other agent-bearing call — step observation, planning, plan +synthesis — invisible. The executor now marks the window where it already +dispatched, which is what keeps a call from being seen twice. + +The stub stands in for a native provider: it invokes the before hooks the way +every provider does and answers without a network. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +from crewai.agent import Agent +from crewai.hooks.dispatch import InterceptionPoint, clear_all, on +from crewai.llms.base_llm import BaseLLM +from crewai.utilities.agent_utils import get_llm_response +from crewai.utilities.types import LLMMessage +from crewai_core.printer import Printer +from pydantic import BaseModel +import pytest + +from ..utils import wait_for_event_handlers + + +class StubProviderLLM(BaseLLM): + def __init__(self) -> None: + super().__init__(model="stub") + + def call( + self, + messages: str | list[LLMMessage], + tools: list[dict[str, Any]] | None = None, + callbacks: list[Any] | None = None, + available_functions: dict[str, Any] | None = None, + from_task: Any | None = None, + from_agent: Any | None = None, + response_model: Any | None = None, + **kwargs: Any, + ) -> str: + formatted: list[LLMMessage] = ( + messages + if isinstance(messages, list) + else [{"role": "user", "content": messages}] + ) + self._invoke_before_llm_call_hooks(formatted, from_agent) + return "Thought: done\nFinal Answer: ok" + + def supports_function_calling(self) -> bool: + return False + + +@pytest.fixture(autouse=True) +def _clean_hooks(): + clear_all() + yield + # A kickoff emits events whose handlers run on a pool; draining them here + # keeps a straggler from firing inside an unrelated test. + wait_for_event_handlers() + clear_all() + + +@pytest.fixture +def seen_agents() -> list[str | None]: + roles: list[str | None] = [] + + @on(InterceptionPoint.PRE_MODEL_CALL) + def record(ctx: Any) -> None: + agent = getattr(ctx, "agent", None) + roles.append(getattr(agent, "role", None)) + + return roles + + +def test_the_executor_loop_is_seen_exactly_once(seen_agents): + agent = Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(), + ) + + assert str(agent.kickoff("say ok")) == "ok" + assert seen_agents == ["Worker"] + + +def test_a_direct_call_carrying_an_agent_is_seen_once(seen_agents): + agent = Agent( + role="Planner", + goal="Plan", + backstory="You plan.", + llm=StubProviderLLM(), + ) + + StubProviderLLM().call([{"role": "user", "content": "hi"}], from_agent=agent) + + assert seen_agents == ["Planner"] + + +def test_a_direct_call_without_an_agent_is_seen_once(seen_agents): + StubProviderLLM().call([{"role": "user", "content": "hi"}]) + + assert seen_agents == [None] + + +def test_a_structured_litellm_call_is_seen_once(seen_agents, monkeypatch): + pytest.importorskip("litellm") + import instructor + + from crewai.llm import LLM + + class Answer(BaseModel): + text: str + + client = SimpleNamespace( + chat=SimpleNamespace( + completions=SimpleNamespace(create=lambda **_: Answer(text="ok")) + ) + ) + monkeypatch.setattr(instructor, "from_litellm", lambda *_, **__: client) + + llm = LLM(model="openai/gpt-4o-mini", is_litellm=True) + llm.call([{"role": "user", "content": "hi"}], response_model=Answer) + + assert seen_agents == [None] + + +def test_an_agent_call_outside_the_executor_loop_is_seen_once(seen_agents): + agent = Agent( + role="Worker", + goal="Answer", + backstory="You answer.", + llm=StubProviderLLM(), + ) + + get_llm_response( + llm=StubProviderLLM(), + messages=[{"role": "user", "content": "hi"}], + callbacks=[], + printer=Printer(), + from_agent=agent, + executor_context=None, + ) + + assert seen_agents == ["Worker"]