iT邦幫忙

2025 iThome 鐵人賽

DAY 21
0
生成式 AI

打造基於 MCP 協議與 n8n 工作流的會議處理 Agent系列 第 21

Day 21 Gradio 介面更新 — 加入歷史會議列表與即時處理狀態

  • 分享至 

  • xImage
  •  

昨天我在郵件系統中加入了連結到 Notion 的按鈕,提升了使用者體驗。但現在 Gradio 前端介面,每次處理完會議後,之前的記錄就消失了,而且在處理過程中,我也不知道系統到底跑了多久。

所以今天的目標就是為 Gradio 介面加入會議歷史追溯功能,讓我能查看和管理之前處理過的會議記錄,同時也加入顯示處理時間,讓我掌握系統的執行狀態。

今天的目標與挑戰

  • 建立會議歷史管理模組,自動儲存每次處理的記錄
  • 在 Gradio 介面中加入歷史會議記錄分頁
  • 製作表格篩選功能
  • 加入刪除記錄功能
  • 顯示處理時間
  • 強化介面配置,讓資訊呈現更加清晰

Step 1:建立會議歷史管理模組

首先需要先建立一個專門管理會議歷史記錄的模組,負責記錄的新增、查詢、刪除等操作。

1-1 建立 src/meeting_history.py

在專案根目錄下建立 src 資料夾,然後在 src/ 目錄中建立 meeting_history.py 檔案,並在裡面撰寫以下程式碼

import json
import os
from datetime import datetime

class MeetingHistory:
    def __init__(self, history_file="data/meeting_history.json"):
        self.history_file = history_file
        self._ensure_history_file()
    
    def _ensure_history_file(self):
        # 確保 data 目錄存在
        os.makedirs(os.path.dirname(self.history_file), exist_ok=True)
        
        if not os.path.exists(self.history_file):
            with open(self.history_file, 'w', encoding='utf-8') as f:
                json.dump([], f, ensure_ascii=False)
    
    def add_record(self, data):
        # 從 n8n 回傳的資料提取欄位
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        
        record = {
            "session_id": data.get("session_id", ""),
            "timestamp": timestamp,
            "project_name": data.get("project_name", "未指定專案"),
            "meeting_type": data.get("meeting_type", "一般會議"),
            "summary": data.get("summary", "無摘要"),
            "tasks": data.get("tasks", "無任務"),
            "notion_url": data.get("url", ""),
            "participants": data.get("participants", []),
            "task_count": data.get("task_count", 0),
            "deadline_date": data.get("deadline_date", "")
        }
        
        # 讀取現有記錄
        records = self.get_all_records()
        records.insert(0, record)
        
        # 保留最近 50 筆
        if len(records) > 50:
            records = records[:50]
        
        # 儲存
        with open(self.history_file, 'w', encoding='utf-8') as f:
            json.dump(records, f, ensure_ascii=False, indent=2)
        
        print(f"✅ 成功儲存會議記錄:{record['session_id']}")
        return record
    
    def get_all_records(self):
        try:
            with open(self.history_file, 'r', encoding='utf-8') as f:
                return json.load(f)
        except:
            return []
    
    def get_record_by_session(self, session_id):
        records = self.get_all_records()
        for record in records:
            if record.get("session_id") == session_id:
                return record
        return None
    
    def delete_record(self, session_id):
        # 刪除指定的記錄
        records = self.get_all_records()
        original_count = len(records)
        
        # 過濾掉要刪除的記錄
        records = [r for r in records if r.get("session_id") != session_id]
        
        if len(records) < original_count:
            # 有記錄被刪除,儲存更新後的列表
            with open(self.history_file, 'w', encoding='utf-8') as f:
                json.dump(records, f, ensure_ascii=False, indent=2)
            print(f"✅ 成功刪除記錄:{session_id}")
            return True
        
        return False
    
    def get_history_dataframe(self):
        # 回傳適合 Gradio Dataframe 的格式
        records = self.get_all_records()
        
        if not records:
            return [["目前沒有歷史記錄", "", "", ""]]
        
        # 格式:[記錄時間, 專案名稱, 會議類型, Session ID]
        data = []
        for record in records:
            data.append([
                record.get("timestamp", "未知時間"),
                record.get("project_name", "未指定專案"),
                record.get("meeting_type", "一般會議"),
                record.get("session_id", "")
            ])
        
        return data

