Lorenze/agent executor flow pattern (#3975)

* WIP gh pr refactor: update agent executor handling and introduce flow-based executor

* wip

* refactor: clean up comments and improve code clarity in agent executor flow

- Removed outdated comments and unnecessary explanations in  and  classes to enhance code readability.
- Simplified parameter updates in the agent executor to avoid confusion regarding executor recreation.
- Improved clarity in the  method to ensure proper handling of non-final answers without raising errors.

* bumping pytest-randomly numpy

* also bump versions of anthropic sdk

* ensure flow logs are not passed if its on executor

* revert anthropic bump

* fix

* refactor: update dependency markers in uv.lock for platform compatibility

- Enhanced dependency markers for , , , and others to ensure compatibility across different platforms (Linux, Darwin, and architecture-specific conditions).
- Removed unnecessary event emission in the  class during kickoff.
- Cleaned up commented-out code in the  class for better readability and maintainability.

* drop dupllicate

* test: enhance agent executor creation and stop word assertions

- Added calls to create_agent_executor in multiple test cases to ensure proper agent execution setup.
- Updated assertions for stop words in the agent tests to remove unnecessary checks and improve clarity.
- Ensured consistency in task handling by invoking create_agent_executor with the appropriate task parameter.

* refactor: reorganize agent executor imports and introduce CrewAgentExecutorFlow

- Removed the old import of CrewAgentExecutorFlow and replaced it with the new import from the experimental module.
- Updated relevant references in the codebase to ensure compatibility with the new structure.
- Enhanced the organization of imports in core.py and base_agent.py for better clarity and maintainability.

* updating name

* dropped usage of printer here for rich console and dropped non-added value logging

* address i18n

* Enhance concurrency control in CrewAgentExecutorFlow by introducing a threading lock to prevent concurrent executions. This change ensures that the executor instance cannot be invoked while already running, improving stability and reliability during flow execution.

* string literal returns

* string literal returns

* Enhance CrewAgentExecutor initialization by allowing optional i18n parameter for improved internationalization support. This change ensures that the executor can utilize a provided i18n instance or fallback to the default, enhancing flexibility in multilingual contexts.

---------

Co-authored-by: Greyson LaLonde <[email protected]>

Author: lorenzejay

lib/crewai/src/crewai/agent/core.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from collections.abc import Sequence
+from collections.abc import Callable, Sequence
 import shutil
 import subprocess
 import time
@@ -44,6 +44,7 @@
     MemoryRetrievalCompletedEvent,
     MemoryRetrievalStartedEvent,
 )
+from crewai.experimental.crew_agent_executor_flow import CrewAgentExecutorFlow
 from crewai.knowledge.knowledge import Knowledge
 from crewai.knowledge.source.base_knowledge_source import BaseKnowledgeSource
 from crewai.lite_agent import LiteAgent
@@ -105,7 +106,7 @@ class Agent(BaseAgent):
     The agent can also have memory, can operate in verbose mode, and can delegate tasks to other agents.
 
     Attributes:
-            agent_executor: An instance of the CrewAgentExecutor class.
+            agent_executor: An instance of the CrewAgentExecutor or CrewAgentExecutorFlow class.
             role: The role of the agent.
             goal: The objective of the agent.
             backstory: The backstory of the agent.
@@ -221,6 +222,10 @@ class Agent(BaseAgent):
         default=None,
         description="A2A (Agent-to-Agent) configuration for delegating tasks to remote agents. Can be a single A2AConfig or a dict mapping agent IDs to configs.",
     )
+    executor_class: type[CrewAgentExecutor] | type[CrewAgentExecutorFlow] = Field(
+        default=CrewAgentExecutor,
+        description="Class to use for the agent executor. Defaults to CrewAgentExecutor, can optionally use CrewAgentExecutorFlow.",
+    )
 
     @model_validator(mode="before")
     def validate_from_repository(cls, v: Any) -> dict[str, Any] | None | Any:  # noqa: N805
@@ -721,29 +726,83 @@ def create_agent_executor(
                 self.response_template.split("{{ .Response }}")[1].strip()
             )
 
