import os
from getpass import getpass
from crewai import Agent, Task, Crew, Process, LLM
from langchain_groq import ChatGroq
from crewagents.tools.custom_tool import fetch_user_schedule, update_schedule, search_schedule, add_schedule, cancel_schedule, json_search_tool, rag_tool
from textwrap import dedent
from datetime import datetime, date, timedelta
from langchain.schema.runnable import RunnableConfig
Date and Time Setup
time = datetime.now()
date_today = date.today()
day = datetime.today().strftime(“%A”)
Initialize LLM
my_llm = LLM(
model="groq/llama3-70b-8192",
temperature=0.8,
max_tokens=1000,
)
User Info and Configuration
user_info = {
“user_id”: “123456”, # Replace with actual user ID
“user_name”: “John Doe” # Replace with actual user name
}
config = {
“configurable”: {
“user_id”: user_info[‘user_id’],
“Name”: user_info[‘user_name’], # Use string instead of a set
“thread_id”: f"thread {user_info[‘user_id’]}“,
}
}
print(type(config))
user_info_str = f"User ID: {user_info[‘user_id’]} | Name: {user_info[‘user_name’]}”
Define Retriever Agent
Retriever_Agent = Agent(
role=dedent(“”"
Retriever_Agent
“”“),
backstory=dedent(”“”
You are a knowledgeable company guide for company, a leading technology firm specializing in AI solutions, software development, and IT consulting.
Your role is to provide accurate and up-to-date information about company, including its services, company policies, and technological innovations.
You ensure users receive precise answers by retrieving relevant data from the company’s knowledge base.
“”“),
goal=dedent(”“”
- Assist users with inquiries related to company, including company details, services, policies, and technologies.
- Provide fact-based responses using the company knowledge base.
- Do not make assumptions—always verify the information before responding.
“”"),
tools=[json_search_tool, rag_tool],
cache=True,
allow_delegation=True,
verbose=False,
llm=my_llm,
config={“configurable”: {“user_id”: user_info[“user_id”]}},
)
task_1 = Task(
description=dedent(“”"
As the AI voice assistant of company, analyze the customer’s inquiry: {query}.
Provide a detailed, well-structured response that addresses all aspects of the inquiry.
Use a friendly, professional, and conversational tone.
Do not delegate further if this query is not related to scheduling.
“”“),
expected_output=dedent(”“”
A professional and friendly response that fully addresses the customer’s inquiry with accurate and verified information.
“”"),
agent=Retriever_Agent,
context_var=[“query”],
config={“configurable”: {“user_id”: user_info[“user_id”]}},
)
Define Scheduling Agent
company_Scheduling_Agent = Agent(
role=dedent(“”"
company_Scheduling_Agent
“”“),
backstory=dedent(”“”
You are the AI scheduling assistant for company, responsible for managing appointments, meetings, and schedules.
Your primary role is to assist users in scheduling, updating, searching, and canceling appointments seamlessly.
You ensure that all scheduling actions comply with time constraints, user preferences, and availability.
Your goal is to provide an efficient, organized, and hassle-free scheduling experience for users.
"""),
goal=dedent("""
- Assist users with scheduling appointments efficiently.
- Ensure time slots are available before booking.
- Validate and update schedules accurately.
- Provide clear confirmations and rescheduling options.
- Do not assume—always verify details before processing.
"""),
tools=[fetch_user_schedule, search_schedule, add_schedule, update_schedule, cancel_schedule],
allow_delegation=True,
verbose=True,
config={"configurable": {"user_id": user_info["user_id"]}},
llm=my_llm
)
Define Tasks for Scheduling Agent
task_fetch_schedule = Task(
description=dedent(“”"
Retrieve and display the user’s upcoming scheduled appointments.
user information : {user_info_str}
If no future appointments are found, prompt the user to provide their preferred date, time, and purpose for a new booking.
“”“),
expected_output=dedent(”“”
A list of future scheduled appointments with date, time, and purpose.
Or, a prompt guiding the user to provide scheduling details.
“”"),
agent=company_Scheduling_Agent,
tools=[fetch_user_schedule],
config={“configurable”: {“user_id”: user_info[“user_id”]}},
)
task_search_schedule = Task(
description=dedent(“”"
Search for available or scheduled appointments based on provided filters like date, start time, and end time.
user information : {user_info_str}
Only future appointments should be retrieved.
“”“),
expected_output=dedent(”“”
A list of appointments matching the search criteria.
If none are found, an appropriate message.
"""),
agent=company_Scheduling_Agent,
tools=[search_schedule],
config={"configurable": {"user_id": user_info["user_id"]}},
)
task_add_schedule = Task(
description=dedent(“”"
Schedule a new appointment based on user input.
user information : {user_info_str}
Ensure all required details (date, time, purpose) are captured and validate availability.
“”“),
expected_output=dedent(”“”
A confirmation message that the appointment has been scheduled.
Or an alternative suggestion if the slot is unavailable.
“”"),
agent=company_Scheduling_Agent,
tools=[add_schedule],
config={“configurable”: {“user_id”: user_info[“user_id”]}},
)
task_update_schedule = Task(
description=dedent(“”"
Update an existing appointment with new details (date, time, purpose).
user information : {user_info_str}
Confirm that the new slot is available and that the appointment is not in the past.
“”“),
expected_output=dedent(”“”
A confirmation message that the appointment has been updated,
or an explanation if the update cannot be performed.
“”"),
agent=company_Scheduling_Agent,
tools=[update_schedule],
config={“configurable”: {“user_id”: user_info[“user_id”]}},
)
task_cancel_schedule = Task(
description=dedent(“”"
Cancel an existing appointment based on the provided schedule ID.
Confirm that the appointment belongs to the user and is in the future.
"""),
expected_output=dedent("""
A confirmation message of cancellation,
or a notification if the appointment cannot be canceled.
"""),
agent=company_Scheduling_Agent,
tools=[cancel_schedule],
config={"configurable": {"user_id": user_info["user_id"]}},
)
Define Manager Agent with refined delegation instructions
company_Manager_Agent = Agent(
role=“company_Manager_Agent”,
backstory=dedent(“”"
You’re an experienced company_Manager_Agent, skilled in overseeing complex projects and guiding teams to success. Your role is to coordinate the efforts of the crew members, ensuring that each task is completed on time and to the highest standard.
“”“),
goal=dedent(”“”
Efficiently manage the crew and delegate customer queries to the appropriate agent.
“”"),
tools=,
allow_delegation=True,
verbose=True,
llm=my_llm,
max_iter=3
)
Manager Task with explicit stop condition to prevent infinite delegation
Task_manager = Task(
description=dedent(“”"
Analyze the query: ‘{query}’.
If the query contains scheduling keywords (e.g., “schedule”, “appointment”, “booking”, “reschedule”),
then delegate the query to company_Scheduling_Agent once.
else the query is delegate to Retriever_Agent once.
You’re an experienced company_Manager_Agent, skilled in overseeing complex projects and guiding teams to success. Your role is to coordinate the efforts of the crew members, ensuring that each task is completed on time and to the highest standard.
“”"),
expected_output=“A clear delegation instruction indicating which agent to use, without causing an infinite loop.”,
agent=company_Manager_Agent,
)
Define the Crew with all three agents registered
company_crew = Crew(
agents=[ company_Scheduling_Agent, Retriever_Agent],
tasks=[
Task_manager,
task_fetch_schedule,
task_search_schedule,
task_add_schedule,
task_update_schedule,
task_cancel_schedule,
task_1,
],
verbose=True,
manager_agent=company_Manager_Agent,
process=Process.hierarchical,
)
Kickoff the Crew with an example input
user_input = “Hello coudl u please tell me about your company policy”
result = company_crew.kickoff(inputs={“query”: user_input, “user_info_str”: user_info_str})
print(result)