pydanticc_remove

app.py

@@ -3,10 +3,9 @@ import gradio as gr
 import requests
 import inspect
 import pandas as pd
-from typing import Dict, Any, List, TypedDict
+from typing import Dict, Any, List, TypedDict, Optional
 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)
@@ -15,15 +14,14 @@ 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(BaseModel):
-    """Schema for the agent's state."""
-    question: str = Field(..., description="The original question")
-    current_step: str = Field(default="analyze", description="Current step in the workflow")
-    tool_output: str = Field(default="", description="Output from the last tool used")
-    final_answer: str = Field(default="", description="The final answer to be returned")
-    history: List[Dict[str, str]] = Field(default_factory=list, description="History of operations performed")
-    needs_more_info: bool = Field(default=False, description="Whether more information is needed")
-    search_query: str = Field(default="", description="Current search query if any")
+class AgentState(TypedDict):
+    question: str
+    current_step: str
+    tool_output: str
+    final_answer: str
+    history: List[Dict[str, str]]
+    needs_more_info: bool
+    search_query: str
 
 class BasicAgent:
     def __init__(self):
@@ -52,7 +50,7 @@ class BasicAgent:
         """Create the agent workflow using LangGraph."""
         # Create the workflow with state schema
         print("Creating Stategraph : error happens here?")
-        workflow = StateGraph(state_schema= AgentState)
+        workflow = StateGraph(state_schema=Dict[str, Any])
         print("Stategraph created")
         # Add nodes
         workflow.add_node("analyze", self._analyze_question)
@@ -68,12 +66,12 @@ class BasicAgent:
         workflow.add_edge("search", "final_answer")
         
         # Define conditional edges
-        def router(state: AgentState) -> str:
-            if state.current_step == 'calculator':
+        def router(state: Dict[str, Any]) -> str:
+            if state["current_step"] == 'calculator':
                 return 'calculator'
-            elif state.current_step == 'search':
+            elif state["current_step"] == 'search':
                 return 'search'
-            elif state.current_step == 'final_answer':
+            elif state["current_step"] == 'final_answer':
                 return 'final_answer'
             return 'analyze'
         
@@ -107,9 +105,9 @@ class BasicAgent:
             print(f"Error calling LLM API: {e}")
             return f"Error getting response from LLM: {str(e)}"
 
-    def _analyze_question(self, state: AgentState) -> AgentState:
+    def _analyze_question(self, state: Dict[str, Any]) -> Dict[str, Any]:
         """Analyze the question and determine the next step."""
-        prompt = f"""Analyze this question and determine what needs to be done: {state.question}
+        prompt = f"""Analyze this question and determine what needs to be done: {state['question']}
         Return your analysis in this format:
         {{
             "needs_calculation": true/false,
@@ -123,71 +121,71 @@ class BasicAgent:
         """
         
         analysis = eval(self._call_llm_api(prompt))
-        state.needs_more_info = analysis.get('needs_search', False)
-        state.search_query = analysis.get('search_query', '')
+        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'])
+            state["current_step"] = 'calculator'
+            state["tool_output"] = str(analysis['calculation'])
         elif analysis.get('needs_search', False):
-            state.current_step = 'search'
+            state["current_step"] = 'search'
         else:
-            state.current_step = 'final_answer'
+            state["current_step"] = 'final_answer'
         
         return state
 
-    def _use_calculator(self, state: AgentState) -> AgentState:
+    def _use_calculator(self, state: Dict[str, Any]) -> Dict[str, Any]:
         """Use the calculator tool."""
         try:
             # Create calculator state with input from tool_output
-            calc_input = eval(state.tool_output)
-            result = self.calculator.invoke(CalculatorState(input=calc_input))
+            calc_input = eval(state["tool_output"])
+            result = self.calculator.invoke({"input": calc_input})
             
-            state.history.append({
+            state["history"].append({
                 'step': 'calculator',
-                'input': state.tool_output,
-                'output': str(result['output'].result)
+                'input': state["tool_output"],
+                'output': str(result['output']['result'])
             })
-            state.current_step = 'final_answer'
+            state["current_step"] = 'final_answer'
         except Exception as e:
-            state.history.append({
+            state["history"].append({
                 'step': 'calculator_error',
                 'error': str(e)
             })
-            state.current_step = 'final_answer'
+            state["current_step"] = 'final_answer'
         return state
 
-    def _use_search(self, state: AgentState) -> AgentState:
+    def _use_search(self, state: Dict[str, Any]) -> Dict[str, Any]:
         """Use the search tool."""
         try:
             # Create search state with input from search_query
-            search_input = SearchInput(
-                query=state.search_query,
-                max_results=3
-            )
-            result = self.search_tool.invoke(SearchState(input=search_input))
+            search_input = {
+                "query": state["search_query"],
+                "max_results": 3
+            }
+            result = self.search_tool.invoke({"input": search_input})
             
-            state.history.append({
+            state["history"].append({
                 'step': 'search',
-                'query': state.search_query,
-                'results': [str(r) for r in result['output'].results]
+                'query': state["search_query"],
+                'results': [str(r) for r in result['output']['results']]
             })
-            state.needs_more_info = False
-            state.current_step = 'final_answer'
+            state["needs_more_info"] = False
+            state["current_step"] = 'final_answer'
         except Exception as e:
-            state.history.append({
+            state["history"].append({
                 'step': 'search_error',
                 'error': str(e)
             })
-            state.current_step = 'final_answer'
+            state["current_step"] = 'final_answer'
         return state
 
-    def _generate_final_answer(self, state: AgentState) -> AgentState:
+    def _generate_final_answer(self, state: Dict[str, Any]) -> Dict[str, Any]:
         """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])
