built management agents with llama_index, recursive self call for task breakdown, fixed LLMFactory to create LLMs from different interfaces

.gitignore

@@ -63,6 +63,9 @@ venv.bak/
 # Local configuration file
 .env
 
+# Rejected by git
+*.png
+
 # Other files
 _to_ignore/
 .github/

alfred.py

@@ -0,0 +1,183 @@
+from langgraph.graph import END, StateGraph
+from langgraph.graph.state import CompiledStateGraph
+
+from typing import Dict, Any, TypedDict, Literal, Optional
+import asyncio
+
+from management import Manager, Assistant
+
+
+# Maximum number of interactions between Assistant and Manager
+MAX_INTERACTIONS = 5
+# Maximum depth of recursion for Manager
+MAX_DEPTH = 3
+# For both Assistant and Manager:
+TEMPERATURE = 0.7
+MAX_TOKENS = 100
+
+
+class State(TypedDict):
+    """State for the agent graph."""
+    initial_query: str
+    current_message: str
+    nr_interactions: int
+    final_answer: Optional[str]
+
+
+class GraphBuilder:
+    def __init__(self):
+        """
+        Initializes the GraphBuilder.
+        """
+        self.assistant_agent = Assistant(TEMPERATURE, MAX_TOKENS)
+        self.manager_agent = Manager(TEMPERATURE, MAX_TOKENS, MAX_DEPTH)
+        self.final_answer_hint = "Final answer:"
+
+    def clear_chat_history(self):
+        self.assistant_agent.clear_context()
+        self.manager_agent.clear_context()
+
+    async def assistant_node(self, state: State) -> State:
+        """
+        Assistant agent that evaluates the query and decides whether to give a final answer
+        or continue the conversation with the Manager.
+        
+        Uses the existing Assistant implementation.
+        """
+        response = await self.assistant_agent.query(state["current_message"])
+        
+        # Check if this is a final answer
+        if self.final_answer_hint in response:
+            # Extract the text after final answer hint
+            state["final_answer"] = response.split(self.final_answer_hint, 1)[1].strip()
+        
+        state["current_message"] = response
+        state["nr_interactions"] += 1
+
+        return state
+
+    async def manager_node(self, state: State) -> State:
+        """
+        Manager agent that handles the queries from the Assistant and provides responses.
+        
+        Uses the existing Manager implementation.
+        """
+        response = await self.manager_agent.query(state["current_message"])
+        
+        state["current_message"] = response
+
+        return state
+
+    async def final_answer_node(self, state: State) -> State:
+        """
+        Final answer node that formats and returns the final response.
+        
+        If there's already a final answer in the state, it uses that.
+        Otherwise, it asks the assistant to formulate a final answer.
+        """
+        # If we already have a final answer, use it
+        if state.get("final_answer") is not None:
+            return state
+        
+        # Otherwise, have the assistant formulate a final answer
+        prompt = f"Based on the conversation so far, provide a final answer to the original query:\n\n{state['initial_query']}"
+        state["current_message"] = prompt
+        response = await self.assistant_agent.query(state["current_message"])
+        
+        # Format the response
+        if self.final_answer_hint not in response:
+            response = f"{self.final_answer_hint}{response}"
+        
+        # Extract the text after final answer hint
+        state["final_answer"] = response.split(self.final_answer_hint, 1)[1].strip()
+        
+        return state
+
+    def should_continue(self, state: State) -> Literal["manager", "final_answer"]:
+        """
+        Decides whether to continue to the Manager or to provide a final answer.
+        
+        Returns:
+            "manager": If the Assistant has decided to continue the conversation
+            "final_answer": If the Assistant has decided to provide a final answer
+        """
+        message = state["current_message"]
+        
+        if state["nr_interactions"] >= MAX_INTERACTIONS or self.final_answer_hint in message:
+            return "final_answer"
+        else:
+            return "manager"
+
+    def build_agent_graph(self) -> CompiledStateGraph:
+        """Build and return the agent graph."""
+        graph = StateGraph(State)
+
+        # Convert async functions to sync functions using asyncio.run
+        def sync_assistant_node(state: State) -> State:
+            return asyncio.run(self.assistant_node(state))
+
+        def sync_manager_node(state: State) -> State:
+            return asyncio.run(self.manager_node(state))
+
+        # Add the nodes with sync wrappers
+        graph.add_node("assistant", sync_assistant_node)
+        graph.add_node("manager", sync_manager_node)
+        graph.add_node("final_answer", self.final_answer_node)
+
+        # Add the edges
+        graph.add_edge("START", "assistant")
+
+        graph.add_conditional_edges(
+            "assistant",
+            self.should_continue,
+            {
+                "manager": "manager",
+                "final_answer": "final_answer"
+            }
+        )
+
+        graph.add_edge("manager", "assistant")
+
+        graph.add_edge("final_answer", END)
+
+        return graph.compile()
+
+
+class Alfred:
+
+    def __init__(self):
+        print("Agent initialized.")
+        self.graph_builder = GraphBuilder()
+        self.agent_graph = self.graph_builder.build_agent_graph()
+
+    async def __call__(self, question: str) -> str:
+        print(f"Agent received question (first 50 chars): {question[:50]}...")
+        result = await self.process_query(question)
+        response = result["final_answer"]
+        print(f"Agent processed the response: {response}")
+
+        return response
+
+    async def process_query(self, query: str) -> Dict[str, Any]:
+        """
+        Process a query through the agent graph.
+        
+        Args:
+            query: The initial query to process
+        
+        Returns:
+            The final state of the graph execution
+        """
+        initial_state: State = {
+            "initial_query": query,
+            "current_message": query,
+            "nr_interactions": 0,
+            "final_answer": None
+        }
+        self.graph_builder.clear_chat_history()
+
+        # Since agent_graph.invoke is synchronous, we don't need to await it
+        # But we might need to run it in an executor if it's computationally intensive
+        loop = asyncio.get_running_loop()
+        result = await loop.run_in_executor(None, self.agent_graph.invoke, initial_state)
+        return result

