From e368c1aa10e74c0d1fdd8d3862a052fc4b43d324 Mon Sep 17 00:00:00 2001 From: Sarah Wooders Date: Tue, 1 Oct 2024 14:28:39 -0700 Subject: [PATCH] chore: remove dead function loading code (#1795) --- letta/agent.py | 38 ------------------ letta/functions/functions.py | 25 +----------- tests/test_agent_function_update.py | 61 ----------------------------- 3 files changed, 1 insertion(+), 123 deletions(-) diff --git a/letta/agent.py b/letta/agent.py index 45ba882f..3217a52d 100644 --- a/letta/agent.py +++ b/letta/agent.py @@ -1117,48 +1117,10 @@ class Agent(BaseAgent): def add_function(self, function_name: str) -> str: # TODO: refactor raise NotImplementedError - # if function_name in self.functions_python.keys(): - # msg = f"Function {function_name} already loaded" - # printd(msg) - # return msg - - # available_functions = load_all_function_sets() - # if function_name not in available_functions.keys(): - # raise ValueError(f"Function {function_name} not found in function library") - - # self.functions.append(available_functions[function_name]["json_schema"]) - # self.functions_python[function_name] = available_functions[function_name]["python_function"] - - # msg = f"Added function {function_name}" - ## self.save() - # self.update_state() - # printd(msg) - # return msg def remove_function(self, function_name: str) -> str: # TODO: refactor raise NotImplementedError - # if function_name not in self.functions_python.keys(): - # msg = f"Function {function_name} not loaded, ignoring" - # printd(msg) - # return msg - - ## only allow removal of user defined functions - # user_func_path = Path(USER_FUNCTIONS_DIR) - # func_path = Path(inspect.getfile(self.functions_python[function_name])) - # is_subpath = func_path.resolve().parts[: len(user_func_path.resolve().parts)] == user_func_path.resolve().parts - - # if not is_subpath: - # raise ValueError(f"Function {function_name} is not user defined and cannot be removed") - - # self.functions = [f_schema for f_schema in self.functions if f_schema["name"] != function_name] - # self.functions_python.pop(function_name) - - # msg = f"Removed function {function_name}" - ## self.save() - # self.update_state() - # printd(msg) - # return msg def update_state(self) -> AgentState: message_ids = [msg.id for msg in self._messages] diff --git a/letta/functions/functions.py b/letta/functions/functions.py index e4600499..43b7f17e 100644 --- a/letta/functions/functions.py +++ b/letta/functions/functions.py @@ -1,17 +1,12 @@ import importlib import inspect import os -import sys from textwrap import dedent # remove indentation from types import ModuleType -from letta.constants import CLI_WARNING_PREFIX, LETTA_DIR +from letta.constants import CLI_WARNING_PREFIX from letta.functions.schema_generator import generate_schema -USER_FUNCTIONS_DIR = os.path.join(LETTA_DIR, "functions") - -sys.path.append(USER_FUNCTIONS_DIR) - def parse_source_code(func) -> str: """Parse the source code of a function and remove indendation""" @@ -68,24 +63,6 @@ def validate_function(module_name, module_full_path): return True, None -def write_function(module_name: str, function_name: str, function_code: str): - """Write a function to a file in the user functions directory""" - # Create the user functions directory if it doesn't exist - if not os.path.exists(USER_FUNCTIONS_DIR): - os.makedirs(USER_FUNCTIONS_DIR) - - # Write the function to a file - file_path = os.path.join(USER_FUNCTIONS_DIR, f"{module_name}.py") - with open(file_path, "w", encoding="utf-8") as f: - f.write(function_code) - succ, error = validate_function(module_name, file_path) - - # raise error if function cannot be loaded - if not succ: - raise ValueError(error) - return file_path - - def load_function_file(filepath: str) -> dict: file = os.path.basename(filepath) module_name = file[:-3] # Remove '.py' from filename diff --git a/tests/test_agent_function_update.py b/tests/test_agent_function_update.py index b861be5d..771a7801 100644 --- a/tests/test_agent_function_update.py +++ b/tests/test_agent_function_update.py @@ -1,10 +1,8 @@ -import inspect import os import pytest from letta import create_client -from letta.functions.functions import USER_FUNCTIONS_DIR from letta.schemas.message import Message from letta.utils import assistant_function_to_tool, json_dumps from tests.utils import create_config, wipe_config @@ -37,12 +35,6 @@ def agent(): return client.server._get_or_load_agent(agent_id=agent_state.id) -@pytest.fixture(scope="module") -def hello_world_function(): - with open(os.path.join(USER_FUNCTIONS_DIR, "hello_world.py"), "w", encoding="utf-8") as f: - f.write(inspect.getsource(hello_world)) - - @pytest.fixture(scope="module") def ai_function_call(): return Message( @@ -57,56 +49,3 @@ def ai_function_call(): } ) ) - - -# TODO: add back once implementation completed -# def test_add_function_happy(agent, hello_world_function, ai_function_call): -# agent.add_function("hello_world") -# -# assert "hello_world" in [f_schema["name"] for f_schema in agent.functions] -# assert "hello_world" in agent.functions_python.keys() -# -# msgs, heartbeat_req, function_failed = agent._handle_ai_response(ai_function_call) -# content = json_loads(msgs[-1].to_openai_dict()["content"]) -# assert content["message"] == "hello, world!" -# assert content["status"] == "OK" -# assert not function_failed - - -# def test_add_function_already_loaded(agent, hello_world_function): -# agent.add_function("hello_world") -# # no exception for duplicate loading -# agent.add_function("hello_world") -# -# -# def test_add_function_not_exist(agent): -# # pytest assert exception -# with pytest.raises(ValueError): -# agent.add_function("non_existent") -# -# -# def test_remove_function_happy(agent, hello_world_function): -# agent.add_function("hello_world") -# -# # ensure function is loaded -# assert "hello_world" in [f_schema["name"] for f_schema in agent.functions] -# assert "hello_world" in agent.functions_python.keys() -# -# agent.remove_function("hello_world") -# -# assert "hello_world" not in [f_schema["name"] for f_schema in agent.functions] -# assert "hello_world" not in agent.functions_python.keys() -# -# -# def test_remove_function_not_exist(agent): -# # do not raise error -# agent.remove_function("non_existent") -# -# -# def test_remove_base_function_fails(agent): -# with pytest.raises(ValueError): -# agent.remove_function("send_message") -# - -if __name__ == "__main__": - pytest.main(["-vv", os.path.abspath(__file__)])