core/ui: 抽出语义化 console 主题, 调用方去硬编码颜色
新增 core/ui.py 集中定义 Rich Theme: - 语义样式名: user / assistant / tool / ok / warn / err / info / muted / accent - 在黑底终端上 readable, 弱化用 grey 而非 dim, 强调走 bright_* - make_console() 统一应用主题, 以后改主题只动这一处 cli.py / main.py / core/loop.py 把内联的 [red] [green] [blue] [yellow] [cyan] [dim] 等替换为语义样式; 调用 make_console() 取代 Console()。
This commit is contained in:
parent
647d92f532
commit
72d2b64c40
74
cli.py
74
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__":
|
||||
|
|
|
|||
13
core/loop.py
13
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
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
6
main.py
6
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:
|
||||
|
|
|
|||
Loading…
Reference in New Issue