170 lines
6.9 KiB
Python
170 lines
6.9 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from dataclasses import asdict
|
|
import shlex
|
|
from typing import Any
|
|
|
|
from rich.markup import escape
|
|
from textual.app import App, ComposeResult
|
|
from textual.containers import Horizontal, Vertical
|
|
from textual.widgets import Footer, Header, Input, RichLog, Static
|
|
|
|
from .config import AppConfig
|
|
from .manager import ModelManager
|
|
from .types import BackendError, ChatParams
|
|
|
|
|
|
class LocalPilotTUI(App[None]):
|
|
CSS = """
|
|
Screen { background: $surface; }
|
|
#body { height: 1fr; }
|
|
#sidebar { width: 30; border: round $panel; padding: 1; }
|
|
#main { width: 1fr; padding: 0 1; }
|
|
#chat { height: 1fr; border: round $panel; padding: 1; }
|
|
#status { height: 1; color: $text-muted; padding: 0 1; }
|
|
#prompt { height: 3; border: round $accent; }
|
|
"""
|
|
BINDINGS = [("ctrl+c", "quit", "退出")]
|
|
|
|
def __init__(self, config: AppConfig) -> None:
|
|
super().__init__()
|
|
self.config = config
|
|
self.manager = ModelManager(config)
|
|
self.messages: list[dict[str, str]] = []
|
|
self.params = ChatParams()
|
|
self.current_model = config.runtime.default_model
|
|
|
|
def compose(self) -> ComposeResult:
|
|
yield Header(show_clock=False)
|
|
with Horizontal(id="body"):
|
|
with Vertical(id="sidebar"):
|
|
yield Static("LocalPilot\n\n模型", id="model_title")
|
|
yield Static(id="models")
|
|
yield Static(id="commands")
|
|
with Vertical(id="main"):
|
|
yield RichLog(id="chat", markup=True, wrap=True, highlight=False)
|
|
yield Static("就绪", id="status")
|
|
yield Input(placeholder="输入消息，/help 查看命令", id="prompt")
|
|
yield Footer()
|
|
|
|
async def on_mount(self) -> None:
|
|
self.query_one("#commands", Static).update(
|
|
"/model ID\n/load ID\n/unload\n/params k=v\n/clear\n/help"
|
|
)
|
|
self._refresh_model_panel()
|
|
self.query_one("#prompt", Input).focus()
|
|
|
|
def _refresh_model_panel(self) -> None:
|
|
rows = []
|
|
for item in self.manager.list_models():
|
|
mark = "*" if item["id"] == self.current_model else " "
|
|
state = "已加载" if item.get("loaded") else "未加载"
|
|
rows.append(f"{mark} {item['id']}\n {item['kind']} | {state}")
|
|
self.query_one("#models", Static).update("\n".join(rows) or "没有启用模型")
|
|
|
|
def _set_status(self, text: str) -> None:
|
|
self.query_one("#status", Static).update(text)
|
|
|
|
async def on_input_submitted(self, event: Input.Submitted) -> None:
|
|
value = event.value.strip()
|
|
event.input.value = ""
|
|
if not value:
|
|
return
|
|
if value.startswith("/"):
|
|
self.run_worker(self._command(value), exclusive=True)
|
|
else:
|
|
self.run_worker(self._chat(value), exclusive=True)
|
|
|
|
async def _command(self, raw: str) -> None:
|
|
try:
|
|
parts = shlex.split(raw)
|
|
except ValueError as exc:
|
|
self._set_status(f"命令解析失败: {exc}")
|
|
return
|
|
command = parts[0].lower()
|
|
if command == "/help":
|
|
self.query_one("#chat", RichLog).write(
|
|
"[bold]命令[/bold]\n/model ID 选择模型\n/load ID 加载模型\n/unload 卸载当前模型\n"
|
|
"/params temperature=0.7 top_p=0.95 max_tokens=512\n/clear 清空对话"
|
|
)
|
|
elif command == "/models":
|
|
self._refresh_model_panel()
|
|
elif command == "/model" and len(parts) > 1:
|
|
self.current_model = parts[1]
|
|
self._refresh_model_panel()
|
|
self._set_status(f"当前模型: {self.current_model}")
|
|
elif command == "/load" and len(parts) > 1:
|
|
await self._load_model(parts[1])
|
|
elif command == "/unload":
|
|
if self.current_model:
|
|
await self.manager.unload(self.current_model)
|
|
self._refresh_model_panel()
|
|
self._set_status("模型已卸载")
|
|
elif command == "/params":
|
|
self._update_params(parts[1:])
|
|
elif command == "/clear":
|
|
self.messages.clear()
|
|
self.query_one("#chat", RichLog).clear()
|
|
self._set_status("对话已清空")
|
|
else:
|
|
self._set_status("未知命令，输入 /help")
|
|
|
|
async def _load_model(self, model_id: str) -> None:
|
|
self._set_status(f"加载 {model_id} ...")
|
|
try:
|
|
await self.manager.load(model_id)
|
|
self.current_model = model_id
|
|
self._refresh_model_panel()
|
|
self._set_status(f"已加载: {model_id}")
|
|
except (BackendError, KeyError) as exc:
|
|
self._set_status(str(exc))
|
|
|
|
def _update_params(self, assignments: list[str]) -> None:
|
|
updates: dict[str, Any] = {}
|
|
for assignment in assignments:
|
|
if "=" not in assignment:
|
|
continue
|
|
key, raw_value = assignment.split("=", 1)
|
|
if not hasattr(self.params, key):
|
|
continue
|
|
old = getattr(self.params, key)
|
|
try:
|
|
if isinstance(old, bool) or key == "enable_thinking":
|
|
updates[key] = raw_value.lower() in {"1", "true", "yes", "on"}
|
|
elif isinstance(old, int):
|
|
updates[key] = int(raw_value)
|
|
elif isinstance(old, float):
|
|
updates[key] = float(raw_value)
|
|
else:
|
|
updates[key] = raw_value.split(",") if key == "stop" else raw_value
|
|
except ValueError:
|
|
self._set_status(f"参数无效: {assignment}")
|
|
return
|
|
self.params = ChatParams(**{**asdict(self.params), **updates})
|
|
self._set_status(f"参数已更新: {', '.join(f'{k}={v}' for k, v in updates.items()) or '无变化'}")
|
|
|
|
async def _chat(self, prompt: str) -> None:
|
|
log = self.query_one("#chat", RichLog)
|
|
log.write(f"[bold cyan]你[/bold cyan] {escape(prompt)}")
|
|
self.messages.append({"role": "user", "content": prompt})
|
|
self._set_status(f"生成中 | {self.current_model or 'default'}")
|
|
pieces: list[str] = []
|
|
try:
|
|
async for chunk in self.manager.chat(self.current_model, self.messages, self.params, stream=True):
|
|
if chunk.text:
|
|
pieces.append(chunk.text)
|
|
self._set_status(f"生成中 | {len(''.join(pieces))} 字符")
|
|
answer = "".join(pieces)
|
|
self.messages.append({"role": "assistant", "content": answer})
|
|
log.write(f"[bold green]模型[/bold green] {escape(answer)}")
|
|
self._refresh_model_panel()
|
|
self._set_status("就绪")
|
|
except (BackendError, KeyError) as exc:
|
|
self._set_status(str(exc))
|
|
log.write(f"[bold red]错误[/bold red] {escape(str(exc))}")
|
|
|
|
async def action_quit(self) -> None:
|
|
await self.manager.shutdown()
|
|
self.exit()
|