Greg's Space
Go back

終端機個人助理（三）：用 tui.py 做出 Textual 聊天介面與核准視窗

這是系列的最後一篇。第一篇準備了設定，第二篇寫好了 Assistant，這篇用 Textual 把它接上畫面：

完整程式碼

檔案位置：src/tui.py

import json
import uuid
from typing import Optional
from textual import on, work
from textual.app import App, ComposeResult
from textual.binding import Binding
from textual.containers import Horizontal, Vertical, VerticalScroll
from textual.screen import ModalScreen
from textual.widgets import Button, Footer, Header, Input, Static
from textual.widget import Widget

from krl import Assistant


class Approval(ModalScreen[bool]):

    def __init__(self, function_name: str, function_variables: dict) -> None:
        """記下要核准的工具名稱和參數"""

        super().__init__()
        self.function_name = function_name
        self.function_variables = function_variables
        return

    def compose(self) -> ComposeResult:
        """對話框：標題、參數、核准與拒絕按鈕"""

        title = f"[b]Agent 想要執行[/b] [cyan]{self.function_name}[/cyan]"
        function_variables_text = json.dumps(
            self.function_variables, ensure_ascii=False, indent=2
        )
        with Vertical(id="dialog"):
            yield Static(title)
            yield Static(
                function_variables_text, classes="variables", markup=False
            )
            with Horizontal(id="buttons"):
                yield Button("核准 (y)", variant="success", id="approve", compact=True)
                yield Button("拒絕 (n)", variant="error", id="reject", compact=True)

    def action_decide(self, approved: bool) -> None:
        """按 y 或 n 時關閉視窗，並回傳結果"""

        self.dismiss(approved)
        return

    @on(Button.Pressed)
    def _pressed(self, event: Button.Pressed) -> None:
        """按下按鈕時關閉視窗，並回傳結果"""

        self.dismiss(event.button.id == "approve")
        return

    BINDINGS = [
        Binding("y", "decide(True)", "核准"),
        Binding("n", "decide(False)", "拒絕"),
    ]
    pass


assistant = Assistant(
    thread_id=str(uuid.uuid4())
)


