用 Textual 和 Rich 打造一個能接 LangChain Agent 的聊天機器人終端機介面

如果你已經用 LangChain 兜出一個能用的 agent,但每次測試都只能在 print() 堆出來的訊息裡找對話紀錄,體驗其實蠻痛苦的。這篇文章要做的事情很單純:用 Textual 蓋一個終端機聊天視窗,把 LangChain agent 接在後面,讓你打字、按 Enter,就能像用 ChatGPT 一樣跟自己的 agent 對話。

先分清楚 Textual 跟 Rich 的分工

很多人第一次聽到這兩個名字會搞混,其實分工很清楚:

實務上兩者是疊在一起用的:Textual 提供「聊天視窗」這個容器,容器裡每一則訊息的內容,交給 Rich 排版(尤其是 agent 回覆常常帶 Markdown、程式碼區塊,Rich 天生就很會處理這些)。

整體架構長什麼樣子

使用者輸入 → Textual Input widget
                │
                ▼
        丟給 LangChain agent(非同步呼叫)
                │
                ▼
   agent 回覆(可能是串流的 token)
                │
                ▼
    用 Rich 渲染成 Markdown → 更新 Textual 畫面上的訊息框

關鍵設計只有一個:agent 呼叫是會花時間的網路 I/O,絕對不能卡住 Textual 的畫面。所以整個串接一定要走非同步(async),這也是後面程式碼的重點。

Step 1:先把 Textual 的骨架搭出來

Textual 的核心概念是:一個 App 裡面放很多 Widget,畫面配置用類似 CSS 的語法寫。先安裝套件:

pip install textual langchain langchain-openai

接著寫一個最小可動的聊天介面:

# chat_app.py
from textual.app import App, ComposeResult
from textual.containers import VerticalScroll
from textual.widgets import Input, Header, Footer, Static


class ChatApp(App):
    CSS = """
    VerticalScroll {
        height: 1fr;
        border: round $accent;
        padding: 1 2;
    }
    Input {
        dock: bottom;
    }
    """

    def compose(self) -> ComposeResult:
        yield Header()
        yield VerticalScroll(id="messages")
        yield Input(placeholder="輸入訊息後按 Enter…")
        yield Footer()

    def on_input_submitted(self, event: Input.Submitted) -> None:
        messages = self.query_one("#messages", VerticalScroll)
        messages.mount(Static(f"你: {event.value}"))
        event.input.value = ""


if __name__ == "__main__":
    ChatApp().run()

這個版本還沒接 agent,先確認畫面能跑:上方是可捲動的訊息區,下方固定一個輸入框,按 Enter 會把你打的字貼到訊息區裡。

python chat_app.py

Step 2:準備一個 LangChain agent

假設你已經有 agent,或者先用最簡單的方式建一個(以 langchain 的 agent 介面為例):

# agent.py
from langchain.agents import create_agent
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini")
agent = create_agent(model=model, tools=[])


async def ask_agent(user_input: str) -> str:
    result = await agent.ainvoke({"messages": [{"role": "user", "content": user_input}]})
    return result["messages"][-1].content

重點是 ainvoke——LangChain 的 agent 大多都提供非同步版本的呼叫方式,這正好可以跟 Textual 的事件迴圈搭在一起,不會互相卡住。

Step 3:把 agent 接進 Textual,並用 Rich 渲染回覆

這一步把兩件事合在一起:呼叫 agent 是非同步的,而 agent 回覆常常帶 Markdown(列表、程式碼區塊),用 Rich 的 Markdown 元件渲染會比純文字好看很多。

# chat_app.py
from textual.app import App, ComposeResult
from textual.containers import VerticalScroll
from textual.widgets import Input, Header, Footer, Static
from rich.markdown import Markdown

from agent import ask_agent


class ChatApp(App):
    CSS = """
    VerticalScroll {
        height: 1fr;
        border: round $accent;
        padding: 1 2;
    }
    Input {
        dock: bottom;
    }
    .user {
        color: $success;
    }
    """

    def compose(self) -> ComposeResult:
        yield Header()
        yield VerticalScroll(id="messages")
        yield Input(placeholder="輸入訊息後按 Enter…")
        yield Footer()

    async def on_input_submitted(self, event: Input.Submitted) -> None:
        user_text = event.value
        event.input.value = ""
        event.input.disabled = True

        messages = self.query_one("#messages", VerticalScroll)
        messages.mount(Static(f"你: {user_text}", classes="user"))

        reply_widget = Static("agent 思考中…")
        messages.mount(reply_widget)
        messages.scroll_end(animate=False)

        reply_text = await ask_agent(user_text)

        reply_widget.update(Markdown(reply_text))
        messages.scroll_end(animate=False)
        event.input.disabled = False
        event.input.focus()


if __name__ == "__main__":
    ChatApp().run()

幾個容易忽略但很重要的細節:

1. on_input_submitted 要宣告成 async def。Textual 的事件處理函式本來就支援 async,只要宣告成非同步,裡面就可以直接 await ask_agent(...),不需要額外開執行緒。
2. 等待 agent 回覆的期間先鎖住輸入框(event.input.disabled = True),避免使用者在還沒收到回覆前又送出下一則訊息,造成多個請求疊在一起。
3. 先掛一個「思考中」的佔位訊息,拿到結果後再 update(),使用者才不會覺得畫面卡住了。
4. agent 的回覆用 rich.markdown.Markdown 包起來再丟給 Static.update()——Textual 的 widget 本來就吃 Rich 的 renderable,所以這行不需要任何轉換,Markdown 語法會自動被渲染成終端機裡的粗體、清單、程式碼區塊。

Step 4(進階):把回覆改成逐字串流

如果你的 agent/模型支援串流(大部分 LangChain 的 chat model 都支援 .astream()),可以讓文字像打字機一樣一個字一個字跑出來,體驗會更接近真正的聊天工具:

async def ask_agent_stream(user_input: str, on_token):
    async for chunk in agent.astream({"messages": [{"role": "user", "content": user_input}]}):
        token = chunk.get("messages", [None])[-1]
        if token:
            on_token(token.content)

在 ChatApp 裡把原本一次性的 await ask_agent(...) 換成邊收 token 邊更新同一個 reply_widget:

        full_text = ""

        async def on_token(token: str):
            nonlocal full_text
            full_text += token
            reply_widget.update(Markdown(full_text))
            messages.scroll_end(animate=False)

        await ask_agent_stream(user_text, on_token)

這裡的技巧一樣是「不要卡住事件迴圈」:每收到一小段文字就立刻更新畫面,而不是等整段回覆完成才顯示。

一句話總結

Textual 負責畫面骨架和使用者互動(輸入框、捲動、事件),Rich 負責把 agent 回覆的內容排版得好看(尤其是 Markdown),而串接 LangChain agent 的關鍵只有一件事:全程走 async,讓等待網路回覆的時間不會卡住整個終端機介面。掌握這三個角色的分工,你就可以把任何 LangChain agent 包成一個可以在終端機裡直接聊天的小工具。