+                               for h in state["history"]])
         
-        prompt = f"""Based on the following information and history, provide a final answer to the question: {state.question}
+        prompt = f"""Based on the following information and history, provide a final answer to the question: {state['question']}
         
         History of steps taken:
         {history_str}
@@ -195,7 +193,7 @@ class BasicAgent:
         Provide a clear, concise answer that addresses the original question.
         """
         
-        state.final_answer = self._call_llm_api(prompt)
+        state["final_answer"] = self._call_llm_api(prompt)
         return state
 
     def __call__(self, question: str) -> str:
@@ -204,19 +202,19 @@ class BasicAgent:
         
         try:
             # Initialize the state
-            initial_state = AgentState(
-                question=question,
-                current_step="analyze",
-                tool_output="",
-                final_answer="",
-                history=[],
-                needs_more_info=False,
-                search_query=""
-            )
+            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
+            return final_state["final_answer"]
             
         except Exception as e:
             print(f"Error in agent processing: {e}")
@@ -289,15 +287,15 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
 
         try:
             # Initialize the state for this question
-            initial_state = AgentState(
-                question=question_text,
-                current_step="analyze",
-                tool_output="",
-                final_answer="",
-                history=[],
-                needs_more_info=False,
-                search_query=""
-            )
+            initial_state = {
+                "question": question_text,
+                "current_step": "analyze",
+                "tool_output": "",
+                "final_answer": "",
+                "history": [],
+                "needs_more_info": False,
+                "search_query": ""
+            }
             
             # Run the workflow for this question
             print(f"\nProcessing question {task_id}: {question_text[:50]}...")
@@ -308,11 +306,11 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
                 f"Step: {h['step']}\n" +
                 f"Input: {h.get('input', h.get('query', ''))}\n" +
                 f"Output: {h.get('output', h.get('results', h.get('error', '')))}"
-                for h in final_state.history
+                for h in final_state["history"]
             ])
             
             # Add to results
-            submitted_answer = final_state.final_answer
+            submitted_answer = final_state["final_answer"]
             answers_payload.append({"task_id": task_id, "submitted_answer": submitted_answer})
             results_log.append({
                 "Task ID": task_id,
@@ -321,7 +319,7 @@ def run_and_submit_all(profile: gr.OAuthProfile | None):
                 "Workflow History": workflow_history
             })
             
-            print(f"Completed question {task_id} with {len(final_state.history)} workflow steps")
+            print(f"Completed question {task_id} with {len(final_state['history'])} workflow steps")
             
         except Exception as e:
             print(f"Error running agent workflow on task {task_id}: {e}")

tools.py

@@ -1,71 +1,60 @@
-from typing import Dict, Any, List
+from typing import Dict, Any, List, TypedDict, Optional
 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 CalculatorInput(TypedDict):
+    operation: str
+    numbers: List[float]
 
-class CalculatorOutput(BaseModel):
-    result: float = Field(..., description="The result of the calculation")
-    operation: str = Field(..., description="The operation that was performed")
+class CalculatorOutput(TypedDict):
+    result: float
+    operation: str
 
-class CalculatorState(BaseModel):
-    input: CalculatorInput
-    output: CalculatorOutput = None
+class SearchResult(TypedDict):
+    title: str
+    link: str
+    snippet: str
 
-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")
-
-class SearchState(BaseModel):
-    input: SearchInput
-    output: SearchOutput = None
+class SearchOutput(TypedDict):
+    results: List[SearchResult]
+    query: str
 
 def create_calculator_tool() -> Graph:
     """Creates a calculator tool using LangGraph that can perform basic arithmetic operations."""
     print("Creating calculator tool")
-    def calculator_function(state: CalculatorState) -> dict:
+    
+    def calculator_function(state: Dict[str, Any]) -> dict:
         print("Calculator function called")
-        if len(state.input.numbers) < 2:
+        input_data = state["input"]
+        if len(input_data["numbers"]) < 2:
             raise ValueError("At least two numbers are required for calculation")
         
-        result = state.input.numbers[0]
+        result = input_data["numbers"][0]
         
-        for num in state.input.numbers[1:]:
-            if state.input.operation == "add":
+        for num in input_data["numbers"][1:]:
+            if input_data["operation"] == "add":
                 result += num
-            elif state.input.operation == "subtract":
+            elif input_data["operation"] == "subtract":
                 result -= num
-            elif state.input.operation == "multiply":
+            elif input_data["operation"] == "multiply":
                 result *= num
-            elif state.input.operation == "divide":
+            elif input_data["operation"] == "divide":
                 if num == 0:
                     raise ValueError("Cannot divide by zero")
                 result /= num
             else:
-                raise ValueError(f"Unsupported operation: {state.input.operation}")
+                raise ValueError(f"Unsupported operation: {input_data['operation']}")
         
         return {
-            "output": CalculatorOutput(
-                result=result,
-                operation=state.input.operation
-            )
+            "output": {
+                "result": result,
+                "operation": input_data["operation"]
+            }
         }
 
     # Create the graph with state schema
-    workflow = StateGraph(state_schema=CalculatorState)
+    workflow = StateGraph(state_schema=Dict[str, Any])
     print("Calculator graph for workflow created")
     # Add the calculator tool node
     workflow.add_node("calculator", ToolNode(calculator_function))
@@ -79,34 +68,34 @@ def create_calculator_tool() -> Graph:
 def create_search_tool() -> Graph:
     """Creates a search tool using DuckDuckGo that can search for information online."""
     
-    def search_function(state: SearchState) -> dict:
+    def search_function(state: Dict[str, Any]) -> dict:
         with DDGS() as ddgs:
             # Run search
             raw_results = list(ddgs.text(
-                state.input.query,
-                max_results=state.input.max_results
+                state["input"]["query"],
+                max_results=state["input"].get("max_results", 3)
             ))
 
         results = []
         for r in raw_results:
             try:
-                results.append(SearchResult(
-                    title=r.get("title", ""),
-                    link=r.get("href", r.get("link", "")),  # DuckDuckGo sometimes uses "href"
-                    snippet=r.get("body", r.get("snippet", ""))
-                ))
+                results.append({
+                    "title": r.get("title", ""),
+                    "link": r.get("href", r.get("link", "")),  # DuckDuckGo sometimes uses "href"
+                    "snippet": r.get("body", r.get("snippet", ""))
+                })
             except Exception as e:
                 print("Skipping malformed search result:", r, "Error:", e)
 
         return {
-            "output": SearchOutput(
-                results=results,
-                query=state.input.query
-            )
+            "output": {
+                "results": results,
+                "query": state["input"]["query"]
+            }
         }
 
     # Create the graph with state schema
-    workflow = StateGraph(state_schema=SearchState)
+    workflow = StateGraph(state_schema=Dict[str, Any])
     
     # Add the search tool node
     workflow.add_node("search", ToolNode(search_function))