Upload 5 files

README.md

@@ -1,15 +1,35 @@
----
-title: Template Final Assignment
-emoji: 🕵🏻‍♂️
-colorFrom: indigo
-colorTo: indigo
-sdk: gradio
-sdk_version: 5.25.2
-app_file: app.py
-pinned: false
-hf_oauth: true
-# optional, default duration is 8 hours/480 minutes. Max duration is 30 days/43200 minutes.
-hf_oauth_expiration_minutes: 480
----
-
-Check out the configuration reference at https://huggingface.co/docs/hub/spaces-config-reference
+# Hugging Face Space - Personal Task Manager Agent
+
+This repository contains a Personal Task Manager Agent created for the Hugging Face AI Agent Course final project.
+
+## Project Overview
+
+The Personal Task Manager Agent helps users manage tasks through natural language commands. It can:
+- Add new tasks with details like priority, due date, and category
+- List all tasks or filter by category/priority
+- Mark tasks as complete
+- Delete tasks
+- Provide help information
+
+## Files in this Repository
+
+- `task_manager_agent.py`: Core agent implementation with task management functionality
+- `app.py`: Hugging Face Space implementation with Gradio UI and GAIA API endpoint
+- `requirements.txt`: Required Python packages
+- `README.md`: This documentation file
+
+## GAIA Benchmark Integration
+
+This agent includes an API endpoint specifically designed for the GAIA benchmark evaluation. The endpoint processes questions from the benchmark and returns answers in the expected format.
+
+## How to Use
+
+Interact with the agent using natural language commands in the chat interface. Here are some example commands:
+
+- Add a task: "Add a new task to buy groceries"
+- Add with details: "Add task to call mom priority:high due:2023-05-20 category:personal"
+- List tasks: "Show me my tasks" or "What do I need to do?"
+- Complete a task: "Mark task 2 as done" or "I completed task 3"
+- Delete a task: "Delete task 1" or "Remove task 4"
+- Filter tasks: "Show high priority tasks" or "List personal tasks"
+- Get help: "Help me" or "What can you do?"

app.py

@@ -1,196 +1,109 @@
-import os
+"""
+Hugging Face Space implementation for Personal Task Manager Agent
+This file serves as the entry point for the Hugging Face Space
+"""
+
 import gradio as gr
-import requests
-import inspect
-import pandas as pd
+from task_manager_agent import TaskManagerAgent
+import json
+import os
+
+# Initialize the agent
+agent = TaskManagerAgent()
 
-# (Keep Constants as is)
-# --- Constants ---
-DEFAULT_API_URL = "https://agents-course-unit4-scoring.hf.space"
+# Try to load existing tasks if available
+if os.path.exists("tasks.json"):
+    agent.load_state("tasks.json")
 
-# --- Basic Agent Definition ---
-# ----- THIS IS WERE YOU CAN BUILD WHAT YOU WANT ------
-class BasicAgent:
-    def __init__(self):
-        print("BasicAgent initialized.")
-    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
+def process_message(message, history):
+    """Process user message and return agent response"""
+    response = agent.process_query(message)
+    
+    # Save state after each interaction
+    agent.save_state("tasks.json")
+    
+    return response
 
-def run_and_submit_all( profile: gr.OAuthProfile | None):
+def get_gaia_answer(question):
     """
-    Fetches all questions, runs the BasicAgent on them, submits all answers,
-    and displays the results.
+    Function to process GAIA benchmark questions
+    This is the function that will be called by the GAIA API
     """
-    # --- 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
-
-    api_url = DEFAULT_API_URL
-    questions_url = f"{api_url}/questions"
-    submit_url = f"{api_url}/submit"
-
-    # 1. Instantiate Agent ( modify this part to create your agent)
-    try:
-        agent = BasicAgent()
-    except Exception as e:
-        print(f"Error instantiating agent: {e}")
-        return f"Error initializing agent: {e}", None
-    # 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)
-
-    # 2. Fetch Questions
-    print(f"Fetching questions from: {questions_url}")
-    try:
-        response = requests.get(questions_url, timeout=15)
-        response.raise_for_status()
-        questions_data = response.json()
-        if not questions_data:
-             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 Exception as e:
-        print(f"An unexpected error occurred fetching questions: {e}")
-        return f"An unexpected error occurred fetching questions: {e}", None
-
-    # 3. Run your Agent
-    results_log = []
-    answers_payload = []
-    print(f"Running agent on {len(questions_data)} questions...")
-    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:
-            print(f"Skipping item with missing task_id or question: {item}")
-            continue
-        try:
-            submitted_answer = agent(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:
-             print(f"Error running agent on task {task_id}: {e}")
-             results_log.append({"Task ID": task_id, "Question": question_text, "Submitted Answer": f"AGENT ERROR: {e}"})
-
-    if not answers_payload:
-        print("Agent did not produce any answers to submit.")
-        return "Agent did not produce any answers to submit.", pd.DataFrame(results_log)
-
-    # 4. Prepare Submission 
-    submission_data = {"username": username.strip(), "agent_code": agent_code, "answers": answers_payload}
-    status_update = f"Agent finished. Submitting {len(answers_payload)} answers for user '{username}'..."
-    print(status_update)
-
-    # 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.")
-        results_df = pd.DataFrame(results_log)
-        return final_status, results_df
-    except requests.exceptions.HTTPError as e:
-        error_detail = f"Server responded with status {e.response.status_code}."
-        try:
-            error_json = e.response.json()
-            error_detail += f" Detail: {error_json.get('detail', e.response.text)}"
-        except requests.exceptions.JSONDecodeError:
-            error_detail += f" Response: {e.response.text[:500]}"
-        status_message = f"Submission Failed: {error_detail}"
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-    except requests.exceptions.Timeout:
-        status_message = "Submission Failed: The request timed out."
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-    except requests.exceptions.RequestException as e:
-        status_message = f"Submission Failed: Network error - {e}"
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-    except Exception as e:
-        status_message = f"An unexpected error occurred during submission: {e}"
-        print(status_message)
-        results_df = pd.DataFrame(results_log)
-        return status_message, results_df
-
-
-# --- Build Gradio Interface using Blocks ---
-with gr.Blocks() as demo:
-    gr.Markdown("# Basic Agent Evaluation Runner")
-    gr.Markdown(
-        """
-        **Instructions:**
-
-        1.  Please clone this space, then modify the code to define your agent's logic, the tools, the necessary packages, etc ...
-        2.  Log in to your Hugging Face account using the button below. This uses your HF username for submission.
-        3.  Click 'Run Evaluation & Submit All Answers' to fetch questions, run your agent, submit answers, and see the score.
-
-        ---
-        **Disclaimers:**
-        Once clicking on the "submit button, it can take quite some time ( this is the time for the agent to go through all the questions).
-        This space provides a basic setup and is intentionally sub-optimal to encourage you to develop your own, more robust solution. For instance for the delay process of the submit button, a solution could be to cache the answers and submit in a seperate action or even to answer the questions in async.
-        """
+    # Process the question with our agent
+    response = agent.process_query(question)
+    
+    # For GAIA benchmark, we need to return just the answer without any formatting
+    # Strip any extra formatting that might be in the response
+    clean_response = response.strip()
+    
+    return clean_response
+
+# Create Gradio interface
+with gr.Blocks(title="Personal Task Manager Agent") as demo:
+    gr.Markdown("# Personal Task Manager Agent")
+    gr.Markdown("""
+    This agent helps you manage tasks through natural language commands.
+    
+    ## Example commands:
+    - Add a task: "Add a new task to buy groceries"
+    - Add with details: "Add task to call mom priority:high due:2023-05-20 category:personal"
+    - List tasks: "Show me my tasks" or "What do I need to do?"
+    - Complete a task: "Mark task 2 as done" or "I completed task 3"
+    - Delete a task: "Delete task 1" or "Remove task 4"
+    - Filter tasks: "Show high priority tasks" or "List personal tasks"
+    - Get help: "Help me" or "What can you do?"
+    """)
+    
+    chatbot = gr.Chatbot(height=400)
+    msg = gr.Textbox(label="Type your command here")
+    clear = gr.Button("Clear")
+    
+    def user(message, history):
+        return "", history + [[message, None]]
+    
+    def bot(history):
+        message = history[-1][0]
+        response = process_message(message, history)
+        history[-1][1] = response
+        return history
+    
+    msg.submit(user, [msg, chatbot], [msg, chatbot], queue=False).then(
+        bot, chatbot, chatbot
     )
-
-    gr.LoginButton()
-
-    run_button = gr.Button("Run Evaluation & Submit All Answers")
-
-    status_output = gr.Textbox(label="Run Status / Submission Result", lines=5, interactive=False)
-    # Removed max_rows=10 from DataFrame constructor
-    results_table = gr.DataFrame(label="Questions and Agent Answers", wrap=True)
-
-    run_button.click(
-        fn=run_and_submit_all,
-        outputs=[status_output, results_table]
-    )
-
+    
+    clear.click(lambda: None, None, chatbot, queue=False)
+    
+    # Add GAIA API endpoint explanation
+    gr.Markdown("""
+    ## GAIA Benchmark API
+    
+    This Space includes an API endpoint for the GAIA benchmark. The API processes questions 
+    and returns answers in the format expected by the benchmark.
+    
+    The endpoint is automatically available when deployed on Hugging Face Spaces.
+    """)
+
+# For GAIA API endpoint
+def gaia_api(question):
+    """API endpoint for GAIA benchmark"""
+    answer = get_gaia_answer(question)
+    return {"answer": answer}
+
+# Launch the app
 if __name__ == "__main__":
-    print("\n" + "-"*30 + " App Starting " + "-"*30)
-    # Check for SPACE_HOST and SPACE_ID at startup for information
-    space_host_startup = os.getenv("SPACE_HOST")
-    space_id_startup = os.getenv("SPACE_ID") # Get SPACE_ID at startup
-
-    if space_host_startup:
-        print(f"✅ SPACE_HOST found: {space_host_startup}")
-        print(f"   Runtime URL should be: https://{space_host_startup}.hf.space")
-    else:
-        print("ℹ️  SPACE_HOST environment variable not found (running locally?).")
-
-    if space_id_startup: # Print repo URLs if SPACE_ID is found
-        print(f"✅ SPACE_ID found: {space_id_startup}")
-        print(f"   Repo URL: https://huggingface.co/spaces/{space_id_startup}")
-        print(f"   Repo Tree URL: https://huggingface.co/spaces/{space_id_startup}/tree/main")
-    else:
-        print("ℹ️  SPACE_ID environment variable not found (running locally?). Repo URL cannot be determined.")
-
-    print("-"*(60 + len(" App Starting ")) + "\n")
-
-    print("Launching Gradio Interface for Basic Agent Evaluation...")
-    demo.launch(debug=True, share=False)
+    # Set up FastAPI for GAIA benchmark
+    from fastapi import FastAPI, Request
+    import uvicorn
+    from pydantic import BaseModel
+    
+    app = FastAPI()
+    
+    class Question(BaseModel):
+        question: str
+    
+    @app.post("/api/gaia")
+    async def api_gaia(question: Question):
+        return gaia_api(question.question)
+    
+    # Mount Gradio app
+    demo.launch(server_name="0.0.0.0", server_port=7860)

requirements.txt

@@ -1,2 +1,4 @@
-gradio
-requests
+gradio==4.0.2
+fastapi==0.104.1
+uvicorn==0.23.2
+pydantic==2.4.2

task_manager_agent.py

@@ -0,0 +1,336 @@
+"""
+Personal Task Manager Agent for Hugging Face AI Agent Course Final Project
+
+This agent helps users manage tasks, set priorities, and organize information
+through natural language understanding and simple task management functionality.
+"""
+
+import json
+import re
+import datetime
+from typing import List, Dict, Any, Optional, Tuple
+
+class Task:
+    """Represents a single task with properties like description, priority, due date, etc."""
+    
+    def __init__(self, description: str, priority: str = "medium", 
+                 due_date: Optional[str] = None, category: str = "general"):
+        self.description = description
+        self.priority = priority.lower()  # "high", "medium", or "low"
+        self.due_date = due_date  # Format: YYYY-MM-DD
+        self.category = category.lower()
+        self.completed = False
+        self.creation_date = datetime.datetime.now().strftime("%Y-%m-%d")
+    
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert task to dictionary for JSON serialization"""
+        return {
+            "description": self.description,
+            "priority": self.priority,
+            "due_date": self.due_date,
+            "category": self.category,
+            "completed": self.completed,
+            "creation_date": self.creation_date
+        }
+    
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> 'Task':
+        """Create a Task object from a dictionary"""
+        task = cls(
+            description=data["description"],
+            priority=data["priority"],
+            due_date=data["due_date"],
+            category=data["category"]
+        )
+        task.completed = data["completed"]
+        task.creation_date = data["creation_date"]
+        return task
+    
+    def __str__(self) -> str:
+        """String representation of the task"""
+        status = "✓" if self.completed else "□"
+        due_str = f" (Due: {self.due_date})" if self.due_date else ""
+        return f"{status} [{self.priority.upper()}] {self.description}{due_str} - {self.category}"
+
+
+class TaskManager:
+    """Manages a collection of tasks with functionality to add, update, and query tasks"""
+    
+    def __init__(self):
+        self.tasks: List[Task] = []
+    
+    def add_task(self, task: Task) -> int:
+        """Add a new task and return its index"""
+        self.tasks.append(task)
+        return len(self.tasks) - 1
+    
+    def update_task(self, index: int, **kwargs) -> bool:
+        """Update task properties at the given index"""
+        if 0 <= index < len(self.tasks):
+            task = self.tasks[index]
+            for key, value in kwargs.items():
+                if hasattr(task, key):
+                    setattr(task, key, value)
+            return True
+        return False
+    
+    def complete_task(self, index: int) -> bool:
+        """Mark a task as completed"""
+        return self.update_task(index, completed=True)
+    
+    def delete_task(self, index: int) -> bool:
+        """Delete a task at the given index"""
+        if 0 <= index < len(self.tasks):
+            self.tasks.pop(index)
+            return True
+        return False
+    
+    def get_tasks(self, 
+                  category: Optional[str] = None, 
+                  priority: Optional[str] = None,
+                  completed: Optional[bool] = None) -> List[Tuple[int, Task]]:
+        """Get tasks with optional filtering, returns list of (index, task) tuples"""
+        result = []
+        for i, task in enumerate(self.tasks):
+            if (category is None or task.category == category.lower()) and \
+               (priority is None or task.priority == priority.lower()) and \
+               (completed is None or task.completed == completed):
+                result.append((i, task))
+        return result
+    
+    def get_all_tasks(self) -> List[Tuple[int, Task]]:
+        """Get all tasks with their indices"""
+        return [(i, task) for i, task in enumerate(self.tasks)]
+    
+    def get_categories(self) -> List[str]:
+        """Get list of unique categories"""
+        return list(set(task.category for task in self.tasks))
+    
+    def save_to_file(self, filename: str) -> bool:
+        """Save tasks to a JSON file"""
+        try:
+            with open(filename, 'w') as f:
+                json.dump([task.to_dict() for task in self.tasks], f, indent=2)
+            return True
+        except Exception as e:
+            print(f"Error saving tasks: {e}")
+            return False
+    
+    def load_from_file(self, filename: str) -> bool:
+        """Load tasks from a JSON file"""
+        try:
+            with open(filename, 'r') as f:
+                data = json.load(f)
+                self.tasks = [Task.from_dict(item) for item in data]
+            return True
+        except Exception as e:
+            print(f"Error loading tasks: {e}")
+            return False
+
+
+class TaskManagerAgent:
+    """
+    AI agent that understands natural language requests related to task management
+    and performs appropriate actions using the TaskManager.
+    """
+    
+    def __init__(self):
+        self.task_manager = TaskManager()
+        # Common patterns for understanding user intent
+        self.patterns = {
+            "add_task": [
+                r"add (?:a )?(?:new )?task(?: to)?(?: do)?(?: called)?(?: to)?(?: about)? (.+)",
+                r"create (?:a )?(?:new )?task(?: to)?(?: do)?(?: called)?(?: to)?(?: about)? (.+)",
+                r"remind me to (.+)",
+            ],
+            "list_tasks": [
+                r"(?:show|list|display|get)(?: all)?(?: my)? tasks",
+                r"what (?:are my|do i have for) tasks",
+                r"show me what i need to do",
+            ],
+            "complete_task": [
+                r"(?:mark|set) task (?:number )?(\d+) (?:as )?(?:done|complete|finished)",
+                r"complete task (?:number )?(\d+)",
+                r"i (?:have )?(?:done|completed|finished) task (?:number )?(\d+)",
+            ],
+            "delete_task": [
+                r"(?:delete|remove) task (?:number )?(\d+)",
+                r"get rid of task (?:number )?(\d+)",
+            ],
+            "filter_tasks": [
+                r"(?:show|list|display|get) (?:all )?(\w+) tasks",
+                r"(?:show|list|display|get) tasks (?:with|that are) (\w+)",
+            ],
+            "help": [
+                r"(?:help|assist|guide) me",
+                r"what can you do",
+                r"how do (?:you|i|this) work",
+            ]
+        }
+    
+    def _extract_task_details(self, description: str) -> Dict[str, Any]:
+        """Extract task details like priority, due date, and category from description"""
+        details = {"description": description}
+        
+        # Extract priority
+        priority_match = re.search(r"(?:priority|important|urgency):?\s*(high|medium|low)", description, re.I)
+        if priority_match:
+            details["priority"] = priority_match.group(1).lower()
+            # Remove the priority text from description
+            details["description"] = re.sub(r"(?:priority|important|urgency):?\s*(high|medium|low)", "", details["description"], flags=re.I).strip()
+        
+        # Extract due date (simple format: YYYY-MM-DD or MM/DD/YYYY)
+        date_match = re.search(r"(?:due|by|on):?\s*(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{4})", description, re.I)
+        if date_match:
+            date_str = date_match.group(1)
+            # Convert MM/DD/YYYY to YYYY-MM-DD if needed
+            if "/" in date_str:
+                month, day, year = date_str.split("/")
+                date_str = f"{year}-{month.zfill(2)}-{day.zfill(2)}"
+            details["due_date"] = date_str
+            # Remove the date text from description
+            details["description"] = re.sub(r"(?:due|by|on):?\s*(\d{4}-\d{2}-\d{2}|\d{1,2}/\d{1,2}/\d{4})", "", details["description"], flags=re.I).strip()
+        
+        # Extract category
+        category_match = re.search(r"(?:category|tag|type):?\s*(\w+)", description, re.I)
+        if category_match:
+            details["category"] = category_match.group(1).lower()
+            # Remove the category text from description
+            details["description"] = re.sub(r"(?:category|tag|type):?\s*(\w+)", "", details["description"], flags=re.I).strip()
+        
+        return details
+    
+    def process_query(self, query: str) -> str:
+        """
+        Process a natural language query and perform the appropriate task management action.
+        Returns a response string.
+        """
+        # Check for intent matches
+        for intent, patterns in self.patterns.items():
+            for pattern in patterns:
+                match = re.search(pattern, query, re.I)
+                if match:
+                    # Call the appropriate method based on intent
+                    if intent == "add_task" and match.group(1):
+                        return self._handle_add_task(match.group(1))
+                    elif intent == "list_tasks":
+                        return self._handle_list_tasks()
+                    elif intent == "complete_task" and match.group(1):
+                        return self._handle_complete_task(int(match.group(1)))
+                    elif intent == "delete_task" and match.group(1):
+                        return self._handle_delete_task(int(match.group(1)))
+                    elif intent == "filter_tasks" and match.group(1):
+                        return self._handle_filter_tasks(match.group(1))
+                    elif intent == "help":
+                        return self._handle_help()
+        
+        # If no pattern matches, try to understand as a general query
+        return self._handle_general_query(query)
+    
+    def _handle_add_task(self, description: str) -> str:
+        """Handle adding a new task"""
+        details = self._extract_task_details(description)
+        task = Task(
+            description=details["description"],
+            priority=details.get("priority", "medium"),
+            due_date=details.get("due_date"),
+            category=details.get("category", "general")
+        )
+        index = self.task_manager.add_task(task)
+        return f"Added task {index}: {task}"
+    
+    def _handle_list_tasks(self) -> str:
+        """Handle listing all tasks"""
+        tasks = self.task_manager.get_all_tasks()
+        if not tasks:
+            return "You don't have any tasks yet."
+        
+        result = "Here are your tasks:\n"
+        for index, task in tasks:
+            result += f"{index}: {task}\n"
+        return result
+    
+    def _handle_complete_task(self, index: int) -> str:
+        """Handle marking a task as complete"""
+        if self.task_manager.complete_task(index):
+            return f"Marked task {index} as completed."
+        return f"Task {index} not found."
+    
+    def _handle_delete_task(self, index: int) -> str:
+        """Handle deleting a task"""
+        if self.task_manager.delete_task(index):
+            return f"Deleted task {index}."
+        return f"Task {index} not found."
+    
+    def _handle_filter_tasks(self, filter_term: str) -> str:
+        """Handle filtering tasks by category or priority"""
+        # Check if filter is a priority
+        if filter_term.lower() in ["high", "medium", "low"]:
+            tasks = self.task_manager.get_tasks(priority=filter_term.lower())
+            filter_type = "priority"
+        else:
+            # Assume it's a category
+            tasks = self.task_manager.get_tasks(category=filter_term.lower())
+            filter_type = "category"
+        
+        if not tasks:
+            return f"No tasks found with {filter_type} '{filter_term}'."
+        
+        result = f"Tasks with {filter_type} '{filter_term}':\n"
+        for index, task in tasks:
+            result += f"{index}: {task}\n"
+        return result
+    
+    def _handle_help(self) -> str:
+        """Handle help request"""
+        return """
+I can help you manage your tasks. Here's what you can ask me to do:
+
+- Add a task: "Add a new task to buy groceries"
+- Add with details: "Add task to call mom priority:high due:2023-05-20 category:personal"
+- List tasks: "Show me my tasks" or "What do I need to do?"
+- Complete a task: "Mark task 2 as done" or "I completed task 3"
+- Delete a task: "Delete task 1" or "Remove task 4"
+- Filter tasks: "Show high priority tasks" or "List personal tasks"
+- Get help: "Help me" or "What can you do?"
+
+Try one of these commands to get started!
+"""
+    
+    def _handle_general_query(self, query: str) -> str:
+        """Handle queries that don't match specific patterns"""
+        # Check if it might be a task addition without explicit "add task" prefix
+        if not any(re.search(pattern, query, re.I) for patterns in self.patterns.values() for pattern in patterns):
+            # If query is short and looks like a task, add it
+            if len(query.split()) <= 10 and not query.endswith("?"):
+                return self._handle_add_task(query)
+        
+        return f"I'm not sure how to help with '{query}'. Type 'help' to see what I can do."
+    
+    def save_state(self, filename: str = "tasks.json") -> bool:
+        """Save the current state of tasks to a file"""
+        return self.task_manager.save_to_file(filename)
+    
+    def load_state(self, filename: str = "tasks.json") -> bool:
+        """Load tasks from a file"""
+        return self.task_manager.load_from_file(filename)
+
+
+# Example usage for testing
+if __name__ == "__main__":
+    agent = TaskManagerAgent()
+    
+    # Test with some example queries
+    test_queries = [
+        "add task to buy groceries",
+        "add task to call mom priority:high due:2023-05-20 category:personal",
+        "show my tasks",
+        "mark task 0 as done",
+        "show high priority tasks",
+        "help"
+    ]
+    
+    for query in test_queries:
+        print(f"\nQuery: {query}")
+        response = agent.process_query(query)
+        print(f"Response: {response}")

taskm.jsonl

@@ -0,0 +1,5 @@
+{"task_id": "task_id_1", "model_answer": "Added task 0: □ [MEDIUM] buy groceries - general", "reasoning_trace": "The user asked to add a task to buy groceries. I processed this as a task addition request, created a new task with default priority (medium) and category (general), and added it to the task list."}
+{"task_id": "task_id_2", "model_answer": "Added task 1: □ [HIGH] call mom (Due: 2023-05-20) - personal", "reasoning_trace": "The user asked to add a task with specific details. I extracted the priority (high), due date (2023-05-20), and category (personal) from the request and created a task with these parameters."}
+{"task_id": "task_id_3", "model_answer": "Here are your tasks:\n0: □ [MEDIUM] buy groceries - general\n1: □ [HIGH] call mom (Due: 2023-05-20) - personal", "reasoning_trace": "The user asked to see their tasks. I retrieved all tasks from the task manager and formatted them as a numbered list showing the status, priority, description, due date (if any), and category."}
+{"task_id": "task_id_4", "model_answer": "Marked task 0 as completed.", "reasoning_trace": "The user asked to mark task 0 as done. I updated the task's completed status to true and confirmed the action."}
+{"task_id": "task_id_5", "model_answer": "Tasks with priority 'high':\n1: □ [HIGH] call mom (Due: 2023-05-20) - personal", "reasoning_trace": "The user asked to see high priority tasks. I filtered the tasks by priority='high' and returned the matching tasks as a formatted list."}