这两天一直在本地环境折腾数据库 Agent 的落地。本来是打算搞
Dify 适配达梦的
MCP-Server 方案,但是搞了许久最后还是没搞成功。 想到少安在群里说可以试试
pyTiDB,于是今天上午换成
TiDB 官方的
pyTiDB 数据访问层,结合
FastAPI 封装微服务,再与
Dify Agent 进行全流程打通,终于成功构建出了一套支持自然语言交互的
TiDB 自动化运维助手。
Dify :一个开源的大语言模型(LLM)应用开发与编排平台,它的目标是降低 AI 应用的开发门槛。比如在 Dify 的画布里,把“大模型节点”、MCP Server、“条件判断分支”像画流程图一样
拖拉拽串联起来,一个复杂的 AI 数据流就可以搭建好。pyTiDB:
TiDB 官方生态社区提供的面向 Python 的 SDK 与工具扩展。在 AI Agent 应用架构中(例如结合 Dify、FastAPI 等),Python 版本的 TiDB 接口模块负责将大模型(LLM)生成的
SQL 指令或自然语言查询请求转化为安全的数据库调用,充当控制层与 TiDB 分布式集群之间的中间对接桥梁。典型分层如下:FastAPI(HTTP 层) ↓ 调用 pytidb(TiDB 数据访问层) ↓ SQL/向量协议 TiDB (数据库集群)
每层作用:
RESTful API 接口,负责接收来自
Dify Agent 的
HTTP 请求,完成参数校验、权限控制以及把 JSON 数据反序列化。相当于一个协议转换器:SQL 组装逻辑,向下屏蔽底层通信细节。pytidb 的 SQL 语句命令并执行,将结果回传。Ollama:
Ollama 是一个开源的本地大语言模型(LLM)运行与管理工具,可以让普通用户能够像使用 Docker 一样,在个人电脑或本地服务器上
一键部署、运行和管理大模型。 Windows 宿主机安装 Ollama ,本地部署了 deepseek-r1:1.5b 大模型,这样就可以不联网使用 AI,然后对接虚拟机里面的 Dify。当然也可以不本地部署,使用在线的模型,前提是
Dify 所在的虚拟机可以连公网。顺便提一句最好是使用 CentOS 8及以上或 Rocky Linux 的操作系统。因为后者核心 Glibc 与底层 C/C++ 运行时以及 Python的版本更高。
| 机器IP | 配置大小 | 部署组件 | 备注 |
|---|---|---|---|
| 192.168.182.153 | 4C8G | Dify/pyTiDB微服务 | Centos7.9 虚拟机 |
| 192.168.182.31 | 4C8G | TiDB-Server/PD/Tikv-Server | Centos7.9 虚拟机 |
| 192.168.182.32 | 4C8G | TiDB-Server/PD/Tikv-Server | Centos7.9 虚拟机 |
| 192.168.182.33 | 4C8G | TiDB-Server/PD/Tikv-Server | Centos7.9 虚拟机 |
| 192.168.182.1 | 64G | Ollama(可选) | 宿主机 |
SQL 生成逻辑。/opt/pytidb-main/venv/bin/python3 /opt/pytidb-main/app.py ),使用
SQLAlchemy 与
PyMySQL 连接池,将数据库查询与系统指令包装为标准
RESTful API 。192.168.182.31/32/33)。Ollama打开官网
https://ollama.com/download ,根据操作系统选择对应的版本,下载完并安装,
Windows 系统默认安装到
C:\Users\用户名\.ollama 目录。
打开一个 cmd 终端,运行
ollama run deepseek-r1:1.5b ,会自动拉取 1.5b 的模型,大小约 1G 左右。下载到
C:\Users\用户名\.ollama\models 目录,如果不能正常运行,提示
GPU 方面的报错,则需配置系统变量。
Win + R 输入
sysdm.cpl 回车,打开系统属性。CUDA_VISIBLE_DEVICES ,
变量值:
-1 (设为 -1 会直接隐藏所有
CUDA 设备,强制走 CPU)OLLAMA_HOST,变量值:
0.0.0.0, (允许局域网其他设备直接访问,
Ollama 默认只监听
127.0.0.1)Ollama。ollama run deepseek-r1:1.5b >>> 你好 你好!很高兴见到你,有什么我可以帮忙的吗?
在 Linux 虚拟机里部署 dify,因此首先要安装 docker 。
# 使用 阿里云的 Centos 源 curl -o /etc/yum.repos.d/CentOS-Base.repo https://mirrors.aliyun.com/repo/Centos-7.repo # 安装必要的系统工具 yum install -y yum-utils device-mapper-persistent-data lvm2 # 添加 Docker CE 镜像源(使用阿里云镜像源加速) yum-config-manager --add-repo https://mirrors.aliyun.com/docker-ce/linux/centos/docker-ce.repo # 安装 Docker CE 引擎和最新的 Compose 插件 yum install -y docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin mkdir -p /etc/systemd/system/docker.service.d # 配置 docker 使用宿主机的代理 cat > /etc/systemd/system/docker.service.d/http-proxy.conf << 'EOF' [Service] Environment="HTTP_PROXY=http://192.168.182.1:10811" Environment="HTTPS_PROXY=http://192.168.182.1:10811" Environment="NO_PROXY=localhost,127.0.0.1,0.0.0.0" EOF [root@tidb3 docker]# systemctl daemon-reload [root@tidb3 docker]# systemctl enable --now docker Created symlink from /etc/systemd/system/multi-user.target.wants/docker.service to /usr/lib/systemd/system/docker.service. [root@tidb3 docker]# systemctl show docker --property=Environment Environment=HTTP_PROXY=http://192.168.182.1:10811 HTTPS_PROXY=http://192.168.182.1:10811 NO_PROXY=localh [root@tidb3 docker]# docker --version Docker version 26.1.4, build 5650f9b [root@tidb3 docker]# docker pull hello-world Using default tag: latest latest: Pulling from library/hello-world 4f55086f7dd0: Pull complete Digest: sha256:5dd0d3e6e255913fc30f90b9f2b1d359cc2cbdb48090cc4b65f1676e203243cc Status: Downloaded newer image for hello-world:latest docker.io/library/hello-world:latest
# 1. 下载 docker-compose, 如果较慢换国内的源 curl -L "https://github.com/docker/compose/releases/download/v2.29.1/docker-compose-$(uname -s)-$(uname -m)" -o /usr/local/bin/docker-compose # 2. 赋予执行权限并创建软链接 chmod +x /usr/local/bin/docker-compose ln -sf /usr/local/bin/docker-compose /usr/bin/docker-compose # 3. 验证版本 [root@tidb3 ~]# docker-compose --version Docker Compose version v2.29.1
# 1. 使用 git 克隆 dify yum -y install git git clone https://github.com/langgenius/dify.git # 2. 进入 Docker 编排目录 cd dify/docker # 3. 复制生成环境变量文件 cp .env.example .env # 4. 一键拉起 Dify 服务集群 [root@mysql docker]# docker compose up -d [+] Running 110/16 ✔ api Pulled 297.3s ✔ worker_beat Pulled 297.3s ✔ db_postgres Pulled 294.2s ✔ api_websocket Pulled 297.3s ✔ sandbox Pulled 255.6s ✔ ssrf_proxy Pulled 88.3s ✔ weaviate Pulled 148.2s ✔ worker Pulled 297.3s ✔ local_sandbox Pulled 123.3s ✔ agent_backend Pulled 278.2s ✔ init_permissions Pulled 58.8s ✔ plugin_daemon Pulled 321.4s ✔ agent_ssrf_proxy Pulled 88.3s ✔ web Pulled 276.8s ✔ redis Pulled 259.9s ✔ nginx Pulled 213.0s [+] Running 22/22 ✔ Network docker_default Created 0.0s ✔ Network docker_local_sandbox_proxy_network Created 0.0s ✔ Network docker_agent_sandbox_network C... 0.0s ✔ Network docker_ssrf_proxy_network Crea... 0.0s ✔ Volume "docker_dify_agent_local_sandbox_home" Created 0.0s ✔ Volume "docker_dify_agent_local_sandbox_workspace" Created 0.0s ✔ Container docker-weaviate-1 Started 1.0s ✔ Container docker-web-1 Started 1.1s ✔ Container docker-init_permissions-1 Ex... 4.6s ✔ Container docker-db_postgres-1 Healthy 4.6s ✔ Container docker-redis-1 Started 1.0s ✔ Container docker-agent_ssrf_proxy-1 St... 1.1s ✔ Container docker-local_sandbox-1 Start... 1.1s ✔ Container docker-ssrf_proxy-1 Started 1.1s ✔ Container docker-sandbox-1 Started 1.0s ✔ Container docker-plugin_daemon-1 Start... 3.8s ✔ Container docker-worker_beat-1 Started 3.8s ✔ Container docker-api_websocket-1 Start... 3.9s ✔ Container docker-agent_backend-1 Start... 4.1s ✔ Container docker-api-1 Started 4.9s ✔ Container docker-worker-1 Started 5.0s ✔ Container docker-nginx-1 Started 5.5s [root@mysql docker]#


