Conversation 多輪對話

Conversation 類提供了在同一會話中進行多輪交互的便捷方法，適合實現引導式操作、資訊收集、對話式問答等場景。

建立對話

透過 Event 物件的 conversation() 方法建立：

from ErisPulse.Core.Event import command

@command("quiz")
async def quiz_handler(event):
    conv = event.conversation(timeout=30)

    await conv.say("🎮 歡迎參加知識問答！")

    answer = await conv.choose("第一題：Python 的創造者是誰？", [
        "Guido van Rossum",
        "James Gosling",
        "Dennis Ritchie",
    ])

    if answer is None:
        await conv.say("超時了，下次再來吧！")
        return

    if answer == 0:
        await conv.say("正確！")
    else:
        await conv.say("錯誤了，正確答案是 Guido van Rossum")

    conv.stop()

核心 API

say(content, **kwargs)

發送訊息，回傳 self 支援鏈式呼叫：

await conv.say("第一行").say("第二行").say("第三行")

也可以指定發送方式：

await conv.say("https://example.com/image.jpg", method="Image")

wait(prompt=None, timeout=None)

等待使用者回覆，回傳 Event 物件或 None（超時）：

# 簡單等待
resp = await conv.wait()
if resp:
    text = resp.get_text()

# 發送提示後等待
resp = await conv.wait(prompt="請輸入你的名字：")

# 使用自訂超時（覆蓋對話預設超時）
resp = await conv.wait(prompt="請在10秒內回覆：", timeout=10)

confirm(prompt=None, **kwargs)

等待使用者確認（是/否），回傳 True / False / None（超時）：

result = await conv.confirm("確定要刪除所有資料嗎？")
if result is True:
    await conv.say("已刪除")
elif result is False:
    await conv.say("已取消")
else:
    await conv.say("超時未回覆")

內建識別的確認詞：是/yes/y/確認/確定/好/ok/true/對/嗯/行/同意/沒問題/可以/當然...

內建識別的否定詞：否/no/n/取消/不/不要/不行/cancel/false/錯/不對/別/拒絕...

choose(prompt, options, **kwargs)

等待使用者從選項中選擇，回傳選項索引（0-based）或 None：

choice = await conv.choose("請選擇顏色：", ["紅色", "綠色", "藍色"])
if choice is not None:
    colors = ["紅色", "綠色", "藍色"]
    await conv.say(f"你選擇了 {colors[choice]}")

使用者可以透過輸入編號（1/2/3）或選項文字（紅色）來選擇。

options_format="auto"（預設）根據 method 自動選擇內建樣式：Markdown→無序列表，Html→有序列表，其他→純文字列表。
也支援 "list"、"inline"、"md"、"html" 或自訂函數。

支援 merge_prompt=True 合併為一條訊息，以及占位符控制選項插入位置（預設 {options}，可透過 placeholder 自訂）：

choice = await conv.choose(
    "## 請選擇\n{options}",
    ["選項A", "選項B"],
    method="Markdown",
    merge_prompt=True,
)

# 自訂占位符
choice = await conv.choose(
    "請選擇: [choices]",
    ["選項A", "選項B"],
    placeholder="[choices]",
)

collect(fields, **kwargs)

多步驟收集資訊，回傳資料字典或 None：

data = await conv.collect([
    {"key": "name", "prompt": "請輸入姓名"},
    {"key": "age", "prompt": "請輸入年齡",
     "validator": lambda e: e.get("alt_message", "").strip().isdigit(),
     "retry_prompt": "年齡必須是數字，請重新輸入"},
    {"key": "city", "prompt": "請輸入城市"},
])

if data:
    await conv.say(f"註冊成功！\n姓名: {data['name']}\n年齡: {data['age']}\n城市: {data['city']}")
else:
    await conv.say("註冊過程中斷")

欄位配置：

條件欄位：使用 condition 可以實現動態表單，只有條件滿足時才收集該欄位：

data = await conv.collect([
    {"key": "has_car", "prompt": "你有車嗎？（是/否）"},
    {"key": "car_brand", "prompt": "請輸入車型",
     "condition": lambda d: d.get("has_car", "").lower() in ("是", "yes", "y")},
])

stop()

手動結束對話，設定 is_active 為 False：

conv.stop()

is_active

對話是否處於活躍狀態：

if conv.is_active:
    await conv.say("對話還在進行中")

活躍狀態管理

stateDiagram-v2
    state "活躍" as active
    state "非活躍" as inactive
    [*] --> active: event.conversation()
    active --> active: say / wait / confirm / choose / collect
    active --> inactive: stop()
    active --> inactive: wait() 超時
    active --> inactive: collect() 超時或重試耗盡
    inactive --> [*]

對話在以下情況會自動變為非活躍狀態：

1. 調用 stop() 方法
2. wait() 超時回傳 None
3. collect() 因任何步驟超時或重試耗盡而回傳 None

非活躍後，所有互動方法（wait/confirm/choose/collect）會立即回傳 None，不會繼續等待使用者輸入。

分支與跳轉

@conv.branch(name) 裝飾器

使用 branch() 註冊對話分支，透過 goto() 在分支間跳轉：

@command("menu")
async def menu_handler(event):
    conv = event.conversation(timeout=60)

    @conv.branch("main")
    async def main_menu():
        await conv.say("=== 主選單 ===\n1. 個人資訊\n2. 設定\n3. 退出")
        resp = await conv.wait()
        if resp is None:
            return
        text = resp.get_text().strip()
        if text == "1":
            await conv.goto("profile")
        elif text == "2":
            await conv.goto("settings")
        elif text == "3":
            await conv.say("再見！")
            conv.stop()

    @conv.branch("profile")
    async def profile():
        await conv.say("=== 個人資訊 ===\n姓名: Alice\n0. 返回")
        resp = await conv.wait()
        if resp and resp.get_text().strip() == "0":
            await conv.goto("main")

    @conv.branch("settings")
    async def settings():
        await conv.say("=== 設定 ===\n1. 通知開關\n0. 返回")
        resp = await conv.wait()
        if resp and resp.get_text().strip() == "0":
            await conv.goto("main")

    await conv.start()  # 從第一個註冊的分支開始