app.py

@@ -2,7 +2,8 @@ import os
 import gradio as gr
 import requests
 import pandas as pd
-from manager import Manager
+
+from alfred import Alfred
 
 
 # --- Constants ---
@@ -12,7 +13,6 @@ QUESTIONS_LIMIT = 3
 
 
 class Application:
-
     def __init__(self):
         self.space_id = os.getenv("SPACE_ID")
         self.username = None
@@ -60,13 +60,13 @@ class Application:
                 print("Fetched questions list is empty.")
                 return "Fetched questions list is empty or invalid format.", None
             print(f"Fetched {len(questions_data)} questions.")
-        except requests.exceptions.RequestException as e:
-            print(f"Error fetching questions: {e}")
-            return f"Error fetching questions: {e}", None
         except requests.exceptions.JSONDecodeError as e:
             print(f"Error decoding JSON response from questions endpoint: {e}")
             print(f"Response text: {response.text[:500]}")
             return f"Error decoding server response for questions: {e}", None
+        except requests.exceptions.RequestException as e:
+            print(f"Error fetching questions: {e}")
+            return f"Error fetching questions: {e}", None
         except Exception as e:
             print(f"An unexpected error occurred fetching questions: {e}")
             return f"An unexpected error occurred fetching questions: {e}", None
@@ -190,7 +190,7 @@ class Application:
 
         # 1. Instantiate the Main Agent
         try:
-            agent = Manager()
+            agent = Alfred()
         except Exception as e:
             print(f"Error instantiating agent: {e}")
             return f"Error initializing agent: {e}", None
@@ -219,7 +219,6 @@ class Application:
 
 
 class UI:
-
     app = Application()
 
     @classmethod

args.py

@@ -2,7 +2,7 @@
 from enum import Enum
 
 
-class InterfaceChoice(Enum):
+class LLMInterface(Enum):
     HUGGINGFACE = "HuggingFace"
     OPENAILIKE = "OpenAILike"
     OPENAI = "OpenAI"
@@ -10,7 +10,9 @@ class InterfaceChoice(Enum):
 
 
 class Args:
