139 lines
4.8 KiB
Python
139 lines
4.8 KiB
Python
"""CLI 入口: 简单 REPL。
|
|
|
|
用法:
|
|
python cli.py chat # 新建一个 session
|
|
python cli.py chat --resume last # 恢复最近一个
|
|
python cli.py chat --resume 20260506_141523
|
|
python cli.py chat --model deepseek_v4.pro
|
|
python cli.py sessions # 列出历史 session
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from rich.console import Console
|
|
from rich.prompt import Prompt
|
|
|
|
from main import build_agent, load_config, resolve_workspace, sessions_dir
|
|
|
|
|
|
@click.group()
|
|
def cli() -> None:
|
|
"""zcbot - 个人任务 agent"""
|
|
|
|
|
|
@cli.command()
|
|
@click.option("--model", default=None, help="模型档案,如 deepseek_v4.flash 或 deepseek_v4.pro")
|
|
@click.option("--workspace", default=None, help="工作目录(存 sessions/)")
|
|
@click.option("--resume", default=None, help="恢复某个 session: 'last' 或 session_id")
|
|
def chat(model: str, workspace: str, resume: str) -> None:
|
|
"""启动交互式 REPL。每次启动默认开新 session,用 --resume 接老的。"""
|
|
console = Console()
|
|
try:
|
|
agent, session, sid = build_agent(
|
|
model_name=model,
|
|
workspace=workspace,
|
|
console=console,
|
|
session_id=resume,
|
|
resume=bool(resume),
|
|
)
|
|
except Exception as e:
|
|
console.print(f"[red]启动失败:[/red] {type(e).__name__}: {e}")
|
|
sys.exit(1)
|
|
|
|
if resume:
|
|
console.print(
|
|
f"[green]恢复 session[/green] [bold]{sid}[/bold] ({len(session.messages)} 条消息) "
|
|
f"model: [bold]{agent.caps.model_id}[/bold]"
|
|
)
|
|
else:
|
|
console.print(
|
|
f"[green]新 session[/green] [bold]{sid}[/bold] "
|
|
f"model: [bold]{agent.caps.model_id}[/bold]"
|
|
)
|
|
console.print("[dim]/exit 退出 /reset 清空当前对话 /new 开一个新 session /id 显示 session id[/dim]\n")
|
|
|
|
while True:
|
|
try:
|
|
user_input = Prompt.ask("[bold blue]you[/bold blue]")
|
|
except (EOFError, KeyboardInterrupt):
|
|
console.print("\n[dim]bye[/dim]")
|
|
break
|
|
|
|
cmd = user_input.strip()
|
|
if cmd in ("/exit", "/quit"):
|
|
break
|
|
if cmd == "/reset":
|
|
session.reset(keep_system=True)
|
|
console.print("[dim]当前 session 已重置(保留 system)[/dim]")
|
|
continue
|
|
if cmd == "/new":
|
|
try:
|
|
agent, session, sid = build_agent(
|
|
model_name=model, workspace=workspace, console=console
|
|
)
|
|
except Exception as e:
|
|
console.print(f"[red]新建失败:[/red] {type(e).__name__}: {e}")
|
|
continue
|
|
console.print(f"[green]新 session[/green] [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]session: {sid} model: {model_disp} cwd: {cwd_disp}[/dim]")
|
|
continue
|
|
if not cmd:
|
|
continue
|
|
|
|
try:
|
|
agent.run(user_input)
|
|
except KeyboardInterrupt:
|
|
console.print("\n[yellow]已中断本轮。下一条输入会继续这个 session。[/yellow]")
|
|
except Exception as e:
|
|
console.print(f"[red]运行错误:[/red] {type(e).__name__}: {e}")
|
|
|
|
|
|
@cli.command()
|
|
@click.option("--workspace", default=None, help="工作目录")
|
|
@click.option("--limit", default=20, help="显示最近 N 个")
|
|
def sessions(workspace: str, limit: int) -> None:
|
|
"""列出已有 session。"""
|
|
cfg = load_config()
|
|
ws = resolve_workspace(workspace, cfg)
|
|
sdir = sessions_dir(ws)
|
|
|
|
items = sorted(sdir.glob("*.json"), reverse=True)[:limit]
|
|
if not items:
|
|
click.echo(f"(no sessions in {sdir})")
|
|
return
|
|
|
|
click.echo(f"{'session id':<18} {'msgs':>4} {'cwd':<32} preview")
|
|
click.echo("-" * 100)
|
|
for p in items:
|
|
try:
|
|
data = json.loads(p.read_text(encoding="utf-8"))
|
|
if isinstance(data, list):
|
|
messages, meta = data, {}
|
|
else:
|
|
messages = data.get("messages", []) or []
|
|
meta = data.get("meta", {}) or {}
|
|
n = len(messages)
|
|
preview = ""
|
|
for m in messages:
|
|
if isinstance(m, dict) and m.get("role") == "user":
|
|
preview = (m.get("content") or "")[:50].replace("\n", " ")
|
|
break
|
|
cwd = meta.get("cwd") or "?"
|
|
if len(cwd) > 32:
|
|
cwd = "..." + cwd[-29:]
|
|
except Exception as e:
|
|
n, preview, cwd = -1, f"[parse error: {e}]", "?"
|
|
click.echo(f"{p.stem:<18} {n:>4} {cwd:<32} {preview}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
cli()
|