diff --git a/src/backend/base/langflow/components/agents/SequentialTaskAgent.py b/src/backend/base/langflow/components/agents/SequentialTaskAgent.py new file mode 100644 index 000000000000..58062508c44e --- /dev/null +++ b/src/backend/base/langflow/components/agents/SequentialTaskAgent.py @@ -0,0 +1,141 @@ +from crewai import Agent, Task + +from langflow.base.agents.crewai.tasks import SequentialTask +from langflow.custom import Component +from langflow.io import BoolInput, DictInput, HandleInput, MultilineInput, Output + + +class SequentialTaskAgentComponent(Component): + display_name = "Sequential Task Agent" + description = "Creates a CrewAI Task and its associated Agent." + documentation = "https://docs.crewai.com/how-to/LLM-Connections/" + icon = "CrewAI" + + inputs = [ + # Agent inputs + MultilineInput(name="role", display_name="Role", info="The role of the agent."), + MultilineInput(name="goal", display_name="Goal", info="The objective of the agent."), + MultilineInput( + name="backstory", + display_name="Backstory", + info="The backstory of the agent.", + ), + HandleInput( + name="tools", + display_name="Tools", + input_types=["Tool"], + is_list=True, + info="Tools at agent's disposal", + value=[], + ), + HandleInput( + name="llm", + display_name="Language Model", + info="Language model that will run the agent.", + input_types=["LanguageModel"], + ), + BoolInput( + name="memory", + display_name="Memory", + info="Whether the agent should have memory or not", + advanced=True, + value=True, + ), + BoolInput( + name="verbose", + display_name="Verbose", + advanced=True, + value=True, + ), + BoolInput( + name="allow_delegation", + display_name="Allow Delegation", + info="Whether the agent is allowed to delegate tasks to other agents.", + value=False, + advanced=True, + ), + BoolInput( + name="allow_code_execution", + display_name="Allow Code Execution", + info="Whether the agent is allowed to execute code.", + value=False, + advanced=True, + ), + DictInput( + name="agent_kwargs", + display_name="Agent kwargs", + info="Additional kwargs for the agent.", + is_list=True, + advanced=True, + ), + # Task inputs + MultilineInput( + name="task_description", + display_name="Task Description", + info="Descriptive text detailing task's purpose and execution.", + ), + MultilineInput( + name="expected_output", + display_name="Expected Task Output", + info="Clear definition of expected task outcome.", + ), + BoolInput( + name="async_execution", + display_name="Async Execution", + value=False, + advanced=True, + info="Boolean flag indicating asynchronous task execution.", + ), + # Chaining input + HandleInput( + name="previous_task", + display_name="Previous Task", + input_types=["SequentialTask"], + info="The previous task in the sequence (for chaining).", + required=False, + ), + ] + + outputs = [ + Output( + display_name="Sequential Task", + name="task_output", + method="build_agent_and_task", + ), + ] + + def build_agent_and_task(self) -> list[SequentialTask]: + # Build the agent + agent_kwargs = self.agent_kwargs or {} + agent = Agent( + role=self.role, + goal=self.goal, + backstory=self.backstory, + llm=self.llm, + verbose=self.verbose, + memory=self.memory, + tools=self.tools if self.tools else [], + allow_delegation=self.allow_delegation, + allow_code_execution=self.allow_code_execution, + **agent_kwargs, + ) + + # Build the task + task = Task( + description=self.task_description, + expected_output=self.expected_output, + agent=agent, + async_execution=self.async_execution, + ) + + # If there's a previous task, create a list of tasks + if self.previous_task: + if isinstance(self.previous_task, list): + tasks = self.previous_task + [task] + else: + tasks = [self.previous_task, task] + else: + tasks = [task] + + self.status = f"Agent: {repr(agent)}\nTask: {repr(task)}" + return tasks