-        self.agent_executor = CrewAgentExecutor(
-            llm=self.llm,  # type: ignore[arg-type]
-            task=task,  # type: ignore[arg-type]
-            agent=self,
-            crew=self.crew,
-            tools=parsed_tools,
-            prompt=prompt,
-            original_tools=raw_tools,
-            stop_words=stop_words,
-            max_iter=self.max_iter,
-            tools_handler=self.tools_handler,
-            tools_names=get_tool_names(parsed_tools),
-            tools_description=render_text_description_and_args(parsed_tools),
-            step_callback=self.step_callback,
-            function_calling_llm=self.function_calling_llm,
-            respect_context_window=self.respect_context_window,
-            request_within_rpm_limit=(
-                self._rpm_controller.check_or_wait if self._rpm_controller else None
-            ),
-            callbacks=[TokenCalcHandler(self._token_process)],
-            response_model=task.response_model if task else None,
+        rpm_limit_fn = (
+            self._rpm_controller.check_or_wait if self._rpm_controller else None
         )
 
+        if self.agent_executor is not None:
+            self._update_executor_parameters(
+                task=task,
+                tools=parsed_tools,
+                raw_tools=raw_tools,
+                prompt=prompt,
+                stop_words=stop_words,
+                rpm_limit_fn=rpm_limit_fn,
+            )
+        else:
+            self.agent_executor = self.executor_class(
+                llm=cast(BaseLLM, self.llm),
+                task=task,
+                i18n=self.i18n,
+                agent=self,
+                crew=self.crew,
+                tools=parsed_tools,
+                prompt=prompt,
+                original_tools=raw_tools,
+                stop_words=stop_words,
+                max_iter=self.max_iter,
+                tools_handler=self.tools_handler,
+                tools_names=get_tool_names(parsed_tools),
+                tools_description=render_text_description_and_args(parsed_tools),
+                step_callback=self.step_callback,
+                function_calling_llm=self.function_calling_llm,
+                respect_context_window=self.respect_context_window,
+                request_within_rpm_limit=rpm_limit_fn,
+                callbacks=[TokenCalcHandler(self._token_process)],
+                response_model=task.response_model if task else None,
+            )
+
+    def _update_executor_parameters(
+        self,
+        task: Task | None,
+        tools: list,
+        raw_tools: list[BaseTool],
+        prompt: dict,
+        stop_words: list[str],
+        rpm_limit_fn: Callable | None,
+    ) -> None:
+        """Update executor parameters without recreating instance.
+
+        Args:
+            task: Task to execute.
+            tools: Parsed tools.
+            raw_tools: Original tools.
+            prompt: Generated prompt.
+            stop_words: Stop words list.
+            rpm_limit_fn: RPM limit callback function.
+        """
+        self.agent_executor.task = task
+        self.agent_executor.tools = tools
+        self.agent_executor.original_tools = raw_tools
+        self.agent_executor.prompt = prompt
+        self.agent_executor.stop = stop_words
+        self.agent_executor.tools_names = get_tool_names(tools)
+        self.agent_executor.tools_description = render_text_description_and_args(tools)
+        self.agent_executor.response_model = task.response_model if task else None
+
+        self.agent_executor.tools_handler = self.tools_handler
+        self.agent_executor.request_within_rpm_limit = rpm_limit_fn
+
+        if self.agent_executor.llm:
+            existing_stop = getattr(self.agent_executor.llm, "stop", [])
+            self.agent_executor.llm.stop = list(
+                set(
+                    existing_stop + stop_words
+                    if isinstance(existing_stop, list)
+                    else stop_words
+                )
+            )
+
     def get_delegation_tools(self, agents: list[BaseAgent]) -> list[BaseTool]:
         agent_tools = AgentTools(agents=agents)
         return agent_tools.tools()

lib/crewai/src/crewai/agents/agent_builder/base_agent.py

@@ -457,7 +457,6 @@ def set_cache_handler(self, cache_handler: CacheHandler) -> None:
         if self.cache:
             self.cache_handler = cache_handler
             self.tools_handler.cache = cache_handler
-        self.create_agent_executor()
 
     def set_rpm_controller(self, rpm_controller: RPMController) -> None:
         """Set the rpm controller for the agent.
@@ -467,7 +466,6 @@ def set_rpm_controller(self, rpm_controller: RPMController) -> None:
         """
         if not self._rpm_controller:
             self._rpm_controller = rpm_controller
-            self.create_agent_executor()
 
     def set_knowledge(self, crew_embedder: EmbedderConfig | None = None) -> None:
         pass

lib/crewai/src/crewai/agents/crew_agent_executor.py

@@ -91,6 +91,7 @@ def __init__(
         request_within_rpm_limit: Callable[[], bool] | None = None,
         callbacks: list[Any] | None = None,
         response_model: type[BaseModel] | None = None,
+        i18n: I18N | None = None,
     ) -> None:
         """Initialize executor.
 
@@ -114,7 +115,7 @@ def __init__(
             callbacks: Optional callbacks list.
             response_model: Optional Pydantic model for structured outputs.
         """
-        self._i18n: I18N = get_i18n()
+        self._i18n: I18N = i18n or get_i18n()
         self.llm = llm
         self.task = task
         self.agent = agent

lib/crewai/src/crewai/experimental/__init__.py

@@ -1,3 +1,4 @@
+from crewai.experimental.crew_agent_executor_flow import CrewAgentExecutorFlow
 from crewai.experimental.evaluation import (
     AgentEvaluationResult,
     AgentEvaluator,
@@ -23,6 +24,7 @@
     "AgentEvaluationResult",
     "AgentEvaluator",
     "BaseEvaluator",
+    "CrewAgentExecutorFlow",
     "EvaluationScore",
     "EvaluationTraceCallback",
     "ExperimentResult",