同時也建立 src/__init__.py 檔案(可以是空檔案),讓 Python 將 src 視為一個模組。

1-2 模組設計要點

meeting_history.py 模組的設計大致可以分為以下幾點

  • 單一職責原則:這個模組只負責會議歷史的管理,不涉及其他邏輯
  • 自動建立目錄:透過 os.makedirs 自動建立 data/ 目錄
  • 資料格式轉換get_history_dataframe() 方法將 JSON 格式轉換為 Gradio Dataframe 所需的格式
  • 記錄數量限制:保留最近 50 筆記錄,避免檔案過大
  • 完整的 CRUD 操作:新增、查詢、刪除功能一應俱全

Step 2:更新 n8n 工作流的回應格式

為了讓 Gradio 能正確接收並顯示資料,我需要調整 n8n 的最後一個節點,讓它回傳更完整的資料結構。

2-1 修改「組合 Webhook 回應資料」節點

在 n8n 工作流中,在「Respond to Webhook」前新增一個 code 節點,並且重新命名為「組合 Webhook 回應資料」,並在其中撰寫以下內容

// 從各節點取得資料
const webhookData = $('Webhook').first().json.body;
const structuredTask = $('提取結構化任務').first().json;
const notionPage = $('建立初始會議記錄').first().json;

// 組合完整的 Webhook 回應資料
return {
  session_id: webhookData.session_id,
  url: notionPage.url,
  summary: structuredTask.meeting_attributes.summary,
  tasks: structuredTask.meeting_info.tasks,
  project_name: structuredTask.meeting_info.project_name,
  meeting_type: structuredTask.meeting_attributes.meeting_type,
  participants: structuredTask.participant_info.participants,
  task_count: structuredTask.task_info.task_count,
  deadline_date: structuredTask.meeting_info.deadline_date
};

這樣 n8n 就會回傳一個包含所有必要資訊的 JSON 物件,包括專案名稱、會議類型、參與人員等,讓 Gradio 能夠完整地儲存和顯示這些資訊。


Step 3:改造 Gradio 介面

現在我要改造 app.py,加入歷史會議記錄的分頁和各種互動功能。

3-1 更新 app.py 的 import 區塊

首先匯入剛才建立的 meeting_history 模組,並在程式啟動時初始化

import gradio as gr
import json
import time
from mcp_agent import MCPAgent
from src.meeting_history import MeetingHistory

# 初始化
print("正在初始化 MCP Agent")
agent = MCPAgent(model="medium")
history_manager = MeetingHistory()
print("✅ MCP Agent 與歷史記錄管理器初始化完成")

3-2 修改處理函式,加入時間計算與記錄儲存

接著修改 process_and_run_agent() 函式,加入處理時間的計算,並在處理完成後自動儲存記錄:

