| """ |
| Console Display Manager for CodeAct Agent. |
| Handles all console output and display formatting. |
| """ |
|
|
| 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.""" |
| |
| 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) |
|
|