LLMdeepseek-r1:1.5b(必须与
ollama list 中显示的名称完全一致)http://192.168.182.1:11434


将 FastAPI 服务和 Dify 在同一台机器,通过 Docker 默认网桥 IP,避开了跨机器防火墙和复杂网络联调。
# Centos7 下载旧版 Miniconda(支持 GLIBC 2.17) wget https://repo.anaconda.com/miniconda/Miniconda3-py311_23.11.0-2-Linux-x86_64.sh -O /opt/miniconda.sh # 安装到 /opt/miniconda3 bash /opt/miniconda.sh -b -p /opt/miniconda3 # 初始化 /opt/miniconda3/bin/conda init bash source ~/.bashrc # 验证 (base) [root@mysql ~]# python --version Python 3.11.5 #创建独立的 Python 3.11 虚拟环境 (base) [root@mysql opt]# conda create -n tidb-env python=3.11 -y (base) [root@mysql opt]# conda activate tidb-env (tidb-env) [root@mysql opt]# # 安装完成后,在创建 `pytidb` 项目虚拟环境时即可直接使用指定的 Python 版本: (tidb-env) [root@mysql pytidb-main]# python3.11 -m venv /opt/pytidb-main/venv (tidb-env) [root@mysql pytidb-main]# source venv/bin/activate (venv) (tidb-env) [root@mysql pytidb-main]# # 安装依赖库 # 安装基础编译工具链 yum install -y gcc gcc-c++ make # 1. 确保 pip 和 wheel 工具是最新的 pip install --upgrade pip wheel setuptools # 2. 指定直接安装预编译的 numpy pip install "numpy<3,>=1" --only-binary=:all: # 3. 重新安装目标依赖 pip install --only-binary=:all: greenlet pytidb pip install pytidb fastapi uvicorn pymysql
碰到了一堆 GCC 版本过低的报错,因此,建议使用 CentOS 8 以上或 Rocky Linux 的操作系统。
下载链接如下: https://codeload.github.com/pingcap/pytidb/zip/refs/heads/main
app.py 本质上是一个使用 FastAPI 框架 编写的 Web 后端程序。它的核心工作有三件:
import os
from typing import Optional
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
app = FastAPI(
title="PyTiDB Service for Dify",
description="TiDB 集群运维与数据查询接口服务",
version="0.2.0",
)
# ==================== 1. 读取环境变量与初始化 Engine ====================
TIDB_HOST = os.getenv("TIDB_HOST", "192.168.182.31")
TIDB_PORT = int(os.getenv("TIDB_PORT", "4000"))
TIDB_USER = os.getenv("TIDB_USER", "root")
TIDB_PASSWORD = os.getenv("TIDB_PASSWORD", "root")
TIDB_DATABASE = os.getenv("TIDB_DATABASE", "information_schema")
# 拼接数据库连接字符串
if TIDB_PASSWORD:
db_url = f"mysql+pymysql://{TIDB_USER}:{TIDB_PASSWORD}@{TIDB_HOST}:{TIDB_PORT}/{TIDB_DATABASE}"
else:
db_url = f"mysql+pymysql://{TIDB_USER}@{TIDB_HOST}:{TIDB_PORT}/{TIDB_DATABASE}"
# 创建 SQLAlchemy Engine
engine = create_engine(
db_url,
pool_recycle=3600,
pool_pre_ping=True,
pool_size=10,
max_overflow=20,
)
def run_query(sql_statement: str, params: dict = None):
"""使用 SQLAlchemy 统一执行查询并返回字典列表"""
try:
with engine.connect() as conn:
result = conn.execute(text(sql_statement), params or {})
if result.returns_rows:
return [dict(row._mapping) for row in result]
return []
except SQLAlchemyError as e:
raise HTTPException(
status_code=500, detail=f"数据库查询失败: {str(e)}"
)
# ==================== 2. API 接口定义 ====================
@app.get("/", summary="心跳检查")
def read_root():
return {"status": "ok", "message": "PyTiDB Service is running"}
# --- 【集群与节点管理】 ---
@app.get("/cluster/nodes", summary="获取集群所有节点 IP 及拓扑信息")
def get_cluster_nodes():
"""从 TiDB 内置元数据表中获取所有节点(TiDB/PD/TiKV等)的 IP、端口和角色信息"""
sql = """
SELECT TYPE, INSTANCE, STATUS, START_TIME
FROM INFORMATION_SCHEMA.CLUSTER_INFO;
"""
nodes = run_query(sql)
return {"status": "success", "count": len(nodes), "nodes": nodes}
@app.get("/cluster/static-ips", summary="获取预设的三台集群主节点 IP 静态列表")
def get_static_ips():
"""快速返回预设的三台节点 IP 列表"""
return {
"status": "success",
"cluster_ips": [
{"ip": "192.168.182.31", "role": "TiDB / PD / Control Node"},
{"ip": "192.168.182.32", "role": "TiKV / Storage Node 1"},
{"ip": "192.168.182.33", "role": "TiKV / Storage Node 2"},
],
}
# --- 【数据库库表结构查询】 ---
@app.get("/db/databases", summary="获取所有的数据库名称")
def list_databases():
"""获取当前 TiDB 集群中的所有数据库名称"""
rows = run_query("SHOW DATABASES;")
dbs = [list(r.values())[0] for r in rows]
return {"status": "success", "databases": dbs}
@app.get("/db/tables", summary="获取指定数据库中的所有表")
def list_tables(db_name: str = "test"):
"""获取指定数据库下的所有表名"""
rows = run_query(f"SHOW TABLES FROM `{db_name}`;")
tables = [list(r.values())[0] for r in rows]
return {"status": "success", "database": db_name, "tables": tables}
# --- 【监控与性能分析】 ---
@app.get("/monitor/slow-queries", summary="查询最近的慢查询日志")
def get_slow_queries(limit: int = 10):
"""获取 TiDB 中最近执行的慢查询记录"""
sql = f"""
SELECT Time, Query_time, Process_time, Wait_time, User, Host, DB, query
FROM INFORMATION_SCHEMA.SLOW_QUERY
ORDER BY Time DESC
LIMIT {limit};
"""
try:
rows = run_query(sql)
return {"status": "success", "count": len(rows), "slow_queries": rows}
except Exception as e:
return {
"status": "error",
"message": f"无法查询慢日志: {str(e)}",
"slow_queries": [],
}
@app.get("/monitor/processlist", summary="查看当前正在执行的连接与会话")
def get_processlist():
"""查看数据库当前的活跃连接数和正在执行的 SQL 线程"""
processes = run_query("SHOW PROCESSLIST;")
return {
"status": "success",
"active_connections": len(processes),
"processes": processes,
}
# --- 【通用动态 SQL 查询】 ---
class DynamicSQLRequest(BaseModel):
db_name: Optional[str] = "test"
sql: str
@app.post("/db/execute-sql", summary="执行自定义 SELECT / SHOW 查询 SQL")
def execute_sql(req: DynamicSQLRequest):
"""出于安全限制,仅允许大模型执行只读查询 (SELECT / SHOW / EXPLAIN / DESC)"""
sql_clean = req.sql.strip().lower()
allowed_prefixes = ("select", "show", "explain", "desc")
if not sql_clean.startswith(allowed_prefixes):
raise HTTPException(
status_code=400,
detail=(
"安全限制:该接口仅允许执行只读查询操作 (SELECT / SHOW / EXPLAIN"
" / DESC)"
),
)
# 如果指定了库名,先切换上下文或直接执行
if req.db_name:
run_query(f"USE `{req.db_name}`;")
result = run_query(req.sql)
return {"status": "success", "row_count": len(result), "data": result}
if __name__ == "__main__":
import uvicorn
uvicorn.run("app:app", host="0.0.0.0", port=8000, reload=True)
vim /etc/systemd/system/pytidb-service.service
[Unit]
Description=PyTiDB FastAPI Service for Dify
After=network.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt/pytidb-main
ExecStart=/opt/pytidb-main/venv/bin/python3 -m uvicorn app:app --host 0.0.0.0 --port 8000
Restart=always
[Install]
WantedBy=multi-user.target
# 重新加载 systemd 配置
systemctl daemon-reload
# 重启 pytidb 服务
systemctl restart pytidb-service.service
# 检查服务运行状态
systemctl status pytidb-service.service
# 测试接口是否能正常响应
(base) [root@mysql pytidb-main]# curl http://127.0.0.1:8000/
{"status":"ok","message":"PyTiDB Service is running"}(base) [root@mysql pytidb-main]#
Dify 选择集成,点击 Swagger API 作为工具,创建并编辑自定义工具
输入如下参数:
{
"openapi": "3.0.0",
"info": {
"title": "PyTiDB Service for Dify",
"description": "TiDB 集群运维与数据查询接口服务",
"version": "0.2.0"
},
"servers": [
{
"url": "http://192.168.182.153:8000"
}
],
"paths": {
"/": {
"get": {
"summary": "心跳检查",
"operationId": "read_root",
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/cluster/nodes": {
"get": {
"summary": "获取集群所有节点 IP 及拓扑信息",
"operationId": "get_cluster_nodes",
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/cluster/static-ips": {
"get": {
"summary": "获取预设的三台集群主节点 IP 静态列表",
"operationId": "get_static_ips",
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/db/databases": {
"get": {
"summary": "获取所有的数据库名称",
"operationId": "list_databases",
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/db/tables": {
"get": {
"summary": "获取指定数据库中的所有表",
"operationId": "list_tables",
"parameters": [
{
"name": "db_name",
"in": "query",
"required": false,
"schema": {
"type": "string",
"default": "test"
},
"description": "要查询的数据库名,默认为 test"
}
],
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/monitor/slow-queries": {
"get": {
"summary": "查询最近的慢查询日志",
"operationId": "get_slow_queries",
"parameters": [
{
"name": "limit",
"in": "query",
"required": false,
"schema": {
"type": "integer",
"default": 10
},
"description": "返回的慢日志条数上限,默认 10 条"
}
],
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/monitor/processlist": {
"get": {
"summary": "查看当前正在执行的连接与会话",
"operationId": "get_processlist",
"responses": {
"200": { "description": "Successful Response" }
}
}
},
"/db/execute-sql": {
"post": {
"summary": "执行自定义 SELECT / SHOW 查询 SQL",
"operationId": "execute_sql",
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"type": "object",
"properties": {
"db_name": {
"type": "string",
"description": "目标数据库名,默认 test",
"default": "test"
},
"sql": {
"type": "string",
"description": "要执行的 SQL 查询语句 (仅限 SELECT / SHOW / EXPLAIN / DESC)"
}
},
"required": ["sql"]
}
}
}
},
"responses": {
"200": { "description": "Successful Response" }
}
}
}
}
}
随后点击保存。点击测试成功。

创建 Agent,模型可选择刚刚的本地大模型,也可选择在线模型(输入 API key),工具选择刚刚新建的工具,点击应用便可询问 AI 相关问题,如图所示(本地模型有点问题,我这里选择的在线模型)。
询问集群拓扑状态:

询问会话情况:

上午那会儿创建了个 test 数据库卡住了,还没来得及排查。

本次实践通过
FastAPI +
PyTiDB 搭建起了底层安全可控的运维微服务,并借助
Dify Agent 的 ReAct 架构与
DeepSeek 的大模型能力,打通了自然语言与私有数据库之间的壁垒。虽然是个很小的 demo,但是可以帮我们理解 AI 的以下几个底层逻辑:
这也为我们后面自己写 skills 提供了指导。 如果本文对你有所帮助,欢迎点赞、推荐和转发,也欢迎关注后续文章,一起探索 AI 如何赋能数据库运维。
来自 “ ITPUB博客 ” ,链接:https://blog.itpub.net/31444449/viewspace-3181032/ ,如需转载,请注明出处,否则将追究法律责任。