def process_and_run_agent(audio_filepath, command_text, progress=gr.Progress()):
    # 驗證輸入
    if audio_filepath is None:
        return (
            "## 錯誤\n您尚未上傳音訊檔案,請先上傳一個音訊檔案!",
            gr.Dataframe(),
            None,
        )

    if not command_text.strip():
        return "## 錯誤\n指令為空,請輸入您希望執行的指令!", gr.Dataframe(), None

    print(f"接收到音訊:{audio_filepath}")
    print(f"執行指令:{command_text}")

    try:
        start_time = time.time()
        progress(None, desc="正在處理中...")

        # 呼叫核心 Agent 邏輯
        result = agent.process_audio(audio_filepath, command_text)
        elapsed_time = time.time() - start_time

        # 處理回傳結果
        data = None
        if isinstance(result, list) and len(result) > 0:
            data = result[0]
        elif isinstance(result, dict):
            data = result

        if data:
            # 儲存到歷史記錄
            history_manager.add_record(data)

            # 取得顯示用的資料
            project_name = data.get("project_name", "未指定專案")
            meeting_type = data.get("meeting_type", "一般會議")
            summary = data.get("summary", "摘要生成失敗")
            tasks = data.get("tasks", "任務提取失敗")
            notion_url = data.get("url", "")
            session_id = data.get("session_id", "")

            # 組裝顯示的 Markdown
            output_markdown = (
                f"## 處理完成\n\n"
                f"**會議記錄已成功建立至 Notion,並且已發送郵件通知!**\n\n"
                f"處理時間:{elapsed_time:.1f} 秒\n\n"
                f"---\n\n"
                f"### 會議資訊\n\n"
                f"* **專案名稱**:{project_name}\n"
                f"* **會議類型**:{meeting_type}\n"
                f"* **Session ID**:`{session_id}`\n\n"
                f"🔗 **[點此前往 Notion 查看完整記錄]({notion_url})**\n\n"
                f"---\n\n"
                f"### 會議摘要\n\n"
                f"{summary}\n\n"
                f"---\n\n"
                f"### 行動項目\n\n"
                f"{tasks}"
            )

            # 更新歷史記錄表格
            updated_table = history_manager.get_history_dataframe()

            return output_markdown, updated_table, None

        else:
            formatted_result = json.dumps(result, indent=2, ensure_ascii=False)
            return (
                (
                    "### 處理異常\n\n"
                    "n8n 工作流已執行,但回傳格式非預期。\n\n"
                    f"``````"
                ),
                gr.Dataframe(),
                None,
            )

    except Exception as e:
        print(f"處理過程發生錯誤:{str(e)}")
        import traceback

        traceback.print_exc()
        return (
            (
                f"## 處理失敗\n\n"
                f"處理過程中發生錯誤:\n\n"
                f"``````\n\n"
                f"請檢查音訊檔案格式是否正確,或稍後再試。"
            ),
            gr.Dataframe(),
            None,
        )

process_and_run_agent() 函式的幾個關鍵修改點有

  1. 使用 time.time() 記錄開始時間
  2. 處理完成後計算 elapsed_time
  3. 在結果中顯示處理時間
  4. 呼叫 history_manager.add_record(data) 儲存記錄
  5. 更新歷史記錄表格並回傳

3-3 加入歷史記錄相關函式

接著加入三個新的函式,分別處理載入記錄刪除記錄篩選記錄

def load_history_record(evt: gr.SelectData):
    # 根據點擊的行載入記錄
    row_index = evt.index[0]

    # 取得所有記錄
    all_records = history_manager.get_all_records()

    if row_index >= len(all_records):
        return "## 錯誤\n\n無效的選擇。", row_index

    record = all_records[row_index]

    if not record:
        return "## 錯誤\n\n找不到指定的會議記錄。", None

    # 組裝顯示內容(加入會議類型)
    timestamp = record.get("timestamp", "未知時間")
    project_name = record.get("project_name", "未指定專案")
    meeting_type = record.get("meeting_type", "一般會議")
    summary = record.get("summary", "無摘要")
    tasks = record.get("tasks", "無任務")
    notion_url = record.get("notion_url", "")
    session_id_display = record.get("session_id", "")

    output_markdown = (
        f"## 歷史會議記錄\n\n"
        f"**記錄時間**:{timestamp}\n\n"
        f"---\n\n"
        f"### 會議資訊\n\n"
        f"* **專案名稱**:{project_name}\n"
        f"* **會議類型**:{meeting_type}\n"
        f"* **Session ID**:`{session_id_display}`\n\n"
        f"🔗 **[點此前往 Notion 查看完整記錄]({notion_url})**\n\n"
        f"---\n\n"
        f"### 會議摘要\n\n"
        f"{summary}\n\n"
        f"---\n\n"
        f"### 行動項目\n\n"
        f"{tasks}"
    )

    return output_markdown, row_index