conv.start(name=None)

啟動對話，預設從第一個註冊的分支開始：

await conv.start()          # 從第一個分支開始
await conv.start("settings") # 從指定分支開始

上下文與持久化

conv.context

每個對話實例內建 context 字典，用於在分支間共享狀態：

@conv.branch("step1")
async def step1():
    conv.context["username"] = resp.get_text().strip()
    await conv.goto("step2")

@conv.branch("step2")
async def step2():
    name = conv.context.get("username", "未知")
    await conv.say(f"你好，{name}！")

save() / resume() / clear_saved()

對話支援持久化，可在超時或中斷後恢復：

# 保存對話狀態（通常無需手動呼叫，見下方"自動檢查點"）
await conv.save()

# ... 之後在同一會話中恢復 ...
conv2 = event.conversation()
if await conv2.resume():
    await conv2.say("歡迎回來！繼續之前的對話")
else:
    await conv2.say("沒有找到之前的對話")

# 清除保存的對話
await conv.clear_saved()

儲存鍵含 target 維度（conversation:{platform}:{user_id}:{target_id}），同一使用者在不同會話中的對話互不覆蓋；舊格式（不含 target）的存檔在 resume() 時自動遷移。

自動檢查點與重啟恢復

自動存檔

框架在以下時機自動維護檢查點，通常無需手動調用 save()：

檢查點 TTL

存檔帶時間戳，超過 ErisPulse.interaction.checkpoint_ttl（預設 24 小時）的存檔會被清理：

• 惰性丟棄：恢復時發現存檔已過期，自動丟棄
• 後台主動清理：框架有周期 GC 任務（首次使用檢查點後惰性啟動）主動枚舉並刪除過期存檔，避免長期運行時 storage 中過期檢查點無限累積。被主動清理的存檔若之後收到會話消息，按"無檢查點"處理

[ErisPulse.interaction]
checkpoint_ttl = 86400  # 秒

重啟自動恢復

框架重啟後，進行中的對話（記憶體中的等待協程）會丟失，但檢查點仍在。透過 register_resume_handler 註冊恢復工廠，框架即可在重啟後收到該會話首條消息時自動續接對話：

from ErisPulse.Core.Event.wrapper import Conversation

@Conversation.register_resume_handler()  # 可傳 platform="onebot11" 限定平台
def make_conversation(event) -> Conversation:
    # 工廠職責：重建對話並重新註冊所有分支
    conv = event.conversation(timeout=60)

    @conv.branch("menu")
    async def menu(conv, event):
        ...

    return conv

註冊後，重啟前處於 menu 分支的使用者發來首條消息時，框架自動：恢復 context → 認領該消息 → 從存檔分支繼續對話。未註冊工廠時此機制零開銷。

恢復即接管

resume() 成功時框架自動完成兩件事：

1. 會話接管：自動 acquire 該會話的互斥租約——其他模組可透過 sdk.interaction.get_owner_of(event) 感知"這個使用者正被對話占用"；會話已被其他模組占用時放棄恢復（返回 False），避免兩個對話打架
2. 歷史帶回：從會話收件箱取最近 10 條消息到 conv.recent_history（AI 模組恢復後 LLM 上下文不斷檔）；resume(with_history=0) 可關閉

if await conv.resume(with_history=20):
    for m in conv.recent_history:
        print(m["role"], ":", m["text"])

手動恢復（不用自動機制時）

@command("continue")
async def continue_handler(event):
    conv = event.conversation()
    # ... 註冊分支 ...
    if await conv.resume():
        conv.goto(conv.get_current_branch())

典型流程模式

引導式註冊

@command("register")
async def register_handler(event):
    conv = event.conversation(timeout=60)

    await conv.say("歡迎註冊！")

    data = await conv.collect([
        {"key": "username", "prompt": "請輸入使用者名稱（3-20個字元）",
         "validator": lambda e: 3 <= len(e.get_text().strip()) <= 20},
        {"key": "email", "prompt": "請輸入電子信箱",
         "validator": lambda e: "@" in e.get_text() and "." in e.get_text(),
         "retry_prompt": "電子信箱格式不正確，請重新輸入"},
    ])

    if not data:
        await event.reply("註冊已取消")
        return

    confirmed = await conv.confirm(
        f"確認註冊資訊？\n使用者名稱: {data['username']}\n電子信箱: {data['email']}"
    )

    if confirmed:
        await conv.say("✅ 註冊成功！")
    else:
        await conv.say("❌ 已取消註冊")

循環對話

@command("chat")
async def chat_handler(event):
    conv = event.conversation(timeout=120)
    await conv.say("進入對話模式，輸入「退出」結束")

    while conv.is_active:
        resp = await conv.wait()
        if resp is None:
            await conv.say("超時，對話結束")
            break

        text = resp.get_text().strip()

        if text == "退出":
            await conv.say("再見！")
            conv.stop()
        elif text == "幫助":
            await conv.say("可用命令：退出、幫助、狀態")
        elif text == "狀態":
            await conv.say("對話活躍中")
        else:
            await conv.say(f"你說的是：{text}")

相關文件

• Event 包裝類 - Event 物件的所有方法
• 事件處理入門 - 事件處理基礎