Refactor agent architecture: replace ReActAgent with LangGraphAgent, streamline tool integration, and enhance Excel parsing and image description functionalities.

agents.py

@@ -1,75 +1,98 @@
 import os
-from llama_index.core.agent.workflow import ReActAgent, FunctionAgent
-from llama_index.core import PromptTemplate
-from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI
+from typing import Any, List, TypedDict
+from langgraph.graph import StateGraph
+from langgraph.prebuilt import ToolNode, tools_condition
+from langchain_core.messages import HumanMessage, SystemMessage
+from langchain_core.rate_limiters import InMemoryRateLimiter
+from langchain_openai import ChatOpenAI
 from tools import (
-    search_tool,
     describe_image_tool,
     parse_excel_tool,
-    access_webpage_tool,
-    string_functions_tool
+    webpage_extraction_tool,
+    brave_web_search,
+    python_code_interpreter_tool,
+    audio_file_transcriber
 )
 
-thinking_agent = ReActAgent(
-    name="Thinking Agent",
-    description="An agent that can think and reason about tasks, and then handoff the task to Technician Agent for execution, or to Manager Agent for review.",
-    llm=HuggingFaceInferenceAPI(
-        model_name="Qwen/Qwen2.5-Coder-32B-Instruct",
-        provider="auto",
-        token=os.environ.get("HF_TOKEN")
-    ),
-    system_prompt="You are a thinking agent that can reason about tasks and communicate the necessary steps to complete them to Technician Agent, if necessary. If you believe the task is completed and the question is answered, you must handoff the answer to Manager Agent for final review.",
-    can_handoff_to=["Technician Agent", "Manager Agent"]
-)
+class AgentState(TypedDict):
+    messages: List[Any]
+    question: str
+    file_path: str
+    final_answer: str
 
-technician_agent = ReActAgent(
-    name="Technician Agent",
-    description="An agent that can perform various technical tasks such as searching the web, describing images, parsing Excel files, string operations, and accessing webpages.",
-    tools=[search_tool, describe_image_tool, parse_excel_tool, access_webpage_tool, string_functions_tool],
-    llm=HuggingFaceInferenceAPI(
-        model_name="Qwen/Qwen2.5-Coder-32B-Instruct",
-        provider="auto",
-        token=os.environ.get("HF_TOKEN")
-    ),
-    system_prompt="You are a helpful agent that answers questions based on the provided tools. Use the tools to gather information and provide accurate answers, and send those answers to Thinking Agent. If the task is too complex or requires further reasoning, handoff the task to Thinking Agent for analysis with the reasoning as why you cannot complete it. You must always handoff to Thinking Agent",
-    can_handoff_to=["Thinking Agent"]
-)
+tools = [
+    describe_image_tool,
+    parse_excel_tool,
+    webpage_extraction_tool,
+    brave_web_search,
+    python_code_interpreter_tool,
+    audio_file_transcriber
+]
 
