-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy path_pretty.py
More file actions
145 lines (114 loc) · 5.35 KB
/
Copy path_pretty.py
File metadata and controls
145 lines (114 loc) · 5.35 KB
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
141
142
143
144
145
"""Pretty terminal rendering for the examples, built on `rich`.
Run the examples with the dev extras installed (which include `rich`):
uv sync --all-extras --dev
python examples/01_stream_events.py
Design:
- **One row per event**, in aligned columns: ``icon type detail meta``.
- **Color means status, everywhere**: running = blue, success = green,
error = red. The *icon* carries the event type; the model name is dim cyan.
- Long LLM reasoning is shown as a single truncated line, with an explicit dim
"reasoning hidden" marker so you know there was more.
"""
from __future__ import annotations
try:
from rich.console import Console
from rich.panel import Panel
from rich.rule import Rule
from rich.table import Table
from rich.text import Text
except ModuleNotFoundError as exc: # pragma: no cover - example-only guard
raise SystemExit(
"examples need 'rich' — run `uv sync --all-extras --dev` "
"(or `pip install rich`)."
) from exc
from cominty_sdk import ThreadSummary, events
console = Console()
# Color = status. The same hue always means the same thing.
_STATUS_STYLE = {"running": "blue", "success": "green", "error": "bold red"}
_LABEL_W = 7
_DETAIL_W = 44 # keeps icon+label+detail+meta on one ~80-col line
def _status_style(status: str) -> str:
return _STATUS_STYLE.get(status, "dim")
def _truncate(text: str, width: int) -> str:
text = " ".join(text.split()) # collapse newlines/runs of whitespace
return text if len(text) <= width else text[: width - 1] + "…"
def header(thread_id: object, message_id: object) -> None:
"""Print the thread / message ids that the run is bound to."""
line = Text(" ")
line.append("thread ", style="dim")
line.append(_short(thread_id))
line.append(" · ", style="dim")
line.append("message ", style="dim")
line.append(_short(message_id))
console.print()
console.print(line)
console.print()
def render(event: events.AnyEvent) -> None:
"""Render one streamed event as a single aligned row."""
icon, label, detail, meta, meta_style = _describe(event)
line = Text(" ")
line.append(f"{icon} ")
line.append(f"{label:<{_LABEL_W}}", style="bold")
line.append(" ")
line.append(f"{_truncate(detail, _DETAIL_W):<{_DETAIL_W}}")
if meta:
line.append(" ")
line.append(meta, style=meta_style)
# One row per event: never wrap; crop with … on a too-narrow terminal.
console.print(line, no_wrap=True, crop=True)
# LLM reasoning can be a long dump; we showed one truncated line above — flag
# that the rest is hidden so the stream stays scannable.
if isinstance(event, events.LlmStep):
full = " ".join(event.data.description.split())
if len(full) > _DETAIL_W:
console.print(
Text(f" ↳ reasoning hidden ({len(full)} chars)", style="dim")
)
def panel(text: str, *, title: str, style: str = "cyan") -> None:
"""Print arbitrary text in a bordered, titled panel."""
console.print()
console.print(Panel(text.strip(), title=title, border_style=style,
padding=(1, 2)))
def answer(text: str, *, title: str = "Answer") -> None:
"""Print the final assistant reply in a bordered panel."""
panel(text, title=title, style="green")
def rule(title: str = "") -> None:
console.print(Rule(title, style="dim"))
def thread_table(threads: list[ThreadSummary], *, title: str) -> None:
"""Render a list of thread summaries as a table."""
table = Table(title=title, title_justify="left", header_style="bold",
expand=False)
table.add_column("created", style="dim", no_wrap=True)
table.add_column("★", justify="center", no_wrap=True)
table.add_column("name")
table.add_column("id", style="dim", no_wrap=True)
for t in threads:
table.add_row(
f"{t.created_at:%Y-%m-%d}",
"[yellow]★[/]" if t.starred else "",
t.name,
_short(t.id),
)
console.print(table)
# --------------------------------------------------------------------------- #
# Per-event formatting: (icon, label, detail, meta, meta_style)
# --------------------------------------------------------------------------- #
def _describe(event: events.AnyEvent) -> tuple[str, str, str, str, str]:
status_style = _status_style(event.status)
if isinstance(event, events.ToolCall):
detail = event.data.error or event.data.message or event.data.name
return "🔧", "tool", detail, event.status, status_style
if isinstance(event, events.LlmStep):
return "🧠", "think", event.data.description, event.data.model, "dim cyan"
if isinstance(event, events.Result):
return "💰", "result", f"${event.data.cost.total}", event.status, status_style
if isinstance(event, events.IntermediaryUpdate):
return "📝", "update", event.data.message, event.status, status_style
if isinstance(event, events.UploadingFile):
return "📤", "upload", event.data.filename, event.status, status_style
if isinstance(event, (events.WaitingForStart, events.SettingUpSandbox)):
return "⚙", "setup", event.name, event.status, status_style
return "•", "event", event.name, event.status, status_style
def _short(value: object) -> str:
s = str(value)
return f"{s[:8]}…{s[-4:]}" if len(s) > 14 else s