def delete_selected_record(selected_index):
    # 根據選中的索引刪除記錄
    if selected_index is None:
        return gr.Dataframe(), "## 提示\n\n請先從表格中點擊選擇一筆要刪除的記錄。", None

    # 取得所有記錄
    all_records = history_manager.get_all_records()

    if selected_index >= len(all_records):
        return gr.Dataframe(), "## 錯誤\n\n無效的選擇。", None

    # 取得要刪除的 session_id
    session_id = all_records[selected_index].get("session_id", "")

    # 執行刪除
    if history_manager.delete_record(session_id):
        # 更新表格
        updated_table = history_manager.get_history_dataframe()
        message = f"## 刪除完成\n\n成功刪除記錄。"
        return updated_table, message, None
    else:
        return gr.Dataframe(), "## 錯誤\n\n刪除失敗。", None


def refresh_history_table():
    # 重新整理歷史記錄表格
    return history_manager.get_history_dataframe(), None


def filter_history_table(time_filter, project_filter, type_filter, session_filter):
    # 根據篩選條件過濾歷史記錄
    all_records = history_manager.get_all_records()

    filtered_records = []
    for record in all_records:
        # 時間篩選
        if time_filter and time_filter.strip():
            if time_filter.lower() not in record.get("timestamp", "").lower():
                continue

        # 專案篩選
        if project_filter and project_filter.strip():
            if project_filter.lower() not in record.get("project_name", "").lower():
                continue

        # 類型篩選
        if type_filter and type_filter.strip():
            if type_filter.lower() not in record.get("meeting_type", "").lower():
                continue

        # Session ID 篩選
        if session_filter and session_filter.strip():
            if session_filter.lower() not in record.get("session_id", "").lower():
                continue

        filtered_records.append(record)

    # 轉換為 Dataframe 格式
    if not filtered_records:
        return [["查無符合條件的記錄", "", "", ""]]

    data = []
    for record in filtered_records:
        data.append(
            [
                record.get("timestamp", "未知時間"),
                record.get("project_name", "未指定專案"),
                record.get("meeting_type", "一般會議"),
                record.get("session_id", ""),
            ]
        )

    return data

3-4 更新 Gradio 介面

最後是更新 Gradio 介面,新的頁面我採用了兩欄式設計,左側顯示表格,右側顯示詳細內容,以下是完整的 Gradio 介面程式碼