class Screen(App):

    def __init__(self) -> None:
        """初始化主畫面，並開一段新的對話"""

        super().__init__()
        return

    def compose(self) -> ComposeResult:
        """版面：標題列、聊天區、輸入框、快捷鍵列"""

        yield Header()
        yield VerticalScroll(id="chat")
        yield Input(placeholder="輸入訊息，Enter 送出", id="prompt")
        yield Footer()

    def on_mount(self) -> None:
        """畫面建立完成後，讓游標停在輸入框"""

        self.query_one("#prompt", Input).focus()
        return

    async def _add(self, widget: Widget) -> None:
        """把元件加到聊天區最下面，並捲到底"""

        chat = self.query_one("#chat", VerticalScroll)
        await chat.mount(widget)
        chat.scroll_end(animate=False)
        return

    @on(Input.Submitted, "#prompt")
    async def _submit(self, event: Input.Submitted) -> None:
        """使用者在輸入框按 Enter 時，顯示訊息並交給 agent"""

        text = event.value.strip()
        if not text:
            return
        
        event.input.value = ""
        event.input.disabled = True
        await self._add(Static(text, classes="user", markup=False))
        _ = self._converse(text)
        return

    def get_tool_widget(self, tool_event: dict) -> Static:
        """把工具資訊變成聊天區裡的一行，失敗的用紅色顯示"""

        tool_name = tool_event["name"]

        # AI 決定呼叫工具：顯示工具名稱和參數。
        if tool_event["type"] == "tool_call":
            tool_args_text = json.dumps(tool_event["args"], ensure_ascii=False)
            tool_content = f"▶ {tool_name}: {tool_args_text[:300]}"
            return Static(tool_content, classes="tool", markup=False)

        # 工具執行完：顯示成功或失敗，以及結果的前 300 個字。
        result_text = tool_event["content"][:300]
        if tool_event["status"] == "error":
            tool_content = f"✗ {tool_name} 失敗\n{result_text}"
            return Static(tool_content, classes="error", markup=False)

        tool_content = f"✓ {tool_name} 成功\n{result_text}"
        return Static(tool_content, classes="tool", markup=False)

    @work(exclusive=True)
    async def _converse(self, question_content: Optional[str]) -> None:
        """把問題交給 agent 並顯示回答，需要審核時逐一跳出核准視窗"""

        chat = self.query_one("#chat", VerticalScroll)

        # 每一次對話，按下 Enter 就是問問題，所以不會有審批結果。
        approval_decisions = None

        # agent 可能停下來好幾次等審批，所以用迴圈一直重複，
        # 直到它不再停下來為止。
        while True:

            # 目前還沒有回答框，等 AI 開始說話才建立。
            answer_widget = None
            answer_content = ""

            # 這一輪 agent 想執行的指令，先當作沒有。
            approval_requests = []

            # 一邊收 agent 的回答，一邊顯示。
            # 收到的是文字就顯示；收到的是 list，代表 agent 停下來等審批；
            # 收到的是 dict，代表 agent 做完了一步。
            async for answer_piece in assistant.get_feedback(
                question_content, approval_decisions
            ):
                if isinstance(answer_piece, list):
                    approval_requests = answer_piece
                    continue

                # 收到 dict，是工具資訊，顯示一行到聊天區。
                # 然後忘掉目前的回答框，AI 接下來的話會寫在新的框裡。
                if isinstance(answer_piece, dict):
                    await self._add(self.get_tool_widget(answer_piece))
                    answer_widget = None
                    continue

                # 收到文字，但目前沒有回答框，就在最下面開一個新的。
                if answer_widget is None:
                    answer_widget = Static(
                        "", classes="assistant", markup=False
                    )
                    await self._add(answer_widget)
                    answer_content = ""

                answer_content += answer_piece
                answer_widget.update(answer_content)
                chat.scroll_end(animate=False)
                continue

            # 沒有要審批的指令，代表 agent 已經回答完，結束這次對話。
            if not approval_requests:
                break

            # 一個一個跳出核准視窗，把使用者的回答照順序記下來。
            approval_decisions = []
            for approval_request in approval_requests:
                approval = Approval(
                    approval_request["name"], approval_request["args"]
                )
                approved = await self.push_screen_wait(approval)
                approval_decisions.append(approved)
                continue

            # 下一輪只送審批結果給 agent，不再送問題。
            question_content = None
            continue

        # 對話結束，讓使用者可以再輸入。
        prompt = self.query_one("#prompt", Input)
        prompt.disabled = False
        prompt.focus()
        return

    TITLE = "Personal Assistant"
    CSS = """
    #chat {
        padding: 0 1;
    }
    .user {
        background: $boost;
        border-left: thick $accent;
        padding: 0 1;
        margin: 1 0 0 0;
    }
    .assistant {
        margin: 1 0 0 0;
    }
    .tool {
        color: $text-muted;
        border-left: thick $warning;
        padding: 0 1;
        margin: 1 0 0 0;
    }
    .error {
        color: $error;
        border-left: thick $error;
        padding: 0 1;
        margin: 1 0 0 0;
    }
    Approval {
        align: center middle;
    }
    #dialog {
        width: 80%;
        max-height: 80%;
        height: auto;
        border: thick $warning;
        background: $surface;
        padding: 1 2;
    }
    #dialog .variables {
        margin: 1 0;
        max-height: 20;
        overflow-y: auto;
    }
    #buttons {
        height: auto;
        align-horizontal: right;
    }
    #buttons Button {
        margin-left: 2;
    }
    """
    BINDINGS = [
        Binding("ctrl+q", "quit", "Quit"),
    ]
    pass

def activate() -> None:
    """啟動 TUI"""

    Screen().run()
    return

if __name__ == "__main__":

    activate()
    pass```

下面逐段說明。

## 1. `Approval`：核准視窗

`Approval` 繼承 `ModalScreen[bool]`，是一個蓋在主畫面上的對話框，關閉時會回傳一個 `bool`。

- `compose()` 顯示工具名稱，並用 `json.dumps(..., indent=2)` 把參數排版好。參數是 agent 產生的內容，可能含有 `[` `]` 這類符號，所以用 `markup=False` 避免被當成 Textual 的格式標記。
- 按按鈕或按 `y` / `n` 都會呼叫 `self.dismiss(...)`，把結果交回給等待的人。

`BINDINGS` 裡的 `"decide(True)"` 會呼叫 `action_decide(True)`，這是 Textual 把按鍵綁到方法的寫法：`action_` 後面接上的名稱就是 action 名稱。

## 2. `Screen`：主畫面

```python
def compose(self) -> ComposeResult:
    yield Header()
    yield VerticalScroll(id="chat")
    yield Input(placeholder="輸入訊息，Enter 送出", id="prompt")
    yield Footer()

