diff --git a/cli.py b/cli.py index 16e2817..5afb9d6 100644 --- a/cli.py +++ b/cli.py @@ -15,11 +15,11 @@ import json import sys import click -from rich.console import Console from rich.prompt import Prompt from rich.table import Table from core.task import TaskState +from core.ui import make_console from main import ( ROOT, build_agent, @@ -43,7 +43,7 @@ def cli() -> None: @click.option("--desc", default="", help="一句话任务描述,便于 tasks 列表识别") def chat(model: str, workspace: str, resume: str, mode: str, desc: str) -> None: """启动交互式 REPL。每次启动默认开新 task,用 --resume 接老的。""" - console = Console() + console = make_console() try: agent, session, sid, task_state, task_dir = build_agent( model_name=model, @@ -55,32 +55,32 @@ def chat(model: str, workspace: str, resume: str, mode: str, desc: str) -> None: description=desc, ) except Exception as e: - console.print(f"[red]启动失败:[/red] {type(e).__name__}: {e}") + console.print(f"[err]启动失败:[/err] {type(e).__name__}: {e}") sys.exit(1) if resume: console.print( - f"[green]恢复 task[/green] [bold]{sid}[/bold] ({len(session.messages)} 条消息) " - f"model: [bold]{agent.caps.model_id}[/bold]" + f"[ok]恢复 task[/ok] [bold]{sid}[/bold] ({len(session.messages)} 条消息) " + f"model: [accent]{agent.caps.model_id}[/accent]" ) else: meta_tail = "" if task_state.mode or task_state.description: meta_tail = f" mode={task_state.mode!r} desc={task_state.description!r}" console.print( - f"[green]新 task[/green] [bold]{sid}[/bold] " - f"model: [bold]{agent.caps.model_id}[/bold]{meta_tail}" + f"[ok]新 task[/ok] [bold]{sid}[/bold] " + f"model: [accent]{agent.caps.model_id}[/accent]{meta_tail}" ) console.print( - "[dim]/exit 退出 /reset 清空对话(保留 task) /new 开新 task /id /status 查看 " - "/done /abandon 改状态 /desc <文本> 设描述[/dim]\n" + "[info]/exit 退出 /reset 清空对话(保留 task) /new 开新 task /id /status 查看 " + "/done /abandon 改状态 /desc <文本> 设描述[/info]\n" ) while True: try: - user_input = Prompt.ask("[bold blue]you[/bold blue]") + user_input = Prompt.ask("[user]you[/user]", console=console) except (EOFError, KeyboardInterrupt): - console.print("\n[dim]bye[/dim]") + console.print("\n[muted]bye[/muted]") break cmd = user_input.strip() @@ -88,7 +88,7 @@ def chat(model: str, workspace: str, resume: str, mode: str, desc: str) -> None: break if cmd == "/reset": session.reset(keep_system=True) - console.print("[dim]当前 task 对话已重置(保留 system 和 state)[/dim]") + console.print("[info]当前 task 对话已重置(保留 system 和 state)[/info]") continue if cmd == "/new": try: @@ -97,39 +97,39 @@ def chat(model: str, workspace: str, resume: str, mode: str, desc: str) -> None: mode=mode, description=desc, ) except Exception as e: - console.print(f"[red]新建失败:[/red] {type(e).__name__}: {e}") + console.print(f"[err]新建失败:[/err] {type(e).__name__}: {e}") continue - console.print(f"[green]新 task[/green] [bold]{sid}[/bold]") + console.print(f"[ok]新 task[/ok] [bold]{sid}[/bold]") continue if cmd == "/id": cwd_disp = session.meta.get("cwd", "?") model_disp = session.meta.get("model", agent.caps.model_id) - console.print(f"[dim]task: {sid} model: {model_disp} cwd: {cwd_disp}[/dim]") + console.print(f"[info]task: {sid} model: {model_disp} cwd: {cwd_disp}[/info]") continue if cmd == "/status": console.print( - f"[dim]task {task_state.task_id} status={task_state.status} " + f"[info]task {task_state.task_id} status={task_state.status} " f"mode={task_state.mode!r} desc={task_state.description!r}\n" f" model={task_state.model} tokens={task_state.tokens_total} " f"(p={task_state.tokens_prompt}/c={task_state.tokens_completion}) " - f"created={task_state.created_at} updated={task_state.updated_at}[/dim]" + f"created={task_state.created_at} updated={task_state.updated_at}[/info]" ) continue if cmd == "/done": task_state.status = "completed" task_state.save(task_dir) - console.print(f"[green]task {sid} marked completed[/green]") + console.print(f"[ok]task {sid} marked completed[/ok]") break if cmd == "/abandon": task_state.status = "abandoned" task_state.save(task_dir) - console.print(f"[yellow]task {sid} marked abandoned[/yellow]") + console.print(f"[warn]task {sid} marked abandoned[/warn]") break if cmd.startswith("/desc"): new_desc = cmd[len("/desc"):].strip() task_state.description = new_desc task_state.save(task_dir) - console.print(f"[dim]description set: {new_desc!r}[/dim]") + console.print(f"[info]description set: {new_desc!r}[/info]") continue if not cmd: continue @@ -137,9 +137,9 @@ def chat(model: str, workspace: str, resume: str, mode: str, desc: str) -> None: try: agent.run(user_input) except KeyboardInterrupt: - console.print("\n[yellow]已中断本轮。下一条输入会继续这个 task。[/yellow]") + console.print("\n[warn]已中断本轮。下一条输入会继续这个 task。[/warn]") except Exception as e: - console.print(f"[red]运行错误:[/red] {type(e).__name__}: {e}") + console.print(f"[err]运行错误:[/err] {type(e).__name__}: {e}") finally: sync_task_tokens(task_state, task_dir, agent.llm) @@ -189,12 +189,12 @@ def tasks(workspace: str, limit: int, status: str) -> None: tbl.add_column("msgs", justify="right") tbl.add_column("tokens", justify="right") tbl.add_column("desc") - sc = {"active": "cyan", "completed": "green", "abandoned": "dim"} + sc = {"active": "status.active", "completed": "status.completed", "abandoned": "status.abandoned"} for _, tid, st, mode, model, tok, n, desc in rows: - c = sc.get(st, "white") + c = sc.get(st, "info") d_show = desc if len(desc) <= 50 else desc[:47] + "..." tbl.add_row(tid, f"[{c}]{st}[/{c}]", mode, model, str(n), str(tok), d_show) - Console().print(tbl) + make_console().print(tbl) @cli.command() @@ -209,25 +209,25 @@ def probe(model: str, long_context: bool) -> None: cfg = load_config() name = model or cfg["default_model"] - console = Console() + console = make_console() try: caps = ModelCapabilities.load(name, ROOT / cfg["models_dir"]) except Exception as e: - console.print(f"[red]档案加载失败:[/red] {type(e).__name__}: {e}") + console.print(f"[err]档案加载失败:[/err] {type(e).__name__}: {e}") sys.exit(1) console.print( - f"[bold]probing[/bold] [cyan]{caps.model_id}[/cyan] (profile: {name}) " - f"[dim]long-context={long_context}[/dim]\n" + f"[bold]probing[/bold] [accent]{caps.model_id}[/accent] (profile: {name}) " + f"[muted]long-context={long_context}[/muted]\n" ) try: llm = LLM(caps) except Exception as e: - console.print(f"[red]LLM 构造失败:[/red] {type(e).__name__}: {e}") + console.print(f"[err]LLM 构造失败:[/err] {type(e).__name__}: {e}") sys.exit(1) - with console.status("[dim]running probes...[/dim]", spinner="dots"): + with console.status("[muted]running probes...[/muted]", spinner="dots"): report = probe_capabilities(caps, llm, include_long_context=long_context) tbl = Table(show_lines=False) @@ -236,9 +236,9 @@ def probe(model: str, long_context: bool) -> None: tbl.add_column("observed") tbl.add_column("status") tbl.add_column("detail") - color = {"ok": "green", "mismatch": "yellow", "error": "red", "skip": "dim"} + color = {"ok": "ok", "mismatch": "warn", "error": "err", "skip": "muted"} for r in report.results: - c = color.get(r.status, "white") + c = color.get(r.status, "info") tbl.add_row( r.name, str(r.declared), @@ -250,14 +250,14 @@ def probe(model: str, long_context: bool) -> None: if report.has_mismatch: console.print( - "\n[yellow]存在能力对账差异 —— 看 detail,必要时改 " - f"config/models/{caps.family}.yaml[/yellow]" + "\n[warn]存在能力对账差异 —— 看 detail,必要时改 " + f"config/models/{caps.family}.yaml[/warn]" ) sys.exit(2) if any(r.status == "error" for r in report.results): - console.print("\n[red]部分探测出错(见 detail)[/red]") + console.print("\n[err]部分探测出错(见 detail)[/err]") sys.exit(3) - console.print("\n[green]全部能力声明与实测一致。[/green]") + console.print("\n[ok]全部能力声明与实测一致。[/ok]") if __name__ == "__main__": diff --git a/core/loop.py b/core/loop.py index 57de9d1..392c33a 100644 --- a/core/loop.py +++ b/core/loop.py @@ -13,6 +13,7 @@ from rich.markdown import Markdown from .capabilities import ModelCapabilities from .llm import LLM from .session import Session +from .ui import make_console def _extract_usage(usage: Any) -> Tuple[int, int]: @@ -43,7 +44,7 @@ class AgentLoop: self.session = session self.caps = capabilities self.max_iterations = max_iterations or capabilities.max_iterations - self.console = console or Console() + self.console = console or make_console() @contextmanager def _thinking(self): @@ -55,7 +56,7 @@ class AgentLoop: elapsed = time.monotonic() - start total = self.llm.token_counter.total tail = f" ctx {total:,} tok" if total else "" - return f"[dim]thinking... {elapsed:.1f}s{tail}[/dim]" + return f"[muted]thinking... {elapsed:.1f}s{tail}[/muted]" class Ctx: elapsed: float = 0.0 @@ -95,13 +96,13 @@ class AgentLoop: pt, ct = _extract_usage(getattr(response, "usage", None)) self.console.print( - f"[dim][in {pt:,} out {ct:,} t {t.elapsed:.1f}s][/dim]" + f"[info][in {pt:,} out {ct:,} t {t.elapsed:.1f}s][/info]" ) tool_calls = getattr(msg, "tool_calls", None) or [] content = getattr(msg, "content", None) if content: - self.console.print("[cyan]assistant>[/cyan]") + self.console.print("[assistant]assistant>[/assistant]") self.console.print(Markdown(content)) if not tool_calls: @@ -130,7 +131,7 @@ class AgentLoop: preview = json.dumps(args, ensure_ascii=False) if len(preview) > 200: preview = preview[:200] + "..." - self.console.print(f"[yellow]tool>[/yellow] {name}({preview})") + self.console.print(f"[tool]tool>[/tool] {name}({preview})") tool = self.tools.get(name) if tool is None: @@ -153,5 +154,5 @@ class AgentLoop: # 给用户预览(截短) preview = result if len(result) < 400 else result[:400] + "..." - self.console.print(f"[dim]{preview}[/dim]") + self.console.print(f"[muted]{preview}[/muted]") return result diff --git a/core/ui.py b/core/ui.py new file mode 100644 index 0000000..2465217 --- /dev/null +++ b/core/ui.py @@ -0,0 +1,38 @@ +"""Rich 控制台主题: 统一 dark 终端友好的配色。 + +设计目标: 在黑底终端上,所有文字 readable,且语义有层次。 +- 弱化用 grey50/grey70 而不是 dim —— 一些终端把 dim 渲染得过暗。 +- 强调色一律走 bright_* —— 8 色版的 blue/cyan 在 Windows 深底太黯。 +- 用语义样式名(user / assistant / tool / ok / warn / err / info / muted), + 避免在调用处硬编码颜色,以后改主题只动这一处。 +""" +from __future__ import annotations + +from rich.console import Console +from rich.theme import Theme + +ZCBOT_THEME = Theme( + { + # 通用层次 + "info": "grey70", # 普通信息(系统反馈、路径、命令提示) + "muted": "grey50", # 真要弱化的次要内容(预览、spinner 文本) + "accent": "bright_cyan", # 强调(模型名等) + # 角色 + "user": "bold bright_magenta", + "assistant": "bold bright_cyan", + "tool": "bold bright_yellow", + # 状态 + "ok": "bright_green", + "warn": "bright_yellow", + "err": "bold bright_red", + # task 状态列 + "status.active": "bright_cyan", + "status.completed": "bright_green", + "status.abandoned": "grey50", + } +) + + +def make_console() -> Console: + """创建带 zcbot 主题的 Console。所有 TUI 入口都应通过这个工厂拿 console。""" + return Console(theme=ZCBOT_THEME) diff --git a/main.py b/main.py index 7c80927..fb34ba3 100644 --- a/main.py +++ b/main.py @@ -103,10 +103,10 @@ def build_agent( saved_cwd = session.meta.get("cwd") if saved_cwd and console is not None and saved_cwd != str(tool_base): console.print( - f"[yellow]提示:[/yellow] 当前 cwd 与 task 记录不同 —— " + f"[warn]提示:[/warn] 当前 cwd 与 task 记录不同 —— " f"工具基于 current cwd,不会自动切回。\n" - f" task cwd: [dim]{saved_cwd}[/dim]\n" - f" current cwd: [dim]{tool_base}[/dim]" + f" task cwd: [info]{saved_cwd}[/info]\n" + f" current cwd: [info]{tool_base}[/info]" ) task_state = TaskState.load(task_dir) if task_state is None: