feat: todo list for agent

examples/agents/todo_tools_example.py

@@ -0,0 +1,71 @@
+"""Example demonstrating the new instance-based todo functionality."""
+
+import asyncio
+
+from ragbits.agents import Agent, AgentOptions
+from ragbits.agents.tools.todo import TodoList, create_todo_manager, get_todo_instruction_tpl
+from ragbits.core.llms import LiteLLM, ToolCall
+
+
+async def main():
+    """Demonstrate the new instance-based todo approach with streaming and logging."""
+
+    # Create a dedicated TodoList instance for this agent
+    my_todo_list = TodoList()
+    my_todo_manager = create_todo_manager(my_todo_list)
+
+    # Create an agent with higher turn limit and todo capabilities
+    my_agent = Agent(
+        llm=LiteLLM("gpt-4o-mini"),
+        prompt="""
+        You are an expert hiking guide. You can either answer questions or
+        create a comprehensive, detailed hiking trip plan.
+
+        WORKFLOW:
+        1. If query is complex you have access to todo_manager tool to create a todo list with specific tasks
+        2. If query is simple question, you work without todo_manager tool, just answer the question
+        3. If you use todo_manager tool, you must follow the todo workflow below
+
+        For hiking plans include:
+        - Specific route names, distances, elevation gain
+        - Detailed gear recommendations with quantities
+        - Transportation details with times, costs, parking info
+        - Weather considerations and backup plans
+        - Safety information and emergency contacts
+        """ + get_todo_instruction_tpl(task_range=(3, 5)),
+        tools=[my_todo_manager],  # Use the instance-specific todo manager
+        default_options=AgentOptions(max_turns=30)
+    )
+
+    query = "Plan a 1-day hiking trip for 2 people in Tatra Mountains, Poland. Focus on scenic routes under 15km, avoiding crowds."
+    # query = "How long is hike to Giewont from Kuźnice?"
+    # query = "Is it difficult to finish Orla Perć? Would you recommend me to go there if I've never been in mountains before?"
+
+    stream = my_agent.run_streaming(query)
+
+    async for response in stream:
+        match response:
+            case str():
+                if response.strip():
+                    print(response, end="", flush=True)
+
+            case ToolCall():
+                if response.name == "todo_manager":
+                    action = response.arguments.get("action", "unknown")
+
+                    if action == "create":
+                        print("=== Enhanced Todo Workflow Example ===\n")
+                        print("🚀 Hiking trip planning with systematic workflow:\n")
+
+                        tasks = response.arguments.get("tasks", [])
+                        tasks_count = len(tasks)
+                        print(f" - Creating {tasks_count} tasks", flush=True)
+                        for i, task in enumerate(tasks, 1):
+                            print(f"   {i}. {task}")
+
+    print("\n\n" + "="*50)
+    print("🎉 Systematic hiking trip planning completed!")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

packages/ragbits-agents/src/ragbits/agents/__init__.py

@@ -7,6 +7,7 @@
     AgentRunContext,
     ToolCallResult,
 )
+from ragbits.agents.tools import get_todo_instruction_tpl, create_todo_manager
 from ragbits.agents.types import QuestionAnswerAgent, QuestionAnswerPromptInput, QuestionAnswerPromptOutput
 
 __all__ = [
@@ -20,4 +21,6 @@
     "QuestionAnswerPromptInput",
     "QuestionAnswerPromptOutput",
     "ToolCallResult",
+    "get_todo_instruction_tpl",
+    "create_todo_manager",
 ]

packages/ragbits-agents/src/ragbits/agents/tools/__init__.py

@@ -1,3 +1,5 @@
-from ragbits.agents.tools.openai import get_code_interpreter_tool, get_image_generation_tool, get_web_search_tool
+"""Agent tools for extending functionality."""
 
-__all__ = ["get_code_interpreter_tool", "get_image_generation_tool", "get_web_search_tool"]
+from .todo import get_todo_instruction_tpl, create_todo_manager
+
+__all__ = ["create_todo_manager", "get_todo_instruction_tpl"]

packages/ragbits-agents/src/ragbits/agents/tools/todo.py

@@ -0,0 +1,219 @@
+"""Todo list management tool for agents."""
+
+import uuid
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any, Literal, Callable
+
+
+class TaskStatus(str, Enum):
+    """Task status options."""
+    PENDING = "pending"
+    IN_PROGRESS = "in_progress"
+    COMPLETED = "completed"
+
+
+@dataclass
+class Task:
+    """Simple task representation."""
+    id: str
+    description: str
+    status: TaskStatus = TaskStatus.PENDING
+    order: int = 0
+    summary: str | None = None
+
+
+@dataclass
+class TodoList:
+    """Simple todo list for one agent run."""
+    tasks: list[Task] = field(default_factory=list)
+    current_index: int = 0
+
+    def get_current_task(self) -> Task | None:
+        """Get current task to work on."""
+        if self.current_index < len(self.tasks):
+            return self.tasks[self.current_index]
+        return None
+
+    def advance_to_next(self):
+        """Move to next task."""
+        self.current_index += 1
+
+    def create_tasks(self, task_descriptions: list[str]) -> dict[str, Any]:
+        """Create tasks from descriptions."""
+        if not task_descriptions:
+            raise ValueError("Tasks required for create action")
+
+        # Clear existing tasks
+        self.tasks.clear()
+        self.current_index = 0
+
+        for i, desc in enumerate(task_descriptions):
+            task = Task(
+                id=str(uuid.uuid4()),
+                description=desc.strip(),
+                order=i
+            )
+            self.tasks.append(task)
+
+        return {
+            "action": "create",
+            "tasks": [{"id": t.id, "description": t.description, "order": t.order} for t in self.tasks],
+            "total_count": len(self.tasks),
+            "message": f"Created {len(task_descriptions)} tasks"
+        }
+
+    def get_current(self) -> dict[str, Any]:
+        """Get current task information."""
+        current = self.get_current_task()
+        if not current:
+            return {
+                "action": "get_current",
+                "current_task": None,
+                "all_completed": True,
+                "message": "All tasks completed!"
+            }
+
+        return {
+            "action": "get_current",
+            "current_task": {"id": current.id, "description": current.description, "status": current.status.value},
+            "progress": f"{self.current_index + 1}/{len(self.tasks)}",
+            "message": f"Current task: {current.description}"
+        }
+
+    def start_current_task(self) -> dict[str, Any]:
+        """Start the current task."""
+        current = self.get_current_task()
+        if not current:
+            raise ValueError("No current task to start")
+
+        current.status = TaskStatus.IN_PROGRESS
+        return {
+            "action": "start_task",
+            "task": {"id": current.id, "description": current.description, "status": current.status.value},
+            "message": f"Started task: {current.description}"
+        }
+
+    def complete_current_task(self, summary: str) -> dict[str, Any]:
+        """Complete the current task with summary."""
+        if not summary:
+            raise ValueError("Summary required for complete_task")
+
+        current = self.get_current_task()
+        if not current:
+            raise ValueError("No current task to complete")
+
+        if current.status != TaskStatus.IN_PROGRESS:
+            raise ValueError("Task must be started before completing")
+
+        current.status = TaskStatus.COMPLETED
+        current.summary = summary.strip()
+        self.advance_to_next()
+
+        next_task = self.get_current_task()
+        completed_count = sum(1 for t in self.tasks if t.status == TaskStatus.COMPLETED)
+
+        return {
+            "action": "complete_task",
+            "completed_task": {"id": current.id, "description": current.description, "summary": current.summary},
+            "next_task": {"id": next_task.id, "description": next_task.description} if next_task else None,
+            "progress": f"{completed_count}/{len(self.tasks)}",
+            "all_completed": next_task is None,
+            "message": f"Completed: {current.description}"
+        }
+
+    def get_final_summary(self) -> dict[str, Any]:
+        """Get comprehensive final summary of all completed work."""
+        completed_tasks = [t for t in self.tasks if t.status == TaskStatus.COMPLETED]
+
+        if not completed_tasks:
+            return {
+                "action": "get_final_summary",
+                "final_summary": "",
+                "message": "No completed tasks found."
+            }
+
+        # Create comprehensive final summary
+        final_content = []
+        for i, task in enumerate(completed_tasks):
+            if task.summary:
+                final_content.append(f"**{i+1}. {task.description}**:\n{task.summary}")
+            else:
+                final_content.append(f"**{i+1}. {task.description}**: Completed")
+
+        final_summary = "\n\n".join(final_content)
+
+        return {
+            "action": "get_final_summary",
+            "final_summary": final_summary,
+            "total_completed": len(completed_tasks),
+            "message": f"Final summary with {len(completed_tasks)} completed tasks."
+        }
+
+
+def create_todo_manager(todo_list: TodoList) -> Callable[..., dict[str, Any]]:
+    """
+    Create a todo_manager function bound to a specific TodoList instance.
+
+    This allows each agent to have its own isolated todo list.
+
+    Args:
+        todo_list: The TodoList instance to bind to
+
+    Returns:
+        A todo_manager function that operates on the provided TodoList
+    """
+    def todo_manager(
+        action: Literal["create", "get_current", "start_task", "complete_task", "get_final_summary"],
+        tasks: list[str] | None = None,
+        summary: str | None = None,
+    ) -> dict[str, Any]:
+        """
+        Todo manager bound to a specific TodoList instance.
+
+        Actions:
+        - create: Create todo list with tasks
+        - get_current: Get current task to work on
+        - start_task: Mark current task as in progress
+        - complete_task: Complete current task with summary
+        - get_final_summary: Get all completed work
+        """
+        if action == "create":
+            return todo_list.create_tasks(tasks or [])
+        elif action == "get_current":
+            return todo_list.get_current()
+        elif action == "start_task":
+            return todo_list.start_current_task()
+        elif action == "complete_task":
+            return todo_list.complete_current_task(summary or "")
+        elif action == "get_final_summary":
+            return todo_list.get_final_summary()
+        else:
+            raise ValueError(f"Unknown action: {action}")
+
+    return todo_manager
+
+
+def get_todo_instruction_tpl(task_range: tuple[int, int] = (3, 5)) -> str:
+    """Generate system prompt instructions for todo workflow."""
+    min_tasks, max_tasks = task_range
+
+    return f"""
+
+    ## Todo Workflow
+
+    Available actions:
+    - `todo_manager(action="create", tasks=[...])`: Create {min_tasks}-{max_tasks} tasks
+    - `todo_manager(action="get_current")`: Get current task
+    - `todo_manager(action="start_task")`: Start current task
+    - `todo_manager(action="complete_task", summary="...")`: Complete with detailed summary
+    - `todo_manager(action="get_final_summary")`: Get comprehensive final results
+
+    WORKFLOW:
+    1. Create todo list
+    2. For each task: get_current → start_task → [do work] → complete_task
+    3. When done: get_final_summary
+
+    IMPORTANT: Task summaries should be DETAILED and COMPREHENSIVE (3-5 sentences).
+    Include specific information, recommendations, and actionable details.
+    """