-    INTERFACE = InterfaceChoice.OPENAILIKE
+    primary_llm_interface=LLMInterface.OPENAILIKE
+    # secondary_llm_interface=LLMInterface.HUGGINGFACE
+    vlm_interface=LLMInterface.HUGGINGFACE
     model_name="Qwen/Qwen2.5-Coder-32B-Instruct"
     api_base="http://127.0.0.1:1234/v1"  # LM Studio local endpoint
     api_key="api_key"

diagrams/diagram.puml

@@ -11,7 +11,7 @@ hide empty members
 ' Agent Definitions (Use class notation for agents with tools as fields)
 
 
-class "Ambassador" as Ambassador <<M>> {
+class "Assistant" as Assistant <<M>> {
 }
 
 class "Manager" as Manager <<M>> {
@@ -63,10 +63,10 @@ class "VideoHandler" as VideoHandler {
 }
 
 ' Agent-to-Agent Connections
-Query --> Ambassador
-Ambassador --> Manager : request
-Manager --> Ambassador : solution
-Ambassador --> Final_Answer : problem solved
+Query --> Assistant
+Assistant --> Manager : request
+Manager --> Assistant : solution
+Assistant --> Final_Answer : problem solved
 
 Manager ..> Manager : complex task
 Manager ..> Solver : trivial task

llm_factory.py

@@ -1,21 +1,22 @@
 from llama_index.llms.openai_like import OpenAILike
 from llama_index.llms.openai import OpenAI
 from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI
-from args import InterfaceChoice, Args
+
+from args import LLMInterface, Args
 
 
 class LLMFactory():
 
     @classmethod
-    def create(cls, system_prompt, temperature = None, max_tokens = None):
-        if Args.INTERFACE == InterfaceChoice.OPENAILIKE:
+    def create(cls, interface: LLMInterface, system_prompt, temperature = None, max_tokens = None):
+        if interface == LLMInterface.OPENAILIKE:
             return cls._openailike_create(system_prompt, temperature, max_tokens)
-        elif Args.INTERFACE == InterfaceChoice.OPENAI:
+        elif interface == LLMInterface.OPENAI:
             return cls._openai_create(system_prompt, temperature, max_tokens)
-        elif Args.INTERFACE == InterfaceChoice.HUGGINGFACE:
+        elif interface == LLMInterface.HUGGINGFACE:
             return cls._hf_create(system_prompt, temperature, max_tokens)
         else:
-            raise ValueError(f"Invalid interface choice: {Args.INTERFACE}")
+            raise ValueError(f"Interface '{interface}' is not supported !")
 
     @staticmethod
     def _openailike_create(system_prompt, temperature=None, max_tokens=None):

management.py

@@ -0,0 +1,154 @@
+from llama_index.core.agent.workflow import AgentWorkflow
+from llama_index.core.tools import FunctionTool
+from llama_index.core.workflow import Context
+
+from typing import List
+import os
+
+from llm_factory import LLMFactory
+from solver import Solver, Summarizer
+from args import Args
+
+
+class Assistant:
+    def __init__(self, temperature, max_tokens):
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "01_assistant.txt")
+        self.system_prompt = ""
+        with open(system_prompt_path, "r") as file:
+            self.system_prompt = file.read().strip()
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
+        self.agent = AgentWorkflow.setup_agent(llm=llm)
+        self.ctx = Context(self.agent)
+
+    async def query(self, question: str) -> str:
+        """
+        Process a user query and return a response using the agent.
+        
+        Args:
+            question: The user's question or input text
+            
+        Returns:
+            The agent's response as a string
+        """
+        response = await self.agent.run(question, ctx=self.ctx)
+        response = str(response)
+        return response
+
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        self.ctx = Context(self.agent)
+
+
+class Manager:
+    def __init__(self, temperature, max_tokens, max_depth):
+        self.max_depth = max_depth
+        self.current_depth = 0
+
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "02_manager.txt")
+        self.system_prompt = ""
+        with open(system_prompt_path, "r") as file:
+            self.system_prompt = file.read().strip()
+
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
+        self.agent = AgentWorkflow.from_tools_or_functions(
+            [
+            FunctionTool.from_defaults(
+                name="require_break_up",
+                description="Break a complex task into simpler subtasks. Use when a task needs to be divided into manageable parts.",
+                fn=self.require_break_up
+            ),
+            FunctionTool.from_defaults(
+                name="require_solution",
+                description="Request direct solutions for specific tasks. Use when a task is simple enough to be solved directly.",
+                fn=self.require_solution
+            )
+            ],
+            llm=llm
+        )
+        self.ctx = Context(self.agent)
+        self.solver = Solver(temperature, max_tokens)
+        self.summarizer = Summarizer(temperature, max_tokens)
+
+    async def query(self, question: str, remember = True) -> str:
+        """
+        Process a question using the manager agent and return a response.
+        
+        Args:
+            question: The question or task to process
+            remember: Whether to maintain context between queries (default: True)
+            
+        Returns:
+            The agent's response as a string
+        """
+        if remember:
+            response = await self.agent.run(question, ctx=self.ctx)
+        else:
+            response = await self.agent.run(question)
+        response = str(response)
+        return response
+
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        self.ctx = Context(self.agent)
+
+    async def require_break_up(self, tasks: List[str], try_solving = False) -> str:
+        """
+        Break down complex tasks into simpler subtasks recursively up to max_depth.
+        
+        Args:
+            tasks: List of tasks to break down
+            try_solving: Whether to attempt solving tasks at max depth (default: False)
+            
+        Returns:
+            Summarized report of the task breakdown
+        """
+        print(f"-> require_break_up tool used (input: {tasks}) !")
+        if not tasks:
+            return "Error: No tasks provided to break up. Please provide at least one task."
+        
+        self.current_depth += 1
+
+        observation = ""
+        if self.current_depth < self.max_depth:
+            for task in tasks:
+                solution = await self.query(task, remember=False)
+                response = f"For task:\n\n{task}\n\nThe following break up has been provided:\n\n{solution}\n\n"
+                observation += response
+        elif try_solving:
+            for task in tasks:
+                response = await self.solver.query(task)
+        else:
+            observation = "Maximum depth for `break_up` tool has been reached ! At this point, you may try to break up the task yourself or try `require_solution`."
+        
+        self.current_depth -= 1
+        report = await self.summarizer.query(observation.strip())
+        return report
+
+    async def require_solution(self, tasks: List[str]) -> str:
+        """
+        Request direct solutions for the provided tasks using the Solver.
+        
+        Args:
+            tasks: List of tasks to solve
+            
+        Returns:
+            Summarized report of solutions for all tasks
+        """
+        print(f"-> require_solution tool used (input: {tasks}) !")
+        if not tasks:
+            return "Error: No tasks provided to solve. Please provide at least one task."
+        
+        observation = ""
+        for task in tasks:
+            solution = await self.solver.query(task)
+            response = f"For task:\n\n{task}\n\nThe Solver provided the solution:\n\n{solution}\n\n"
+            observation += response
+
+        report = await self.summarizer.query(observation.strip())
+        return report

