🤖 MCP SQL 聊天機器人和儀表板：讓你的資料庫會說話！ 🌟 這個神奇的工具是什麼？ 想像一下，你可以直接問你的資料庫：「今年哪個月銷售最好？」或是「我有多少位客戶？」然後它就像朋友一樣回答你！不僅如此，它還能幫你畫出漂亮的圖表，讓數字變成一目瞭然的視覺故事。 這就是我們這個 MCP SQL 聊天機器人 的威力！它結合了： 🗣️ 自然語言對話（像跟人聊天一樣問問題） 🎯 智能 SQL 轉換（AI 幫你寫 SQL 語句） 📊 自動儀表板生成（美麗的圖表自動出現） ⚡ 即時回應（秒速得到答案） ✨ 它有哪些超能力？ 1. 🎤 自然語言界面 - 像聊天一樣問問題 再也不用記 SQL 語法！ 直接用中文問：「給我看看最近一個月的訂單」 AI 自動翻譯：把你的問題變成專業的 SQL 查詢 智能理解：就算你的問題不夠精確，AI 也能猜到你想要什麼 2. 🔄 SQL 查詢生成 - AI 當你的 SQL 專家 自動生成：根據你的問題寫出完美的 SQL 安全第一：只執行讀取操作，不會破壞你的資料 語法優化：生成的 SQL 又快又準確 3. 📈 互動儀表板 - 數據變身美麗圖表 自動分析：掃描你的資料庫，找出重要趨勢 智能視覺化：根據資料類型選擇最適合的圖表 互動體驗：可以點擊、縮放、篩選的動態圖表 4. ⚡ 即時結果 - 問了就有答案 秒速回應：不用等待，問題馬上有解答 持續對話：可以連續提問，就像真正的對話 5. 💾 可下載儀表板 - 帶走你的分析成果 一鍵下載：把漂亮的儀表板存成 HTML 檔案 離線瀏覽：沒網路也能看你的數據分析 分享方便：輕鬆分享給同事或老闆 🎒 開始前的準備工作 你需要準備這些東西： 🐍 軟體環境 Python 3.10 以上（就像需要一個聰明的大腦） UV 套件管理器（幫你整理所有工具的管家） 可以連接的 SQL 資料庫（你的數據寶庫） 🔑 API 金鑰（AI 的通行證） 你需要以下其中一個： OpenAI API 金鑰（ChatGPT 家族的通行證） Groq API 金鑰（另一個超快 AI 服務） 🔧 設定你的秘密檔案 在專案資料夾裡建立一個 .env 檔案（這是存放秘密的地方）： # 🤖 AI 模型設定 MODEL_API_KEY=你的_API_金鑰        # AI 的身份證 MODEL_ID=你要用的模型名稱           # 例如：gpt-4 或 llama-3.3-70b-versatile # 🗄️ 資料庫連線設定 DB_HOST=你的資料庫主機地址          # 資料庫住在哪裡 DB_USER=資料庫使用者名稱           # 你的資料庫帳號 DB_PASSWORD=資料庫密碼             # 你的資料庫密碼 DB_NAME=資料庫名稱                 # 你要連接的資料庫 💡 小提醒：這個檔案包含敏感資訊，千萬不要分享給別人或上傳到網路上！ 🚀 安裝和設定步驟 第一步：取得程式碼 # 📥 下載專案 git clone cd mcp-agent-experiment 第二步：安裝所有需要的工具 # 🔧 使用 UV 安裝所有依賴套件 uv sync 這個指令會自動： ✅ 建立虛擬環境（如果還沒有的話） ✅ 安裝所有需要的 Python 套件 ✅ 確保版本相容性 第三步：建立設定檔 在專案根目錄建立 .env 檔案，填入你的資料庫和 API 資訊： DB_HOST=localhost                    # 如果資料庫在你的電腦上 DB_USER=your_username               # 你的資料庫使用者名稱   DB_PASSWORD=your_secret_password    # 你的資料庫密碼 DB_NAME=your_database_name          # 資料庫名稱 MODEL_API_KEY=sk-xxx...             # 你的 OpenAI 或 Groq API 金鑰 MODEL_ID=gpt-4                      # 想使用的 AI 模型 🎮 開始使用 啟動你的數據助手 1. 啟動虛擬環境 # 🐧 Linux/macOS 用戶 source .venv/bin/activate # 🪟 Windows 用戶   .venv\Scripts\activate 2. 啟動應用程式 # 🚀 發射！ streamlit run app.py 3. 開啟瀏覽器 看到終端機顯示的網址（通常是 http://localhost:8501），用瀏覽器打開就能看到你的數據助手了！ 📱 使用教學 🗣️ 聊天機器人模式 如何和你的資料庫聊天： 打開聊天機器人頁籤 用自然語言提問，例如： 「我有多少筆訂單？」 「哪個產品賣得最好？」 「上個月的營收是多少？」 「給我看看所有客戶的資料」 「最近一週有什麼趨勢？」 看 AI 回答：它會： 🧠 理解你的問題 🔄 轉換成 SQL 查詢 📊 執行查詢並分析結果 💬 用人話告訴你答案 💡 提問小技巧： 具體一點：「這個月的銷售額」比「銷售情況」更容易理解 可以追問：「那上個月呢？」、「可以看看詳細資料嗎？」 別怕犯錯：AI 很聰明，會盡力理解你的意思 📊 儀表板模式 自動生成美麗的數據視覺化： 切換到儀表板頁籤 點擊「生成儀表板」按鈕 等待魔法發生： 🔍 AI 掃描你的整個資料庫 🎯 找出最重要的數據指標 📈 選擇最適合的圖表類型 🎨 生成漂亮的互動式儀表板 探索你的數據： 📊 各種圖表：長條圖、圓餅圖、折線圖 📋 重要數據表格 🎛️ 互動功能：點擊、篩選、縮放 下載儀表板： 點擊「下載儀表板 HTML」按鈕 💾 儲存成 HTML 檔案 📤 可以分享給同事或客戶 🧩 程式架構解密 🎭 三個主要角色： 1. 🤖 agent.py - 聊天機器人大腦 這是什麼： 負責理解你的問題並轉換成 SQL 的智慧助手 它的工作： 🧠 理解自然語言：把你的中文問題理解成資料庫查詢需求 🔄 轉換成 SQL：自動寫出正確的 SQL 語句 🗄️ 執行查詢：使用 MCP 工具包安全地查詢資料庫 ⚡ 非同步處理：同時處理多個請求，速度超快 💬 自然語言回應：把查詢結果翻譯成人話告訴你 # 匯入必要的套件 import asyncio # 用於異步操作 import os # 用於與作業系統互動，特別是讀取環境變數 from dotenv import load_dotenv # 用於從 .env 檔案載入環境變數 from textwrap import dedent # 用於處理多行字串的縮排 from agno.agent import Agent, RunResponse # 從 agno 套件匯入 Agent 和 RunResponse from agno.models.groq import Groq # 從 agno 套件匯入 Groq 模型 from agno.models.openai import OpenAIChat # 從 agno 套件匯入 OpenAI 模型 from agno.tools.mcp import MCPTools # 從 agno 套件匯入 MCP 工具 from mcp import ClientSession, StdioServerParameters # 從 mcp 套件匯入客戶端會話和標準輸入輸出伺服器參數 from mcp.client.stdio import stdio_client # 從 mcp 套件匯入標準輸入輸出客戶端 from agno.utils.log import logger # 從 agno 套件匯入日誌記錄器 from typing import Optional # 從 typing 模組匯入 Optional from llm_model import get_model # 從 llm_model 模組匯入 get_model 函數 # 定義給語言模型的指令 INSTRUCTIONS = dedent( """\ 你是一個智慧 SQL 助理，可以透過 MCP 工具存取資料庫。 你的工作是： 1. 理解使用者關於資料的問題或請求。 2. 如有需要，使用 `get_schema` 工具來檢索資料庫的結構。 3. 使用正確的資料表和欄位名稱，生成一個有效的 SQL `SELECT` 查詢。 4. 使用 `read_query` 工具來執行你的查詢並檢索結果。 5. 分析結果，並以**清晰、自然的語言摘要**回應資料。 - 這個回應應該簡單、準確且易於理解。 - 專注於關鍵見解、趨勢、計數、比較或重點。 - 包含數字或觀察結果，而不僅僅是重述。 - 除非特別要求，否則避免使用技術術語或原始資料。 限制： - 僅使用 SELECT 查詢。 - 不要執行 INSERT、UPDATE、DELETE 或任何修改操作。 - 除非使用者明確要求，否則不要返回原始 SQL 或結果表。 - 即使結果為空或為零，也始終返回一個友善的人類可讀解釋。 你可以使用的工具： - `get_schema`: 檢索資料庫的完整結構。 - `read_query`: 執行一個 SELECT 查詢並以字典列表的形式返回結果。 使用者查詢範例： - "上週有多少新使用者註冊？" - "今年哪些產品的收入最高？" - "顯示 2023 年訂單的每月明細。" - "按客戶數量排名前 5 的國家是哪些？" 你必須回應： - 一個寫得很好、友善的結果摘要。 - 如果適用，可以包含一個簡短的圖表描述（例如，“這可以用長條圖顯示。”）。 - 沒有其他內容 — 不解釋你是如何得到結果的。 從閱讀使用者的問題開始，並據此進行操作。 """ ) # 從 .env 檔案載入環境變數 load_dotenv() MODEL_ID = os.getenv("MODEL_ID") if not MODEL_ID: raise ValueError("環境變數 GROQ_API_KEY 未設定。") MODEL_API_KEY = os.getenv("MODEL_API_KEY") if not MODEL_API_KEY: raise ValueError("環境變數 MODEL_API_KEY 未設定。") # 如果環境變數中未指定，則使用預設模型 ID DEFAULT_MODEL_ID = "llama-3.3-70b-versatile" async def run_agent(message: str, model_id: Optional[str] = None) -> RunResponse: """ 使用給定的訊息執行代理並返回回應。 Args: message (str): 要傳送給代理的訊息。 model_id (Optional[str]): 要使用的語言模型的 ID。預設為 DEFAULT_MODEL_ID。 Returns: RunResponse: 代理的回應。 Raises: ValueError: 如果任何必要的資料庫環境變數未設定。 RuntimeError: 如果連接到 MCP 伺服器時發生錯誤。 """ # 檢查必要的資料庫環境變數 required_vars = ["DB_HOST", "DB_USER", "DB_PASSWORD", "DB_NAME"] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: raise ValueError(f"缺少必要的環境變數：{', '.join(missing_vars)}") try: cmd = f'uvx mcp-sql-server --db-host {os.getenv("DB_HOST")} --db-user {os.getenv("DB_USER")} --db-password {os.getenv("DB_PASSWORD")} --db-database {os.getenv("DB_NAME")}' async with MCPTools(command=cmd) as mcp_tools: agent = Agent( model=get_model(MODEL_ID, MODEL_API_KEY), tools=[mcp_tools], instructions=INSTRUCTIONS, markdown=True, show_tool_calls=True, ) response = await agent.arun(message) return response except Exception as e: raise RuntimeError(f"連接 MCP 伺服器或執行代理時發生錯誤：{e}") from e async def main(): """ 代理的範例用法。 """ try: # 使用範例查詢執行代理 response = await run_agent("取得所有員工的列表") logger.info(f"代理回應: {response.content}") except ValueError as ve: logger.error(f"配置錯誤: {ve}") except RuntimeError as re: logger.error(f"執行時錯誤: {re}") except Exception as e: logger.error(f"發生未預期的錯誤: {e}") if __name__ == "__main__": asyncio.run(main()) 2. 📊 dashboard_agent.py - 儀表板設計師 這是什麼： 專門負責視覺化和儀表板生成的藝術家 它的超能力： 🔍 架構分析：掃描你的資料庫，找出重要資料表和欄位 🎯 指標識別：自動發現關鍵績效指標（KPI） 📈 智能視覺化：根據資料類型選擇最佳圖表 🎨 美化設計：使用 Chart.js 和 Tailwind CSS 製作漂亮的介面 🌐 HTML 生成：產出可以獨立運行的互動式網頁 # 匯入必要的套件 import asyncio # 用於異步操作 import os # 用於與作業系統互動，特別是讀取環境變數 import json # 用於處理 JSON 格式的資料 from textwrap import dedent # 用於處理多行字串的縮排 from typing import Optional, Dict, Any # 從 typing 模組匯入類型提示 import traceback # 用於追蹤錯誤堆疊 from agno.models.openai import OpenAIChat # 從 agno 套件匯入 OpenAI 模型 from dotenv import load_dotenv # 用於從 .env 檔案載入環境變數 from agno.agent import Agent, RunResponse # 從 agno 套件匯入 Agent 和 RunResponse from agno.models.groq import Groq # 從 agno 套件匯入 Groq 模型 from agno.tools.mcp import MCPTools # 從 agno 套件匯入 MCP 工具 from mcp import ClientSession, StdioServerParameters # 從 mcp 套件匯入客戶端會話和標準輸入輸出伺服器參數 from mcp.client.stdio import stdio_client # 從 mcp 套件匯入標準輸入輸出客戶端 from agno.utils.log import logger # 從 agno 套件匯入日誌記錄器 from llm_model import get_model # 從 llm_model 模組匯入 get_model 函數 # --- 組態設定 --- MARKDOWN = False # 設定 Markdown 輸出為 False SHOW_TOOL_CALLS = False # 設定不顯示工具呼叫 # 從 .env 檔案載入環境變數 load_dotenv() MODEL_ID = os.getenv("MODEL_ID") if not MODEL_ID: raise ValueError("環境變數 GROQ_API_KEY 未設定。") MODEL_API_KEY = os.getenv("MODEL_API_KEY") if not MODEL_API_KEY: raise ValueError("環境變數 MODEL_API_KEY 未設定。") # 定義視覺化類型 VISUALIZATION_TYPES = { "time_series": "隨時間變化的資料（銷售趨勢、使用者增長）", "bar_chart": "比較類別或群組（按地區銷售、按類別產品）", "pie_chart": "顯示組成或比例（市場份額、預算分配）", "scatter_plot": "兩個變數之間的關係（價格與評分、年齡與薪資）", "heatmap": "顯示多個維度的模式或強度（按小時/天的活動）", "table": "需要精確值的詳細單筆記錄或匯總", "gauge": "具有目標值的 KPI（銷售目標、客戶滿意度）", "funnel": "具有流失的順序流程步驟（銷售漏斗、使用者旅程）", } # --- 提示 --- # 資料庫分析和 SQL 生成的指令 INSTRUCTIONS_DB_ANALYSIS_AND_SQL = dedent( """\ 你是一位專業的 SQL 資料分析師和儀表板設計師。請分析資料庫結構並提供一份全面的 JSON 報告，包含： 1. **資料庫領域：** 根據資料表和欄位名稱，識別最可能的領域（例如，銷售、人力資源、庫存、旅遊）。 2. **關鍵指標：** 列出與此領域相關的最重要的 KPI/指標。 3. **視覺化：** 為每個指標推薦一個合適的圖表類型，並簡要解釋其適用原因。 4. **SQL 查詢：** 根據資料庫結構為每個指標生成 SQL 查詢。 5. **儀表板元件：** 建議儀表板中應包含哪些元件（例如，圖表、表格、篩選器）。 **流程：** - 使用 `get_schema` 工具檢索結構。 - 分析資料表和欄位名稱以確定領域。 - 根據領域識別相關指標，並為每個指標提供： - 名稱 - 描述 - 視覺化類型（從 {visualization_types} 中選擇） - 視覺化理由 - 使用正確的資料表/欄位名稱的 SQL 查詢 - 以有效的 JSON 格式返回所有輸出，不要添加任何額外文字： ```json {{ "domain": "識別出的領域", "key_metrics": [ {{ "metric": "指標名稱", "description": "此指標顯示的內容", "visualization_type": "例如 bar_chart", "visualization_rationale": "此圖表適合的原因", "sql": "SELECT ... FROM ... WHERE ... GROUP BY ..." }} ], "dashboard_components": ["元件1", "元件2"] }} ``` **指南：** - 簡潔且具體。 - 確保 SQL 查詢有效、乾淨且與結構相符。 - 僅使用 `get_schema` 工具 — 不要超出此範圍做任何假設。 - 僅輸出 JSON。不要有額外的評論。 **可用工具：** - `get_schema`: 檢索資料庫結構。 """ ).format(visualization_types=json.dumps(VISUALIZATION_TYPES)) # 僅限 SQL 指標資料 JSON 的指令 INSTRUCTIONS_SQL_METRIC_DATA_JSON_ONLY = dedent( """\ 你是一位資深資料分析師。 你將收到： - 一個包含多個指標的 JSON 物件，每個指標都有名稱、描述、視覺化類型和一個 SQL 查詢。 - 使用 `read_query` 工具存取 SQL 資料庫。 你的任務是： 1. 使用 `read_query` 工具執行每個 SQL 查詢。 2. 對於每個指標： - 擷取名稱、描述、視覺化類型和結果資料。 3. 如果結果資料為空，則不要將該項目添加到 JSON 中。 4. 返回一個包含所有指標及其對應結果資料的最終 JSON 回應。 **輸出格式：** 以以下結構返回一個單一的 JSON 物件： ```json {{ "metrics": [ {{ "metric": "指標名稱", "description": "指標的描述", "visualization_type": "bar_chart | line_chart | pie_chart | table", "data": [ {{"column1": value, "column2": value }}, ... ] }} ] }} ``` **可用工具：** - `read_query`: 執行一個 SELECT SQL 查詢並以字典列表的形式返回結果。 **重要：** - 僅返回有效的 JSON。 - 不要返回 HTML、解釋或任何其他文字。 - 如果查詢未返回任何資料，則在 `data` 中返回一個空列表。 """ ) # 從資料渲染儀表板的指令 INSTRUCTIONS_RENDER_DASHBOARD_FROM_DATA = dedent( """\ 你是一位資深的儀表板 UI 工程師。 你將收到： - 一個包含指標陣列的 JSON 物件。 - 每個指標包括：名稱、描述、視覺化類型和一個資料列列表（已從 SQL 取得）。 你的任務是： 1. 渲染一個完整、響應式的 HTML 儀表板。 2. 對於每個指標： - 顯示指標標題和描述。 - 如果 `visualization_type` 是 `bar_chart`、`line_chart` 或 `pie_chart`，使用 Chart.js 渲染一個響應式圖表。 - 如果資料不適合圖表（或類型是 `table`），則渲染一個帶有樣式的 HTML 表格。 3. 使用 **Tailwind CSS** 進行佈局、響應式設計和視覺美化。 4. 確保每個圖表或表格都在一個獨立的卡片式區塊內。 5. 使佈局對行動裝置友善、優雅且易於閱讀。 **要使用的工具：** - 使用 **Chart.js**（來自 CDN 的內聯腳本標籤）。 - 使用 **Tailwind CSS**（透過 CDN）。 **輸出格式：** 僅返回一個有效的、完整的 HTML 文件作為單一字串。不要有解釋或 JSON — 只有 HTML。 **重要：** - 確保 HTML 有效並能在現代瀏覽器中乾淨地渲染。 - 所有圖表必須是響應式的。 - 使用直觀的顏色和佈局。 - 不要包含額外的解釋或評論 — 只有 HTML。 """ ) # --- 輔助函數 --- def clean_json(data): """從 JSON 字串的開頭移除多餘的文字並解析它。""" start_index = data.find('{') if start_index == -1: start_index = data.find('[') if start_index == -1: raise ValueError("在字串中找不到 JSON 結構") cleaned_data = data[start_index:] return cleaned_data def validate_dashboard_json(json_str: str) -> Dict[str, Any]: """驗證儀表板 JSON 的結構。 Args: json_str: 要驗證的 JSON 字串。 Returns: 如果有效，則為解析後的 JSON 資料。 Raises: ValueError: 如果 JSON 無效或缺少必要鍵。 """ try: data = json.loads(json_str) # 檢查必要鍵 if not all(key in data for key in ["domain", "key_metrics", "dashboard_components"]): raise ValueError("JSON 中缺少必要鍵") # 檢查 key_metrics 是否為列表 if not isinstance(data["key_metrics"], list): raise ValueError("key_metrics 必須是列表") # 檢查 dashboard_components 是否為列表 if not isinstance(data["dashboard_components"], list): raise ValueError("dashboard_components 必須是列表") # 檢查每個指標是否具有正確的鍵 for metric in data["key_metrics"]: if not all(key in metric for key in ["metric", "description", "visualization_type", "visualization_rationale", "sql"]): raise ValueError("指標中缺少必要鍵") return data except json.JSONDecodeError as e: raise ValueError(f"無效的 JSON 格式：{e}") except ValueError as e: raise ValueError(f"無效的 JSON 結構：{e}") async def run_mcp_agent(message: str, instructions: str, max_retries: int = 3) -> RunResponse: """使用重試邏輯執行 MCP 代理。 Args: message: 要傳送給代理的訊息。 instructions: 代理的指令。 max_retries: 最大重試次數。 Returns: 代理的回應。 """ retries = 0 while retries ###draft_code_symbol_lessthen### max_retries: try: cmd = f'uvx mcp-sql-server --db-host {os.getenv("DB_HOST")} --db-user {os.getenv("DB_USER")} --db-password {os.getenv("DB_PASSWORD")} --db-database {os.getenv("DB_NAME")}' async with MCPTools(command=cmd) as mcp_tools: agent = Agent( model=get_model(MODEL_ID, MODEL_API_KEY), tools=[mcp_tools], instructions=instructions, markdown=MARKDOWN, show_tool_calls=SHOW_TOOL_CALLS, ) response = await agent.arun(message) return response except json.JSONDecodeError as e: logger.error(f"JSON 解碼錯誤：{e}\n{traceback.format_exc()}") return RunResponse(content=json.dumps({"error": "無效的 JSON 格式", "details": str(e)})) except ValueError as e: logger.error(f"值錯誤：{e}\n{traceback.format_exc()}") retries += 1 if retries == max_retries: return RunResponse(content=json.dumps({"error": "多次重試後 JSON 結構無效", "details": str(e)})) logger.warning(f"JSON 結構錯誤後重試 ({retries}/{max_retries})：{e}") except Exception as e: logger.error(f"未預期的錯誤：{e}\n{traceback.format_exc()}") return RunResponse(content=json.dumps({"error": "未預期的錯誤", "details": str(e)})) return RunResponse(content=json.dumps({"error": "多次重試後無法獲取有效的 JSON"})) # --- 主要函數 --- async def analyze_database(message: str, max_retries: int = 3) -> RunResponse: """分析資料庫結構並返回 JSON 報告。 Args: message: 要傳送給代理的訊息。 max_retries: 最大重試次數。 Returns: 代理的回應。 """ return await run_mcp_agent( message=message, instructions=INSTRUCTIONS_DB_ANALYSIS_AND_SQL, max_retries=max_retries, ) async def get_data_from_database(analysis_json: str, max_retries: int = 3) -> RunResponse: """根據分析 JSON 從資料庫獲取資料。 Args: analysis_json: 包含 SQL 查詢的 JSON。 max_retries: 最大重試次數。 Returns: 代理的回應。 """ return await run_mcp_agent( message=analysis_json, instructions=INSTRUCTIONS_SQL_METRIC_DATA_JSON_ONLY, max_retries=max_retries, ) async def generate_html_dashboard(data_json: str) -> RunResponse: """從資料 JSON 生成 HTML 儀表板。 Args: data_json: 包含儀表板資料的 JSON。 Returns: 包含 HTML 的代理回應。 """ logger.info(f"從資料生成 HTML 儀表板：{data_json}") agent = Agent( model=get_model(MODEL_ID, MODEL_API_KEY), instructions=INSTRUCTIONS_RENDER_DASHBOARD_FROM_DATA, markdown=MARKDOWN, show_tool_calls=SHOW_TOOL_CALLS, ) response = await agent.arun(data_json) return response async def run_agent(message: str, max_retries: int = 3) -> RunResponse: """儀表板代理的主要進入點。 Args: message: 要傳送給代理的訊息。 max_retries: 資料庫分析的最大重試次數。 Returns: 包含 HTML 的代理回應。 Raises: ValueError: 如果任何必要的資料庫環境變數未設定。 """ required_vars = ["DB_HOST", "DB_USER", "DB_PASSWORD", "DB_NAME"] missing_vars = [var for var in required_vars if not os.getenv(var)] if missing_vars: raise ValueError(f"缺少必要的環境變數：{', '.join(missing_vars)}") analysis_response = await analyze_database(message, max_retries) analysis_json_str = analysis_response.content.strip() logger.info(f"分析 JSON: {analysis_json_str}") # 如果存在，移除程式碼區塊標記 if analysis_json_str.startswith("```json"): analysis_json_str = analysis_json_str[7:] if analysis_json_str.endswith("```"): analysis_json_str = analysis_json_str[:-3] cleaned_json_str = clean_json(analysis_json_str) try: validate_dashboard_json(cleaned_json_str) except ValueError as e: logger.error(f"無效的 JSON: {e}\n{analysis_json_str}") if "多次重試後 JSON 結構無效" in str(e): return RunResponse(content=json.dumps({"error": "多次重試後無法獲取有效的 JSON", "details": str(e)})) else: logger.warning("正在重試以獲取正確的 JSON") analysis_response = await analyze_database(message, max_retries) analysis_json_str = analysis_response.content.strip() if analysis_json_str.startswith("```json"): analysis_json_str = analysis_json_str[7:] if analysis_json_str.endswith("```"): analysis_json_str = analysis_json_str[:-3] try: cleaned_json_str = clean_json(analysis_json_str) validate_dashboard_json(cleaned_json_str) except ValueError as e: logger.error(f"無效的 JSON: {e}\n{analysis_json_str}") return RunResponse(content=json.dumps({"error": "多次重試後無法獲取有效的 JSON", "details": str(e)})) data_response = await get_data_from_database(analysis_json_str) data_json_str = data_response.content.strip() logger.info(f"資料 JSON: {data_json_str}") if data_json_str.startswith("```json"): data_json_str = data_json_str[7:] if data_json_str.endswith("```"): data_json_str = data_json_str[:-3] html_response = await generate_html_dashboard(data_json_str) logger.info(f"HTML 程式碼: {html_response}") html_str = html_response.content.strip() if html_str.startswith("```html"): html_str = html_str[7:] if html_str.endswith("```"): html_str = html_str[:-3] return RunResponse(content=html_str) # --- 範例用法 --- async def main(): """儀表板代理的範例用法。""" try: response = await run_agent("分析我的資料庫並建議一個儀表板") logger.info(f"代理回應: {response.content}") except ValueError as ve: logger.error(f"配置錯誤: {ve}") except RuntimeError as re: logger.error(f"執行時錯誤: {re}") except Exception as e: logger.error(f"發生未預期的錯誤: {e}") if __name__ == "__main__": asyncio.run(main()) 3. 🎪 app.py - 舞台總監 這是什麼： 主要的 Streamlit 應用程式，負責整個使用者體驗 它負責： 🖥️ 使用者介面：提供聊天機器人和儀表板的操作界面 🔄 事件循環管理：確保非同步操作在 Streamlit 中正常運作 💾 會話狀態：記住你的對話歷史和設定 🤝 互動處理：處理按鈕點擊、文字輸入等使用者操作 # 匯入必要的套件 import streamlit as st # 用於建立 Web App 的主要套件 import asyncio # 用於異步操作 import json # 用於處理 JSON 格式的資料 from agent import run_agent # 從 agent 模組匯入 run_agent 函數 from dashboard_agent import run_agent as run_dashboard_agent # 從 dashboard_agent 模組匯入 run_agent 函數並重新命名 from typing import Dict, Any # 從 typing 模組匯入 Dict 和 Any # 為異步操作建立一個事件循環 loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # 設定 Streamlit 頁面配置 st.set_page_config( page_title="MCP SQL 聊天機器人 & 儀表板", page_icon="🧊", layout="wide", initial_sidebar_state="expanded", menu_items={ "Report a bug": "https://github.com/vivekpathania/ai-experiments/issues" } ) # 標題和頁首 st.title("🧊 MCP SQL 聊天機器人 & 儀表板") st.write( "使用自然語言與 SQL 資料庫互動並視覺化資料庫趨勢。" "此應用程式連接到 MCP SQL 伺服器以執行您的查詢、提供結果並生成儀表板。" ) # 側邊欄 with st.sidebar: st.header("ℹ️ 關於") st.markdown( "此應用程式允許您使用自然語言與 SQL 資料庫互動並視覺化資料。" "它利用 MCP (模組化控制協議) 連接到 SQL 伺服器，" "執行您的查詢、呈現結果並生成互動式儀表板。" ) st.markdown( "**如何使用：**\n" "1. 確保 MCP SQL 伺服器正在運行。\n" "2. 在下方的聊天框中提出您的問題。\n" "3. 應用程式將生成並執行適當的 SQL 查詢。\n" "4. 結果將顯示在聊天中。\n" "5. 導航到「儀表板」標籤頁以查看資料庫洞察。" ) st.markdown("👨‍💻 作為 AI 實驗開發。在 [GitHub](https://github.com/vivekpathania/ai-experiments) 上貢獻") def display_html_dashboard(html_content: str): """在 Streamlit 中顯示 HTML 儀表板。""" st.components.v1.html(html_content, height=800, scrolling=True) # --- 主要應用程式邏輯 --- # 初始化會話狀態 if "messages" not in st.session_state: st.session_state.messages = [] if "conversation_state" not in st.session_state: st.session_state.conversation_state = {} if "dashboard_html" not in st.session_state: st.session_state.dashboard_html = "" # 標籤頁 tab1, tab2 = st.tabs(["聊天機器人", "儀表板"]) # 聊天機器人標籤頁 with tab1: # 為聊天訊息建立一個容器 chat_container = st.container() # 為聊天容器添加一個 id chat_container.markdown('###draft_code_symbol_lessthen###div id="chat-container">###draft_code_symbol_lessthen###/div>', unsafe_allow_html=True) # 在容器內顯示聊天記錄 with chat_container: for message in st.session_state.messages: with st.chat_message(message["role"]): st.markdown(message["content"]) # 使用者輸入 user_query = st.chat_input("詢問有關資料庫的問題（例如，「列出所有員工」）", key="chat_input") if user_query: st.session_state.messages.append({"role": "user", "content": user_query}) # 立即顯示使用者訊息 with chat_container: with st.chat_message("user"): st.markdown(user_query) with st.spinner("正在從資料庫獲取回應..."): try: resp = loop.run_until_complete(run_agent(user_query)) except Exception as e: st.error(f"發生錯誤：{e}") resp = None if resp: # 將助理的回應添加到會話狀態 st.session_state.messages.append({"role": "assistant", "content": resp.content}) # 立即顯示助理訊息 with chat_container: with st.chat_message("assistant"): st.markdown(resp.content) # 儀表板標籤頁 with tab2: st.header("資料庫儀表板") if st.button("生成儀表板"): with st.spinner("正在分析資料庫並生成儀表板..."): try: dashboard_response = loop.run_until_complete(run_dashboard_agent("分析我的資料庫並建議一個儀表板")) # 如果存在，移除 ```html 標籤 __html = dashboard_response.content if __html.startswith("```html"): __html = __html.replace("```html", "") if __html.endswith("```"): __html = __html.replace("```", "") st.session_state.dashboard_html = __html display_html_dashboard(__html) except Exception as e: st.error(f"發生錯誤：{e}") if st.session_state.dashboard_html: st.download_button( label="下載儀表板 HTML", data=st.session_state.dashboard_html, file_name="dashboard.html", mime="text/html", ) ⚙️ 技術亮點 🚀 非同步操作 - 讓一切都很快 同時處理多件事：一邊分析資料，一邊準備圖表 不會卡住：UI 永遠保持響應，不會讓你等到天荒地老 效率最大化：充分利用電腦資源 🔄 事件循環管理 - 解決相容性問題 Streamlit 友善：讓非同步程式碼在 Streamlit 中完美運作 持久連線：維持穩定的資料庫連接 記憶體優化：避免記憶體洩漏和效能問題 🤝 MCP 整合 - 現代化的資料庫連接 最新標準：使用最先進的 MCP 協定 安全可靠：內建安全機制，保護你的資料 擴展性強：未來可以輕鬆加入更多功能 🧠 多模型支援 - 選擇最適合的 AI OpenAI 模型：GPT-4、GPT-3.5 等經典選擇 Groq 模型：超快速度，適合即時互動 靈活切換：隨時更換不同的 AI 模型 🛠️ 自訂化和擴展 🎨 想要修改界面？ 編輯 app.py 中的 Streamlit 組件 修改 CSS 樣式讓界面更符合你的品味 加入新的頁籤或功能 📊 想要新的圖表類型？ 在 dashboard_agent.py 中加入新的視覺化邏輯 使用其他圖表庫如 Plotly、Matplotlib 客製化你專屬的數據展示方式 🤖 想要更聰明的 AI？ 在 agent.py 中調整提示詞（prompt） 加入更多上下文資訊 訓練 AI 理解你特定領域的術語 🚨 疑難排解 常見問題解決方案： 🔌 連線問題 症狀：無法連接到資料庫 解法： 檢查 .env 檔案中的資料庫設定 確認資料庫服務正在運行 檢查防火牆設定 🤖 AI 回應異常 症狀：AI 給出奇怪的答案 解法： 確認 API 金鑰正確且有額度 試試更具體的問題 檢查網路連線 📊 圖表顯示問題 症狀：儀表板空白或錯誤 解法： 確認資料庫中有足夠的資料 檢查瀏覽器控制台的錯誤訊息 重新整理頁面 🎉 開始你的數據探索之旅！ 現在你已經有了這個強大的數據助手，可以： 📈 商業分析師的最愛 快速生成業績報告 找出銷售趨勢和異常 監控關鍵績效指標 👩‍💻 開發者的好幫手 快速檢查資料庫內容 除錯資料問題 生成資料檔案 📊 管理者的儀表板 一鍵了解業務狀況 美麗的視覺化報告 隨時掌握數據脈動 🎓 學習者的實驗室 練習 SQL 查詢技巧 了解資料視覺化 體驗 AI 應用開發 💫 最後的話 這個工具就像給你的資料庫安裝了一個聰明的翻譯機和畫家。不管你是技術新手還是資深專家，都能輕鬆地從數據中獲得見解。 記住： 🤔 大膽提問：沒有愚蠢的問題，只有尚未被發現的見解 🎨 享受視覺化：讓數字說故事，讓圖表傳達訊息 🔍 持續探索：每個數據點都可能隱藏著寶藏 🚀 不斷實驗：試試不同的問法，發現新的可能性 現在就開始你的數據對話之旅吧！你的資料庫正等著和你聊天呢！ 🎉 💡 小提醒：這個專案使用 MIT 授權，代表你可以自由使用、修改和分享。如果你做出了很酷的改進，也歡迎貢獻回來讓大家一起受益！