diff --git a/BotOps/README.md b/BotOps/README.md new file mode 100644 index 0000000..ff6303f --- /dev/null +++ b/BotOps/README.md @@ -0,0 +1,46 @@ +# BotOps — aibox 智能体 TUI 前端 + +复刻并增强 [aibox/client.py](../aibox/client.py),基于 [Textual](https://textual.textualize.io/) 的现代化终端 UI,连接 aibox 微服务(WebSocket,默认 `ws://localhost:8765`)。 + +## 相比 client.py 的增强 + +| 能力 | client.py | BotOps | +|------|-----------|--------| +| AI 回复渲染 | 纯文本流式 / 一次性静态渲染 | **Markdown 增量流式渲染**(无闪烁) | +| 终端管理 | 手搓光标/提示符 | Textual 部件框架(多面板布局) | +| 任务事件 | 混在对话流里 | **独立侧栏**(task_process / task_status / task_error) | +| 自动重连 | ✅(已移植) | ✅ 指数退避 2s→30s,`/disconnect` 抑制 | +| 快捷键 | Ctrl+C 中断 | Ctrl+C 中断 / Ctrl+Q 退出 / Ctrl+L 清屏 | + +## 安装与运行 + +```bash +cd BotOps +python3 -m venv venv && source venv/bin/activate +pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple +python app.py # 默认 ws://localhost:8765 +python app.py ws://your-host:8765 # 指定 aibox 地址 +``` + +## 命令 + +- 本地命令(不发给服务器):`/connect` `/disconnect` `/interrupt` `/clear` `/status` `/help` `/quit` +- 其余 `/` 命令透传给 aibox 服务器(与 client.py 行为一致) + +## 结构 + +``` +BotOps/ +├── app.py # Textual App:布局 + 消息路由 + 快捷键 +├── ws_client.py # WebSocket 协议层(自动重连) +├── commands.py # 斜杠命令注册表 +└── widgets/ + ├── chat_view.py # 聊天区:AI 回复 Markdown 增量流式渲染 + └── status_bar.py # 底部状态栏(连接状态) +``` + +## 协议说明 + +与 aibox 服务端协议完全兼容(详见 aibox/client.py 注释): +- 发送:`{"type":"message","content":...}` / `{"type":"command","command":"/..."}` / `{"type":"interrupt",...}` +- 接收:`chunk`(**累计全文**,增量由前端计算)/ `complete` / `dialogue_ended` / `task_process` / `task_status` / `task_error` / `interrupt_ack` / `error` / `info` / `command_result` / `connection_ack` / `notification` diff --git a/BotOps/__pycache__/app.cpython-312.pyc b/BotOps/__pycache__/app.cpython-312.pyc new file mode 100644 index 0000000..dca1f4d Binary files /dev/null and b/BotOps/__pycache__/app.cpython-312.pyc differ diff --git a/BotOps/__pycache__/commands.cpython-312.pyc b/BotOps/__pycache__/commands.cpython-312.pyc new file mode 100644 index 0000000..bf2f733 Binary files /dev/null and b/BotOps/__pycache__/commands.cpython-312.pyc differ diff --git a/BotOps/__pycache__/ws_client.cpython-312.pyc b/BotOps/__pycache__/ws_client.cpython-312.pyc new file mode 100644 index 0000000..14441e9 Binary files /dev/null and b/BotOps/__pycache__/ws_client.cpython-312.pyc differ diff --git a/BotOps/app.py b/BotOps/app.py new file mode 100644 index 0000000..62fdf10 --- /dev/null +++ b/BotOps/app.py @@ -0,0 +1,230 @@ +#!/usr/bin/env python3 +"""BotOps — aibox 智能体的现代化 TUI 前端(Textual)。 + +复刻并增强 aibox/client.py: + - AI 回复流式 Markdown 渲染(Markdown 部件增量追加,无闪烁) + - 断线指数退避自动重连(/disconnect 抑制) + - 任务事件独立侧栏(task_process / task_status / task_error) + - Ctrl+C 中断生成、Ctrl+Q 退出、Ctrl+L 清屏 + +用法: + python app.py [ws://host:port] # 默认 ws://localhost:8765 +""" +import sys + +from textual.app import App, ComposeResult +from textual.binding import Binding +from textual.containers import Horizontal +from textual.widgets import Footer, Header, Input, RichLog + +from commands import CommandRegistry +from widgets.chat_view import ChatView +from widgets.status_bar import StatusBar +from ws_client import BotWSClient + +DEFAULT_URI = "ws://localhost:8765" + + +class BotOpsApp(App): + TITLE = "BotOps" + SUB_TITLE = "aibox TUI" + + CSS = """ + #chat { + width: 3fr; + border: solid $primary; + } + #tasks { + width: 1fr; + border: solid $secondary; + } + .msg-user { + color: cyan; + padding: 0 1; + margin-top: 1; + } + .msg-ai { + padding: 0 1; + } + .msg-system { + color: $text-muted; + padding: 0 1; + } + .msg-error { + color: red; + padding: 0 1; + } + .msg-task { + color: yellow; + padding: 0 1; + } + #status { + height: 1; + } + """ + + BINDINGS = [ + Binding("ctrl+c", "interrupt", "中断", show=True), + Binding("ctrl+q", "quit", "退出", show=True), + Binding("ctrl+l", "clear", "清屏", show=True), + ] + + def __init__(self, uri: str = DEFAULT_URI): + super().__init__() + self.ws = BotWSClient(uri) + self.ws.on_message = self._on_ws_message + self.ws.on_state = self._on_ws_state + self.registry = CommandRegistry(self) + + # ── 布局 ───────────────────────────────────────────────────────────── + def compose(self) -> ComposeResult: + yield Header() + with Horizontal(): + yield ChatView(id="chat") + yield RichLog(id="tasks", max_lines=1000) + yield Input(placeholder="输入消息,/help 查看命令,Enter 发送", id="input") + yield StatusBar(id="status") + yield Footer() + + async def on_mount(self): + self.query_one("#input", Input).focus() + self._refresh_status("") + self.run_worker(self.connect_to(self.ws.uri)) + + @property + def chat(self) -> ChatView: + return self.query_one("#chat", ChatView) + + @property + def tasks(self) -> RichLog: + return self.query_one("#tasks", RichLog) + + # ── 连接管理 ────────────────────────────────────────────────────────── + async def connect_to(self, uri: str): + if self.ws.connected: + await self.ws.disconnect() + self.ws.uri = uri + await self.notify_system(f"正在连接 {uri} …") + await self.ws.connect() + + async def disconnect(self): + await self.ws.disconnect() + + async def send_interrupt(self): + if await self.ws.send_interrupt(): + await self.notify_system("🛑 已发送中断信号") + else: + await self.notify_system("未连接到服务器", level="error") + + # ── UI 操作 ─────────────────────────────────────────────────────────── + async def notify_system(self, text: str, level: str = "info"): + css = {"info": "msg-system", "error": "msg-error", "task": "msg-task"}.get(level, "msg-system") + await self.chat.add_system(text, css) + + async def clear_chat(self): + await self.chat.reset() + + def _refresh_status(self, extra: str): + self.query_one("#status", StatusBar).set_state( + self.ws.connected, self.ws.uri if self.ws.connected else "", extra) + + # ── 输入处理 ────────────────────────────────────────────────────────── + async def on_input_submitted(self, event: Input.Submitted): + text = event.value.strip() + event.input.value = "" + if not text: + return + if text.startswith("/") and await self.registry.execute(text): + return + if not self.ws.connected: + await self.notify_system("未连接到服务器,先 /connect", level="error") + return + await self.chat.add_user(text) + await self.chat.start_ai() # 预建 AI 消息框,即时反馈 + await self.ws.send_message(text) + + # ── 快捷键 ──────────────────────────────────────────────────────────── + async def action_interrupt(self): + await self.send_interrupt() + + async def action_clear(self): + await self.clear_chat() + + def action_quit(self): + self.exit() + + # ── aibox 协议消息路由(对应 client.py 的 _handle_message)───────────── + async def _on_ws_message(self, data: dict): + t = data.get("type", "") + content = data.get("content", "") + + if t == "chunk": + if content.startswith("🔧 调用工具"): + await self.notify_system(content, level="task") + elif content.startswith(("✅ 对话记忆已保存", "💾 保存记忆", "❌", "⚠️")): + await self.notify_system(content, level="task") + else: + await self.chat.append_ai(content) + elif t == "complete": + await self.chat.finish_ai(content or self.chat.ai_text) + elif t == "dialogue_ended": + await self.chat.finish_ai(self.chat.ai_text) + if content: + await self.notify_system(f"✅ {content}") + elif t == "task_process": + sub = data.get("msg_type", "") + title = data.get("task_title", "未知任务") + if sub == "chunk": + tail = str(data.get("content", ""))[-200:] + if tail: + self.tasks.write(f"⚡ {title}: {tail}") + elif sub == "complete": + self.tasks.write(f"✅ 任务完成: {title}") + elif t == "task_status": + icon = {"executing": "⚙️", "complete": "✅", "failed": "❌", + "added": "➕", "batch_start": "🚀", "batch_complete": "🏁" + }.get(data.get("status", ""), "📋") + self.tasks.write( + f"{icon} [{data.get('task_id', '?')}] {data.get('title', '')}: " + f"{data.get('message', content)}") + elif t == "task_error": + self.tasks.write( + f"❌ [{data.get('task_id', '?')}] {data.get('task_title', '?')}: " + f"{data.get('error', '未知错误')}") + elif t == "interrupt_ack": + await self.notify_system(f"🛑 {content}") + elif t == "error": + await self.notify_system(f"❌ {content}", level="error") + elif t == "info": + await self.notify_system(f"ℹ️ {content}") + elif t == "command_result": + await self.notify_system(content) + elif t == "connection_ack": + await self.notify_system(f"✅ {content}") + elif t == "notification": + await self.notify_system(f"🔔 {content}") + elif content: + await self.chat.append_ai(content) + + # ── 连接状态回调 ────────────────────────────────────────────────────── + async def _on_ws_state(self, state: str, detail: str): + if state == "connected": + self._refresh_status("") + await self.notify_system(f"✅ 已连接到 {self.ws.uri}") + elif state == "disconnected": + self._refresh_status("") + await self.notify_system(f"❌ {detail}") + elif state == "reconnecting": + self._refresh_status(detail) + elif state == "error": + self._refresh_status("") + await self.notify_system(detail, level="error") + + +def main(): + uri = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_URI + BotOpsApp(uri).run() + + +if __name__ == "__main__": + main() diff --git a/BotOps/commands.py b/BotOps/commands.py new file mode 100644 index 0000000..02599d0 --- /dev/null +++ b/BotOps/commands.py @@ -0,0 +1,62 @@ +"""斜杠命令注册表:/connect /disconnect /interrupt /clear /status /help /quit。 + +本地命令在此拦截处理;未注册的 / 命令透传给 aibox 服务器(与 client.py 行为一致)。 +""" +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from app import BotOpsApp + + +class CommandRegistry: + def __init__(self, app: "BotOpsApp"): + self.app = app + self._commands = { + "/connect": ("连接服务器:/connect ws://host:port", self._connect), + "/disconnect": ("断开连接(不自动重连)", self._disconnect), + "/interrupt": ("中断当前生成(或按 Ctrl+C)", self._interrupt), + "/clear": ("清空聊天区", self._clear), + "/status": ("显示连接状态", self._status), + "/help": ("显示帮助", self._help), + "/quit": ("退出 BotOps", self._quit), + } + + async def execute(self, line: str) -> bool: + """是本地命令则执行并返回 True;否则返回 False(调用方应透传服务器)。""" + cmd = line.split()[0].lower() + entry = self._commands.get(cmd) + if entry is None: + return False + await entry[1](line) + return True + + async def _connect(self, line: str): + parts = line.split() + if len(parts) < 2: + await self.app.notify_system("用法: /connect ws://host:port", level="error") + return + await self.app.connect_to(parts[1]) + + async def _disconnect(self, _line: str): + await self.app.disconnect() + + async def _interrupt(self, _line: str): + await self.app.send_interrupt() + + async def _clear(self, _line: str): + await self.app.clear_chat() + + async def _status(self, _line: str): + if self.app.ws.connected: + await self.app.notify_system(f"✅ 已连接: {self.app.ws.uri}") + else: + await self.app.notify_system("○ 未连接") + + async def _help(self, _line: str): + lines = ["可用命令:"] + [f" {k:<14} {v[0]}" for k, v in self._commands.items()] + lines.append(" 其余 / 命令将透传给 aibox 服务器") + await self.app.notify_system("\n".join(lines)) + + async def _quit(self, _line: str): + await self.app.ws.disconnect() + self.app.exit() diff --git a/BotOps/requirements.txt b/BotOps/requirements.txt new file mode 100644 index 0000000..f89aaf5 --- /dev/null +++ b/BotOps/requirements.txt @@ -0,0 +1,2 @@ +textual>=0.60.0 +websockets>=11.0 diff --git a/BotOps/widgets/__init__.py b/BotOps/widgets/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/BotOps/widgets/__pycache__/__init__.cpython-312.pyc b/BotOps/widgets/__pycache__/__init__.cpython-312.pyc new file mode 100644 index 0000000..5d10886 Binary files /dev/null and b/BotOps/widgets/__pycache__/__init__.cpython-312.pyc differ diff --git a/BotOps/widgets/__pycache__/chat_view.cpython-312.pyc b/BotOps/widgets/__pycache__/chat_view.cpython-312.pyc new file mode 100644 index 0000000..b5501fc Binary files /dev/null and b/BotOps/widgets/__pycache__/chat_view.cpython-312.pyc differ diff --git a/BotOps/widgets/__pycache__/status_bar.cpython-312.pyc b/BotOps/widgets/__pycache__/status_bar.cpython-312.pyc new file mode 100644 index 0000000..2b41b65 Binary files /dev/null and b/BotOps/widgets/__pycache__/status_bar.cpython-312.pyc differ diff --git a/BotOps/widgets/chat_view.py b/BotOps/widgets/chat_view.py new file mode 100644 index 0000000..59b8517 --- /dev/null +++ b/BotOps/widgets/chat_view.py @@ -0,0 +1,69 @@ +"""聊天视图:用户/AI/系统消息 + AI 回复流式 Markdown 渲染。""" +from textual.containers import VerticalScroll +from textual.widgets import Markdown, Static + + +def _delta(prev: str, new: str) -> str: + """计算累计文本的增量(服务端 chunk 为累计全文;前缀失配时取最长公共前缀后的部分)。""" + if new.startswith(prev): + return new[len(prev):] + i = 0 + for a, b in zip(prev, new): + if a != b: + break + i += 1 + return new[i:] + + +class ChatView(VerticalScroll): + """可滚动聊天区。AI 回复用单个 Markdown 部件增量 append,实现流式渲染。""" + + def __init__(self, **kwargs): + super().__init__(**kwargs) + self._ai_widget: Markdown | None = None + self._ai_prev = "" + + @property + def ai_text(self) -> str: + return self._ai_prev + + async def add_user(self, text: str): + await self.mount(Static(f"👤 你: {text}", classes="msg-user", markup=False)) + + async def add_system(self, text: str, css_class: str = "msg-system"): + await self.mount(Static(text, classes=css_class, markup=False)) + + async def start_ai(self): + """预建一条 AI 消息(发送用户消息后即可调用,给出即时反馈)。""" + if self._ai_widget is not None: + return + self._ai_prev = "" + self._ai_widget = Markdown("", classes="msg-ai") + await self.mount(self._ai_widget) + self.scroll_end(animate=False) + + async def append_ai(self, full_text: str): + """以累计全文增量渲染(服务端 chunk 为累计文本)。""" + if self._ai_widget is None: + await self.start_ai() + d = _delta(self._ai_prev, full_text) + if d: + self._ai_widget.append(d) + self._ai_prev = full_text + self.scroll_end(animate=False) + + async def finish_ai(self, full_text: str = ""): + """收尾:用最终全文对齐一次(防中途丢帧),随后关闭本条 AI 消息。""" + if self._ai_widget is None and full_text: + await self.start_ai() + if full_text and full_text != self._ai_prev: + await self.append_ai(full_text) + self._ai_widget = None + self._ai_prev = "" + self.scroll_end(animate=False) + + async def reset(self): + """清空聊天区(/clear)。""" + self._ai_widget = None + self._ai_prev = "" + await self.remove_children() diff --git a/BotOps/widgets/status_bar.py b/BotOps/widgets/status_bar.py new file mode 100644 index 0000000..04ddd2a --- /dev/null +++ b/BotOps/widgets/status_bar.py @@ -0,0 +1,14 @@ +"""底部状态栏:连接状态 + 服务器地址 + 附加信息。""" +from textual.widgets import Static + + +class StatusBar(Static): + def set_state(self, connected: bool, uri: str = "", extra: str = ""): + dot = "●" if connected else "○" + color = "green" if connected else "red" + text = f"[{color}]{dot}[/] {'已连接' if connected else '未连接'}" + if uri: + text += f" {uri}" + if extra: + text += f" | {extra}" + self.update(text) diff --git a/BotOps/ws_client.py b/BotOps/ws_client.py new file mode 100644 index 0000000..01b3d83 --- /dev/null +++ b/BotOps/ws_client.py @@ -0,0 +1,146 @@ +"""BotOps WebSocket 协议层(提炼自 aibox/client.py,增加状态回调与自动重连)。 + +协议:JSON 消息。 + 发送:{"type": "message", "content": ...} + {"type": "command", "command": "/..."} + {"type": "interrupt", "content": ...} + 接收:chunk / complete / dialogue_ended / task_process / task_status / task_error / + interrupt_ack / error / info / command_result / connection_ack / notification + +说明:服务端 chunk 为**累计全文**(与 aibox/client.py 前缀匹配逻辑一致)。 +""" +import asyncio +import json +from typing import Awaitable, Callable, Optional + +import websockets + +MessageHandler = Callable[[dict], Awaitable[None]] +StateHandler = Callable[[str, str], Awaitable[None]] # (state, detail) + + +class BotWSClient: + """aibox 协议 WebSocket 客户端:断线指数退避自动重连,手动断开则抑制。""" + + def __init__(self, uri: str): + self.uri = uri + self.ws = None + self.connected = False + self._manual_disconnect = False + self._listen_task: Optional[asyncio.Task] = None + self._reconnect_task: Optional[asyncio.Task] = None + self.on_message: Optional[MessageHandler] = None + self.on_state: Optional[StateHandler] = None + + async def _emit_state(self, state: str, detail: str = ""): + if self.on_state: + await self.on_state(state, detail) + + async def connect(self) -> bool: + """连接并开始监听;成功返回 True。""" + await self._close_ws() + try: + self.ws = await websockets.connect(self.uri) + except Exception as e: + await self._emit_state("error", f"连接失败: {e}") + return False + self.connected = True + self._manual_disconnect = False + self._listen_task = asyncio.create_task(self._listen()) + await self._emit_state("connected", self.uri) + return True + + async def disconnect(self): + """手动断开(抑制自动重连)。""" + self._manual_disconnect = True + if self._reconnect_task and not self._reconnect_task.done(): + self._reconnect_task.cancel() + self.connected = False + await self._close_ws() + if self._listen_task and not self._listen_task.done(): + self._listen_task.cancel() + await self._emit_state("disconnected", "已手动断开") + + async def _close_ws(self): + if self.ws is not None: + try: + await self.ws.close() + except Exception: + pass + self.ws = None + + async def send_message(self, text: str) -> bool: + """发送用户消息(/ 开头按 command 发送,与 aibox 协议一致)。""" + if not self.connected or not self.ws: + return False + data = ({"type": "command", "command": text} if text.startswith("/") + else {"type": "message", "content": text}) + try: + await self.ws.send(json.dumps(data, ensure_ascii=False)) + return True + except Exception as e: + await self._emit_state("error", f"发送失败: {e}") + return False + + async def send_interrupt(self) -> bool: + """发送中断信号(对应 client.py 的 Ctrl+C 行为)。""" + if not self.connected or not self.ws: + return False + try: + await self.ws.send(json.dumps( + {"type": "interrupt", "content": "用户按下了 Ctrl+C 中断当前操作"}, + ensure_ascii=False)) + return True + except Exception: + return False + + async def _listen(self): + """接收循环:断线时在 finally 中触发自动重连。""" + try: + async for raw in self.ws: + try: + data = json.loads(raw) + except json.JSONDecodeError: + data = {"type": "chunk", "content": raw} + if self.on_message: + await self.on_message(data) + except websockets.exceptions.ConnectionClosed: + pass + except asyncio.CancelledError: + raise + except Exception as e: + await self._emit_state("error", f"接收错误: {e}") + finally: + was_connected = self.connected + self.connected = False + if was_connected: + await self._emit_state("disconnected", "连接已断开") + self._schedule_reconnect() + + def _schedule_reconnect(self): + if self._manual_disconnect: + return + if self._reconnect_task and not self._reconnect_task.done(): + return + self._reconnect_task = asyncio.create_task(self._reconnect_loop()) + + async def _reconnect_loop(self): + """指数退避 2s→30s 封顶,直到成功或手动断开。""" + delay = 2 + while not self._manual_disconnect and not self.connected: + await self._emit_state("reconnecting", f"{delay}s 后自动重连 {self.uri} …") + try: + await asyncio.sleep(delay) + except asyncio.CancelledError: + return + if self._manual_disconnect: + return + try: + self.ws = await websockets.connect(self.uri) + except Exception: + delay = min(delay * 2, 30) + continue + self.connected = True + self._listen_task = asyncio.create_task(self._listen()) + await self._emit_state("connected", self.uri) + return