通過 MCP（Multi-Agent Collaboration Platform，多智能體協作平臺）+ 數據庫，實現一個AI檢索和分析系統。

一、系統目標

實現通過 AI 多智能體對結構化（數據庫）和非結構化（文檔、文本）數據進行：

• ? 智能檢索（自然語言問答、SQL 自動生成）

• ? 智能分析（總結、趨勢識別、異常檢測、風險分析等）

• ? 多模型協同分析 + 決策建議

• ? 結果推送（如 ECM、BPM 系統對接、前端儀表盤展示）?

二、技術組成與框架結構

整體結構如下：

[ 用戶 ] ? [ MCP Server ] ? [ Agent / LangChain Tool ] ? [ 數據庫 / 文檔系統 ]?[ 多模型分析引擎 ]

1. MCP Server

• 支持用戶通過自然語言輸入查詢意圖（如：“分析近三月銷售異常”）

• 調用 Agent 工廠，動態分配合適的 AI Agent 進行任務分解與協作

• 負責中控調度、緩存上下文、流程記錄和結果集成

2. 數據檢索模塊（LangChain + 數據庫工具）

• 使用 LangChain SQLDatabaseChain 或 SQLAgent 工具對接數據庫

• 支持自然語言轉 SQL（通過 GPT/Claude 等模型）

• 數據庫返回結構化數據（DataFrame）

• 對于多表分析，使用 SQLGraph 工具做自動聯表推理

3. AI 分析 Agent

• 負責對結構化數據進行統計分析、趨勢識別、聚類等

• 可通過 Python Agent（LangChain 的 Pandas DataFrame Agent）進行數據分析（用模型調用 pandas、matplotlib 等進行分析）

4. 多模型投票機制（可選）

• 通過調用 Claude/GPT/Mistral 等多個模型

• 對分析結果進行比對和投票（如多個模型都認為某條數據是風險項，則置信度更高）

5. 分析結果輸出模塊

• 可以將結果推送到：

  • ECM 系統（如 IBM FileNet）做文檔記錄

  • BPM 系統（如 IBM BPM）發起審批流程

  • 前端儀表盤展示（如 React + Tailwind + Charts）?

三、功能示例

示例 1：自然語言問數據庫

用戶輸入：「幫我查一下過去3個月每個地區的銷售趨勢，并找出異常波動。」

執行流程：

1. MCP 調用 SQLAgent 將用戶意圖翻譯為 SQL

2. 查詢數據庫中相關銷售表

3. 將數據交給 Pandas Agent 進行趨勢圖繪制 + 異常檢測

4. 返回分析結果 + 圖表

5. 可選：將異常記錄推送到 BPM 發起“銷售異常核查”流程?

示例 2：合同數據分析（結合 ECM）

用戶輸入：「找出近30天簽署的合同中，金額超過100萬的，并做風險分析。」

1. 查詢數據庫 + ECM 中文檔（合同 PDF 或條款摘要）

2. 使用 NLP 模型（或 Agent）讀取文本內容進行分析（如違約風險）

3. 多模型評估結果（Claude/GPT/Mistral 投票）

4. 推送分析結果到 ECM 系統的備注字段或 BPM 流程中?

四、數據庫接入方式（MySQL/PostgreSQL 示例）

可以使用 LangChain 中的 SQLDatabase 對象連接數據庫：

from langchain.utilities import SQLDatabasedb = SQLDatabase.from_uri("mysql+pymysql://user:password@host:3306/database_name")

然后使用 SQLDatabaseChain 或 SQLAgent 對接：

from langchain.chains import SQLDatabaseChain
from langchain.chat_models import ChatOpenAIllm = ChatOpenAI(model_name="gpt-4")
db_chain = SQLDatabaseChain.from_llm(llm, db)response = db_chain.run("顯示最近三個月銷售額趨勢")

五、搭建一個本地 MCP + SQLite 的 AI 數據分析原型系統，使用 LangChain + GPT（或本地模型）實現自然語言到 SQL，再通過 Pandas 分析數據。

第一步：系統核心結構設計

我們先構建一個最簡 MVP（最小可行產品）版本，包含以下模塊：

系統模塊劃分

