For a chatbot using Crewai, is it better to use a sequential or hierarchical process? Or is there no silver bullet?
The idea of the agent is to search for products, add them to the cart, and then process the payment.
Does anyone have an example using the hierarchical method? It’s a bit hard to understand. I created one, but it executes all tasks at once.
Example:
python
@CrewBase
class AgentHierarchical():
"""AgentHierarchical crew"""
agents: List[BaseAgent]
tasks: List[Task]
# Manager Agent that coordinates the entire service
@agent
def agent_manager(self) -> Agent:
return Agent(
config=self.agents_config['agent_manager'],
verbose=True,
allow_delegation=True # Allows task delegation
)
# Agent specialized in product search
@agent
def agent_sales(self) -> Agent:
return Agent(
config=self.agents_config['agent_sales'],
tools=[ConsultaProdutosTool()],
verbose=True
)
# Agent specialized in cart management
@agent
def agent_cart(self) -> Agent:
return Agent(
config=self.agents_config['agent_cart'],
tools=[CarrinhoTool()],
verbose=True
)
# Agent specialized in payment processing
@agent
def agent_payment(self) -> Agent:
return Agent(
config=self.agents_config['agent_payment'],
tools=[PagamentoTool()],
verbose=True
)
# Task for product search
@task
def task_search_products(self) -> Task:
return Task(
config=self.tasks_config['task_search_products']
)
# Task for managing cart
@task
def task_manage_cart(self) -> Task:
return Task(
config=self.tasks_config['task_manage_cart']
)
# Task for payment processing
@task
def task_process_payment(self) -> Task:
return Task(
config=self.tasks_config['task_process_payment']
)
@crew
def crew(self) -> Crew:
"""Creates the AgentHierarchical crew"""
return Crew(
agents=[self.agent_sales(), self.agent_cart(), self.agent_payment()],
tasks=self.tasks,
process=Process.hierarchical, # Hierarchical process
manager_agent=self.agent_manager(), # Defines the manager
verbose=True
)
To me that sounds like a pretty straightforward and well-defined process – exactly the kind of thing Flows are great for mapping out. Here’s the documentation on that, and here’s a quick mock-up of the code:
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel, Field
from typing import List, Optional
class ShoppingState(BaseModel):
"""
Represents the state of the shopping process.
"""
product_query: Optional[str] = None
found_product_name: Optional[str] = None
cart: List[str] = Field(default_factory=list)
payment_processed: bool = False
class ShoppingChatbotFlow(Flow[ShoppingState]):
"""
A simple CrewAI Flow for a shopping chatbot.
This flow models the core shopping process, assuming an initial product
query is provided:
1. Search for the product (using the provided query).
2. Add the found product to the cart.
3. Process payment.
4. Finalize order.
"""
@start()
def search_for_product(self):
"""
Step 1: Search for the product.
This is the flow's entry point. It uses the initial `product_query`
(passed via `kickoff`) to simulate finding a product.
"""
query = self.state.product_query
print(f"\n[Chatbot] Searching for products matching: '{query}'...")
simulated_found_product = f"Premium Quality {query}"
self.state.found_product_name = simulated_found_product
print(f"[Chatbot] Found: '{self.state.found_product_name}'")
print(f"[State] found_product_name set to: '{self.state.found_product_name}'\n")
@listen(search_for_product)
def add_to_cart(self):
"""
Step 2: Add the found product to the cart.
This method runs after 'search_for_product' completes.
It simulates adding the product identified in the previous step to the cart.
"""
product_to_add = self.state.found_product_name
if not product_to_add:
print("[Chatbot] No product found to add to cart. Skipping.")
return
print(f"\n[Chatbot] Adding '{product_to_add}' to the cart...")
self.state.cart.append(product_to_add)
print(f"[Chatbot] Product added. Current cart: {self.state.cart}")
print(f"[State] cart updated: {self.state.cart}\n")
@listen(add_to_cart)
def process_payment(self):
"""
Step 3: Process the payment for items in the cart.
This method runs after 'add_to_cart' completes.
It simulates the payment process for the items currently in the cart.
"""
if not self.state.cart:
print("[Chatbot] Cart is empty. Skipping payment processing.")
return
print(f"\n[Chatbot] Processing payment for cart: {self.state.cart}...")
self.state.payment_processed = True
print("[Chatbot] Payment successful!")
print(f"[State] payment_processed set to: {self.state.payment_processed}\n")
@listen(process_payment)
def finalize_order(self):
"""
Final Step: Finalize the order.
This method runs after 'process_payment' completes.
It confirms the order details to the user.
"""
print("\n[Chatbot] Your order is complete. Thank you for shopping!")
print("\n--- Shopping Flow Final Summary ---")
print(f" Searched for: '{self.state.product_query}'")
print(f" Product found: '{self.state.found_product_name or 'N/A'}'")
print(f" Items in cart: {self.state.cart if self.state.cart else 'Empty'}")
payment_status = "Successful" if self.state.payment_processed else "Not Processed or Failed"
print(f" Payment status: {payment_status}")
print("--- End of Flow ---\n")
if __name__ == "__main__":
print("Initializing Shopping Chatbot Flow...")
shopping_flow = ShoppingChatbotFlow()
product_to_find = "Wireless Headphones"
print(f"Kicking off the flow to find: '{product_to_find}'.\n")
initial_state = {"product_query": product_to_find}
shopping_flow.kickoff(inputs=initial_state)
Would I need an agent for each step? For example, an agent to search for products, an agent to add to the cart, and another to process the payment?
Yeah, so with the approach I sketched out, at each step, you could have some straightforward Python code – you know, maybe hitting an API, doing some data validation, that kind of thing. Or, you could go for a direct LLM.call("messages"), a simple Agent.kickoff("messages"), or even a full-blown, complex Crew.kickoff() with dozens of agents, each loaded with tools. The beauty of it is you get the flexibility to really shape the solution to match how complex your actual problem is.
Anyway, that’s just one way to go about it with CrewAI. I’m positive that some of the more seasoned folks here will jump in with some great insights on other ways to tackle your specific use case.