Agents
This Langflow feature is currently in public preview. Development is ongoing, and the features and functionality are subject to change. Langflow, and the use of such, is subject to the DataStax Preview Terms. |
Agent components are used to define the behavior and capabilities of AI agents in your flow. Agents can interact with APIs, databases, and other services, but can also use LLMs as a reasoning engine to decide which course to take in your flow.
Agent component
This component creates an agent that can use tools to answer questions and perform tasks based on given instructions. For more information on this component, see the agent component documentation.
Parameters
Name | Type | Description |
---|---|---|
agent_llm |
Dropdown |
The provider of the language model that the agent will use to generate responses. |
system_prompt |
String |
Initial instructions and context provided to guide the agent’s behavior. |
tools |
List |
List of tools available for the agent to use. |
input_value |
String |
The input task or question for the agent to process. |
add_current_date_tool |
Boolean |
If true, adds a tool to the agent that returns the current date. |
Name | Type | Description |
---|---|---|
response |
Message |
The agent’s response to the given input task. |
Component code
agent.py
from langchain_core.tools import StructuredTool
from langflow.base.agents.agent import LCToolsAgentComponent
from langflow.base.models.model_input_constants import (
ALL_PROVIDER_FIELDS,
MODEL_PROVIDERS_DICT,
)
from langflow.base.models.model_utils import get_model_name
from langflow.components.helpers import CurrentDateComponent
from langflow.components.helpers.memory import MemoryComponent
from langflow.components.langchain_utilities.tool_calling import (
ToolCallingAgentComponent,
)
from langflow.io import BoolInput, DropdownInput, MultilineInput, Output
from langflow.schema.dotdict import dotdict
from langflow.schema.message import Message
def set_advanced_true(component_input):
component_input.advanced = True
return component_input
class AgentComponent(ToolCallingAgentComponent):
display_name: str = "Agent"
description: str = "Define the agent's instructions, then enter a task to complete using tools."
icon = "bot"
beta = False
name = "Agent"
memory_inputs = [set_advanced_true(component_input) for component_input in MemoryComponent().inputs]
inputs = [
DropdownInput(
name="agent_llm",
display_name="Model Provider",
info="The provider of the language model that the agent will use to generate responses.",
options=[*sorted(MODEL_PROVIDERS_DICT.keys()), "Custom"],
value="OpenAI",
real_time_refresh=True,
input_types=[],
),
*MODEL_PROVIDERS_DICT["OpenAI"]["inputs"],
MultilineInput(
name="system_prompt",
display_name="Agent Instructions",
info="System Prompt: Initial instructions and context provided to guide the agent's behavior.",
value="You are a helpful assistant that can use tools to answer questions and perform tasks.",
advanced=False,
),
*LCToolsAgentComponent._base_inputs,
*memory_inputs,
BoolInput(
name="add_current_date_tool",
display_name="Current Date",
advanced=True,
info="If true, will add a tool to the agent that returns the current date.",
value=True,
),
]
outputs = [Output(name="response", display_name="Response", method="message_response")]
async def message_response(self) -> Message:
llm_model, display_name = self.get_llm()
self.model_name = get_model_name(llm_model, display_name=display_name)
if llm_model is None:
msg = "No language model selected"
raise ValueError(msg)
self.chat_history = await self.get_memory_data()
if self.add_current_date_tool:
if not isinstance(self.tools, list): # type: ignore[has-type]
self.tools = []
# Convert CurrentDateComponent to a StructuredTool
current_date_tool = CurrentDateComponent().to_toolkit()[0]
if isinstance(current_date_tool, StructuredTool):
self.tools.append(current_date_tool)
else:
msg = "CurrentDateComponent must be converted to a StructuredTool"
raise ValueError(msg)
if not self.tools:
msg = "Tools are required to run the agent."
raise ValueError(msg)
self.set(
llm=llm_model,
tools=self.tools,
chat_history=self.chat_history,
input_value=self.input_value,
system_prompt=self.system_prompt,
)
agent = self.create_agent_runnable()
return await self.run_agent(agent)
async def get_memory_data(self):
memory_kwargs = {
component_input.name: getattr(self, f"{component_input.name}") for component_input in self.memory_inputs
}
return await MemoryComponent().set(**memory_kwargs).retrieve_messages()
def get_llm(self):
if isinstance(self.agent_llm, str):
try:
provider_info = MODEL_PROVIDERS_DICT.get(self.agent_llm)
if provider_info:
component_class = provider_info.get("component_class")
display_name = component_class.display_name
inputs = provider_info.get("inputs")
prefix = provider_info.get("prefix", "")
return (
self._build_llm_model(component_class, inputs, prefix),
display_name,
)
except Exception as e:
msg = f"Error building {self.agent_llm} language model"
raise ValueError(msg) from e
return self.agent_llm, None
def _build_llm_model(self, component, inputs, prefix=""):
model_kwargs = {input_.name: getattr(self, f"{prefix}{input_.name}") for input_ in inputs}
return component.set(**model_kwargs).build_model()
def delete_fields(self, build_config: dotdict, fields: dict | list[str]) -> None:
"""Delete specified fields from build_config."""
for field in fields:
build_config.pop(field, None)
def update_input_types(self, build_config: dotdict) -> dotdict:
"""Update input types for all fields in build_config."""
for key, value in build_config.items():
if isinstance(value, dict):
if value.get("input_types") is None:
build_config[key]["input_types"] = []
elif hasattr(value, "input_types") and value.input_types is None:
value.input_types = []
return build_config
async def aupdate_build_config(
self, build_config: dotdict, field_value: str, field_name: str | None = None
) -> dotdict:
# Iterate over all providers in the MODEL_PROVIDERS_DICT
# Existing logic for updating build_config
if field_name == "agent_llm":
provider_info = MODEL_PROVIDERS_DICT.get(field_value)
if provider_info:
component_class = provider_info.get("component_class")
if component_class and hasattr(component_class, "aupdate_build_config"):
# Call the component class's aupdate_build_config method
build_config = await component_class.aupdate_build_config(build_config, field_value, field_name)
provider_configs: dict[str, tuple[dict, list[dict]]] = {
provider: (
MODEL_PROVIDERS_DICT[provider]["fields"],
[
MODEL_PROVIDERS_DICT[other_provider]["fields"]
for other_provider in MODEL_PROVIDERS_DICT
if other_provider != provider
],
)
for provider in MODEL_PROVIDERS_DICT
}
if field_value in provider_configs:
fields_to_add, fields_to_delete = provider_configs[field_value]
# Delete fields from other providers
for fields in fields_to_delete:
self.delete_fields(build_config, fields)
# Add provider-specific fields
if field_value == "OpenAI" and not any(field in build_config for field in fields_to_add):
build_config.update(fields_to_add)
else:
build_config.update(fields_to_add)
# Reset input types for agent_llm
build_config["agent_llm"]["input_types"] = []
elif field_value == "Custom":
# Delete all provider fields
self.delete_fields(build_config, ALL_PROVIDER_FIELDS)
# Update with custom component
custom_component = DropdownInput(
name="agent_llm",
display_name="Language Model",
options=[*sorted(MODEL_PROVIDERS_DICT.keys()), "Custom"],
value="Custom",
real_time_refresh=True,
input_types=["LanguageModel"],
)
build_config.update({"agent_llm": custom_component.to_dict()})
# Update input types for all fields
build_config = self.update_input_types(build_config)
# Validate required keys
default_keys = [
"code",
"_type",
"agent_llm",
"tools",
"input_value",
"add_current_date_tool",
"system_prompt",
"agent_description",
"max_iterations",
"handle_parsing_errors",
"verbose",
]
missing_keys = [key for key in default_keys if key not in build_config]
if missing_keys:
msg = f"Missing required keys in build_config: {missing_keys}"
raise ValueError(msg)
if isinstance(self.agent_llm, str) and self.agent_llm in MODEL_PROVIDERS_DICT:
provider_info = MODEL_PROVIDERS_DICT.get(self.agent_llm)
if provider_info:
component_class = provider_info.get("component_class")
prefix = provider_info.get("prefix")
if component_class and hasattr(component_class, "aupdate_build_config"):
# Call each component class's aupdate_build_config method
# remove the prefix from the field_name
if isinstance(field_name, str) and isinstance(prefix, str):
field_name = field_name.replace(prefix, "")
build_config = await component_class.aupdate_build_config(build_config, field_value, field_name)
return build_config
CSV Agent
This component creates a CSV agent from a CSV file and LLM.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
path |
File |
Path to the CSV file |
agent_type |
String |
Type of agent to create (zero-shot-react-description, openai-functions, or openai-tools) |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
CSV agent instance |
Component code
csv.py
from langchain_experimental.agents.agent_toolkits.csv.base import create_csv_agent
from langflow.base.agents.agent import LCAgentComponent
from langflow.field_typing import AgentExecutor
from langflow.inputs import DropdownInput, FileInput, HandleInput
from langflow.inputs.inputs import MessageTextInput
from langflow.schema.message import Message
from langflow.template.field.base import Output
class CSVAgentComponent(LCAgentComponent):
display_name = "CSVAgent"
description = "Construct a CSV agent from a CSV and tools."
documentation = "https://python.langchain.com/docs/modules/agents/toolkits/csv"
name = "CSVAgent"
icon = "LangChain"
inputs = [
*LCAgentComponent._base_inputs,
HandleInput(
name="llm",
display_name="Language Model",
input_types=["LanguageModel"],
required=True,
info="An LLM Model Object (It can be found in any LLM Component).",
),
FileInput(
name="path",
display_name="File Path",
file_types=["csv"],
input_types=["str", "Message"],
required=True,
info="A CSV File or File Path.",
),
DropdownInput(
name="agent_type",
display_name="Agent Type",
advanced=True,
options=["zero-shot-react-description", "openai-functions", "openai-tools"],
value="openai-tools",
),
MessageTextInput(
name="input_value",
display_name="Text",
info="Text to be passed as input and extract info from the CSV File.",
),
]
outputs = [
Output(display_name="Response", name="response", method="build_agent_response"),
Output(display_name="Agent", name="agent", method="build_agent"),
]
def _path(self) -> str:
if isinstance(self.path, Message) and isinstance(self.path.text, str):
return self.path.text
return self.path
def build_agent_response(self) -> Message:
agent_kwargs = {
"verbose": self.verbose,
"allow_dangerous_code": True,
}
agent_csv = create_csv_agent(
llm=self.llm,
path=self._path(),
agent_type=self.agent_type,
handle_parsing_errors=self.handle_parsing_errors,
**agent_kwargs,
)
result = agent_csv.invoke({"input": self.input_value})
return Message(text=str(result["output"]))
def build_agent(self) -> AgentExecutor:
agent_kwargs = {
"verbose": self.verbose,
"allow_dangerous_code": True,
}
agent_csv = create_csv_agent(
llm=self.llm,
path=self._path(),
agent_type=self.agent_type,
handle_parsing_errors=self.handle_parsing_errors,
**agent_kwargs,
)
self.status = Message(text=str(agent_csv))
return agent_csv
CrewAI Agent
This component represents an Agent of CrewAI, allowing for the creation of specialized AI agents with defined roles, goals, and capabilities within a crew.
For more information, see the CrewAI documentation.
Parameters
Name | Display Name | Info |
---|---|---|
role |
Role |
The role of the agent |
goal |
Goal |
The objective of the agent |
backstory |
Backstory |
The backstory of the agent |
tools |
Tools |
Tools at agent’s disposal |
llm |
Language Model |
Language model that will run the agent |
memory |
Memory |
Whether the agent should have memory or not |
verbose |
Verbose |
Enables verbose output |
allow_delegation |
Allow Delegation |
Whether the agent is allowed to delegate tasks to other agents |
allow_code_execution |
Allow Code Execution |
Whether the agent is allowed to execute code |
kwargs |
kwargs |
Additional keyword arguments for the agent |
Name | Display Name | Info |
---|---|---|
output |
Agent |
The constructed CrewAI Agent object |
Component code
crewai.py
from crewai import Agent
from langflow.base.agents.crewai.crew import convert_llm, convert_tools
from langflow.custom import Component
from langflow.io import BoolInput, DictInput, HandleInput, MultilineInput, Output
class CrewAIAgentComponent(Component):
"""Component for creating a CrewAI agent.
This component allows you to create a CrewAI agent with the specified role, goal, backstory, tools,
and language model.
Args:
Component (Component): Base class for all components.
Returns:
Agent: CrewAI agent.
"""
display_name = "CrewAI Agent"
description = "Represents an agent of CrewAI."
documentation: str = "https://docs.crewai.com/how-to/LLM-Connections/"
icon = "CrewAI"
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 agents 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=False,
),
BoolInput(
name="allow_delegation",
display_name="Allow Delegation",
info="Whether the agent is allowed to delegate tasks to other agents.",
value=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="kwargs",
display_name="kwargs",
info="kwargs of agent.",
is_list=True,
advanced=True,
),
]
outputs = [
Output(display_name="Agent", name="output", method="build_output"),
]
def build_output(self) -> Agent:
kwargs = self.kwargs or {}
# Define the Agent
agent = Agent(
role=self.role,
goal=self.goal,
backstory=self.backstory,
llm=convert_llm(self.llm),
verbose=self.verbose,
memory=self.memory,
tools=convert_tools(self.tools),
allow_delegation=self.allow_delegation,
allow_code_execution=self.allow_code_execution,
**kwargs,
)
self.status = repr(agent)
return agent
Hierarchical Crew
This component represents a group of agents, managing how they should collaborate and the tasks they should perform in a hierarchical structure. This component allows for the creation of a crew with a manager overseeing the task execution.
For more information, see the CrewAI documentation.
Parameters
Name | Display Name | Info |
---|---|---|
agents |
Agents |
List of Agent objects representing the crew members |
tasks |
Tasks |
List of HierarchicalTask objects representing the tasks to be executed |
manager_llm |
Manager LLM |
Language model for the manager agent (optional) |
manager_agent |
Manager Agent |
Specific agent to act as the manager (optional) |
verbose |
Verbose |
Enables verbose output for detailed logging |
memory |
Memory |
Specifies the memory configuration for the crew |
use_cache |
Use Cache |
Enables caching of results |
max_rpm |
Max RPM |
Sets the maximum requests per minute |
share_crew |
Share Crew |
Determines if the crew information is shared among agents |
function_calling_llm |
Function Calling LLM |
Specifies the language model for function calling |
Name | Display Name | Info |
---|---|---|
crew |
Crew |
The constructed Crew object with hierarchical task execution |
Component code
hierarchical_crew.py
from crewai import Crew, Process
from langflow.base.agents.crewai.crew import BaseCrewComponent
from langflow.io import HandleInput
class HierarchicalCrewComponent(BaseCrewComponent):
display_name: str = "Hierarchical Crew"
description: str = (
"Represents a group of agents, defining how they should collaborate and the tasks they should perform."
)
documentation: str = "https://docs.crewai.com/how-to/Hierarchical/"
icon = "CrewAI"
inputs = [
*BaseCrewComponent._base_inputs,
HandleInput(name="agents", display_name="Agents", input_types=["Agent"], is_list=True),
HandleInput(name="tasks", display_name="Tasks", input_types=["HierarchicalTask"], is_list=True),
HandleInput(name="manager_llm", display_name="Manager LLM", input_types=["LanguageModel"], required=False),
HandleInput(name="manager_agent", display_name="Manager Agent", input_types=["Agent"], required=False),
]
def build_crew(self) -> Crew:
tasks, agents = self.get_tasks_and_agents()
manager_llm = self.get_manager_llm()
return Crew(
agents=agents,
tasks=tasks,
process=Process.hierarchical,
verbose=self.verbose,
memory=self.memory,
cache=self.use_cache,
max_rpm=self.max_rpm,
share_crew=self.share_crew,
function_calling_llm=self.function_calling_llm,
manager_agent=self.manager_agent,
manager_llm=manager_llm,
step_callback=self.get_step_callback(),
task_callback=self.get_task_callback(),
)
JSON Agent
This component creates a JSON agent from a JSON or YAML file and an LLM.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
path |
File |
Path to the JSON or YAML file |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
JSON agent instance |
Component code
json.py
from pathlib import Path
import yaml
from langchain.agents import AgentExecutor
from langchain_community.agent_toolkits import create_json_agent
from langchain_community.agent_toolkits.json.toolkit import JsonToolkit
from langchain_community.tools.json.tool import JsonSpec
from langflow.base.agents.agent import LCAgentComponent
from langflow.inputs import FileInput, HandleInput
class JsonAgentComponent(LCAgentComponent):
display_name = "JsonAgent"
description = "Construct a json agent from an LLM and tools."
name = "JsonAgent"
legacy: bool = True
inputs = [
*LCAgentComponent._base_inputs,
HandleInput(
name="llm",
display_name="Language Model",
input_types=["LanguageModel"],
required=True,
),
FileInput(
name="path",
display_name="File Path",
file_types=["json", "yaml", "yml"],
required=True,
),
]
def build_agent(self) -> AgentExecutor:
path = Path(self.path)
if path.suffix in {"yaml", "yml"}:
with path.open(encoding="utf-8") as file:
yaml_dict = yaml.safe_load(file)
spec = JsonSpec(dict_=yaml_dict)
else:
spec = JsonSpec.from_file(path)
toolkit = JsonToolkit(spec=spec)
return create_json_agent(llm=self.llm, toolkit=toolkit, **self.get_agent_kwargs())
OpenAI Tools Agent
This component creates an OpenAI Tools Agent using LangChain. For more information, see the LangChain documentation.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent (must be tool-enabled) |
system_prompt |
String |
System prompt for the agent |
user_prompt |
String |
User prompt template (must contain 'input' key) |
chat_history |
List[Data] |
Optional chat history for the agent |
tools |
List[Tool] |
List of tools available to the agent |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
OpenAI Tools Agent instance |
Component code
openai_tools.py
from langchain.agents import create_openai_tools_agent
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, PromptTemplate
from langflow.base.agents.agent import LCToolsAgentComponent
from langflow.inputs import MultilineInput
from langflow.inputs.inputs import DataInput, HandleInput
from langflow.schema import Data
class OpenAIToolsAgentComponent(LCToolsAgentComponent):
display_name: str = "OpenAI Tools Agent"
description: str = "Agent that uses tools via openai-tools."
icon = "LangChain"
name = "OpenAIToolsAgent"
inputs = [
*LCToolsAgentComponent._base_inputs,
HandleInput(
name="llm",
display_name="Language Model",
input_types=["LanguageModel", "ToolEnabledLanguageModel"],
required=True,
),
MultilineInput(
name="system_prompt",
display_name="System Prompt",
info="System prompt for the agent.",
value="You are a helpful assistant",
),
MultilineInput(
name="user_prompt", display_name="Prompt", info="This prompt must contain 'input' key.", value="{input}"
),
DataInput(name="chat_history", display_name="Chat History", is_list=True, advanced=True),
]
def get_chat_history_data(self) -> list[Data] | None:
return self.chat_history
def create_agent_runnable(self):
if "input" not in self.user_prompt:
msg = "Prompt must contain 'input' key."
raise ValueError(msg)
messages = [
("system", self.system_prompt),
("placeholder", "{chat_history}"),
HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=["input"], template=self.user_prompt)),
("placeholder", "{agent_scratchpad}"),
]
prompt = ChatPromptTemplate.from_messages(messages)
return create_openai_tools_agent(self.llm, self.tools, prompt)
OpenAPI Agent
This component creates an OpenAPI Agent to interact with APIs defined by OpenAPI specifications. For more information, see the LangChain documentation on OpenAPI Agents.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
path |
File |
Path to the OpenAPI specification file (JSON or YAML) |
allow_dangerous_requests |
Boolean |
Whether to allow potentially dangerous API requests |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
OpenAPI Agent instance |
Component code
openapi.py
from pathlib import Path
import yaml
from langchain.agents import AgentExecutor
from langchain_community.agent_toolkits import create_openapi_agent
from langchain_community.agent_toolkits.openapi.toolkit import OpenAPIToolkit
from langchain_community.tools.json.tool import JsonSpec
from langchain_community.utilities.requests import TextRequestsWrapper
from langflow.base.agents.agent import LCAgentComponent
from langflow.inputs import BoolInput, FileInput, HandleInput
class OpenAPIAgentComponent(LCAgentComponent):
display_name = "OpenAPI Agent"
description = "Agent to interact with OpenAPI API."
name = "OpenAPIAgent"
icon = "LangChain"
inputs = [
*LCAgentComponent._base_inputs,
HandleInput(name="llm", display_name="Language Model", input_types=["LanguageModel"], required=True),
FileInput(name="path", display_name="File Path", file_types=["json", "yaml", "yml"], required=True),
BoolInput(name="allow_dangerous_requests", display_name="Allow Dangerous Requests", value=False, required=True),
]
def build_agent(self) -> AgentExecutor:
path = Path(self.path)
if path.suffix in {"yaml", "yml"}:
with path.open(encoding="utf-8") as file:
yaml_dict = yaml.safe_load(file)
spec = JsonSpec(dict_=yaml_dict)
else:
spec = JsonSpec.from_file(path)
requests_wrapper = TextRequestsWrapper()
toolkit = OpenAPIToolkit.from_llm(
llm=self.llm,
json_spec=spec,
requests_wrapper=requests_wrapper,
allow_dangerous_requests=self.allow_dangerous_requests,
)
agent_args = self.get_agent_kwargs()
# This is bit weird - generally other create_*_agent functions have max_iterations in the
# `agent_executor_kwargs`, but openai has this parameter passed directly.
agent_args["max_iterations"] = agent_args["agent_executor_kwargs"]["max_iterations"]
del agent_args["agent_executor_kwargs"]["max_iterations"]
return create_openapi_agent(llm=self.llm, toolkit=toolkit, **agent_args)
SQL Agent
This component creates a SQL Agent to interact with SQL databases.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
database_uri |
String |
URI of the SQL database to connect to |
extra_tools |
List[Tool] |
Additional tools to provide to the agent (optional) |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
SQL Agent instance |
Component code
sql.py
from langchain.agents import AgentExecutor
from langchain_community.agent_toolkits import SQLDatabaseToolkit
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_community.utilities import SQLDatabase
from langflow.base.agents.agent import LCAgentComponent
from langflow.inputs import HandleInput, MessageTextInput
class SQLAgentComponent(LCAgentComponent):
display_name = "SQLAgent"
description = "Construct an SQL agent from an LLM and tools."
name = "SQLAgent"
icon = "LangChain"
inputs = [
*LCAgentComponent._base_inputs,
HandleInput(name="llm", display_name="Language Model", input_types=["LanguageModel"], required=True),
MessageTextInput(name="database_uri", display_name="Database URI", required=True),
HandleInput(
name="extra_tools",
display_name="Extra Tools",
input_types=["Tool", "BaseTool"],
is_list=True,
advanced=True,
),
]
def build_agent(self) -> AgentExecutor:
db = SQLDatabase.from_uri(self.database_uri)
toolkit = SQLDatabaseToolkit(db=db, llm=self.llm)
agent_args = self.get_agent_kwargs()
agent_args["max_iterations"] = agent_args["agent_executor_kwargs"]["max_iterations"]
del agent_args["agent_executor_kwargs"]["max_iterations"]
return create_sql_agent(llm=self.llm, toolkit=toolkit, extra_tools=self.extra_tools or [], **agent_args)
Sequential Crew
This component represents a group of agents with tasks that are executed sequentially. This component allows for the creation of a crew that performs tasks in a specific order.
For more information, see the CrewAI documentation.
Parameters
Name | Display Name | Info |
---|---|---|
tasks |
Tasks |
List of SequentialTask objects representing the tasks to be executed |
verbose |
Verbose |
Enables verbose output for detailed logging |
memory |
Memory |
Specifies the memory configuration for the crew |
use_cache |
Use Cache |
Enables caching of results |
max_rpm |
Max RPM |
Sets the maximum requests per minute |
share_crew |
Share Crew |
Determines if the crew information is shared among agents |
function_calling_llm |
Function Calling LLM |
Specifies the language model for function calling |
Name | Display Name | Info |
---|---|---|
crew |
Crew |
The constructed Crew object with sequential task execution |
Component code
sequential_crew.py
from crewai import Agent, Crew, Process, Task
from langflow.base.agents.crewai.crew import BaseCrewComponent
from langflow.io import HandleInput
from langflow.schema.message import Message
class SequentialCrewComponent(BaseCrewComponent):
display_name: str = "Sequential Crew"
description: str = "Represents a group of agents with tasks that are executed sequentially."
documentation: str = "https://docs.crewai.com/how-to/Sequential/"
icon = "CrewAI"
inputs = [
*BaseCrewComponent._base_inputs,
HandleInput(name="tasks", display_name="Tasks", input_types=["SequentialTask"], is_list=True),
]
def get_tasks_and_agents(self, agents_list=None) -> tuple[list[Task], list[Agent]]:
if not agents_list:
agents_list = [task.agent for task in self.tasks] or []
# Use the superclass implementation, passing the customized agents_list
return super().get_tasks_and_agents(agents_list=agents_list)
def build_crew(self) -> Message:
tasks, agents = self.get_tasks_and_agents()
return Crew(
agents=agents,
tasks=tasks,
process=Process.sequential,
verbose=self.verbose,
memory=self.memory,
cache=self.use_cache,
max_rpm=self.max_rpm,
share_crew=self.share_crew,
function_calling_llm=self.function_calling_llm,
step_callback=self.get_step_callback(),
task_callback=self.get_task_callback(),
)
Sequential task agent
This component creates a CrewAI Task and its associated Agent, allowing for the definition of sequential tasks with specific agent roles and capabilities.
For more information, see the CrewAI documentation.
Parameters
Name | Display Name | Info |
---|---|---|
role |
Role |
The role of the agent |
goal |
Goal |
The objective of the agent |
backstory |
Backstory |
The backstory of the agent |
tools |
Tools |
Tools at agent’s disposal |
llm |
Language Model |
Language model that will run the agent |
memory |
Memory |
Whether the agent should have memory or not |
verbose |
Verbose |
Enables verbose output |
allow_delegation |
Allow Delegation |
Whether the agent is allowed to delegate tasks to other agents |
allow_code_execution |
Allow Code Execution |
Whether the agent is allowed to execute code |
agent_kwargs |
Agent kwargs |
Additional kwargs for the agent |
task_description |
Task Description |
Descriptive text detailing task’s purpose and execution |
expected_output |
Expected Task Output |
Clear definition of expected task outcome |
async_execution |
Async Execution |
Boolean flag indicating asynchronous task execution |
previous_task |
Previous Task |
The previous task in the sequence (for chaining) |
Name | Display Name | Info |
---|---|---|
task_output |
Sequential Task |
List of SequentialTask objects representing the created task(s) |
Component code
sequential_task.py
from langflow.base.agents.crewai.tasks import SequentialTask
from langflow.custom import Component
from langflow.io import BoolInput, HandleInput, MultilineInput, Output
class SequentialTaskComponent(Component):
display_name: str = "Sequential Task"
description: str = "Each task must have a description, an expected output and an agent responsible for execution."
icon = "CrewAI"
inputs = [
MultilineInput(
name="task_description",
display_name="Description",
info="Descriptive text detailing task's purpose and execution.",
),
MultilineInput(
name="expected_output",
display_name="Expected Output",
info="Clear definition of expected task outcome.",
),
HandleInput(
name="tools",
display_name="Tools",
input_types=["Tool"],
is_list=True,
info="List of tools/resources limited for task execution. Uses the Agent tools by default.",
required=False,
advanced=True,
),
HandleInput(
name="agent",
display_name="Agent",
input_types=["Agent"],
info="CrewAI Agent that will perform the task",
required=True,
),
HandleInput(
name="task",
display_name="Task",
input_types=["SequentialTask"],
info="CrewAI Task that will perform the task",
),
BoolInput(
name="async_execution",
display_name="Async Execution",
value=True,
advanced=True,
info="Boolean flag indicating asynchronous task execution.",
),
]
outputs = [
Output(display_name="Task", name="task_output", method="build_task"),
]
def build_task(self) -> list[SequentialTask]:
tasks: list[SequentialTask] = []
task = SequentialTask(
description=self.task_description,
expected_output=self.expected_output,
tools=self.agent.tools,
async_execution=False,
agent=self.agent,
)
tasks.append(task)
self.status = task
if self.task:
if isinstance(self.task, list) and all(isinstance(task, SequentialTask) for task in self.task):
tasks = self.task + tasks
elif isinstance(self.task, SequentialTask):
tasks = [self.task, *tasks]
return tasks
Tool Calling Agent
This component creates a Tool Calling Agent using LangChain.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
system_prompt |
String |
System prompt for the agent |
user_prompt |
String |
User prompt template (must contain 'input' key) |
chat_history |
List[Data] |
Optional chat history for the agent |
tools |
List[Tool] |
List of tools available to the agent |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
Tool Calling Agent instance |
Component code
tool_calling.py
from langchain.agents import create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate
from langflow.base.agents.agent import LCToolsAgentComponent
from langflow.inputs import MessageTextInput
from langflow.inputs.inputs import DataInput, HandleInput
from langflow.schema import Data
class ToolCallingAgentComponent(LCToolsAgentComponent):
display_name: str = "Tool Calling Agent"
description: str = "An agent designed to utilize various tools seamlessly within workflows."
icon = "LangChain"
name = "ToolCallingAgent"
inputs = [
*LCToolsAgentComponent._base_inputs,
HandleInput(
name="llm",
display_name="Language Model",
input_types=["LanguageModel"],
required=True,
info="Language model that the agent utilizes to perform tasks effectively.",
),
MessageTextInput(
name="system_prompt",
display_name="System Prompt",
info="System prompt to guide the agent's behavior.",
value="You are a helpful assistant that can use tools to answer questions and perform tasks.",
),
DataInput(
name="chat_history",
display_name="Chat Memory",
is_list=True,
advanced=True,
info="This input stores the chat history, allowing the agent to remember previous conversations.",
),
]
def get_chat_history_data(self) -> list[Data] | None:
return self.chat_history
def create_agent_runnable(self):
messages = [
("system", "{system_prompt}"),
("placeholder", "{chat_history}"),
("human", "{input}"),
("placeholder", "{agent_scratchpad}"),
]
prompt = ChatPromptTemplate.from_messages(messages)
self.validate_tool_names()
try:
return create_tool_calling_agent(self.llm, self.tools or [], prompt)
except NotImplementedError as e:
message = f"{self.display_name} does not support tool calling. Please try using a compatible model."
raise NotImplementedError(message) from e
Vector Store Agent
This component creates a Vector Store Agent using LangChain.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
vectorstore |
VectorStoreInfo |
Vector store information for the agent to use |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
Vector Store Agent instance |
Component code
vector_store.py
from langchain_core.vectorstores import VectorStoreRetriever
from langflow.custom import CustomComponent
from langflow.field_typing import VectorStore
class VectoStoreRetrieverComponent(CustomComponent):
display_name = "VectorStore Retriever"
description = "A vector store retriever"
name = "VectorStoreRetriever"
legacy: bool = True
icon = "LangChain"
def build_config(self):
return {
"vectorstore": {"display_name": "Vector Store", "type": VectorStore},
}
def build(self, vectorstore: VectorStore) -> VectorStoreRetriever:
return vectorstore.as_retriever()
Vector Store Router Agent
This component creates a Vector Store Router Agent using LangChain.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
vectorstores |
List[VectorStoreInfo] |
List of vector store information for the agent to route between |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
Vector Store Router Agent instance |
Component code
vector_store_router.py
from langchain.agents import AgentExecutor, create_vectorstore_router_agent
from langchain.agents.agent_toolkits.vectorstore.toolkit import VectorStoreRouterToolkit
from langflow.base.agents.agent import LCAgentComponent
from langflow.inputs import HandleInput
class VectorStoreRouterAgentComponent(LCAgentComponent):
display_name = "VectorStoreRouterAgent"
description = "Construct an agent from a Vector Store Router."
name = "VectorStoreRouterAgent"
legacy: bool = True
inputs = [
*LCAgentComponent._base_inputs,
HandleInput(
name="llm",
display_name="Language Model",
input_types=["LanguageModel"],
required=True,
),
HandleInput(
name="vectorstores",
display_name="Vector Stores",
input_types=["VectorStoreInfo"],
is_list=True,
required=True,
),
]
def build_agent(self) -> AgentExecutor:
toolkit = VectorStoreRouterToolkit(vectorstores=self.vectorstores, llm=self.llm)
return create_vectorstore_router_agent(llm=self.llm, toolkit=toolkit, **self.get_agent_kwargs())
XML Agent
This component creates an XML Agent using LangChain. The agent uses XML formatting for tool instructions to the Language Model.
Parameters
Name | Type | Description |
---|---|---|
llm |
LanguageModel |
Language model to use for the agent |
user_prompt |
String |
Custom prompt template for the agent (includes XML formatting instructions) |
tools |
List[Tool] |
List of tools available to the agent |
Name | Type | Description |
---|---|---|
agent |
AgentExecutor |
XML Agent instance |
Component code
xml.py
from langchain.agents import create_xml_agent
from langchain_core.prompts import ChatPromptTemplate, HumanMessagePromptTemplate, PromptTemplate
from langflow.base.agents.agent import LCToolsAgentComponent
from langflow.inputs import MultilineInput
from langflow.inputs.inputs import DataInput, HandleInput
from langflow.schema import Data
class XMLAgentComponent(LCToolsAgentComponent):
display_name: str = "XML Agent"
description: str = "Agent that uses tools formatting instructions as xml to the Language Model."
icon = "LangChain"
beta = True
name = "XMLAgent"
inputs = [
*LCToolsAgentComponent._base_inputs,
HandleInput(name="llm", display_name="Language Model", input_types=["LanguageModel"], required=True),
DataInput(name="chat_history", display_name="Chat History", is_list=True, advanced=True),
MultilineInput(
name="system_prompt",
display_name="System Prompt",
info="System prompt for the agent.",
value="""You are a helpful assistant. Help the user answer any questions.
You have access to the following tools:
{tools}
In order to use a tool, you can use <tool></tool> and <tool_input></tool_input> tags. You will then get back a response in the form <observation></observation>
For example, if you have a tool called 'search' that could run a google search, in order to search for the weather in SF you would respond:
<tool>search</tool><tool_input>weather in SF</tool_input>
<observation>64 degrees</observation>
When you are done, respond with a final answer between <final_answer></final_answer>. For example:
<final_answer>The weather in SF is 64 degrees</final_answer>
Begin!
Question: {input}
{agent_scratchpad}
""", # noqa: E501
),
MultilineInput(
name="user_prompt", display_name="Prompt", info="This prompt must contain 'input' key.", value="{input}"
),
]
def get_chat_history_data(self) -> list[Data] | None:
return self.chat_history
def create_agent_runnable(self):
if "input" not in self.user_prompt:
msg = "Prompt must contain 'input' key."
raise ValueError(msg)
messages = [
("system", self.system_prompt),
("placeholder", "{chat_history}"),
HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=["input"], template=self.user_prompt)),
("ai", "{agent_scratchpad}"),
]
prompt = ChatPromptTemplate.from_messages(messages)
return create_xml_agent(self.llm, self.tools, prompt)