feat: todo list for agent

examples/agents/todo_tools_example.py

@@ -0,0 +1,66 @@
+"""Example demonstrating the new single tool-based todo functionality."""
+
+import asyncio
+
+from ragbits.agents import Agent, AgentOptions, ToolCallResult, get_todo_instruction_tpl, todo_manager
+from ragbits.core.llms import LiteLLM, ToolCall
+
+
+async def main():
+    """Demonstrate the new single tool-based todo approach with streaming and logging."""
+
+    # 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=[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, todo_manager
 from ragbits.agents.types import QuestionAnswerAgent, QuestionAnswerPromptInput, QuestionAnswerPromptOutput
 
 __all__ = [
@@ -20,4 +21,6 @@
     "QuestionAnswerPromptInput",
     "QuestionAnswerPromptOutput",
     "ToolCallResult",
+    "get_todo_instruction_tpl",
+    "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, todo_manager
+
+__all__ = ["todo_manager", "get_todo_instruction_tpl"]

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

@@ -0,0 +1,199 @@
+"""Todo list management tool for agents."""
+
+import uuid
+from dataclasses import dataclass, field
+from datetime import datetime
+from enum import Enum
+from typing import Any, Literal
+
+
+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
+
+
+# Storage - just one todo list per agent run
+_current_todo: TodoList | None = None
+
+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]:
+    """
+    Simplified todo manager for agent runs.
+
+    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
+    """
+    global _current_todo
+
+    if action == "create":
+        if not tasks:
+            raise ValueError("Tasks required for create action")
+
+        _current_todo = TodoList()
+        for i, desc in enumerate(tasks):
+            task = Task(
+                id=str(uuid.uuid4()),
+                description=desc.strip(),
+                order=i
+            )
+            _current_todo.tasks.append(task)
+
+        return {
+            "action": "create",
+            "tasks": [{"id": t.id, "description": t.description, "order": t.order} for t in _current_todo.tasks],
+            "total_count": len(_current_todo.tasks),
+            "message": f"Created {len(tasks)} tasks"
+        }
+
+    if not _current_todo:
+        raise ValueError("No todo list exists. Create one first.")
+
+    if action == "get_current":
+        current = _current_todo.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"{_current_todo.current_index + 1}/{len(_current_todo.tasks)}",
+            "message": f"Current task: {current.description}"
+        }
+
+    elif action == "start_task":
+        current = _current_todo.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}"
+        }
+
+    elif action == "complete_task":
+        if not summary:
+            raise ValueError("Summary required for complete_task")
+
+        current = _current_todo.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()
+        _current_todo.advance_to_next()
+
+        next_task = _current_todo.get_current_task()
+        completed_count = sum(1 for t in _current_todo.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(_current_todo.tasks)}",
+            "all_completed": next_task is None,
+            "message": f"Completed: {current.description}"
+        }
+
+    elif action == "get_final_summary":
+        completed_tasks = [t for t in _current_todo.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)
+
+        # Clean up after getting final summary
+        _current_todo = None
+
+        return {
+            "action": "get_final_summary",
+            "final_summary": final_summary,
+            "total_completed": len(completed_tasks),
+            "message": f"Final summary with {len(completed_tasks)} completed tasks."
+        }
+
+    else:
+        raise ValueError(f"Unknown action: {action}")
+
+
+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.
+    """