Jarvis AI review of src/ai/agents/general/generic_tools_agent.py (ref: main)
The file src/ai/agents/general/generic_tools_agent.py was reviewed by Jarvis AI with the following findings:
-
The class
GenericToolsAgentcontains multiple class-level attributes that are mutable. This can lead to unexpected behavior when instances of the class share the same data. Consider using instance-level attributes or providing a clear warning about the shared state.Existing code snippet:
class GenericToolsAgent(BaseSingleActionAgent): ... wrong_tool_calls: list = []Suggested code snippet:
class GenericToolsAgent(BaseSingleActionAgent): ... def __init__(self): ... self.wrong_tool_calls = [] -
The
GenericToolsAgentclass is quite large and contains many methods. Consider breaking it down into smaller classes or modules to improve maintainability and readability.Existing code snippet:
class GenericToolsAgent(BaseSingleActionAgent): ...Suggested code snippet:
Consider refactoring into smaller classes or modules. -
The class attributes are not documented. Adding docstrings or comments explaining the purpose of each attribute would improve code maintainability.
Existing code snippet:
class GenericToolsAgent(BaseSingleActionAgent): ...Suggested code snippet:
Add comments or docstrings for each class attribute. -
The class attributes are initialized with
Noneor primitive types, which could lead to tightly coupled code if they are expected to be specific types. Use dependency injection or factory patterns to improve flexibility.Existing code snippet:
model_configuration: ModelConfiguration = NoneSuggested code snippet:
Use dependency injection for `model_configuration` and other attributes. -
Class attributes are initialized to
Noneor primitive types, which could lead to unintentional sharing of mutable objects across instances if they are changed to mutable types in the future. Consider initializing such attributes inside the__init__method to ensure each instance has its own copy.Existing code snippet:
model_configuration: ModelConfiguration = None conversation_manager: ConversationManager = None tools: list = None previous_work: str = None llm: BaseLanguageModel = None streaming: bool = True step_plans: dict = None step_index: int = -1 current_retries: int = 0 wrong_tool_calls: list = []Suggested code snippet:
def __init__(self): self.model_configuration = None self.conversation_manager = None self.tools = None self.previous_work = None self.llm = None self.streaming = True self.step_plans = None self.step_index = -1 self.current_retries = 0 self.wrong_tool_calls = [] -
The
planmethod is very long and contains deeply nested code. Refactor to reduce complexity and improve readability.Existing code snippet:
def plan(self, intermediate_steps: Tuple[AgentAction, str], **kwargs: Any) -> Union[AgentAction, AgentFinish]: ...Suggested code snippet:
Split the method into smaller, more focused methods. -
Magic numbers are used in the
planmethod (e.g.,self.step_index = -1). Replace them with named constants to improve code clarity.Existing code snippet:
self.step_index = -1Suggested code snippet:
INITIAL_STEP_INDEX = -1 -
The
planmethod contains duplicate logic for constructing prompts and handling actions. Consider abstracting common logic into helper methods.Existing code snippet:
plan_steps_prompt = self.get_plan_steps_prompt(...)Suggested code snippet:
Refactor to create a method for common prompt construction logic. -
The
planmethod contains hard-coded strings for logging and error messages. Consider using a centralized approach for message templates to facilitate changes and localization.Existing code snippet:
log="Agent finished."Suggested code snippet:
Use a centralized message template system. -
The
planmethod has inconsistent indentation and formatting, making it difficult to read. Ensure consistent formatting for better readability.Existing code snippet:
if (self.step_index < len(self.step_plans['steps']) and len(self.step_plans['steps']) > 0):Suggested code snippet:
Reformat code for consistent indentation and line breaks. -
The
planmethod contains commented-out code, which is a form of code clutter. Remove or document the reason for keeping the commented-out code.Existing code snippet:
# TODO: Refactor this so we can execute multiple actions at once (and handle dependencies)Suggested code snippet:
Remove or address the TODO comment. -
The
planmethod uses over-complicated expressions, such as checking the length ofself.step_plans['steps']multiple times. Simplify the logic for better readability.Existing code snippet:
if (self.step_index < len(self.step_plans['steps']) and len(self.step_plans['steps']) > 0):Suggested code snippet:
Refactor to simplify the expression. -
The
get_llmfunction is called without exception handling, which could lead to an unhandled exception if the function raises one. Consider adding a try-except block to handle potential exceptions.Existing code snippet:
self.llm = get_llm( model_configuration=self.model_configuration, tags=['generic_tools'], callbacks=self.conversation_manager.agent_callbacks, streaming=self.streaming, )Suggested code snippet:
try: self.llm = get_llm( model_configuration=self.model_configuration, tags=['generic_tools'], callbacks=self.conversation_manager.agent_callbacks, streaming=self.streaming, ) except Exception as e: logging.error(f'Failed to get LLM: {e}') # Handle the exception appropriately -
The
predictmethod ofself.llmis called without exception handling, which could lead to an unhandled exception if the method raises one. Consider adding a try-except block to handle potential exceptions.Existing code snippet:
text = self.llm.predict( plan_steps_prompt, callbacks=self.conversation_manager.agent_callbacks, )Suggested code snippet:
try: text = self.llm.predict( plan_steps_prompt, callbacks=self.conversation_manager.agent_callbacks, ) except Exception as e: logging.error(f'LLM prediction failed: {e}') # Handle the exception appropriately -
The
parse_jsonfunction is called without exception handling, which could lead to an unhandled exception if the function raises one, especially since JSON parsing is prone toValueErrorif the input is not valid JSON. Consider adding a try-except block to handle potential exceptions.Existing code snippet:
self.step_plans = parse_json( text, llm=self.llm, )Suggested code snippet:
try: self.step_plans = parse_json( text, llm=self.llm, ) except ValueError as e: logging.error(f'JSON parsing failed: {e}') # Handle the exception appropriately -
The method
remove_steps_without_toolis called within theplanmethod, which could be called multiple times. This method modifies theself.step_plansattribute directly, which can lead to side effects ifplanis called more than once. Consider returning a new list of steps without modifying the originalself.step_plans.Existing code snippet:
self.step_plans['steps'] = self.remove_steps_without_tool(self.step_plans['steps'], self.tools)Suggested code snippet:
filtered_steps = self.remove_steps_without_tool(self.step_plans['steps'], self.tools) -
The condition
self.step_index < len(self.step_plans['steps']) and len(self.step_plans['steps']) > 0is redundant. The second part of the condition is unnecessary because ifself.step_indexis less than the length ofself.step_plans['steps'], it implies that the list is not empty.Existing code snippet:
if (self.step_index < len(self.step_plans['steps']) and len(self.step_plans['steps']) > 0):Suggested code snippet:
if self.step_index < len(self.step_plans['steps']): -
The method
remove_steps_without_tooluses a list comprehension to createtool_namesbut then iterates overstepsin a for-loop to filter them. This could be optimized by using a set fortool_namesfor O(1) lookups and a list comprehension for filtering.Existing code snippet:
tool_names = [tool.name for tool in tools]Suggested code snippet:
tool_names = set(tool.name for tool in tools) filtered_steps = [step for step in steps if step['tool'] in tool_names] -
The
predictmethod ofself.llmis called without exception handling, which could lead to an unhandled exception if the method raises one. Consider adding a try-except block to handle potential exceptions.Existing code snippet:
text = self.llm.predict( tool_use_prompt, callbacks=self.conversation_manager.agent_callbacks )Suggested code snippet:
try: text = self.llm.predict( tool_use_prompt, callbacks=self.conversation_manager.agent_callbacks ) except Exception as e: logging.error(f'LLM prediction failed: {e}') # Handle the exception appropriately -
The
parse_jsonfunction is called without exception handling, which could lead to an unhandled exception if the function raises one. Consider adding a try-except block to handle potential exceptions.Existing code snippet:
action_json = parse_json( text, llm=self.llm, )Suggested code snippet:
try: action_json = parse_json( text, llm=self.llm, ) except ValueError as e: logging.error(f'JSON parsing failed: {e}') # Handle the exception appropriately -
The condition
self.step_index == -1is used to handle the case where no steps could be found. However, this condition is checked after attempting to accessself.step_plans['steps'][self.step_index], which could result in anIndexErrorifself.step_indexis -1. This check should be performed before attempting to access the list.Existing code snippet:
step = self.step_plans['steps'][self.step_index]Suggested code snippet:
if self.step_index == -1: # Handle the case where no steps could be found step = {...} else: step = self.step_plans['steps'][self.step_index] -
The
predictmethod ofself.llmis called within theparse_jsonfunction without exception handling, which could lead to an unhandled exception if the method raises one. Consider adding a try-except block to handle potential exceptions.Existing code snippet:
action_json = parse_json( text=self.llm.predict( tool_use_prompt, callbacks=self.conversation_manager.agent_callbacks ), llm=self.llm, )Suggested code snippet:
try: action_json = parse_json( text=self.llm.predict( tool_use_prompt, callbacks=self.conversation_manager.agent_callbacks ), llm=self.llm, ) except Exception as e: logging.error(f'LLM prediction or JSON parsing failed: {e}') # Handle the exception appropriately -
The method
get_tool_calls_from_failed_stepsconstructs a string by concatenating JSON dumps in a loop. This is inefficient and can be improved by using a list to collect the strings and then joining them at the end.Existing code snippet:
for step in intermediate_steps: context += json.dumps(...)Suggested code snippet:
context_parts = [json.dumps(...) for step in intermediate_steps] context = '\n'.join(context_parts) -
The
tryblock is used to append a string representation ofstep[1]tocontext. However, theexceptblock catches all exceptions and does not log or re-raise them, which could hide bugs. It is recommended to catch specific exceptions and handle them accordingly.Existing code snippet:
try: if step[1] is not None: context += '\nReturned: ' + str(step[1]) else: context += '\nReturned: None' except Exception as e: context += '\nReturned: An unknown exception.'Suggested code snippet:
if step[1] is not None: context += '\nReturned: ' + str(step[1]) else: context += '\nReturned: None' -
The exception handling in the loop is too generic and could mask different types of exceptions. It's better to catch specific exceptions and handle them accordingly. Additionally, the exception message is not informative. Consider logging the actual exception message.
Existing code snippet:
try: if step[1] is not None: context += "\nReturned: " + str(step[1]) else: context += "\nReturned: None" except Exception as e: context += "\nReturned: An unknown exception."Suggested code snippet:
try: if step[1] is not None: context += "\nReturned: " + str(step[1]) else: context += "\nReturned: None" except Exception as e: logging.error(f'Error while processing step: {e}') context += f"\nReturned: Exception occurred: {e}" -
The method
get_helpful_contextuses string concatenation in a loop to build the context. This is inefficient as it creates a new string object on each iteration. Use''.join()for better performance.Existing code snippet:
return '\n----\n'.join(...)Suggested code snippet:
return '\n----\n'.join(f"using the `{s[0].tool}` tool returned:\n'{s[1]}'" for s in intermediate_steps if s[1] is not None)