|
|
""" |
|
|
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.""" |
|
|
|
|
|
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) |