initial

app.py

@@ -3,21 +3,217 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
+from typing import Dict, Any, List, TypedDict, Annotated
+from langgraph.graph import Graph, StateGraph
+from langgraph.prebuilt import ToolNode
+from pydantic import BaseModel, Field
+from tools import create_calculator_tool, create_search_tool
 print("trial")
 # (Keep Constants as is)
 # --- Constants ---
 DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+MODEL_API_URL = "https://api-inference.huggingface.co/models/Qwen/Qwen2.5-Coder-32B-Instruct"
+HF_TOKEN = os.getenv("HF_TOKEN")  # Make sure to set this environment variable
+
+class AgentState(TypedDict):
+    """Type definition for the agent's state."""
+    question: str
+    current_step: str
+    tool_output: str
+    final_answer: str
+    history: List[Dict[str, str]]
+    needs_more_info: bool
+    search_query: str
 
 # --- Basic Agent Definition ---
 # ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
 class BasicAgent:
     def __init__(self):
-        print("BasicAgent initialized.")
+        print("Initializing BasicAgent with Qwen2.5-Coder-32B-Instruct API...")
+        if not HF_TOKEN:
+            raise ValueError("HF_TOKEN environment variable not set. Please set your Hugging Face API token.")
+        
+        # Initialize tools
+        self.calculator = create_calculator_tool()
+        self.search_tool = create_search_tool()
+        
+        # Set up headers for API calls
+        self.headers = {
+            "Authorization": f"Bearer {HF_TOKEN}",
+            "Content-Type": "application/json"
+        }
+        
+        # Create the agent workflow
+        self.workflow = self._create_workflow()
+        print("BasicAgent initialization complete.")
+
+    def _call_llm_api(self, prompt: str) -> str:
+        """Call the Qwen model through the Hugging Face API."""
+        try:
+            response = requests.post(
+                MODEL_API_URL,
+                headers=self.headers,
+                json={"inputs": prompt, "parameters": {"max_length": 200}}
+            )
+            response.raise_for_status()
+            return response.json()[0]["generated_text"]
+        except Exception as e:
+            print(f"Error calling LLM API: {e}")
+            return f"Error getting response from LLM: {str(e)}"
+
+    def _analyze_question(self, state: AgentState) -> AgentState:
+        """Analyze the question and determine the next step."""
+        prompt = f"""Analyze this question and determine what needs to be done: {state['question']}
+        Return your analysis in this format:
+        {{
+            "needs_calculation": true/false,
+            "needs_search": true/false,
+            "search_query": "query if needed",
+            "calculation": {{
+                "operation": "add/subtract/multiply/divide",
+                "numbers": [numbers if needed]
+            }}
+        }}
+        """
+        
+        analysis = eval(self._call_llm_api(prompt))
+        state['needs_more_info'] = analysis.get('needs_search', False)
+        state['search_query'] = analysis.get('search_query', '')
+        
+        if analysis.get('needs_calculation', False):
+            state['current_step'] = 'calculator'
+            state['tool_output'] = str(analysis['calculation'])
+        elif analysis.get('needs_search', False):
+            state['current_step'] = 'search'
+        else:
+            state['current_step'] = 'final_answer'
+        
+        return state
+
+    def _use_calculator(self, state: AgentState) -> AgentState:
+        """Use the calculator tool."""
+        try:
+            result = self.calculator.invoke({"input": eval(state['tool_output'])})
+            state['history'].append({
+                'step': 'calculator',
+                'input': state['tool_output'],
+                'output': str(result['output'].result)
+            })
+            state['current_step'] = 'final_answer'
+        except Exception as e:
+            state['history'].append({
+                'step': 'calculator_error',
+                'error': str(e)
+            })
+            state['current_step'] = 'final_answer'
+        return state
+
+    def _use_search(self, state: AgentState) -> AgentState:
+        """Use the search tool."""
+        try:
+            result = self.search_tool.invoke({
+                "input": {
+                    "query": state['search_query'],
+                    "max_results": 3
+                }
+            })
+            state['history'].append({
+                'step': 'search',
+                'query': state['search_query'],
+                'results': [str(r) for r in result['output'].results]
+            })
+            state['needs_more_info'] = False
+            state['current_step'] = 'final_answer'
+        except Exception as e:
+            state['history'].append({
+                'step': 'search_error',
+                'error': str(e)
+            })
+            state['current_step'] = 'final_answer'
+        return state
+
+    def _generate_final_answer(self, state: AgentState) -> AgentState:
+        """Generate the final answer based on all gathered information."""
+        history_str = "\n".join([f"{h['step']}: {h.get('output', h.get('results', h.get('error', '')))}" 
+                               for h in state['history']])
+        
+        prompt = f"""Based on the following information and history, provide a final answer to the question: {state['question']}
+        
+        History of steps taken:
+        {history_str}
+        
+        Provide a clear, concise answer that addresses the original question.
+        """
+        
+        state['final_answer'] = self._call_llm_api(prompt)
+        return state
+
+    def _create_workflow(self) -> Graph:
+        """Create the agent workflow using LangGraph."""
+        workflow = StateGraph(AgentState)
+        
+        # Add nodes
+        workflow.add_node("analyze", self._analyze_question)
+        workflow.add_node("calculator", self._use_calculator)
+        workflow.add_node("search", self._use_search)
+        workflow.add_node("final_answer", self._generate_final_answer)
+        
+        # Define edges
+        workflow.add_edge("analyze", "calculator")
+        workflow.add_edge("analyze", "search")
+        workflow.add_edge("analyze", "final_answer")
+        workflow.add_edge("calculator", "final_answer")
+        workflow.add_edge("search", "final_answer")
+        
+        # Define conditional edges
+        def router(state: AgentState) -> str:
+            if state['current_step'] == 'calculator':
+                return 'calculator'
+            elif state['current_step'] == 'search':
+                return 'search'
+            elif state['current_step'] == 'final_answer':
+                return 'final_answer'
+            return 'analyze'
+        
+        workflow.add_conditional_edges(
+            "analyze",
+            router,
+            {
+                "calculator": "calculator",
+                "search": "search",
+                "final_answer": "final_answer"
+            }
+        )
+        
+        # Set entry and exit points
+        workflow.set_entry_point("analyze")
+        workflow.set_finish_point("final_answer")
+        
+        return workflow.compile()
+
     def __call__(self, question: str) -> str:
-        print(f"Agent received question (first 50 chars): {question[:50]}...")
-        fixed_answer = "This is a default answer."
-        print(f"Agent returning fixed answer: {fixed_answer}")
-        return fixed_answer
+        """Process a question through the agent workflow."""
+        print(f"Agent received question: {question[:50]}...")
+        
+        try:
+            # Initialize the state
+            initial_state = {
+                "question": question,
+                "current_step": "analyze",
+                "tool_output": "",
+                "final_answer": "",
+                "history": [],
+                "needs_more_info": False,
+                "search_query": ""
+            }
+            
+            # Run the workflow
+            final_state = self.workflow.invoke(initial_state)
+            return final_state['final_answer']
+            
+        except Exception as e:
+            print(f"Error in agent processing: {e}")
+            return f"I encountered an error while processing your question: {str(e)}"
 
 def run_and_submit_all( profile: gr.OAuthProfile | None):
     """

requirements.txt

@@ -1,2 +1,5 @@
 gradio
-requests
+requests
+langgraph>=0.0.10
+pydantic>=2.0.0
+duckduckgo-search>=4.1.1

tools.py

@@ -0,0 +1,130 @@
+from typing import Dict, Any, List
+from langgraph.graph import Graph, StateGraph
+from langgraph.prebuilt import ToolNode
+from pydantic import BaseModel, Field
+from duckduckgo_search import DDGS
+
+class CalculatorInput(BaseModel):
+    operation: str = Field(..., description="The operation to perform (add, subtract, multiply, divide)")
+    numbers: List[float] = Field(..., description="List of numbers to perform the operation on")
+
+class CalculatorOutput(BaseModel):
+    result: float = Field(..., description="The result of the calculation")
+    operation: str = Field(..., description="The operation that was performed")
+
+class SearchInput(BaseModel):
+    query: str = Field(..., description="The search query to look up")
+    max_results: int = Field(default=3, description="Maximum number of results to return")
+
+class SearchResult(BaseModel):
+    title: str = Field(..., description="Title of the search result")
+    link: str = Field(..., description="URL of the search result")
+    snippet: str = Field(..., description="Brief description of the search result")
+
+class SearchOutput(BaseModel):
+    results: List[SearchResult] = Field(..., description="List of search results")
+    query: str = Field(..., description="The original search query")
+
+def create_calculator_tool() -> Graph:
+    """Creates a calculator tool using LangGraph that can perform basic arithmetic operations."""
+    
+    def calculator_function(state: Dict[str, Any]) -> Dict[str, Any]:
+        input_data = CalculatorInput(**state["input"])
+        
+        if len(input_data.numbers) < 2:
+            raise ValueError("At least two numbers are required for calculation")
+        
+        result = input_data.numbers[0]
+        
+        for num in input_data.numbers[1:]:
+            if input_data.operation == "add":
+                result += num
+            elif input_data.operation == "subtract":
+                result -= num
+            elif input_data.operation == "multiply":
+                result *= num
+            elif input_data.operation == "divide":
+                if num == 0:
+                    raise ValueError("Cannot divide by zero")
+                result /= num
+            else:
+                raise ValueError(f"Unsupported operation: {input_data.operation}")
+        
+        output = CalculatorOutput(
+            result=result,
+            operation=input_data.operation
+        )
+        
+        return {"output": output}
+
+    # Create the graph
+    workflow = StateGraph()
+    
+    # Add the calculator tool node
+    workflow.add_node("calculator", ToolNode(calculator_function))
+    
+    # Set the entry and exit points
+    workflow.set_entry_point("calculator")
+    workflow.set_finish_point("calculator")
+    
+    return workflow.compile()
+
+def create_search_tool() -> Graph:
+    """Creates a search tool using DuckDuckGo that can search for information online."""
+    
+    def search_function(state: Dict[str, Any]) -> Dict[str, Any]:
+        input_data = SearchInput(**state["input"])
+        
+        # Initialize DuckDuckGo search
+        with DDGS() as ddgs:
+            # Perform the search
+            search_results = list(ddgs.text(
+                input_data.query,
+                max_results=input_data.max_results
+            ))
+            
+            # Convert results to our model
+            results = [
+                SearchResult(
+                    title=result["title"],
+                    link=result["link"],
+                    snippet=result["body"]
+                )
+                for result in search_results
+            ]
+            
+            output = SearchOutput(
+                results=results,
+                query=input_data.query
+            )
+            
+            return {"output": output}
+
+    # Create the graph
+    workflow = StateGraph()
+    
+    # Add the search tool node
+    workflow.add_node("search", ToolNode(search_function))
+    
+    # Set the entry and exit points
+    workflow.set_entry_point("search")
+    workflow.set_finish_point("search")
+    
+    return workflow.compile()
+
+# Example usage:
+# if __name__ == "__main__":
+#     # Create the calculator tool
+#     calculator = create_calculator_tool()
+    
+#     # Example calculation
+#     result = calculator.invoke({
+#         "input": {
+#             "operation": "add",
+#             "numbers": [1, 2, 3, 4]
+#         }
+#     })
+    
+#     print(f"Result: {result['output'].result}")
+
+