版面由上到下是：標題列、可捲動的聊天區、輸入框、快捷鍵列。聊天區一開始是空的，每則訊息都是用 _add() 動態加進去的 Static，加完就捲到最底下。

Assistant 在模組層級建立一次，並給它一個隨機的 thread_id。整個程式執行期間都是同一段對話，agent 記得前面說過的話。

3. 送出訊息

@on(Input.Submitted, "#prompt")
async def _submit(self, event: Input.Submitted) -> None:
    ...
    event.input.disabled = True
    await self._add(Static(text, classes="user", markup=False))
    _ = self._converse(text)

按下 Enter 後先停用輸入框，避免 agent 還在回答時又送出新問題，再把使用者的訊息顯示出來，接著交給 _converse()。

_converse() 加了 @work(exclusive=True)，會在背景的 worker 執行，所以 _submit() 不用等它跑完，畫面也不會卡住。exclusive=True 表示同一時間只會有一個對話在跑。

4. _converse()：對話迴圈

這是整個介面最核心的部分。agent 可能「回答一點 → 停下來等核准 → 繼續回答 → 又停下來…」好幾次，所以用 while True 一直重複，直到 agent 不再停下來：

第 1 輪：get_feedback(問題, None)
         → 收到文字、工具資訊…最後收到 list（等待核准）
         → 逐一跳出核准視窗，收集 [True, False, ...]
第 2 輪：get_feedback(None, [True, False, ...])
         → 繼續收文字、工具資訊…沒有 list
         → 結束迴圈，重新啟用輸入框

每一輪收到的東西照型別處理，正好對應第二篇 get_feedback() 的三種輸出：

收到處理方式
list先記下來，等串流結束再跳出核准視窗
dict用 get_tool_widget() 顯示一行工具資訊，並把 answer_widget 設回 None
str接到目前的回答框後面；沒有回答框就新開一個

把 answer_widget 設回 None 的作用是：AI 說了一段話、呼叫工具、再說一段話時，第二段話會出現在工具資訊下方的新框裡，畫面順序和實際發生的順序一致。

等待核准

approved = await self.push_screen_wait(approval)

push_screen_wait() 會顯示視窗，並等到它 dismiss() 才回傳結果。這個方法只能在 worker 裡使用，這也是 _converse() 需要 @work 的另一個原因。

5. 工具資訊的顯示

get_tool_widget() 把一筆工具資訊變成一行文字：

▶ execute: {"command": "ls -la"}
✓ execute 成功
total 8 ...
✗ read_file 失敗
Error: file not found

參數和結果都只取前 300 個字，避免讀到大檔案時把畫面塞滿。樣式寫在 CSS 裡：使用者訊息用主題的強調色（$accent）當左框，工具資訊用警告色（$warning），錯誤用錯誤色（$error）。

6. 啟動

def activate() -> None:
    Screen().run()

第一篇的 pyproject.toml 已經把 lpa 指令指向 tui:activate。確認安裝過專案後，到任何放有 .env 的資料夾執行：

lpa

助理會在這個資料夾底下建立 workspace/ 開始工作，按 Ctrl+Q 離開。

可以再加強的地方

系列回顧

檔案重點
cfg.py在哪裡啟動就在哪裡工作：workspace/ 與 .env 都跟著目前資料夾
krl.pydeep agent ＋ 對話壓縮 ＋ 執行前審核，用 str / dict / list 三種型別串流給介面
tui.py依型別顯示文字、工具資訊或跳出核准視窗，用迴圈處理多輪核准
Previous Post
用 deepagents + Textual 打造一個會寫程式的終端機個人助理
Next Post
終端機個人助理：逐行看懂 pyproject.toml，把三個 .py 檔變成 lpa 指令