Update 2 - Functional Code Using Flows
I’d like to contribute to the discussion by presenting an alternative approach to solving the proposed problem. As in many real-world scenarios, there are multiple ways to tackle an issue.
Personally, I’m not a fan of using Process.hierarchical. To me (and I emphasize, this is just my personal opinion), it feels more like a chaotic situation with a bunch of people trying to solve the same problem simultaneously, much like my group projects back in fifth grade. 
In this draft I’m presenting, I’ve adopted the flows paradigm from CrewAI. I created a step where it’s decided who will participate in the decision-making process, and only those chosen entities will influence the final answer.
Of course, there’s a lot that can (and should!) be improved in the code presented: the meeting with each participant could be asynchronous for speed; better software engineering practices could be adopted, such as isolating agents and their tasks in separate files, using docstrings, using blank lines to improve readability, etc. I hope you understand that the idea was to have a self-contained, simple, didactic, and functional draft. I’m using OpenRouter, but feel free to use the LLM provider of your choice.
from crewai import Agent, Task, LLM
from crewai.flow.flow import Flow, listen, start
from pydantic import BaseModel
from typing import cast, List
import os
os.environ["OPENROUTER_API_KEY"] = ""
class MeetingPlan(BaseModel):
chosen_specialists: List[str] = []
class CustomerServiceState(BaseModel):
ticket: str = ""
chosen_specialists: List[str] = []
opinions: List[str] = []
response: str = ""
specialists_llm = LLM(
model='openrouter/google/gemini-2.0-flash',
temperature=0.5,
max_tokens=512
)
manager_llm = LLM(
model='openrouter/google/gemini-2.0-flash',
temperature=0.7,
max_tokens=512
)
def technical_support_specialist(question: str) -> str:
agent = Agent(
role="Technical Support Specialist",
goal="Provide expert technical support to users.",
backstory="You are a highly skilled technical support "
"specialist with years of experience in resolving "
"complex technical issues.",
llm=specialists_llm,
allow_delegation=False,
max_iter=2,
max_execution_time=30,
verbose=True
)
task = Task(
description="Analyze the user question and provide your "
"opinion only on technical issues. If the question "
"is not related to your expertise, state 'no opinion'. "
"Answer objectively in 400 characters or less.\n\n"
f"User question: {question}",
expected_output="Objective technical opinion on the user "
"question or 'no opinion'.",
agent=agent
)
opinion = task.execute_sync()
return opinion.raw
def billing_support_specialist(question: str) -> str:
agent = Agent(
role="Billing Support Specialist",
goal="Provide expert opinion on billing issues.",
backstory="You are an experienced billing support "
"specialist adept at resolving billing "
"inquiries and providing clear opinions.",
llm=specialists_llm,
allow_delegation=False,
max_iter=2,
max_execution_time=30,
verbose=True
)
task = Task(
description="Analyze the user question and provide "
"your opinion only on billing issues. If "
"the question is not related to billing, "
"state 'no opinion'. Answer objectively in "
"400 characters or less.\n\n"
f"User question: {question}",
expected_output="Objective billing opinion or 'no opinion'.",
agent=agent
)
opinion = task.execute_sync()
return opinion.raw
def manager(team: str, ticket: str) -> List[str]:
agent = Agent(
role="Customer Support Manager",
goal="Select the best team members for customer "
"support meetings to resolve customer issues efficiently.",
backstory="You are an experienced customer support "
"manager with a proven track record of leading "
"high-performing support teams and resolving "
"complex customer issues effectively.",
llm=manager_llm,
allow_delegation=False,
max_iter=3,
max_execution_time=40,
verbose=True
)
task = Task(
description="Analyze the customer's ticket and determine "
"which specialists from your team should attend "
"a meeting to address the issue. Provide only a "
"list of names. Exclude experts who are not "
"relevant. If no specialist is needed, return an "
"empty list.\n\n"
f"Team: {team}\nTicket: {ticket}",
expected_output="List of names of relevant experts from "
"the team or an empty list if no expert is needed.",
agent=agent,
output_pydantic=MeetingPlan,
)
meeting = task.execute_sync()
if meeting.pydantic:
meeting.pydantic = cast(MeetingPlan, meeting.pydantic)
return meeting.pydantic.chosen_specialists
else:
raise ValueError("Invalid list of specialists: {meeting.raw}")
def client_outcome_architect(ticket: str, opinions: str) -> str:
agent = Agent(
role="Customer Response Architect",
goal="Craft helpful customer responses from expert input.",
backstory="Expert in customer service and synthesizing "
"information to create clear, friendly replies.",
llm=specialists_llm,
allow_delegation=False,
max_iter=2,
max_execution_time=30,
verbose=True
)
task = Task(
description="Generate customer response from ticket and "
"expert opinions. Be friendly and accurate. "
"Rely on expert input only. Answer in "
"500 characters or less. If no input, say: "
"I'm sorry, but our team can't help you.\n\n"
f"Ticket: {ticket}\nSpecialists opinions: {opinions}",
expected_output="Friendly, accurate customer response or "
"I'm sorry, but our team can't help you.",
agent=agent
)
outcome = task.execute_sync()
return outcome.raw
class CustomerServiceFlow(Flow[CustomerServiceState]):
available_specialists = {
'technical_support_specialist': technical_support_specialist,
'billing_support_specialist': billing_support_specialist,
}
@start()
def schedule_meeting(self):
team = ', '.join(
[repr(name) for name in self.available_specialists.keys()]
)
ticket = self.state.ticket
chosen_specialists = manager(team, ticket)
self.state.chosen_specialists = chosen_specialists
@listen(schedule_meeting)
def conduct_meeting(self):
opinions: List[str] = []
for specialist in self.state.chosen_specialists:
try:
opinion = self.available_specialists[specialist](
self.state.ticket
)
opinions.append(f"{specialist} stated: {opinion}")
except:
print(f"\nError: '{specialist}' key is not available.\n")
continue
self.state.opinions = opinions
@listen(conduct_meeting)
def generate_client_response(self):
opinions = '; '.join(self.state.opinions)
client_response = client_outcome_architect(
self.state.ticket, opinions
)
self.state.response = client_response
return client_response
if __name__ == '__main__':
question = input("[🤖 Help Desk]: Hi! How can I help you today?\n")
flow = CustomerServiceFlow()
result = flow.kickoff(
inputs={
"ticket": question
}
)
print(f"\n[🤖 Final Answer]:\n{result}")
print(f"\n[🤖 Flow State]:\n{flow.state}\n")