Fimeg/letta-server
1
0
Fork 0
Code Issues Pull Requests Actions 2 Packages Projects Releases Wiki Activity

chore: remove dead function loading code (#1795)

Browse Source
This commit is contained in:
Sarah Wooders
2024-10-01 14:28:39 -07:00
committed by GitHub
parent ed7e3d1482
commit e368c1aa10
3 changed files with 1 additions and 123 deletions
Download Patch File Download Diff File Expand all files Collapse all files

38
letta/agent.py
View File

@@ -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]

25
letta/functions/functions.py
View File

@@ -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

61
tests/test_agent_function_update.py
View File

@@ -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__)])