File size: 5,346 Bytes
f580ad3 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 |
"""
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) |