-manager_agent = ReActAgent(
-    name="Manager Agent",
-    description="A high-level agent that can manage tasks and coordinate between other agents.",
-    llm=HuggingFaceInferenceAPI(
-        model_name="Qwen/Qwen2.5-Coder-32B-Instruct",
-        provider="auto",
-        token=os.environ.get("HF_TOKEN")
-    ),
-    can_handoff_to=["Thinking Agent"],
-    # system_prompt="You are a manager agent that oversees tasks and coordinates between other agents. Do not include thoughts in your response. Just the answer. " \
-    # "You will be given a question and you need to respond with the correct answer provided by the other agents. " \
-    # "Your response should be a number OR as few words as possible OR a comma separated list of numbers and/or strings. " \
-    # "For example, if the question is 'What color are the stars on the American flag?' Your response would be 'White'. " \
-    # "If you are asked for a number, don't use comma to write your number neither use units such as $ or percent sign unless specified otherwise. " \
-    # "If you are asked for a string, don't use articles, neither abbreviations (e.g. for cities), and write the digits in plain text unless specified otherwise. " \
-    # "If you are asked for a comma separated list, apply the above rules depending of whether the element to be put in the list is a number or a string. " \
-    # "Do not engage in conversation, just respond with the answer unless the question explicitly asks for a certain style of response. " \
-    # "If the results are questionable, you can send the task back to Thinking Agent for further analysis. If the answer can not be concluded, respond with 'I don't know'" 
+rate_limiter = InMemoryRateLimiter(
+    requests_per_second=0.1,  # <-- Can only make a request once every 10 seconds!!
+    check_every_n_seconds=0.1,  # Wake up every 100 ms to check whether allowed to make a request,
+    max_bucket_size=10,  # Controls the maximum burst size.
 )
 
-minimal_system_prompt = """
-You are the Manager Agent. 💼
-Respond with **only the final answer**, in as few words as possible.
-Do **not** include any reasoning, thoughts, or tool calls.
-If it's a number: use plain digits (no commas, %, etc.).
-If it's a string: no articles, no abbreviations.
-If it's a list: comma-separated, each element following the rules above.
-If unsure, reply: I don't know.
-Below is the current conversation consisting of interleaving human and assistant messages.
-"""
+class LangGraphAgent:
+    def __init__(self, model_name: str = "gpt-4o",):
+        self.llm = ChatOpenAI(model=model_name, max_tokens=3000, temperature=0, streaming=True, rate_limiter=rate_limiter)
+        self.llm_with_tools = self.llm.bind_tools(tools)
+        self.tool_node = ToolNode(tools)
+        self.graph = self.create_graph().compile()
+
+    def create_graph(self) -> StateGraph:
+        """Creates a state graph for the agent's workflow."""
+        # Define the tools and their respective states
+        graph = StateGraph(AgentState)
+        graph.add_node("agent", self.agent_node)
+        graph.add_node("tools", self.tool_node)
+
+        graph.set_entry_point("agent")
+
+        graph.add_conditional_edges("agent", tools_condition)
+        graph.add_edge("tools", "agent") 
+         
+        return graph
+
+    def agent_node(self, state: AgentState):
+        """Creates a node for the agent that uses the model to respond to user queries."""
+        messages = state['messages']
+
+        if len(messages) == 1:
+            system_prompt = ("You are a helpful assistant that can answer questions using various tools. "
+            "You must answer the given question using as few words as possible, or the given format, if any."
+            " If the answer is a number, you must return the number only, do not include symbols or commas. If you don't know the answer, say 'I don't know'. "
+            "These questions are nontrivial and may require advanced critical thinking and multiple tool calls."
+            "If you need to search the web for information, always use a search tool before using a wepbage extraction tool so you always have a legit website."
+            "If given a Python file, execute it with the code interpreter tool (riza_exec_python)")
+            messages = [SystemMessage(system_prompt)] + messages
+        
+        while True:
+            response = self.llm_with_tools.invoke(messages)
+            messages.append(response)
+
+            if not response.tool_calls:
+                break
+
+            tool_outputs = []
+            for call in response.tool_calls:
+                tool_output = self.tool_node.invoke({"messages": [response], "tool_call": call})
+                tool_outputs.extend(tool_output["messages"])
 
-manager_prompt = PromptTemplate(template=minimal_system_prompt)
+            messages.extend(tool_outputs)
 
-manager_agent.update_prompts({"react_header", manager_prompt.get_template()})
+        state["messages"] = messages
+        return {"messages": messages}
 
-prompt_dict = manager_agent.get_prompts()
-for k, v in prompt_dict.items():
-    print(f"Prompt: {k}\n\nValue: {v.template}")
+    def run(self, question: str) -> str:
+        state = AgentState(messages=[HumanMessage(content=question)], question=question, final_answer=None)
+        result = self.graph.invoke(state)
+        
+        final_message = result["messages"][-1]
+        if hasattr(final_message, 'content'):
+            result['final_answer'] = final_message.content
+        else:
+            result['final_answer'] = str(final_message)
+        return result['final_answer']

app.py

@@ -6,62 +6,14 @@ import inspect
 import dotenv
 dotenv.load_dotenv()  # Load environment variables from .env file
 import pandas as pd
-from llama_index.llms.huggingface_api import HuggingFaceInferenceAPI
-from llama_index.core.agent.workflow import AgentWorkflow, AgentOutput, ToolCall, ToolCallResult, AgentInput, AgentStream
-from agents import thinking_agent, technician_agent, manager_agent
-import asyncio
+from agents import LangGraphAgent
+
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
 
-llm = HuggingFaceInferenceAPI(model_name="deepseek-ai/DeepSeek-R1", provider="auto", token=os.environ.get("HF_TOKEN"))
-
-
-#def run_and_submit_all( profile: gr.OAuthProfile | None):
-
-async def run_agent_query(agent: AgentWorkflow, question: str):
-    """
-    Runs the agent on a single question and returns the answer.
-    This function is intended to be used with Gradio for interactive querying.
-    """
-    
-    handler = agent.run(user_msg=question)
-    current_agent = None
-    final_response = None
-    current_tool_calls = ""
-    async for event in handler.stream_events():
-        if (
-            hasattr(event, "current_agent_name")
-            and event.current_agent_name != current_agent
-        ):
-            current_agent = event.current_agent_name
-            print(f"\n{'='*50}")
-            print(f"🤖 Agent: {current_agent}")
-            print(f"{'='*50}\n")
 
-        if isinstance(event, AgentStream):
-            if event.delta:
-                print(event.delta, end="", flush=True)
-        elif isinstance(event, AgentInput):
-            print("📥 Input:", event.input)
-        elif isinstance(event, AgentOutput):
-            if event.response.content:
-                print("📤 Output:", event.response.content)
-                final_response = event.response.content
-            if event.tool_calls:
-                print(
-                    "🛠️  Planning to use tools:",
-                    [call.tool_name for call in event.tool_calls],
-                )
-        elif isinstance(event, ToolCallResult):
-            print(f"🔧 Tool Result ({event.tool_name}):")
-            print(f"  Arguments: {event.tool_kwargs}")
-            print(f"  Output: {event.tool_output}")
-        elif isinstance(event, ToolCall):
-            print(f"🔨 Calling Tool: {event.tool_name}")
-            print(f"  With arguments: {event.tool_kwargs}")
-    return final_response if final_response else "No response from agent."
-def run_and_submit_all():
+def run_and_submit_all( profile: gr.OAuthProfile | None):
     """
     Fetches all questions, runs the BasicAgent on them, submits all answers,
     and displays the results.
@@ -69,26 +21,19 @@ def run_and_submit_all():
     # --- Determine HF Space Runtime URL and Repo URL ---
     space_id = os.getenv("SPACE_ID") # Get the SPACE_ID for sending link to the code
 
-    # if profile:
-    #     username= f"{profile.username}"
-    #     print(f"User logged in: {username}")
-    # else:
-    #     print("User not logged in.")
-    #     return "Please Login to Hugging Face with the button.", None
+    if profile:
+        username= f"{profile.username}"
+        print(f"User logged in: {username}")
+    else:
+        print("User not logged in.")
+        return "Please Login to Hugging Face with the button.", None
+
 
     api_url = DEFAULT_API_URL
     questions_url = f"{api_url}/questions"
     submit_url = f"{api_url}/submit"
 
-    try:
-        agent = AgentWorkflow(
-                agents=[thinking_agent, technician_agent, manager_agent],
-                root_agent=manager_agent.name,
-                handoff_output_prompt="handoff_result: Passed to {to_agent}. Reason: {reason}. Please continue processing using the original user question."
-            )
-    except Exception as e:
-        print(f"Error instantiating agent: {e}")
-        return f"Error initializing agent: {e}", None
+    agent = LangGraphAgent()
     # In the case of an app running as a hugging Face space, this link points toward your codebase ( usefull for others so please keep it public)
     agent_code = f"https://huggingface.co/spaces/{space_id}/tree/main"
     print(agent_code)
@@ -118,7 +63,7 @@ def run_and_submit_all():
     results_log = []
     answers_payload = []
     print(f"Running agent on {len(questions_data)} questions...")
-    for item in questions_data[:4]:
+    for item in questions_data:
         task_id = item.get("task_id")
         question_text = item.get("question")
         if not task_id or question_text is None:
@@ -145,7 +90,7 @@ def run_and_submit_all():
                 return f"An unexpected error occurred fetching file {file_name}: {e}", None
         try:
             print(f"Running agent on task {task_id} with question: {question_text}")
-            submitted_answer = asyncio.run(run_agent_query(agent, question_text))
+            submitted_answer = agent.run(question_text)
             answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": submitted_answer})
         except Exception as e:
@@ -164,20 +109,19 @@ def run_and_submit_all():
     # 5. Submit
     print(f"Submitting {len(answers_payload)} answers to: {submit_url}")
     try:
-        # response = requests.post(submit_url, json=submission_data, timeout=60)
-        # response.raise_for_status()
-        # result_data = response.json()
-        # final_status = (
-        #     f"Submission Successful!\n"
-        #     f"User: {result_data.get('username')}\n"
-        #     f"Overall Score: {result_data.get('score', 'N/A')}% "
-        #     f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
-        #     f"Message: {result_data.get('message', 'No message received.')}"
-        # )
-        # print("Submission successful.")
+        response = requests.post(submit_url, json=submission_data, timeout=60)
+        response.raise_for_status()
+        result_data = response.json()
+        final_status = (
+            f"Submission Successful!\n"
+            f"User: {result_data.get('username')}\n"
+            f"Overall Score: {result_data.get('score', 'N/A')}% "
+            f"({result_data.get('correct_count', '?')}/{result_data.get('total_attempted', '?')} correct)\n"
+            f"Message: {result_data.get('message', 'No message received.')}"
+        )
+        print("Submission successful.")
         results_df = pd.DataFrame(results_log)
-        return "Nothing submitted", results_df
-        # return final_status, results_df
+        return final_status, results_df
     except requests.exceptions.HTTPError as e:
         error_detail = f"Server responded with status {e.response.status_code}."
         try:

tools.py

@@ -1,119 +1,154 @@
-from llama_index.core.tools import FunctionTool
-from llama_index.tools.tavily_research import TavilyToolSpec
-from llama_index.core.schema import ImageDocument
-from llama_index.readers.whisper import WhisperReader
+from langchain_core.tools import tool
+import dotenv
+dotenv.load_dotenv()
+from langchain_tavily import TavilySearch, TavilyExtract
+from langchain_openai import ChatOpenAI
+from langchain_core.messages import HumanMessage
+from langchain_community.tools.riza.command import ExecPython
+from langchain_community.tools import BraveSearch
+from langchain_community.document_loaders import AssemblyAIAudioTranscriptLoader
+import requests
+import base64
+import pandas as pd
 
-import os
+# def search_tool(query: str) -> str:
+#     """
+#     Search the web for information using Tavily API (or similar service).
+#     """
+#     api_key = os.environ.get("TAVILY_API_KEY")
+#     if not api_key:
+#         return "TAVILY_API_KEY not set."
+#     url = f"https://api.tavily.com/search?q={query}&key={api_key}"
+#     try:
+#         response = requests.get(url)
+#         response.raise_for_status()
+#         return response.text
+#     except Exception as e:
+#         return f"Error searching web: {e}"
 
-tool_spec = TavilyToolSpec(
-    api_key=os.environ.get("TAVILY_API_KEY"),
-)
+@tool
+def describe_image_tool(image_path: str, prompt: str = "Describe the following image:") -> str:
+    """
+    Describe an image using a vision language model.
+    Args:
+        image_path (str): Path to the image file.
+        prompt (str): Prompt to guide the description.
+    Returns:
+        str: Description of the image.
+    """
+    print(f"Describing image: {image_path} with prompt: {prompt}")
+    vision_llm = ChatOpenAI(model="gpt-4o")
+    with open(image_path, "rb") as image_file:
+        image_content = image_file.read()
 
-search_tool = FunctionTool.from_defaults(tool_spec.search)
+    image_base64 = base64.b64encode(image_content).decode('utf-8')
 
-from llama_index.multi_modal_llms.huggingface import HuggingFaceMultiModal
+    message = [
+        HumanMessage(
+            content=[
+                {"type": "text", "text": prompt},
+                {"type": "image_url", "image_url": {"url": f"data:image/jpeg;base64,{image_base64}"}}
+            ]
+        )
+    ]
+    response = vision_llm.invoke(message)
+    return response.content if response else f"[Image description for {image_path} with prompt '{prompt}' not available."
 
-def describe_image(image_path: str, prompt: str = "Describe the following image:") -> str:
+@tool
+def parse_excel_tool(file_path: str) -> str:
     """
-    Function to describe an image using a multi-modal LLM.
-    :param image_path: Path to the image file or base64 encoded image data.
-    :param prompt: Prompt to use for the description. Defaults to "Describe the following image:".
-    :return: Description of the image.
+    Parse an Excel file and return its content as a string.
+    Args:
+        file_path (str): Path to the Excel file.
+    Returns:
+        str: Content of the Excel file as a string.
     """
-    image = ImageDocument(image_path=image_path)
     try:
-        llm = HuggingFaceMultiModal.from_model_name("Qwen/Qwen2-VL-2B-Instruct", use_fast=True)
-        return llm.complete(
-            prompt=f"{prompt}",
-            image_documents=[image]
-        ).text
+        df = pd.read_excel(io=file_path)
+        return df.to_string() if not df.empty else "The Excel file is empty."
     except Exception as e:
-        return f"Error describing image: {e}"   
-
-describe_image_tool = FunctionTool.from_defaults(describe_image)
-
-# Tool to parse xls/xlsx files
-def parse_excel(file_path: str) -> str:
-    """
-    Function to parse an Excel file and return its content as a string.
-    :param file_path: Path to the Excel file (xls or xlsx).
-    :return: Content of the Excel file as a string.
-    """
-    import pandas as pd
-    
-    df = pd.read_excel(io=file_path)
-    
-    # Convert DataFrame to string
-    return df.to_string() if not df.empty else "The Excel file is empty."
-
-parse_excel_tool = FunctionTool.from_defaults(parse_excel)
+        return f"Error parsing Excel file: {e}"
 
-def access_webpage(url: str) -> str:
+@tool
+def access_webpage_tool(url: str) -> str:
     """
-    Function to access a webpage and return its content.
-    :param url: URL of the webpage to access.
-    :return: Content of the webpage.
+    Access a webpage and return its content. This should not be used for YouTube
+    Args:
+        url (str): URL of the webpage to access.
+    Returns:
+        str: Content of the webpage or an error message.
     """
-    import requests
+    print(f"Accessing webpage: {url}")
     try:
-        print(f"Accessing webpage: {url}")
         response = requests.get(url)
-        print(f"Response status code: {response.status_code}")
-        response.raise_for_status()  # Raise an error for bad responses
+        response.raise_for_status()
         return response.text
-    except requests.RequestException as e:
-        return f"Error accessing {url}: {e}"
     except Exception as e:
-        return f"An unexpected error occurred: {e}"
-
-access_webpage_tool = FunctionTool.from_defaults(access_webpage)
+        return f"Error accessing {url}: {e}"
+# @tool
+# def string_functions_tool(input_string: str, operation: str) -> str:
+#     """
+#     Perform various string operations based on the specified operation.
+#     Args:
+#         input_string (str): The string to operate on.
+#         operation (str): The operation to perform. Supported operations: 
+#                          'uppercase', 'lowercase', 'reverse', 'length', 
+#                          'count_vowels', 'count_consonants', 'count_words', 
+#                          'count_sentences'.
+#     Returns:
+#         str: Result of the operation or an error message if the operation is invalid.
+#     """
+#     if operation == "uppercase":
+#         return input_string.upper()
+#     elif operation == "lowercase":
+#         return input_string.lower()
+#     elif operation == "reverse":
+#         return input_string[::-1]
+#     elif operation == "length":
+#         return str(len(input_string))
+#     elif operation == "count_vowels":
+#         vowels = "aeiouAEIOU"
+#         return str(sum(1 for char in input_string if char in vowels))
+#     elif operation == "count_consonants":
+#         vowels = "aeiouAEIOU"
+#         return str(sum(1 for char in input_string if char.isalpha() and char not in vowels))
+#     elif operation == "count_words":
+#         return str(len(input_string.split()))
+#     elif operation == "count_sentences":
+#         import re
+#         sentences = re.split(r'[.!?]+', input_string)
+#         return str(len([s for s in sentences if s.strip()]))
+#     else:
+#         return "Invalid operation. Supported operations: uppercase, lowercase, reverse."
 
-def string_functions(input_string: str, operation: str) -> str:
+@tool
+def audio_file_transcriber(file_path: str) -> str :
     """
-    Function to perform string operations.
-    :param input_string: The input string to operate on.
-    :param operation: The operation to perform (e.g., "uppercase", "lowercase", "reverse", "length", "count_vowels", "count_consonants", "count_words", "count_sentences").
-    :return: Result of the string operation.
+    Transcribe an audio file
+    Args:
+        file_path (str): the file path of the audio file
+    Returns:
+        str: A transcription of the audio
     """
-    if operation == "uppercase":
-        return input_string.upper()
-    elif operation == "lowercase":
-        return input_string.lower()
-    elif operation == "reverse":
-        return input_string[::-1]
-    elif operation == "length":
-        return str(len(input_string))
-    elif operation == "count_vowels":
-        vowels = "aeiouAEIOU"
-        return str(sum(1 for char in input_string if char in vowels))
-    elif operation == "count_consonants":
-        vowels = "aeiouAEIOU"
-        return str(sum(1 for char in input_string if char.isalpha() and char not in vowels))
-    elif operation == "count_words":
-        return str(len(input_string.split()))
-    elif operation == "count_sentences":
-        import re
-        sentences = re.split(r'[.!?]+', input_string)
-        return str(len([s for s in sentences if s.strip()]))
-    else:
-        return "Invalid operation. Supported operations: uppercase, lowercase, reverse."
+    loader = AssemblyAIAudioTranscriptLoader(file_path=file_path)
+    docs = loader.load()
+    return docs[0].page_content
 
-string_functions_tool = FunctionTool.from_defaults(string_functions)
 
-def transcribe_audio(audio_path: str) -> str:
-    """
-    Function to transcribe audio using a multi-modal LLM.
-    :param audio_path: Path to the audio file.
-    :return: Transcription of the audio.
-    """
-    try:
-        reader = WhisperReader(api_key=os.environ.get("OPENAI_API_KEY"))
-        documents = reader.load_data(file=audio_path)
-        if not documents:
-            return "No audio content found."
-        # Assuming the first document contains the transcription
-        return documents[0].text if documents else "No transcription available."
-    except Exception as e:
-        return f"Error transcribing audio: {e}"
-    
-transcribe_audio_tool = FunctionTool.from_defaults(transcribe_audio)
+brave_web_search = BraveSearch.from_search_kwargs({"max_results": 5})
+
+
+# search_tool = TavilySearch(
+#     max_results=5,
+#     include_answer="basic",
+#     include_images=True,
+#     include_image_descriptions=True,
+#     search_depth="advanced"
+# )
+
+webpage_extraction_tool = TavilyExtract(
+    extract_depth="advanced",
+    include_images=True
+)
+
+python_code_interpreter_tool = ExecPython()