模塊名	功能說明
用戶界面	輸入自然語言查詢
MCP Server	調度 Agent，管理上下文和執行流程
SQL Agent	自然語言轉 SQL，并查詢 SQLite
分析 Agent	用 Pandas 對結果做分析、趨勢識別、圖表輸出等
輸出模塊	將分析結果展示在終端或網頁

?

第二步：環境準備

安裝依賴（建議新建虛擬環境）

pip install langchain openai sqlite3 pandas matplotlib streamlit

你可以用 gpt-3.5-turbo 作為 LLM，也可以對接本地模型（如 Ollama）

?第三步：原型代碼結構

我們先建立如下文件結構：

mcp_sqlite_demo/
│
├── main.py                # 主入口
├── db_init.py             # 創建并填充 SQLite 示例數據
├── mcp_server.py          # 模擬 MCP 調度器
├── agent_sql.py           # SQL Agent 實現
├── agent_analysis.py      # 數據分析 Agent 實現

第四步：編寫關鍵代碼

1. 創建示例數據庫（db_init.py）

import sqlite3
import random
from datetime import datetime, timedeltaconn = sqlite3.connect("sales.db")
cur = conn.cursor()cur.execute("DROP TABLE IF EXISTS sales")
cur.execute("""
CREATE TABLE sales (id INTEGER PRIMARY KEY,region TEXT,amount REAL,sale_date TEXT
)
""")regions = ["華東", "華南", "華北", "西南", "東北"]
start_date = datetime(2023, 12, 1)for i in range(90):for region in regions:date = (start_date + timedelta(days=i)).strftime("%Y-%m-%d")amount = round(random.uniform(5000, 50000), 2)cur.execute("INSERT INTO sales (region, amount, sale_date) VALUES (?, ?, ?)",(region, amount, date))conn.commit()
conn.close()
print("數據庫初始化完成 ?")

?運行后會生成一個 sales.db 的銷售數據庫。

2. SQL Agent（agent_sql.py）

from langchain.utilities import SQLDatabase
from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
from langchain.chat_models import ChatOpenAIdef get_sql_agent():db = SQLDatabase.from_uri("sqlite:///sales.db")llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)toolkit = SQLDatabaseToolkit(db=db, llm=llm)agent_executor = create_sql_agent(llm=llm,toolkit=toolkit,verbose=True,)return agent_executor

?3. 分析 Agent（agent_analysis.py）

import pandas as pd
import matplotlib.pyplot as pltdef analyze_sales(df):print("\n🔍 正在分析銷售趨勢...")trend = df.groupby("sale_date")["amount"].sum()trend.plot(kind="line", title="每日銷售趨勢", figsize=(10, 5))plt.tight_layout()plt.savefig("sales_trend.png")max_day = trend.idxmax()print(f"\n📈 銷售峰值日期：{max_day}，金額：{trend.max():,.2f}")

?4. MCP Server 主控（mcp_server.py）

from agent_sql import get_sql_agent
from agent_analysis import analyze_sales
import pandas as pd
import sqlite3def run_query(natural_language_query):print(f"\n🧠 MCP 收到請求：{natural_language_query}")agent = get_sql_agent()sql_response = agent.run(natural_language_query)print("\n📊 SQL Agent 執行結果如下：")print(sql_response)# 執行 SQL 再交給分析模塊db = sqlite3.connect("sales.db")df = pd.read_sql_query("SELECT * FROM sales", db)analyze_sales(df)

5. 啟動主程序（main.py）

from mcp_server import run_queryif __name__ == "__main__":print("歡迎使用 MCP + SQLite 智能分析系統")while True:query = input("\n💬 請輸入分析請求（或 q 退出）：")if query.lower() in ['q', 'exit']:breakrun_query(query)

第五步：運行測試

先初始化數據庫：

python db_init.py

?然后運行主程序：

python main.py

示例輸入：

過去30天每個地區銷售趨勢如何？有沒有異常波動？

六、可運行的本地系統模板

把這個 MCP + SQLite 的 AI 分析系統打造成一個完整可運行的本地系統模板，包括：

• ? 完整代碼結構

• ? Streamlit 前端界面

• ? Docker 容器化部署支持

• ? 可擴展的多 Agent 結構?

1.項目結構設計（最終版）

我們建立如下目錄結構：

