Lorenze/agent executor flow pattern #3975

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged

lorenzejay merged 30 commits into main from lorenze/agent-executor-flow-pattern

Dec 28, 2025

+2,024 −609

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

-Original file line number
+Diff line change
@@ -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
@@ Expand Down Expand Up / @@ -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
@@ Expand Down Expand Up / @@ -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.
@@ Expand Down Expand Up / @@ -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
@@ Expand Down Expand Up / @@ -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()
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ Expand Up @@
             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.
@@ Expand All @@
             """
             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

-Original file line number
+Diff line change
@@ Expand Up / @@ -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.
@@ Expand All / @@ -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
@@ Expand Down @@

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

-Original file line number
+Diff line change
@@ -1,3 +1,4 @@
+    from crewai.experimental.crew_agent_executor_flow import CrewAgentExecutorFlow
     from crewai.experimental.evaluation import (
         AgentEvaluationResult,
         AgentEvaluator,
@@ Expand All / @@ -23,6 +24,7 @@ @@
         "AgentEvaluationResult",
         "AgentEvaluator",
         "BaseEvaluator",
+        "CrewAgentExecutorFlow",
         "EvaluationScore",
         "EvaluationTraceCallback",
         "ExperimentResult",
@@ Expand Down @@

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Lorenze/agent executor flow pattern #3975

Diff view

Diff view

There are no files selected for viewing

Uh oh!

Uh oh!