Ability to integrate external agents into the framework

src/crewai/agent.py

@@ -131,7 +131,7 @@ def check_agent_executor(self) -> "Agent":
     def execute_task(
         self,
         task: Any,
-        context: Optional[str] = None,
+        context: Optional[List[str]] = None,
         tools: Optional[List[Any]] = None,
     ) -> str:
         """Execute a task with the agent.
@@ -147,6 +147,7 @@ def execute_task(
         task_prompt = task.prompt()
 
         if context:
+            context = "\n".join(context)
             task_prompt = self.i18n.slice("task_with_context").format(
                 task=task_prompt, context=context
             )
@@ -215,9 +216,9 @@ def create_agent_executor(self) -> None:
         }
 
         if self._rpm_controller:
-            executor_args[
-                "request_within_rpm_limit"
-            ] = self._rpm_controller.check_or_wait
+            executor_args["request_within_rpm_limit"] = (
+                self._rpm_controller.check_or_wait
+            )
 
         if self.memory:
             summary_memory = ConversationSummaryMemory(

src/crewai/agents/agent_interface.py

@@ -0,0 +1,52 @@
+from abc import ABC, abstractmethod
+from typing import Optional, List, Any, Dict
+
+from pydantic import BaseModel, PrivateAttr, Field
+
+from crewai.utilities import I18N
+
+
+class AgentWrapperParent(ABC, BaseModel):
+    _i18n: I18N = PrivateAttr(default=I18N())
+    data: Dict[str, Any] = Field(
+        default_factory=dict,
+        description="Data storage for children, as pydantic doesn't play well with inheritance.",
+    )
+    role: str = Field(description="Role of the agent", default="")
+    allow_delegation: bool = Field(
+        description="Allow delegation of tasks to other agents?", default=False
+    )
+
+    @property
+    def i18n(self) -> I18N:
+        if hasattr(self, "_agent") and hasattr(self._agent, "i18n"):
+            return self._agent.i18n
+        else:
+            return self._i18n
+
+    @i18n.setter
+    def i18n(self, value: I18N) -> None:
+        if hasattr(self, "_agent") and hasattr(self._agent, "i18n"):
+            self._agent.i18n = value
+        else:
+            self._i18n = value
+
+    @abstractmethod
+    def execute_task(
+        self,
+        task: str,
+        context: Optional[str] = None,
+        tools: Optional[List[Any]] = None,
+    ) -> str:
+        pass
+
+    @property
+    @abstractmethod
+    def tools(self) -> List[Any]:
+        pass
+
+    def set_cache_handler(self, cache_handler: Any) -> None:
+        pass
+
+    def set_rpm_controller(self, rpm_controller: Any) -> None:
+        pass

src/crewai/agents/langchain_agent.py

@@ -0,0 +1,52 @@
+from typing import Any, Optional, List, Callable
+import warnings
+
+from langchain_core.messages import AIMessage
+from crewai.agents.agent_interface import AgentWrapperParent
+
+
+class LangchainCrewAgent(AgentWrapperParent):
+
+    def __init__(
+        self,
+        agent: Any,
+        role: str,
+        allow_delegation: bool = False,
+        tools: List[Any] | None = None,
+        **data: Any,
+    ):
+        super().__init__(role=role, allow_delegation=allow_delegation, **data)
+        self.data.update(data)
+        self.data["agent"] = agent
+        # store tools by name to eliminate duplicates
+        self.data["tools"] = {}
+        self.tools = tools or []
+
+    def execute_task(
+        self,
+        task: str,
+        context: Optional[List[str]] = None,
+        tools: Optional[List[Any]] = None,
+    ) -> str:
+        used_tools = self.tools + (tools or [])
+
+        if context:
+            context = [AIMessage(content=ctx) for ctx in context]
+        else:
+            context = []
+        # https://github.com/langchain-ai/langchain/discussions/17403
+        return self.data["agent"].invoke(
+            {"input": task, "chat_history": context, "tools": used_tools}
+        )["output"]
+
+    @property
+    def tools(self) -> List[Any]:
+        return list(self.data["tools"].values())
+
+    @tools.setter
+    def tools(self, tools: List[Any]) -> None:
+        for tool in tools:
+            if tool.name not in self.data["tools"]:
+                self.data["tools"][tool.name] = tool
+            else:
+                warnings.warn(f"Tool {tool.name} already exists in the agent.")

src/crewai/crew.py

@@ -15,6 +15,8 @@
 )
 from pydantic_core import PydanticCustomError
 
+from crewai.agents.agent_interface import AgentWrapperParent
+
 from crewai.agent import Agent
 from crewai.agents.cache import CacheHandler
 from crewai.process import Process
@@ -51,7 +53,7 @@ class Crew(BaseModel):
     _cache_handler: InstanceOf[CacheHandler] = PrivateAttr(default=CacheHandler())
     model_config = ConfigDict(arbitrary_types_allowed=True)
     tasks: List[Task] = Field(default_factory=list)
-    agents: List[Agent] = Field(default_factory=list)
+    agents: List[Agent | AgentWrapperParent] = Field(default_factory=list)
     process: Process = Field(default=Process.sequential)
     verbose: Union[int, bool] = Field(default=0)
     full_output: Optional[bool] = Field(

src/crewai/task.py

@@ -8,6 +8,7 @@
 from crewai.agent import Agent
 from crewai.tasks.task_output import TaskOutput
 from crewai.utilities import I18N
+from crewai.agents.agent_interface import AgentWrapperParent
 
 
 class Task(BaseModel):
@@ -24,7 +25,7 @@ class Config:
     callback: Optional[Any] = Field(
         description="Callback to be executed after the task is completed.", default=None
     )
-    agent: Optional[Agent] = Field(
+    agent: Optional[Agent | AgentWrapperParent] = Field(
         description="Agent responsible for execution the task.", default=None
     )
     expected_output: Optional[str] = Field(
@@ -91,7 +92,6 @@ def execute(
                 if task.async_execution:
                     task.thread.join()
                 context.append(task.output.result)
-            context = "\n".join(context)
 
         tools = tools or self.tools
 

src/crewai/tools/agent_tools.py

@@ -7,11 +7,15 @@
 from crewai.task import Task
 from crewai.utilities import I18N
 
+from crewai.agents.agent_interface import AgentWrapperParent
+
 
 class AgentTools(BaseModel):
     """Default tools around agent delegation"""
 
-    agents: List[Agent] = Field(description="List of agents in this crew.")
+    agents: List[Agent | AgentWrapperParent] = Field(
+        description="List of agents in this crew."
+    )
     i18n: I18N = Field(default=I18N(), description="Internationalization settings.")
 
     def tools(self):