|
1 | | -import re |
| 1 | +"""CLI-specific visualization configuration. |
2 | 2 |
|
3 | | -from rich.console import Console |
4 | | -from rich.panel import Panel |
5 | | -from rich.text import Text |
| 3 | +This module customizes the SDK's default visualizer for CLI usage by: |
| 4 | +- Skipping SystemPromptEvent (only relevant for SDK internals) |
| 5 | +- Re-exporting DefaultConversationVisualizer for use in CLI |
| 6 | +""" |
6 | 7 |
|
7 | | -from openhands.sdk.conversation.visualizer.base import ( |
8 | | - ConversationVisualizerBase, |
| 8 | +from openhands.sdk.conversation.visualizer.default import ( |
| 9 | + EVENT_VISUALIZATION_CONFIG, |
| 10 | + DefaultConversationVisualizer as CLIVisualizer, |
| 11 | + EventVisualizationConfig, |
9 | 12 | ) |
10 | | -from openhands.sdk.event import ( |
11 | | - ActionEvent, |
12 | | - AgentErrorEvent, |
13 | | - MessageEvent, |
14 | | - ObservationEvent, |
15 | | - PauseEvent, |
16 | | - SystemPromptEvent, |
17 | | - UserRejectObservation, |
18 | | -) |
19 | | -from openhands.sdk.event.base import Event |
20 | | -from openhands.sdk.event.condenser import Condensation |
21 | | - |
22 | | - |
23 | | -# These are external inputs |
24 | | -_OBSERVATION_COLOR = "yellow" |
25 | | -_MESSAGE_USER_COLOR = "gold3" |
26 | | -_PAUSE_COLOR = "bright_yellow" |
27 | | -# These are internal system stuff |
28 | | -_SYSTEM_COLOR = "magenta" |
29 | | -_THOUGHT_COLOR = "bright_black" |
30 | | -_ERROR_COLOR = "red" |
31 | | -# These are agent actions |
32 | | -_ACTION_COLOR = "blue" |
33 | | -_MESSAGE_ASSISTANT_COLOR = _ACTION_COLOR |
34 | | - |
35 | | -DEFAULT_HIGHLIGHT_REGEX = { |
36 | | - r"^Reasoning:": f"bold {_THOUGHT_COLOR}", |
37 | | - r"^Thought:": f"bold {_THOUGHT_COLOR}", |
38 | | - r"^Action:": f"bold {_ACTION_COLOR}", |
39 | | - r"^Arguments:": f"bold {_ACTION_COLOR}", |
40 | | - r"^Tool:": f"bold {_OBSERVATION_COLOR}", |
41 | | - r"^Result:": f"bold {_OBSERVATION_COLOR}", |
42 | | - r"^Rejection Reason:": f"bold {_ERROR_COLOR}", |
43 | | - # Markdown-style |
44 | | - r"\*\*(.*?)\*\*": "bold", |
45 | | - r"\*(.*?)\*": "italic", |
46 | | -} |
47 | | - |
48 | | -_PANEL_PADDING = (1, 1) |
49 | | - |
50 | | - |
51 | | -class CLIVisualizer(ConversationVisualizerBase): |
52 | | - """Handles visualization of conversation events with Rich formatting. |
53 | | -
|
54 | | - Provides Rich-formatted output with panels and complete content display. |
55 | | - """ |
56 | | - |
57 | | - _console: Console |
58 | | - _skip_user_messages: bool |
59 | | - _highlight_patterns: dict[str, str] |
60 | | - |
61 | | - def __init__( |
62 | | - self, |
63 | | - name: str | None = None, |
64 | | - highlight_regex: dict[str, str] | None = DEFAULT_HIGHLIGHT_REGEX, |
65 | | - skip_user_messages: bool = False, |
66 | | - ): |
67 | | - """Initialize the visualizer. |
68 | | -
|
69 | | - Args: |
70 | | - name: Optional name to prefix in panel titles to identify |
71 | | - which agent/conversation is speaking. |
72 | | - highlight_regex: Dictionary mapping regex patterns to Rich color styles |
73 | | - for highlighting keywords in the visualizer. |
74 | | - For example: {"Reasoning:": "bold blue", |
75 | | - "Thought:": "bold green"} |
76 | | - skip_user_messages: If True, skip displaying user messages. Useful for |
77 | | - scenarios where user input is not relevant to show. |
78 | | - """ |
79 | | - super().__init__( |
80 | | - name=name, |
81 | | - ) |
82 | | - self._console = Console() |
83 | | - self._skip_user_messages = skip_user_messages |
84 | | - self._highlight_patterns = highlight_regex or {} |
85 | | - |
86 | | - def on_event(self, event: Event) -> None: |
87 | | - """Main event handler that displays events with Rich formatting.""" |
88 | | - panel = self._create_event_panel(event) |
89 | | - if panel: |
90 | | - self._console.print(panel) |
91 | | - self._console.print() # Add spacing between events |
92 | | - |
93 | | - def _apply_highlighting(self, text: Text) -> Text: |
94 | | - """Apply regex-based highlighting to text content. |
95 | | -
|
96 | | - Args: |
97 | | - text: The Rich Text object to highlight |
98 | | -
|
99 | | - Returns: |
100 | | - A new Text object with highlighting applied |
101 | | - """ |
102 | | - if not self._highlight_patterns: |
103 | | - return text |
104 | | - |
105 | | - # Create a copy to avoid modifying the original |
106 | | - highlighted = text.copy() |
| 13 | +from openhands.sdk.event import SystemPromptEvent |
107 | 14 |
|
108 | | - # Apply each pattern using Rich's built-in highlight_regex method |
109 | | - for pattern, style in self._highlight_patterns.items(): |
110 | | - pattern_compiled = re.compile(pattern, re.MULTILINE) |
111 | | - highlighted.highlight_regex(pattern_compiled, style) |
112 | 15 |
|
113 | | - return highlighted |
114 | | - |
115 | | - def _create_event_panel(self, event: Event) -> Panel | None: |
116 | | - """Create a Rich Panel for the event with appropriate styling.""" |
117 | | - # Use the event's visualize property for content |
118 | | - content = event.visualize |
119 | | - |
120 | | - if not content.plain.strip(): |
121 | | - return None |
122 | | - |
123 | | - # Apply highlighting if configured |
124 | | - if self._highlight_patterns: |
125 | | - content = self._apply_highlighting(content) |
126 | | - |
127 | | - # Don't emit system prompt in CLI |
128 | | - if isinstance(event, SystemPromptEvent): |
129 | | - title = f"[bold {_SYSTEM_COLOR}]" |
130 | | - if self._name: |
131 | | - title += f"{self._name} " |
132 | | - title += f"System Prompt[/bold {_SYSTEM_COLOR}]" |
133 | | - return None |
134 | | - elif isinstance(event, ActionEvent): |
135 | | - # Check if action is None (non-executable) |
136 | | - title = f"[bold {_ACTION_COLOR}]" |
137 | | - if self._name: |
138 | | - title += f"{self._name} " |
139 | | - if event.action is None: |
140 | | - title += f"Agent Action (Not Executed)[/bold {_ACTION_COLOR}]" |
141 | | - else: |
142 | | - title += f"Agent Action[/bold {_ACTION_COLOR}]" |
143 | | - return Panel( |
144 | | - content, |
145 | | - title=title, |
146 | | - subtitle=self._format_metrics_subtitle(), |
147 | | - border_style=_ACTION_COLOR, |
148 | | - padding=_PANEL_PADDING, |
149 | | - expand=True, |
150 | | - ) |
151 | | - elif isinstance(event, ObservationEvent): |
152 | | - title = f"[bold {_OBSERVATION_COLOR}]" |
153 | | - if self._name: |
154 | | - title += f"{self._name} " |
155 | | - title += f"Observation[/bold {_OBSERVATION_COLOR}]" |
156 | | - return Panel( |
157 | | - content, |
158 | | - title=title, |
159 | | - border_style=_OBSERVATION_COLOR, |
160 | | - padding=_PANEL_PADDING, |
161 | | - expand=True, |
162 | | - ) |
163 | | - elif isinstance(event, UserRejectObservation): |
164 | | - title = f"[bold {_ERROR_COLOR}]" |
165 | | - if self._name: |
166 | | - title += f"{self._name} " |
167 | | - title += f"User Rejected Action[/bold {_ERROR_COLOR}]" |
168 | | - return Panel( |
169 | | - content, |
170 | | - title=title, |
171 | | - border_style=_ERROR_COLOR, |
172 | | - padding=_PANEL_PADDING, |
173 | | - expand=True, |
174 | | - ) |
175 | | - elif isinstance(event, MessageEvent): |
176 | | - if ( |
177 | | - self._skip_user_messages |
178 | | - and event.llm_message |
179 | | - and event.llm_message.role == "user" |
180 | | - ): |
181 | | - return |
182 | | - assert event.llm_message is not None |
183 | | - # Role-based styling |
184 | | - role_colors = { |
185 | | - "user": _MESSAGE_USER_COLOR, |
186 | | - "assistant": _MESSAGE_ASSISTANT_COLOR, |
187 | | - } |
188 | | - role_color = role_colors.get(event.llm_message.role, "white") |
189 | | - |
190 | | - # "User Message To [Name] Agent" for user |
191 | | - # "Message from [Name] Agent" for agent |
192 | | - agent_name = f"{self._name} " if self._name else "" |
193 | | - |
194 | | - if event.llm_message.role == "user": |
195 | | - title_text = ( |
196 | | - f"[bold {role_color}]User Message to " |
197 | | - f"{agent_name}Agent[/bold {role_color}]" |
198 | | - ) |
199 | | - else: |
200 | | - title_text = ( |
201 | | - f"[bold {role_color}]Message from " |
202 | | - f"{agent_name}Agent[/bold {role_color}]" |
203 | | - ) |
204 | | - return Panel( |
205 | | - content, |
206 | | - title=title_text, |
207 | | - subtitle=self._format_metrics_subtitle(), |
208 | | - border_style=role_color, |
209 | | - padding=_PANEL_PADDING, |
210 | | - expand=True, |
211 | | - ) |
212 | | - elif isinstance(event, AgentErrorEvent): |
213 | | - title = f"[bold {_ERROR_COLOR}]" |
214 | | - if self._name: |
215 | | - title += f"{self._name} " |
216 | | - title += f"Agent Error[/bold {_ERROR_COLOR}]" |
217 | | - return Panel( |
218 | | - content, |
219 | | - title=title, |
220 | | - subtitle=self._format_metrics_subtitle(), |
221 | | - border_style=_ERROR_COLOR, |
222 | | - padding=_PANEL_PADDING, |
223 | | - expand=True, |
224 | | - ) |
225 | | - elif isinstance(event, PauseEvent): |
226 | | - title = f"[bold {_PAUSE_COLOR}]" |
227 | | - if self._name: |
228 | | - title += f"{self._name} " |
229 | | - title += f"User Paused[/bold {_PAUSE_COLOR}]" |
230 | | - return Panel( |
231 | | - content, |
232 | | - title=title, |
233 | | - border_style=_PAUSE_COLOR, |
234 | | - padding=_PANEL_PADDING, |
235 | | - expand=True, |
236 | | - ) |
237 | | - elif isinstance(event, Condensation): |
238 | | - title = f"[bold {_SYSTEM_COLOR}]" |
239 | | - if self._name: |
240 | | - title += f"{self._name} " |
241 | | - title += f"Condensation[/bold {_SYSTEM_COLOR}]" |
242 | | - return Panel( |
243 | | - content, |
244 | | - title=title, |
245 | | - subtitle=self._format_metrics_subtitle(), |
246 | | - border_style=_SYSTEM_COLOR, |
247 | | - expand=True, |
248 | | - ) |
249 | | - else: |
250 | | - # Fallback panel for unknown event types |
251 | | - title = f"[bold {_ERROR_COLOR}]" |
252 | | - if self._name: |
253 | | - title += f"{self._name} " |
254 | | - title += f"UNKNOWN Event: {event.__class__.__name__}[/bold {_ERROR_COLOR}]" |
255 | | - return Panel( |
256 | | - content, |
257 | | - title=title, |
258 | | - subtitle=f"({event.source})", |
259 | | - border_style=_ERROR_COLOR, |
260 | | - padding=_PANEL_PADDING, |
261 | | - expand=True, |
262 | | - ) |
263 | | - |
264 | | - def _format_metrics_subtitle(self) -> str | None: |
265 | | - """Format LLM metrics as a visually appealing subtitle string with icons, |
266 | | - colors, and k/m abbreviations using conversation stats.""" |
267 | | - stats = self.conversation_stats |
268 | | - if not stats: |
269 | | - return None |
270 | | - |
271 | | - combined_metrics = stats.get_combined_metrics() |
272 | | - if not combined_metrics or not combined_metrics.accumulated_token_usage: |
273 | | - return None |
274 | | - |
275 | | - usage = combined_metrics.accumulated_token_usage |
276 | | - cost = combined_metrics.accumulated_cost or 0.0 |
277 | | - |
278 | | - # helper: 1234 -> "1.2K", 1200000 -> "1.2M" |
279 | | - def abbr(n: int | float) -> str: |
280 | | - n = int(n or 0) |
281 | | - if n >= 1_000_000_000: |
282 | | - val, suffix = n / 1_000_000_000, "B" |
283 | | - elif n >= 1_000_000: |
284 | | - val, suffix = n / 1_000_000, "M" |
285 | | - elif n >= 1_000: |
286 | | - val, suffix = n / 1_000, "K" |
287 | | - else: |
288 | | - return str(n) |
289 | | - return f"{val:.2f}".rstrip("0").rstrip(".") + suffix |
290 | | - |
291 | | - input_tokens = abbr(usage.prompt_tokens or 0) |
292 | | - output_tokens = abbr(usage.completion_tokens or 0) |
293 | | - |
294 | | - # Cache hit rate (prompt + cache) |
295 | | - prompt = usage.prompt_tokens or 0 |
296 | | - cache_read = usage.cache_read_tokens or 0 |
297 | | - cache_rate = f"{(cache_read / prompt * 100):.2f}%" if prompt > 0 else "N/A" |
298 | | - reasoning_tokens = usage.reasoning_tokens or 0 |
299 | | - |
300 | | - # Cost |
301 | | - cost_str = f"{cost:.4f}" if cost > 0 else "0.00" |
302 | | - |
303 | | - # Build with fixed color scheme |
304 | | - parts: list[str] = [] |
305 | | - parts.append(f"[cyan]↑ input {input_tokens}[/cyan]") |
306 | | - parts.append(f"[magenta]cache hit {cache_rate}[/magenta]") |
307 | | - if reasoning_tokens > 0: |
308 | | - parts.append(f"[yellow] reasoning {abbr(reasoning_tokens)}[/yellow]") |
309 | | - parts.append(f"[blue]↓ output {output_tokens}[/blue]") |
310 | | - parts.append(f"[green]$ {cost_str}[/green]") |
| 16 | +# CLI-specific customization: skip SystemPromptEvent |
| 17 | +# (not needed in CLI output, only relevant for SDK internals) |
| 18 | +EVENT_VISUALIZATION_CONFIG[SystemPromptEvent] = EventVisualizationConfig( |
| 19 | + **{**EVENT_VISUALIZATION_CONFIG[SystemPromptEvent].model_dump(), "skip": True} |
| 20 | +) |
311 | 21 |
|
312 | | - return "Tokens: " + " • ".join(parts) |
| 22 | +__all__ = ["CLIVisualizer"] |
0 commit comments