manager.py

@@ -1,32 +0,0 @@
-import asyncio
-import os
-from llm_factory import LLMFactory
-
-
-# TODO: Langgraph graph that process the following:
-# 1. Thakes the Query Node (Start Node) and sends it to the Assistant Node (has memory - access to the conversation hystory stored in the State)
-# 2. The Assistant Node decides whether the querry is ready to be delivered (the solution is available in the conversation hystory, or the `MAX_DEPTH` has been reached)
-#   - if yes: formulates a concise final answer and sends it to the Final_Answer Node (End Node)
-#   - if no: formulates a querry (could be the first one or a follow-up) and sends it to the Manager Node (also has access to the conversation hystory)
-#       (!) This communication happens back and forth until the querry gets solved (or up to a maximum depth defined by a `MAX_DEPTH` variable)
-
-
-class Manager:
-
-    def __init__(self):
-        print("Agent initialized.")
-
-    async def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        self.final_answer = ""
-        response = await self.query(question)
-        print(f"Agent processed the response: {response}")
-        return response
-    
-    async def query(self, question: str) -> str:
-        # TODO
-        pass
-
-
-if __name__ == "__main__":
-    print("---__main__---")

solver.py

@@ -1,20 +1,70 @@
 from llama_index.core.agent.workflow import AgentWorkflow
 from llama_index.core.tools import FunctionTool
 from llama_index.core.workflow import Context