# --- Gradio 介面定義 ---
with gr.Blocks(
    theme=gr.themes.Soft(),
    title="M2A Agent 會議處理平台",
    css="""
    #delete-btn {
        background-color: #dc3545 !important;
        color: white !important;
        border: none !important;
        font-weight: bold !important;
    }
    #delete-btn:hover {
        background-color: #c82333 !important;
    }
    """
) as demo:
    
    gr.Markdown("# M2A Agent 會議處理平台")
    gr.Markdown("這是一個智慧會議助理系統,能夠自動轉錄音訊、提取任務、建立 Notion 記錄並發送通知郵件。")
    
    # 使用 State 來追蹤當前選中的記錄索引
    selected_record_index = gr.State(value=None)
    
    with gr.Tabs():
        # 分頁 1:新會議處理
        with gr.Tab("新會議處理"):
            gr.Markdown(
                "### 上傳會議音訊並輸入處理指令\n"
                "**範例指令**:請生成會議摘要與提取行動任務,特別注意時間相關資訊,與出席的人員有誰,被指派任務的人有誰"
            )
            
            with gr.Row():
                with gr.Column(scale=1):
                    audio_input = gr.Audio(
                        type="filepath", 
                        label="會議音訊檔案(支援 MP3、WAV、M4A 等格式)"
                    )
                    command_input = gr.Textbox(
                        lines=4,
                        label="處理指令",
                        placeholder="請在此輸入處理指令",
                        value="請生成會議摘要與提取行動任務,特別注意時間相關資訊,與出席的人員有誰,被指派任務的人有誰"
                    )
                    submit_button = gr.Button("開始處理", variant="primary", size="lg")
                
                with gr.Column(scale=2):
                    output_display = gr.Markdown(
                        value="## 歡迎使用\n\n請上傳音訊檔案並點擊「開始處理」按鈕。",
                        label="處理結果"
                    )
        
        # 分頁 2:歷史會議記錄
        with gr.Tab("歷史會議記錄"):
            gr.Markdown("### 查看與管理過往會議記錄")
            gr.Markdown("**使用說明**:使用下方篩選器快速找到目標記錄,點擊表格中的任一行即可在右側查看詳細內容。")
            
            # 篩選器區域
            with gr.Row():
                time_filter_input = gr.Textbox(
                    label="篩選記錄時間",
                    placeholder="例如:2025-10-02",
                    scale=1
                )
                project_filter_input = gr.Textbox(
                    label="篩選專案名稱",
                    placeholder="例如:員工績效管理系統",
                    scale=1
                )
                type_filter_input = gr.Textbox(
                    label="篩選會議類型",
                    placeholder="例如:技術討論",
                    scale=1
                )
                session_filter_input = gr.Textbox(
                    label="篩選 Session ID",
                    placeholder="例如:session_",
                    scale=1
                )
            
            with gr.Row():
                filter_button = gr.Button("套用篩選", variant="secondary")
                clear_filter_button = gr.Button("清除篩選")
                refresh_button = gr.Button("重新整理")
            
            # 使用兩欄式配置
            with gr.Row():
                with gr.Column(scale=2):
                    gr.Markdown("**歷史會議記錄(點擊表格選擇記錄)**")
                    
                    history_table = gr.Dataframe(
                        value=history_manager.get_history_dataframe(),
                        headers=["記錄時間", "專案名稱", "會議類型", "Session ID"],
                        interactive=False,
                        wrap=True
                    )
                    
                    delete_button = gr.Button(
                        "🗑️ 刪除此記錄",
                        elem_id="delete-btn"
                    )
                
                with gr.Column(scale=3):
                    history_output = gr.Markdown(
                        value="## 提示\n\n請從左側表格中點擊選擇一筆歷史會議記錄。"
                    )
    
    # 設定元件互動
    submit_button.click(
        fn=process_and_run_agent,
        inputs=[audio_input, command_input],
        outputs=[output_display, history_table, selected_record_index]
    )
    
    # 點擊表格行時載入記錄並更新選中的索引
    history_table.select(
        fn=load_history_record,
        inputs=[],
        outputs=[history_output, selected_record_index]
    )
    
    # 篩選按鈕
    filter_button.click(
        fn=filter_history_table,
        inputs=[time_filter_input, project_filter_input, type_filter_input, session_filter_input],
        outputs=[history_table]
    )
    
    # 清除篩選按鈕
    clear_filter_button.click(
        fn=lambda: ("", "", "", "", history_manager.get_history_dataframe()),
        inputs=[],
        outputs=[time_filter_input, project_filter_input, type_filter_input, session_filter_input, history_table]
    )
    
    # 刪除按鈕
    delete_button.click(
        fn=delete_selected_record,
        inputs=[selected_record_index],
        outputs=[history_table, history_output, selected_record_index]
    )
    
    # 重新整理按鈕
    refresh_button.click(
        fn=refresh_history_table,
        inputs=[],
        outputs=[history_table, selected_record_index]
    )

# 啟動應用程式
if __name__ == "__main__":
    demo.launch(share=True, server_name="0.0.0.0", server_port=7860)

Step 4:測試與驗證

完成所有修改後,我進行了測試。

4-1 測試新會議處理與時間顯示

  1. 在終端機執行 python app.py 啟動 Gradio 介面
  2. 在「新會議處理」分頁上傳音訊並輸入指令
  3. 點擊「開始處理」按鈕
  4. 等待處理完成,觀察結果

結果驗證

  • 顯示「處理完成」訊息
  • 顯示處理時間(例如:「處理時間:176.5 秒」)
  • 包含專案名稱、會議類型、Notion 連結等資訊
  • 記錄自動加入歷史列表
    Gradio 1

4-2 測試歷史記錄功能

  1. 切換到「歷史會議記錄」分頁
  2. 在左側表格中看到剛才處理的記錄
  3. 點擊表格中的任一行
  4. 在右側查看詳細內容

結果驗證

  • 左側表格清楚顯示所有歷史記錄(記錄時間、專案名稱、會議類型、Session ID)
  • 點擊後右側立即顯示完整的會議摘要和任務
  • 包含 Notion 連結,可以直接點擊前往查看
    Gradio 2-1
    Gradio 2-2

4-3 測試篩選功能

  1. 在「篩選專案名稱」輸入框中輸入「雲端」
  2. 點擊「套用篩選」按鈕
  3. 觀察表格只顯示符合條件的記錄
  4. 點擊「清除篩選」按鈕,確認表格是否恢復顯示所有記錄

結果驗證

  • 表格只顯示專案名稱包含「雲端基礎架構遷移」的記錄
  • 支援模糊搜尋,不區分大小寫
  • 清除篩選後恢復顯示所有記錄
    Gradio 3-1
    Gradio 3-2
    Gradio 3-3

4-4 測試刪除功能

  1. 在表格中點擊某一行記錄
  2. 點擊「🗑️ 刪除此記錄」按鈕
  3. 確認記錄已從表格中消失

結果驗證

  • 紅色的刪除按鈕清楚可見
  • 點擊後成功刪除記錄
  • 表格自動更新,不再顯示已刪除的記錄
    Gradio 4

今天的成果總結

完成項目

  • 建立了 src/meeting_history.py 模組,實現完整的歷史記錄管理功能(新增、查詢、刪除)
  • 在 Gradio 介面中加入「歷史會議記錄」分頁,採用兩欄式設計提升使用體驗
  • 實作表格篩選功能,支援四種篩選條件(時間、專案、類型、Session ID),並支援模糊搜尋
  • 加入刪除記錄功能,使用紅色按鈕提升辨識度
  • 顯示處理時間,讓使用者清楚了解系統執行狀態
  • 在 n8n 建立「組合 Webhook 回應資料」節點,提供完整且結構化的資料
  • 採用 gr.State 追蹤選中的記錄,讓刪除功能更加穩定

心得

今天透過將會議歷史管理抽離成獨立的 MeetingHistory 類別,不僅讓程式碼更加專注,如果未來要加入新功能,也會變得非常容易。

在「會議處理」與「歷史會議記錄」的部分採用兩欄式介面的設計,讓使用者體驗大幅提升,左側瀏覽、右側詳情的配置,非常直覺,而且新增了篩選功能,可以解決當記錄變多之後難以查找的問題,也在處理會議時透過顯示正在的處理時間,使得使用者了解系統的執行效率,不會因為等待而感到焦慮。

不只如此,我在開發的過程中,我也學到了 Gradio 的一些眉角,例如使用 elem_id 配合 CSS 來自訂按鈕樣式、使用 gr.SelectData 來處理表格點擊事件等,這些都是實戰的寶貴經驗。

🎯 明天計劃

嘗試實作持續性對話,讓 Agent 能針對載入的歷史會議進行補充提問與修改,同時當指令或內容不明確時,它會主動反問,實現更智慧的互動。


上一篇
Day 20 郵件優化 — 加入直達按鈕與強化迴圈穩定性
系列文
打造基於 MCP 協議與 n8n 工作流的會議處理 Agent21
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言