import gc, sys, lancedb import config, lib.ragroleplay as ragroleplay_lib # --- USER TOOLS --- schema_users_store = { "type": "function", "function": { "name": "users_store", "description": "Create and store a new user profile in the knowledge database.", "parameters": { "type": "object", "properties": { "fullname": {"type": "string", "description": "Full name of the user"}, "nickname": {"type": "string", "description": "Nickname of the user"}, "alias": {"type": "string", "description": "Alias of the user"}, "persona": {"type": "string", "description": "Detailed persona or description of the user"}, "telegram_id": {"type": "string", "description": "Telegram user ID"}, "telegram_username": {"type": "string", "description": "Telegram username"}, "xampp_username": {"type": "string", "description": "XAMPP username"} }, "required": ["fullname", "nickname", "telegram_id"] } } } schema_users_filter = { "type": "function", "function": { "name": "users_filter", "description": "Retrieve user profiles based on flexible filters like names, IDs, or persona keywords.", "parameters": { "type": "object", "properties": { "query_name": {"type": "string", "description": "Search by fullname, nickname, or alias"}, "unique_id": {"type": "string", "description": "Search by telegram_id, telegram_username, or xampp_username"}, "persona_keyword": {"type": "string", "description": "Search for a keyword within the user's persona"}, "user_id": {"type": "string", "description": "Search by absolute user UUID"} } } } } schema_users_update = { "type": "function", "function": { "name": "users_update", "description": "Update an existing user profile using their unique user ID.", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "The unique UUID of the user to update"}, "fullname": {"type": "string", "description": "Updated full name"}, "nickname": {"type": "string", "description": "Updated nickname"}, "alias": {"type": "string", "description": "Updated alias"}, "persona": {"type": "string", "description": "Updated persona description"}, "telegram_id": {"type": "string", "description": "Updated telegram ID"}, "telegram_username": {"type": "string", "description": "Updated telegram username"}, "xampp_username": {"type": "string", "description": "Updated xampp username"} }, "required": ["user_id"] } } } schema_users_delete = { "type": "function", "function": { "name": "users_delete", "description": "Permanently delete a user profile from the database using their unique user ID.", "parameters": { "type": "object", "properties": { "user_id": {"type": "string", "description": "The unique UUID of the user to delete"} }, "required": ["user_id"] } } } # --- MEMORIES TOOLS --- schema_memories_check = { "type": "function", "function": { "name": "memories_check", "description": "Search and retrieve relevant memories from the knowledge database based on the provided prompt text.", "parameters": { "type": "object", "properties": { "prompt_text": {"type": "string", "description": "The text prompt to search for in memories"}, "search_limit": {"type": "integer", "description": "The maximum number of relevant memories to return"} }, "required": ["prompt_text", "search_limit"] } } } schema_memories_store = { "type": "function", "function": { "name": "memories_store", "description": "Store a new memory entry into the knowledge database.", "parameters": { "type": "object", "properties": { "character": {"type": "string", "description": "The active character name"}, "user": {"type": "string", "description": "The active user name"}, "relative_time": {"type": "string", "description": "Description of when the event happened (e.g., '2 hours ago', 'yesterday')"}, "event": {"type": "string", "description": "Brief summary of the event"}, "category": {"type": "string", "description": "Category of the memory (e.g., 'Emotional', 'Physical', 'Relationship')"}, "detail": {"type": "string", "description": "Detailed description of the event"}, "physical": {"type": "string", "description": "Physical state of the character during the event"}, "emotional": {"type": "string", "description": "Emotional state of the character during the event"} }, "required": ["character", "user", "relative_time", "event", "category", "detail", "physical", "emotional"] } } } schema_memories_filter = { "type": "function", "function": { "name": "memories_filter", "description": "Retrieve memories based on specific filters such as category, character, or user.", "parameters": { "type": "object", "properties": { "category": {"type": "string", "description": "Filter by memory category"}, "character": {"type": "string", "description": "Filter by character name"}, "user": {"type": "string", "description": "Filter by user name"} } } } } schema_memories_summarize = { "type": "function", "function": { "name": "memories_summarize", "description": "Get a condensed summary of the most relevant memories based on the prompt text.", "parameters": { "type": "object", "properties": { "prompt_text": {"type": "string", "description": "The text prompt to summarize relevant memories for"}, "search_limit": {"type": "integer", "description": "The number of memories to include in the summary", "default": 5} }, "required": ["prompt_text"] } } } schema_memories_update = { "type": "function", "function": { "name": "memories_update", "description": "Update an existing memory entry. Use the memory ID to identify the record.", "parameters": { "type": "object", "properties": { "memory_id": {"type": "string", "description": "The unique ID of the memory to update"}, "relative_time": {"type": "string", "description": "Updated relative time"}, "event": {"type": "string", "description": "Updated event summary"}, "category": {"type": "string", "description": "Updated category"}, "detail": {"type": "string", "description": "Updated detail"}, "physical": {"type": "string", "description": "Updated physical state"}, "emotional": {"type": "string", "description": "Updated emotional state"} }, "required": ["memory_id"] } } } schema_memories_delete = { "type": "function", "function": { "name": "memories_delete", "description": "Permanently delete a memory entry from the database using its unique ID.", "parameters": { "type": "object", "properties": { "memory_id": {"type": "string", "description": "The unique ID of the memory to delete"} }, "required": ["memory_id"] } } } # --- WORLD TOOLS --- schema_worlds_store = { "type": "function", "function": { "name": "worlds_store", "description": "Store a new world location/description in the knowledge database.", "parameters": { "type": "object", "properties": { "category": {"type": "string", "description": "Category of the world/location"}, "location": {"type": "string", "description": "Name of the place or location"}, "description": {"type": "string", "description": "Detailed description of the area"} }, "required": ["category", "location", "description"] } } } schema_worlds_filter = { "type": "function", "function": { "name": "worlds_filter", "description": "Search for world locations by category, name, or unique ID.", "parameters": { "type": "object", "properties": { "category": {"type": "string", "description": "Filter by category"}, "location": {"type": "string", "description": "Filter by location name"}, "world_id": {"type": "string", "description": "Filter by absolute UUID"} } } } } schema_worlds_update = { "type": "function", "function": { "name": "worlds_update", "description": "Update an existing world location entry.", "parameters": { "type": "object", "properties": { "world_id": {"type": "string", "description": "The unique UUID of the world entry"}, "category": {"type": "string", "description": "Updated category"}, "location": {"type": "string", "description": "Updated location name"}, "description": {"type": "string", "description": "Updated description"} }, "required": ["world_id"] } } } schema_worlds_delete = { "type": "function", "function": { "name": "worlds_delete", "description": "Delete a world location from the database.", "parameters": { "type": "object", "properties": { "world_id": {"type": "string", "description": "The unique UUID of the world entry"} }, "required": ["world_id"] } } } # --- OBJECT TOOLS --- schema_objects_store = { "type": "function", "function": { "name": "objects_store", "description": "Store a new object in the knowledge database.", "parameters": { "type": "object", "properties": { "keyword": {"type": "string", "description": "Keyword to trigger the object"}, "when": {"type": "string", "description": "When the object is used or appears"}, "where": {"type": "string", "description": "Where the object can be found"}, "name": {"type": "string", "description": "Name of the object"}, "description": {"type": "string", "description": "Detailed description of the object"}, "shape": {"type": "string", "description": "Physical description or shape"}, "usage": {"type": "string", "description": "How the object is used"} }, "required": ["keyword", "when", "name"] } } } schema_objects_filter = { "type": "function", "function": { "name": "objects_filter", "description": "Search for objects by keyword or unique ID.", "parameters": { "type": "object", "properties": { "keyword": {"type": "string", "description": "The object's keyword"}, "object_id": {"type": "string", "description": "The unique UUID of the object"} } } } } schema_objects_update = { "type": "function", "function": { "name": "objects_update", "description": "Update an existing object entry.", "parameters": { "type": "object", "properties": { "object_id": {"type": "string", "description": "The unique UUID of the object"}, "keyword": {"type": "string", "description": "Updated keyword"}, "when": {"type": "string", "description": "Updated usage context"}, "where": {"type": "string", "description": "Updated location"}, "name": {"type": "string", "description": "Updated name"}, "description": {"type": "string", "description": "Updated description"}, "shape": {"type": "string", "description": "Updated shape"}, "usage": {"type": "string", "description": "Updated usage"} }, "required": ["object_id"] } } } schema_objects_delete = { "type": "function", "function": { "name": "objects_delete", "description": "Delete an object from the database.", "parameters": { "type": "object", "properties": { "object_id": {"type": "string", "description": "The unique UUID of the object"}, }, "required": ["object_id"] } } } # --- OUTFIT TOOLS --- schema_outfits_store = { "type": "function", "function": { "name": "outfits_store", "description": "Store a new outfit set in the knowledge database.", "parameters": { "type": "object", "properties": { "keyword": {"type": "string", "description": "Keyword for the outfit"}, "when": {"type": "string", "description": "When this outfit is worn"}, "outfit": {"type": "string", "description": "Description of the outfit set"} }, "required": ["keyword", "when", "outfit"] } } } schema_outfits_filter = { "type": "function", "function": { "name": "outfits_filter", "description": "Search for outfits by keyword or unique ID.", "parameters": { "type": "object", "properties": { "keyword": {"type": "string", "description": "The outfit keyword"}, "outfit_id": {"type": "string", "description": "The unique UUID of the outfit"} } } } } schema_outfits_update = { "type": "function", "function": { "name": "outfits_update", "description": "Update an existing outfit entry.", "parameters": { "type": "object", "properties": { "outfit_id": {"type": "string", "description": "The unique UUID of the outfit"}, "keyword": {"type": "string", "description": "Updated keyword"}, "when": {"type": "string", "description": "Updated usage context"}, "outfit": {"type": "string", "description": "Updated outfit description"} }, "required": ["outfit_id"] } } } schema_outfits_delete = { "type": "function", "function": { "name": "outfits_delete", "description": "Delete an outfit from the database.", "parameters": { "type": "object", "properties": { "outfit_id": {"type": "string", "description": "The unique UUID of the outfit"}, }, "required": ["outfit_id"] } } } # --- TODO TOOLS --- schema_todos_store = { "type": "function", "function": { "name": "todos_store", "description": "Store a new to-do item in the knowledge database.", "parameters": { "type": "object", "properties": { "keyword": {"type": "string", "description": "Trigger keyword for the to-do"}, "when": {"type": "string", "description": "When the to-do should be performed"}, "do": {"type": "string", "description": "Action to perform"} }, "required": ["keyword", "when", "do"] } } } schema_todos_filter = { "type": "function", "function": { "name": "todos_filter", "description": "Search for to-dos by keyword or unique ID.", "parameters": { "type": "object", "properties": { "keyword": {"type": "string", "description": "The to-do keyword"}, "todo_id": {"type": "string", "description": "The unique UUID of the to-do"} } } } } schema_todos_update = { "type": "function", "function": { "name": "todos_update", "description": "Update an existing to-do entry.", "parameters": { "type": "object", "properties": { "todo_id": {"type": "string", "description": "The unique UUID of the to-do"}, "keyword": {"type": "string", "description": "Updated keyword"}, "when": {"type": "string", "description": "Updated when context"}, "do": {"type": "string", "description": "Updated action"} }, "required": ["todo_id"] } } } schema_todos_delete = { "type": "function", "function": { "name": "todos_delete", "description": "Delete a to-do from the database.", "parameters": { "type": "object", "properties": { "todo_id": {"type": "string", "description": "The unique UUID of the to-do"}, }, "required": ["todo_id"] } } } def memories_check(prompt_text, search_limit): try: db = lancedb.connect(config.ragroleplay_db_path) table = db.open_table("knowledge_memories") query_vector = ragroleplay_lib.embed_text(config.ragroleplay_model_url, config.ragroleplay_model_name, prompt_text) results = table.search(query_vector, vector_column_name="vector_context").limit(search_limit).to_list() if not results: return "Tidak ada memori yang ditemukan." output = [] output.append(f"Prompt: {prompt_text}\n") output.append(f"Ditemukan {len(results)} memori yang relevan berdasarkan KONTEKS:\n") for i, row in enumerate(results, 1): mem_info = ( f"Memori #{i}\n" f"Kapan: {row['relative_time']}\n" f"Kejadian: {row['event']}\n" f"Detail: {row['detail']}\n" f"Kondisi Fisik: {row['physical']}\n" f"Emosi: {row['emotional']}\n" f"Kategori: {row['category']}\n" f"Skor Jarak: {row.get('_distance', 'N/A')}\n" ) output.append(mem_info) result_text = "\n".join(output) del table del db gc.collect() return result_text except Exception as e: return f"Error searching memories: {str(e)}" def memories_store(character, user, relative_time, event, category, detail, physical, emotional): try: success = ragroleplay_lib.memories_store( config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, character, user, relative_time, event, category, detail, physical, emotional ) if success: return "Berhasil menyimpan memori baru ke dalam database." else: return "Gagal menyimpan memori. Silakan periksa log." except Exception as e: return f"Error while storing memory: {str(e)}" def memories_filter(category=None, character=None, user=None): try: results = ragroleplay_lib.memories_filter(config.ragroleplay_db_path, category, character, user) if not results: return "Tidak ada memori yang sesuai dengan filter tersebut." output = [] output.append(f"Ditemukan {len(results)} memori berdasarkan filter:\n") for i, row in enumerate(results, 1): mem_info = ( f"Memori #{i}\n" f"Kapan: {row['relative_time']}\n" f"Kejadian: {row['event']}\n" f"Detail: {row['detail']}\n" f"Kategori: {row['category']}\n" ) output.append(mem_info) return "\n".join(output) except Exception as e: return f"Error filtering memories: {str(e)}" def memories_summarize(prompt_text, search_limit=5): try: result = ragroleplay_lib.memories_summarize( config.ragroleplay_db_path, prompt_text, config.ragroleplay_model_url, config.ragroleplay_model_name, search_limit ) return result except Exception as e: return f"Error summarizing memories: {str(e)}" def memories_update(memory_id, **updates): try: success = ragroleplay_lib.memories_update( config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, memory_id, **updates ) if success: return "Berhasil memperbarui memori." else: return "Gagal memperbarui memori. Pastikan ID memori benar." except Exception as e: return f"Error while updating memory: {str(e)}" def memories_delete(memory_id): try: success = ragroleplay_lib.memories_delete(config.ragroleplay_db_path, memory_id) if success: return "Berhasil menghapus memori dari database." else: return "Gagal menghapus memori." except Exception as e: return f"Error while deleting memory: {str(e)}" def users_store(fullname, nickname, alias=None, persona=None, telegram_id=None, telegram_username=None, xampp_username=None): try: success = ragroleplay_lib.users_store( config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, fullname, nickname, alias, persona, telegram_id, telegram_username, xampp_username ) if success: return "Berhasil menyimpan profil user baru." else: return "Gagal menyimpan profil user." except Exception as e: return f"Error while storing user: {str(e)}" def users_filter(query_name=None, unique_id=None, persona_keyword=None, user_id=None): try: results = ragroleplay_lib.users_filter(config.ragroleplay_db_path, query_name, unique_id, persona_keyword, user_id) if not results: return "Tidak ada user yang sesuai dengan filter tersebut." output = [] output.append(f"Ditemukan {len(results)} user berdasarkan filter:\n") for i, row in enumerate(results, 1): user_info = ( f"User #{i}\n" f"ID: {row['id']}\n" f"Nama: {row['fullname']} ({row['nickname']})\n" f"Persona: {row['persona']}\n" ) output.append(user_info) return "\n".join(output) except Exception as e: return f"Error filtering users: {str(e)}" def users_update(user_id, **updates): try: success = ragroleplay_lib.users_update( config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, user_id, **updates ) if success: return "Berhasil memperbarui profil user." else: return "Gagal memperbarui profil user. Pastikan ID benar." except Exception as e: return f"Error while updating user: {str(e)}" def users_delete(user_id): try: success = ragroleplay_lib.users_delete(config.ragroleplay_db_path, user_id) if success: return "Berhasil menghapus profil user." else: return "Gagal menghapus profil user." except Exception as e: return f"Error while deleting user: {str(e)}" # --- OBJECTS IMPLEMENTATION --- def objects_store(keyword, when, where, name, description, shape, usage): try: success = ragroleplay_lib.objects_store(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, keyword, when, where, name, description, shape, usage) return "Berhasil menyimpan objek baru." if success else "Gagal menyimpan objek." except Exception as e: return f"Error: {str(e)}" def objects_filter(keyword=None, object_id=None): try: results = ragroleplay_lib.objects_filter(config.ragroleplay_db_path, keyword, object_id) if not results: return "Tidak ada objek yang ditemukan." output = [f"Ditemukan {len(results)} objek:"] for i, row in enumerate(results, 1): output.append(f"Objek #{i}\nID: {row['id']}\nNama: {row['name']}\nKeyword: {row['keyword']}\nPenggunaan: {row['usage']}") return "\n".join(output) except Exception as e: return f"Error: {str(e)}" def objects_update(object_id, **updates): try: success = ragroleplay_lib.objects_update(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, object_id, **updates) return "Berhasil memperbarui objek." if success else "Gagal memperbarui objek." except Exception as e: return f"Error: {str(e)}" def objects_delete(object_id): try: success = ragroleplay_lib.objects_delete(config.ragroleplay_db_path, object_id) return "Berhasil menghapus objek." if success else "Gagal menghapus objek." except Exception as e: return f"Error: {str(e)}" # --- OUTFIT IMPLEMENTATION --- def outfits_store(keyword, when, outfit): try: success = ragroleplay_lib.outfits_store(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, keyword, when, outfit) return "Berhasil menyimpan outfit baru." if success else "Gagal menyimpan outfit." except Exception as e: return f"Error: {str(e)}" def outfits_filter(keyword=None, outfit_id=None): try: results = ragroleplay_lib.outfits_filter(config.ragroleplay_db_path, keyword, outfit_id) if not results: return "Tidak ada outfit yang ditemukan." output = [f"Ditemukan {len(results)} outfit:"] for i, row in enumerate(results, 1): output.append(f"Outfit #{i}\nID: {row['id']}\nKeyword: {row['keyword']}\nDeskripsi: {row['outfit']}") return "\n".join(output) except Exception as e: return f"Error: {str(e)}" def outfits_update(outfit_id, **updates): try: success = ragroleplay_lib.outfits_update(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, outfit_id, **updates) return "Berhasil memperbarui outfit." if success else "Gagal memperbarui outfit." except Exception as e: return f"Error: {str(e)}" def outfits_delete(outfit_id): try: success = ragroleplay_lib.outfits_delete(config.ragroleplay_db_path, outfit_id) return "Berhasil menghapus outfit." if success else "Gagal menghapus outfit." except Exception as e: return f"Error: {str(e)}" # --- TODO IMPLEMENTATION --- def todos_store(keyword, when, do): try: success = ragroleplay_lib.todos_store(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, keyword, when, do) return "Berhasil menyimpan to-do baru." if success else "Gagal menyimpan to-do." except Exception as e: return f"Error: {str(e)}" def todos_filter(keyword=None, todo_id=None): try: results = ragroleplay_lib.todos_filter(config.ragroleplay_db_path, keyword, todo_id) if not results: return "Tidak ada to-do yang ditemukan." output = [f"Ditemukan {len(results)} to-do:"] for i, row in enumerate(results, 1): output.append(f"To-do #{i}\nID: {row['id']}\nKeyword: {row['keyword']}\nAction: {row['do']}") return "\n".join(output) except Exception as e: return f"Error: {str(e)}" def todos_update(todo_id, **updates): try: success = ragroleplay_lib.todos_update(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, todo_id, **updates) return "Berhasil memperbarui to-do." if success else "Gagal memperbarui to-do." except Exception as e: return f"Error: {str(e)}" def todos_delete(todo_id): try: success = ragroleplay_lib.todos_delete(config.ragroleplay_db_path, todo_id) return "Berhasil menghapus to-do." if success else "Gagal menghapus to-do." except Exception as e: return f"Error: {str(e)}" # --- WORLD IMPLEMENTATION --- def worlds_store(category, location, description): try: success = ragroleplay_lib.worlds_store(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, category, location, description) return "Berhasil menyimpan lokasi dunia baru." if success else "Gagal menyimpan lokasi dunia." except Exception as e: return f"Error: {str(e)}" def worlds_filter(category=None, location=None, world_id=None): try: results = ragroleplay_lib.worlds_filter(config.ragroleplay_db_path, category, location, world_id) if not results: return "Tidak ada lokasi dunia yang ditemukan." output = [f"Ditemukan {len(results)} lokasi:"] for i, row in enumerate(results, 1): output.append(f"Lokasi #{i}\nID: {row['id']}\nNama: {row['location']}\nKategori: {row['category']}\nDeskripsi: {row['description']}") return "\n".join(output) except Exception as e: return f"Error: {str(e)}" def worlds_update(world_id, **updates): try: success = ragroleplay_lib.worlds_update(config.ragroleplay_db_path, config.ragroleplay_model_url, config.ragroleplay_model_name, world_id, **updates) return "Berhasil memperbarui lokasi dunia." if success else "Gagal memperbarui lokasi dunia." except Exception as e: return f"Error: {str(e)}" def worlds_delete(world_id): try: success = ragroleplay_lib.worlds_delete(config.ragroleplay_db_path, world_id) return "Berhasil menghapus lokasi dunia." if success else "Gagal menghapus lokasi dunia." except Exception as e: return f"Error: {str(e)}"