+
 import asyncio
 import os
+
 from llm_factory import LLMFactory
 from toolbox import Toolbox
+from args import Args
 
 
-class MathExpert:
+class Solver:
+    def __init__(self, temperature, max_tokens):
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "06_math_expert.txt")
+        self.system_prompt = ""
+        with open(system_prompt_path, "r") as file:
+            self.system_prompt = file.read().strip()
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
+        self.agent = AgentWorkflow.from_tools_or_functions(
+            [
+            Toolbox.math.symbolic_calc,
+            Toolbox.math.unit_converter,
+            ],
+            llm=llm
+        )
+        self.ctx = Context(self.agent)
+    
+    def get_system_prompt(self):
+        return self.system_prompt
+
+    async def query(self, question: str) -> str:
+        response = await self.agent.run(question, ctx=self.ctx)
+        response = str(response)
+        return response
 
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        self.ctx = Context(self.agent)
+
+
+class Summarizer:
+    def __init__(self, temperature, max_tokens):
+        system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "01_assistant.txt")
+        self.system_prompt = ""
+        with open(system_prompt_path, "r") as file:
+            self.system_prompt = file.read().strip()
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
+        self.agent = AgentWorkflow.setup_agent(llm=llm)
+        self.ctx = Context(self.agent)
+
+    async def query(self, question: str) -> str:
+        response = await self.agent.run(question, ctx=self.ctx)
+        response = str(response)
+        return response
+
+
+class MathExpert:
     def __init__(self, temperature, max_tokens):
         system_prompt_path = os.path.join(os.getcwd(), "system_prompts", "06_math_expert.txt")
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
-        llm = LLMFactory.create(self.system_prompt, temperature, max_tokens)
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
         self.agent = AgentWorkflow.from_tools_or_functions(
             [
             Toolbox.math.symbolic_calc,
@@ -32,6 +82,13 @@ class MathExpert:
         response = str(response)
         return response
 
+    def clear_context(self):
+        """
+        Clears the current context of the agent, resetting any conversation history.
+        This is useful when starting a new conversation or when the context needs to be refreshed.
+        """
+        self.ctx = Context(self.agent)
+
 
 class Researcher:
     def __init__(self, temperature, max_tokens):
@@ -39,7 +96,7 @@ class Researcher:
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
-        llm = LLMFactory.create(self.system_prompt, temperature, max_tokens)
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
         
         self.agent = AgentWorkflow.from_tools_or_functions(
             Toolbox.web_search.duck_duck_go_tools,
@@ -62,7 +119,7 @@ class EncryptionExpert:
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
-        llm = LLMFactory.create(self.system_prompt, temperature, max_tokens)
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
         
         self.agent = AgentWorkflow.from_tools_or_functions(
             [
@@ -95,7 +152,7 @@ class RecursiveSolverAgent:
     pass
 
 
-class Solver:
+class Solver_2:
 
     def __init__(self, temperature, max_tokens):
         print("Agent initialized.")
@@ -103,7 +160,7 @@ class Solver:
         self.system_prompt = ""
         with open(system_prompt_path, "r") as file:
             self.system_prompt = file.read().strip()
-        llm = LLMFactory.create(self.system_prompt, temperature, max_tokens)
+        llm = LLMFactory.create(Args.primary_llm_interface, self.system_prompt, temperature, max_tokens)
         self.agent = AgentWorkflow.from_tools_or_functions(
             [
             FunctionTool.from_defaults(self.delegate_to_math_expert),

toolbox.py

@@ -1,9 +1,9 @@
-import pint
-import sympy as sp
-
 from llama_index.core.tools import FunctionTool
 from llama_index.tools.duckduckgo import DuckDuckGoSearchToolSpec
 
+import pint
+import sympy as sp
+
 
 class _Math: