Add LLM Task Manger

bindings/ceylon/ceylon/base/playground.py

@@ -36,6 +36,7 @@ def __init__(self, name="playground", port=8888):
         self.llm_agents: Dict[str, AgentConnectedStatus] = {}
         self._connected_event = None
         self._stop_event = None
+        self._all_tasks_completed_events: Dict[str, asyncio.Event] = {}
         self._completed_tasks: Dict[str, TaskOutput] = {}
         self._task_results: Dict[str, Any] = {}
 
@@ -55,6 +56,11 @@ def get_completed_tasks(self) -> Dict[str, TaskOutput]:
         """Get all completed tasks"""
         return self._completed_tasks.copy()
 
+    async def wait_and_get_completed_tasks(self) -> Dict[str, TaskOutput]:
+        for event in self._all_tasks_completed_events.values():
+            await event.wait()
+        return self.get_completed_tasks()
+
     def get_task_results(self) -> Dict[str, Any]:
         """Get all task results"""
         return self._task_results.copy()
@@ -109,11 +115,12 @@ async def play(self, workers: Optional[List[BaseAgent]] = None):
             for task_id, output in self._completed_tasks.items():
                 if output.completed:
                     duration = output.end_time - output.start_time if output.end_time and output.start_time else None
-                    logger.info(f"Task {task_id} ({output.name}) completed in {duration:.2f}s" if duration else f"Task {task_id} ({output.name}) completed")
+                    logger.info(
+                        f"Task {task_id} ({output.name}) completed in {duration:.2f}s" if duration else f"Task {task_id} ({output.name}) completed")
                 else:
                     logger.warning(f"Task {task_id} ({output.name}) failed: {output.error}")
 
             # Cleanup
             self._connected_event = None
             self._stop_event = None
-            await self.stop()
+            await self.stop()

bindings/ceylon/ceylon/llm/agent.py

@@ -70,10 +70,10 @@ async def execute_task(self, task: TaskMessage) -> None:
         Execute an LLM task with retry logic and error handling
         """
         try:
-            print(f"\n{'='*80}")
-            print(f"Task: {task.name}")
-            print(f"Description: {task.description}")
-            print(f"{'='*80}\n")
+            logger.info(f"\n{'='*80}")
+            logger.info(f"Task: {task.name}")
+            logger.info(f"Description: {task.description}")
+            logger.info(f"{'='*80}\n")
 
             async with self.processing_lock:
                 response = await self._execute_with_retry(task)
@@ -83,10 +83,10 @@ async def execute_task(self, task: TaskMessage) -> None:
                 self.response_cache[task.task_id] = response
 
                 # Print the response
-                print("\nGenerated Content:")
-                print(f"{'-'*80}")
-                print(response.content)
-                print(f"{'-'*80}\n")
+                logger.info("\nGenerated Content:")
+                logger.info(f"{'-'*80}")
+                logger.info(response.content)
+                logger.info(f"{'-'*80}\n")
 
                 # Update task with completion info
                 task.completed = True

bindings/ceylon/ceylon/task/playground.py

@@ -2,6 +2,8 @@
 #  Licensed under the Apache License, Version 2.0 (See LICENSE.md or http://www.apache.org/licenses/LICENSE-2.0).
 # 
 #
+import asyncio
+
 from loguru import logger
 from ceylon import on, on_connect
 from ceylon.base.playground import BasePlayGround, TaskOutput
@@ -34,6 +36,7 @@ async def handle_task_completion(self, task: TaskMessage, time: int):
             if hasattr(task, 'result'):
                 self.add_task_result(task.task_id, task.result)
 
+            self._all_tasks_completed_events[task.task_id].set()
             # Broadcast status update
             await self.broadcast_message(TaskStatusUpdate(
                 task_id=task.task_id,
@@ -97,6 +100,7 @@ async def assign_task_groups(self, groups):
         assignments = await self.task_manager.assign_task_groups(groups)
         for assignment in assignments:
             await self.broadcast_message(assignment)
+            self._all_tasks_completed_events[assignment.task_id] = asyncio.Event()
 
     async def print_all_statistics(self):
         """Print statistics for all task groups"""

bindings/ceylon/examples/llm/app.py

@@ -1,12 +1,14 @@
 import asyncio
 import uuid
-from typing import List, Dict
 from datetime import datetime
+from typing import List, Dict
+
+from ceylon.llm.agent import LLMAgent, LLMConfig
+from ceylon.llm.models.ollama import OllamaModel
 from ceylon.task import TaskPlayGround
-from ceylon.task.data import TaskMessage, TaskGroup, TaskGroupGoal, GoalStatus
+from ceylon.task.data import TaskMessage, TaskGroupGoal, GoalStatus
 from ceylon.task.manager import TaskManager
-from ceylon.llm.models.ollama import OllamaModel
-from ceylon.llm.agent import LLMAgent, LLMConfig
+
 
 def print_header(text: str):
     print(f"\n{'='*80}")

bindings/ceylon/examples/llm/simple_app.py

@@ -70,15 +70,9 @@ async def main():
         async with playground.play(workers=[llm_agent]) as active_playground:
             # Assign task group
             await active_playground.assign_task_groups([task_group])
-
             # Wait for completion
-            while True:
-                await asyncio.sleep(1)
-                if task.task_id in active_playground.get_completed_tasks():
-                    break
-
             # Get and display results
-            completed_task = active_playground.get_completed_tasks()[task.task_id]
+            completed_task = (await active_playground.wait_and_get_completed_tasks())[task.task_id]
             if completed_task.completed:
                 print("\nTask Completed Successfully!")
                 print(f"Duration: {completed_task.end_time - completed_task.start_time:.2f}s")