Spaces:
Running
Running
| """ | |
| Console Display Manager for CodeAct Agent. | |
| Handles all console output and display formatting. | |
| """ | |
| from typing import List, Dict | |
| from rich import box | |
| from rich.console import Console | |
| from rich.panel import Panel | |
| from rich.rule import Rule | |
| from rich.syntax import Syntax | |
| from rich.table import Table | |
| from rich.text import Text | |
| from ..workflow.plan_manager import PlanManager | |
| class ConsoleDisplay: | |
| """Handles all console output and display formatting.""" | |
| def __init__(self): | |
| self.console = Console() | |
| def print_task_header(self, query: str): | |
| """Print the task header panel.""" | |
| self.console.print( | |
| Panel( | |
| f"\n[bold]{query}\n", | |
| title="[bold]Code Agent - New Run", | |
| subtitle="Starting agent execution...", | |
| border_style="yellow", | |
| subtitle_align="left", | |
| ) | |
| ) | |
| def print_tools_info(self, total_tools: int, regular_tools: List[str], mcp_tools: Dict): | |
| """Print information about loaded tools.""" | |
| self.console.print(f"π οΈ Loaded {total_tools} total functions:") | |
| if regular_tools: | |
| self.console.print(f" π Regular tools ({len(regular_tools)}): {regular_tools}") | |
| if mcp_tools: | |
| self.console.print(f" π MCP tools ({len(mcp_tools)}):") | |
| for tool_name, tool_info in mcp_tools.items(): | |
| server_name = tool_info.get('server', 'unknown') | |
| self.console.print(f" β’ {tool_name} (from {server_name} server)") | |
| def print_packages_info(self, imported_packages: List[str], failed_packages: List[tuple]): | |
| """Print information about imported packages.""" | |
| if imported_packages: | |
| self.console.print(f"π¦ Imported {len(imported_packages)} packages: {imported_packages}") | |
| for package_name, error in failed_packages: | |
| self.console.print(f"β οΈ Failed to import {package_name}: {error}") | |
| def print_reasoning(self, content: str, step_count: int): | |
| """Print agent reasoning panel.""" | |
| self.console.print( | |
| Panel( | |
| Text(content, style="italic"), | |
| title=f"[bold]π§ Agent Reasoning - Step {step_count}", | |
| border_style="yellow", | |
| box=box.SIMPLE, | |
| ) | |
| ) | |
| def print_plan(self, plan: str): | |
| """Print current plan panel with progress statistics.""" | |
| # Get plan progress statistics | |
| progress = PlanManager.get_plan_progress(plan) | |
| progress_text = f"({progress['completed']}/{progress['total']} completed)" | |
| self.console.print( | |
| Panel( | |
| Text(plan, style="cyan"), | |
| title=f"[bold]π Current Plan {progress_text}", | |
| border_style="cyan", | |
| box=box.SIMPLE, | |
| ) | |
| ) | |
| def print_code_execution(self, code: str, step_count: int): | |
| """Print code execution panel.""" | |
| self.console.print(f"β‘ Step {step_count}: Executing code...") | |
| self.console.print( | |
| Panel( | |
| Syntax(code, "python", theme="monokai", word_wrap=True), | |
| title=f"[bold]π» Code Execution - Step {step_count}", | |
| title_align="left", | |
| box=box.HORIZONTALS, | |
| ) | |
| ) | |
| def print_solution(self, solution: str, step_count: int): | |
| """Print final solution panel.""" | |
| self.console.print(f"π― Step {step_count}: Providing final solution...") | |
| self.console.print( | |
| Panel( | |
| Text(solution, style="bold green"), | |
| title=f"[bold]β Final Solution - Step {step_count}", | |
| border_style="green", | |
| ) | |
| ) | |
| def print_error(self, error: str, step_count: int): | |
| """Print error panel.""" | |
| self.console.print(f"β Step {step_count}: Error encountered...") | |
| self.console.print( | |
| Panel( | |
| Text(error, style="bold red"), | |
| title=f"[bold]β οΈ Error - Step {step_count}", | |
| border_style="red", | |
| ) | |
| ) | |
| def print_execution_result(self, result: str, step_count: int): | |
| """Print execution result panel.""" | |
| self.console.print( | |
| Panel( | |
| Syntax(result, "text", theme="github-dark", word_wrap=True), | |
| title=f"[bold]π Execution Result - Step {step_count}", | |
| border_style="cyan", | |
| title_align="left", | |
| ) | |
| ) | |
| def print_execution_summary(self, step_count: int, error_count: int, duration: float): | |
| """Print execution summary table.""" | |
| self.console.print(Rule(title="Execution Summary", style="yellow")) | |
| summary_table = Table(show_header=False, box=box.SIMPLE) | |
| summary_table.add_column("Metric", style="cyan") | |
| summary_table.add_column("Value", style="bold") | |
| summary_table.add_row("β Total steps", str(step_count)) | |
| summary_table.add_row("β οΈ Total errors", str(error_count)) | |
| summary_table.add_row("β±οΈ Total execution time", f"{duration:.2f} seconds") | |
| if step_count > 0: | |
| summary_table.add_row("π Average time per step", f"{duration/step_count:.2f}s") | |
| self.console.print(summary_table) |