mcp_sqlite_ai/
│
├── app/
│   ├── __init__.py
│   ├── db_init.py               # 初始化 SQLite 示例數據
│   ├── mcp_server.py            # MCP 調度模塊
│   ├── agent_sql.py             # SQL Agent
│   ├── agent_analysis.py        # 分析 Agent
│   └── config.py                # 模型配置
│
├── frontend/
│   └── dashboard.py             # Streamlit 前端交互界面
│
├── Dockerfile                   # Docker 鏡像構建
├── requirements.txt             # 依賴列表
└── run.sh                       # 一鍵運行腳本

2.文件內容逐個說明

? app/db_init.py

# 初始化數據庫
import sqlite3
import random
from datetime import datetime, timedeltadef init_db():conn = sqlite3.connect("sales.db")cur = conn.cursor()cur.execute("DROP TABLE IF EXISTS sales")cur.execute("""CREATE TABLE sales (id INTEGER PRIMARY KEY,region TEXT,amount REAL,sale_date TEXT)""")regions = ["華東", "華南", "華北", "西南", "東北"]start_date = datetime(2023, 12, 1)for i in range(90):for region in regions:date = (start_date + timedelta(days=i)).strftime("%Y-%m-%d")amount = round(random.uniform(5000, 50000), 2)cur.execute("INSERT INTO sales (region, amount, sale_date) VALUES (?, ?, ?)",(region, amount, date))conn.commit()conn.close()print("? SQLite 數據庫初始化完成")if __name__ == "__main__":init_db()

?? app/config.py

OPENAI_API_KEY = "your-openai-key"  # 替換成你自己的，或使用環境變量加載

?? app/agent_sql.py

from langchain.utilities import SQLDatabase
from langchain.agents import create_sql_agent
from langchain.agents.agent_toolkits import SQLDatabaseToolkit
from langchain.chat_models import ChatOpenAI
from app.config import OPENAI_API_KEYdef get_sql_agent():db = SQLDatabase.from_uri("sqlite:///sales.db")llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0, openai_api_key=OPENAI_API_KEY)toolkit = SQLDatabaseToolkit(db=db, llm=llm)return create_sql_agent(llm=llm, toolkit=toolkit, verbose=True)

? app/agent_analysis.py?

import pandas as pd
import matplotlib.pyplot as pltdef analyze_sales(df, output_path="frontend/sales_plot.png"):trend = df.groupby("sale_date")["amount"].sum()trend.plot(kind="line", title="每日銷售趨勢", figsize=(10, 5))plt.tight_layout()plt.savefig(output_path)plt.close()return f"趨勢圖已生成：{output_path}"

?? app/mcp_server.py

import sqlite3
import pandas as pd
from app.agent_sql import get_sql_agent
from app.agent_analysis import analyze_salesdef run_mcp_query(query):agent = get_sql_agent()response = agent.run(query)conn = sqlite3.connect("sales.db")df = pd.read_sql_query("SELECT * FROM sales", conn)plot_path = analyze_sales(df)return response, plot_path

? frontend/dashboard.py （Streamlit 前端）?

import streamlit as st
from app.mcp_server import run_mcp_queryst.set_page_config(page_title="AI 智能銷售分析", layout="centered")st.title("📊 AI 銷售數據分析平臺")
query = st.text_input("請輸入分析請求（如：過去30天銷售趨勢）", "")if st.button("提交分析") and query.strip():with st.spinner("正在分析，請稍候..."):response, img_path = run_mcp_query(query)st.success("分析完成")st.image(img_path, caption="銷售趨勢圖", use_column_width=True)st.markdown("### 🤖 SQL Agent 結果")st.markdown(response)

?? requirements.txt

langchain>=0.1.0
openai>=1.0.0
pandas
matplotlib
streamlit
sqlite3

?? Dockerfile

FROM python:3.10-slimWORKDIR /appCOPY . /app
RUN pip install --upgrade pip && \pip install -r requirements.txtEXPOSE 8501CMD ["streamlit", "run", "frontend/dashboard.py", "--server.port=8501", "--server.address=0.0.0.0"]

?? run.sh

#!/bin/bash
echo "初始化數據庫..."
python3 app/db_init.pyecho "啟動前端界面..."
streamlit run frontend/dashboard.py

3.啟動方式

👉 本地運行：

chmod +x run.sh
./run.sh

?瀏覽器訪問：http://localhost:8501

🐳 Docker 容器運行：

docker build -t mcp-ai-agent .
docker run -p 8501:8501 mcp-ai-agent