Spaces:
Sleeping
Sleeping
File size: 5,711 Bytes
3973360 bec62f4 3973360 0171bb1 3973360 53916e6 3973360 0171bb1 3973360 0171bb1 3973360 0171bb1 3973360 0171bb1 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 | from langchain_core.tools import tool
from langchain_core.runnables.config import RunnableConfig
from datetime import datetime
from src.utils.logger import logger
from src.apis.controllers.scheduling_controller import (
create_a_activity_controller,
update_a_activity_controller,
delete_activities_controller,
search_activities_controller,
)
@tool
async def create_a_activity(
description: str,
activity_category: str,
start_time: datetime,
end_time: datetime,
config: RunnableConfig,
) -> str:
"""Create an activity by extracting details from conversation context if available.
Note: Call tool when user creates schedule for a single location
Args:
description (str): Concise description of the activity. Not too detailing.
activity_category (str): value in list ["work", "study", "relax", "exercise", "other"] classifying the activity based on user's description. Not required user typing the category, it can be extracted from the description. If not sure then return 'other'.
start_time (datetime): Activity's start time in the format "YYYY-MM-DD HH:MM:SS".
end_time (datetime): Activity's end time in the format "YYYY-MM-DD HH:MM:SS".
"""
try:
# Convert start_time and end_time to the required format
configuration = config.get("configurable", {})
user_id = configuration.get("user_id", None)
if not user_id:
logger.warning(f"User not login")
return "You are not logged in. Please log in to use this feature"
response = await create_a_activity_controller(
None,
activity_category,
description,
start_time,
end_time,
user_id,
overlap_allow=False,
)
return response["message"]
except Exception as e:
logger.error(f"Error creating activity: {e}")
return f"Error creating activity {e}"
@tool
async def search_activities(
start_time: datetime,
end_time: datetime,
config: RunnableConfig,
) -> str:
"""Search for activities by extracting details from conversation context if available or asking the user for the details.
Using
Args:
start_time (datetime): Activity's start time in the format "YYYY-MM-DD HH:MM:SS". Ask the user for the start time of the activity.
end_time (datetime): Activity's end time in the format "YYYY-MM-DD HH:MM:SS". Ask the user if they not mentioned the end time of the activity.
"""
try:
configuration = config.get("configurable", {})
user_id = configuration.get("user_id", None)
if not user_id:
logger.warning(f"User not login")
return "You are not logged in. Please log in to use this feature"
response = await search_activities_controller(start_time, end_time, user_id)
return response["message"]
except Exception as e:
logger.error(f"Error searching activities: {e}")
return f"Error searching activities {e}"
@tool
async def update_a_activiy(
description: str,
activity_category: str,
start_time: datetime,
end_time: datetime,
config: RunnableConfig,
) -> str:
"""Update an activity by extracting details from conversation context if available or asking the user for the details.
description (str): Concise description of the activity. Not too detailing.
activity_category (str): value in list ["work", "study", "relax", "exercise", "other"] classifying the activity based on user's description. Not required user typing the category, it can be extracted from the description. If not sure then return 'other'.
start_time (datetime): Activity's start time in the format "YYYY-MM-DD HH:MM:SS".
end_time (datetime): Activity's end time in the format "YYYY-MM-DD HH:MM:SS".
"""
try:
configuration = config.get("configurable", {})
user_id = configuration.get("user_id", None)
if not user_id:
logger.warning(f"User not login")
return "You are not logged in. Please log in to use this feature"
response = await update_a_activity_controller(
None,
activity_category,
description,
start_time,
end_time,
user_id,
)
return response["message"]
except Exception as e:
logger.error(f"Error updating activity: {e}")
return f"Error updating activity {e}"
@tool
async def delete_a_activity(
start_time: datetime,
end_time: datetime,
config: RunnableConfig,
) -> str:
"""Delete an activity by extracting details from conversation context if available or asking the user for the details.
Args:
start_time (datetime): Activity's start time in the format "YYYY-MM-DD HH:MM:SS".
end_time (datetime): Activity's end time in the format "YYYY-MM-DD HH:MM:SS".
"""
try:
configuration = config.get("configurable", {})
user_id = configuration.get("user_id", None)
if not user_id:
logger.warning(f"User not login")
return "You are not logged in. Please log in to use this feature"
response = await delete_activities_controller(
None, start_time, end_time, user_id
)
return response["message"]
except Exception as e:
logger.error(f"Error deleting activity: {e}")
return f"Error deleting activity {e}"
logger.error(f"Error deleting activity: {e}")
return f"Error deleting activity {e}"
|