DeepChat实战教程：为DeepChat添加SQL查询能力，连接本地数据库实现自然语言查数

DeepChat实战教程：为DeepChat添加SQL查询能力，连接本地数据库实现自然语言查数

1. 为什么需要给DeepChat加上SQL能力

你有没有遇到过这样的场景：手边有一份销售数据表，想快速知道“上个月华东区销售额最高的产品是什么”，却得先打开数据库工具、写SELECT语句、再反复调试WHERE条件？或者，刚入职的运营同事面对一张20个字段的用户行为表，连“活跃用户数”都算不出来，只能一次次找工程师帮忙。

DeepChat本身已经很强大——它能用本地Llama 3模型和你聊物理、写诗、分析逻辑。但它的知识停留在“通用理解”层面，对你的具体数据一无所知。就像一个博学的教授，能讲透量子力学，却不知道你公司MySQL里那张orders_2024表长什么样。

本教程要解决的，就是这个关键断层：让DeepChat真正读懂你的数据库，把自然语言提问，直接变成可执行的SQL查询，并返回结构化结果。不是调用外部API，不上传任何数据，所有操作都在你自己的机器里完成。

整个过程不需要改DeepChat源码，不碰Ollama底层，只用几段清晰的Python胶水代码+少量配置，就能让对话框变成你的“智能数据库助手”。你会看到：输入“显示最近7天下单金额超过500的客户”，它自动写出SQL、执行、再把表格结果用中文总结给你。

这不只是功能叠加，而是把AI从“聊天伙伴”升级为“业务协作者”。

2. 准备工作：确认环境与安装依赖

在动手前，请确保你已成功运行DeepChat镜像——也就是能看到那个简洁的聊天界面。如果还没启动，建议先按官方说明完成首次部署（会自动下载llama3:8b模型，约4.7GB）。

本教程的增强能力基于Python生态，所有操作都在容器内部完成，不暴露端口、不修改原有服务架构。我们采用“轻量代理”思路：在DeepChat WebUI和数据库之间加一层智能翻译层。

2.1 检查当前环境

DeepChat镜像默认已预装Python 3.11、pip及基础库。我们先验证一下：

# 进入正在运行的DeepChat容器（假设容器名为deepchat） docker exec -it deepchat bash # 检查Python和pip python3 --version # 应输出 Python 3.11.x pip list | grep -i "sqlalchemy\|pymysql\|duckdb"

如果没看到sqlalchemy、pymysql（或psycopg2，根据你用的数据库）、duckdb，就需要安装。别担心，一行命令搞定：

# 安装核心依赖（以MySQL为例，PostgreSQL用户请替换pymysql为psycopg2-binary） pip install sqlalchemy pymysql duckdb python-dotenv

小贴士：DuckDB是我们的“安全沙箱”。它是一个嵌入式分析型数据库，无需单独服务进程，所有数据存于内存或单个文件中。我们用它来临时存储和验证生成的SQL，避免直接执行可能出错的语句到生产库——这是保障安全的关键一步。

2.2 创建数据库连接配置

在容器内创建一个安全的配置文件，存放你的数据库连接信息。绝不硬编码在代码里！

# 在容器内创建配置目录 mkdir -p /app/config # 创建 .env 文件（用nano或echo） cat > /app/config/.env << 'EOF' # 数据库类型：mysql / postgresql / sqlite DB_TYPE=mysql # MySQL连接信息（请替换成你的真实地址） DB_HOST=host.docker.internal DB_PORT=3306 DB_NAME=your_business_db DB_USER=readonly_user DB_PASSWORD=your_secure_password # SQLite示例（如用本地SQLite，取消下面三行注释，注释掉上面MySQL部分） # DB_TYPE=sqlite # DB_PATH=/app/data/app.db # 可选：设置只读模式，防止AI误删数据 DB_READONLY=true EOF

安全提醒：host.docker.internal是Docker内置DNS，指向宿主机。如果你的数据库也在Docker中（比如叫mysql-db），请将DB_HOST改为该容器名。readonly_user必须是数据库中仅拥有SELECT权限的账号——这是防误操作的底线。

2.3 获取数据库元数据（让AI“认识”你的表）

AI不能凭空猜表结构。我们需要把数据库的“说明书”喂给它。执行以下脚本，自动生成一份简洁的表描述：

# 创建元数据提取脚本 cat > /app/scripts/generate_schema.py << 'EOF' import os import sys from dotenv import load_dotenv from sqlalchemy import create_engine, text # 加载配置 load_dotenv("/app/config/.env") db_type = os.getenv("DB_TYPE", "mysql") if db_type == "mysql": engine = create_engine( f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}" ) elif db_type == "postgresql": engine = create_engine( f"postgresql+psycopg2://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}" ) else: engine = create_engine(f"sqlite:///{os.getenv('DB_PATH')}") # 提取所有表名和字段 with engine.connect() as conn: if db_type in ["mysql", "postgresql"]: schema_sql = """ SELECT table_name, column_name, data_type, is_nullable, column_comment FROM information_schema.columns WHERE table_schema = %s ORDER BY table_name, ordinal_position """ tables_info = conn.execute(text(schema_sql), (os.getenv("DB_NAME"),)).fetchall() else: schema_sql = """ SELECT name as table_name, sql as ddl FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' """ tables_info = conn.execute(text(schema_sql)).fetchall() # 生成易读描述 schema_desc = [] current_table = "" for row in tables_info: if db_type in ["mysql", "postgresql"]: table, col, dtype, nullable, comment = row if table != current_table: schema_desc.append(f"\n表名：{table}") current_table = table desc = f" - {col} ({dtype})" if comment and comment.strip(): desc += f" // {comment.strip()}" if nullable == "YES": desc += " [可为空]" schema_desc.append(desc) else: table, ddl = row schema_desc.append(f"\n表名：{table}\n 结构：{ddl}") # 输出到文件 with open("/app/config/db_schema.md", "w", encoding="utf-8") as f: f.write("# 数据库结构说明书\n\n") f.write("以下是当前数据库中所有表的字段说明，供AI理解使用：\n\n") f.writelines(schema_desc) print(" 数据库结构说明书已生成：/app/config/db_schema.md") EOF # 执行脚本 python3 /app/scripts/generate_schema.py

运行后，你会在/app/config/db_schema.md里看到类似这样的内容：

# 数据库结构说明书 以下是当前数据库中所有表的字段说明，供AI理解使用： 表名：orders - id (bigint) // 订单唯一ID - user_id (int) // 下单用户ID - amount (decimal(10,2)) // 订单金额 - status (varchar(20)) // 订单状态：pending/paid/shipped/cancelled - created_at (datetime) // 创建时间 表名：users - id (int) // 用户ID - name (varchar(100)) // 用户姓名 - region (varchar(50)) // 所属区域：华东/华北/华南... - last_login (datetime) // 最后登录时间

这份文件，就是DeepChat理解你数据的“词典”。

3. 核心实现：编写SQL生成与执行模块

现在进入最关键的环节：让DeepChat不仅能“看懂”你的表，还能“写出”正确的SQL，并安全地执行它。我们不修改前端，而是通过一个独立的Python服务，接收DeepChat发来的自然语言问题，返回结构化结果。

3.1 创建SQL智能代理服务

在容器内创建主服务文件：

mkdir -p /app/sql_agent cat > /app/sql_agent/main.py << 'EOF' import os import json import time from datetime import datetime from typing import Dict, List, Optional, Tuple from dotenv import load_dotenv from sqlalchemy import create_engine, text from sqlalchemy.exc import SQLAlchemyError import duckdb # 加载配置 load_dotenv("/app/config/.env") DB_TYPE = os.getenv("DB_TYPE", "mysql") DB_READONLY = os.getenv("DB_READONLY", "true").lower() == "true" # 初始化主数据库引擎 def get_main_engine(): if DB_TYPE == "mysql": return create_engine( f"mysql+pymysql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}" ) elif DB_TYPE == "postgresql": return create_engine( f"postgresql+psycopg2://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}" ) else: return create_engine(f"sqlite:///{os.getenv('DB_PATH')}") # 初始化DuckDB（用于安全验证） def get_duckdb_conn(): return duckdb.connect(database=':memory:') # 读取数据库结构说明书 def load_schema_desc() -> str: try: with open("/app/config/db_schema.md", "r", encoding="utf-8") as f: return f.read() except FileNotFoundError: return " 未找到数据库结构说明书，请先运行 generate_schema.py" # 构建给大模型的提示词（Prompt） def build_prompt(question: str) -> str: schema_desc = load_schema_desc() return f"""你是一个专业的数据库SQL生成助手。用户会用中文提出一个查询需求，请你严格按以下规则执行： 1. **只输出纯SQL语句**，不要任何解释、不要```sql```包裹、不要额外字符。 2. **必须使用标准SQL语法**，兼容MySQL 8.0+。 3. **只允许使用SELECT语句**，禁止INSERT/UPDATE/DELETE/DROP等任何修改操作。 4. **优先使用JOIN关联表**，避免子查询（除非必要）。 5. **日期处理**：如需“最近7天”，用 `WHERE created_at >= DATE_SUB(NOW(), INTERVAL 7 DAY)`；如需“上个月”，用 `WHERE YEAR(created_at) = YEAR(DATE_SUB(NOW(), INTERVAL 1 MONTH)) AND MONTH(created_at) = MONTH(DATE_SUB(NOW(), INTERVAL 1 MONTH))`。 6. **数值排序**：如需“最高”，用 `ORDER BY amount DESC LIMIT 1`。 以下是数据库结构说明书： {schema_desc} 用户问题： {question} """ # 调用Llama 3生成SQL（模拟DeepChat调用，实际中由WebUI触发） def generate_sql(question: str) -> str: # 这里是与DeepChat集成的关键点 # 实际部署时，你需要将此函数接入DeepChat的API钩子 # 本教程为演示，我们用一个简化版：直接调用Ollama API import requests try: response = requests.post( "http://localhost:11434/api/generate", json={ "model": "llama3:8b", "prompt": build_prompt(question), "stream": False, "options": {"temperature": 0.1, "num_predict": 256} } ) if response.status_code == 200: result = response.json() sql = result.get("response", "").strip() # 清理可能的多余字符 if sql.startswith("```sql"): sql = sql[6:].split("```")[0].strip() elif sql.startswith("```"): sql = sql[3:].split("```")[0].strip() return sql else: return "ERROR: Ollama API call failed" except Exception as e: return f"ERROR: {str(e)}" # 在DuckDB中安全验证SQL def validate_sql(sql: str) -> Tuple[bool, str, Optional[List]]: if not sql.upper().startswith("SELECT"): return False, " 错误：只允许SELECT语句", None try: # 尝试在DuckDB中解析（不执行） conn = get_duckdb_conn() # DuckDB不支持所有MySQL语法，我们做最小化验证 # 简单检查是否有明显危险关键词 dangerous_keywords = ["INSERT", "UPDATE", "DELETE", "DROP", "ALTER", "CREATE"] for kw in dangerous_keywords: if kw in sql.upper(): return False, f" 错误：检测到危险关键词 '{kw}'，拒绝执行", None # 尝试编译（基本语法检查） conn.execute("EXPLAIN " + sql) return True, " SQL语法初步验证通过", None except Exception as e: return False, f" SQL语法错误：{str(e)}", None # 执行SQL并返回结果 def execute_sql(sql: str) -> Dict: start_time = time.time() # 第一步：DuckDB安全验证 is_valid, msg, _ = validate_sql(sql) if not is_valid: return { "success": False, "error": msg, "sql": sql, "data": [], "columns": [], "execution_time": 0 } # 第二步：在真实数据库执行 try: engine = get_main_engine() with engine.connect() as conn: result = conn.execute(text(sql)) rows = result.fetchall() columns = list(result.keys()) # 转为字典列表（便于JSON序列化） data = [dict(zip(columns, row)) for row in rows] execution_time = round(time.time() - start_time, 3) return { "success": True, "sql": sql, "data": data, "columns": columns, "row_count": len(data), "execution_time": execution_time } except SQLAlchemyError as e: return { "success": False, "error": f"数据库执行错误：{str(e)}", "sql": sql, "data": [], "columns": [], "execution_time": 0 } # 主函数：接收问题，返回完整结果 def handle_question(question: str) -> Dict: print(f"[{datetime.now().strftime('%H:%M:%S')}] 收到问题：{question}") # 1. 生成SQL sql = generate_sql(question) print(f"→ 生成SQL：{sql[:100]}...") # 2. 执行SQL result = execute_sql(sql) # 3. 如果失败，尝试更保守的提示词重试（可选） if not result["success"] and "语法错误" in result["error"]: # 简化提示词，要求更基础的SQL simple_prompt = f"""请将以下中文问题，转换为最简单的MySQL SELECT语句，只用基础语法（无子查询、无复杂函数）： {question}""" # 此处可调用Ollama重试，本教程省略 pass return result # 测试入口（供调试用） if __name__ == "__main__": test_q = "显示最近7天下单金额超过500的客户" result = handle_question(test_q) print(json.dumps(result, ensure_ascii=False, indent=2)) EOF

3.2 启动SQL代理服务

这个服务是后台守护进程，不占用Web端口，只监听内部请求：

# 创建启动脚本 cat > /app/sql_agent/start.sh << 'EOF' #!/bin/bash cd /app/sql_agent echo " 启动SQL智能代理服务..." python3 main.py EOF chmod +x /app/sql_agent/start.sh # 在后台运行（实际部署建议用supervisord，此处简化） nohup /app/sql_agent/start.sh > /app/sql_agent/log.txt 2>&1 & echo " SQL代理服务已启动，日志见 /app/sql_agent/log.txt"

关键设计说明：

• 双保险验证：先用DuckDB做语法和安全扫描，再在真实库执行。
• 只读强制：DB_READONLY=true配置确保即使SQL生成有偏差，也无法修改数据。
• 错误友好：返回清晰的错误原因（如“检测到UPDATE关键词”），方便调试。
• 性能透明：返回execution_time，让你知道每次查询花了多久。

4. 与DeepChat前端集成（零代码修改方案）

DeepChat的WebUI是静态前端，我们不修改它，而是利用其“扩展能力”——通过浏览器控制台注入一段轻量JavaScript，让它在发送消息前，先调用我们的SQL代理。

4.1 创建前端注入脚本

mkdir -p /app/web_extensions cat > /app/web_extensions/sql_extension.js << 'EOF' // DeepChat SQL扩展插件 (function() { // 检查是否在DeepChat页面 if (!document.querySelector('.chat-container')) return; // 添加SQL开关按钮 const toolbar = document.querySelector('.input-toolbar'); if (toolbar && !document.getElementById('sql-toggle-btn')) { const btn = document.createElement('button'); btn.id = 'sql-toggle-btn'; btn.className = 'btn btn-sm btn-outline-secondary'; btn.textContent = ' SQL模式'; btn.title = '开启后，所有提问将尝试转为SQL查询'; btn.style.marginLeft = '8px'; btn.addEventListener('click', function() { const isActive = this.classList.toggle('active'); this.textContent = isActive ? ' SQL已启用' : ' SQL模式'; localStorage.setItem('deepchat_sql_enabled', isActive); }); toolbar.insertBefore(btn, toolbar.firstChild); } // 监听发送事件 const originalSend = window.sendMessage; window.sendMessage = function(message) { if (localStorage.getItem('deepchat_sql_enabled') === 'true') { // 发送前拦截，调用SQL代理 fetch('/api/sql-query', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ question: message }) }) .then(r => r.json()) .then(data => { let reply = ''; if (data.success) { reply = ` 查询成功（${data.execution_time}s）\n\n`; reply += `**SQL语句：**\n\`\`\`sql\n${data.sql}\n\`\`\`\n\n`; if (data.row_count === 0) { reply += ' 未找到匹配数据。'; } else { reply += `**查询结果（共${data.row_count}条）：**\n\n`; // 简单表格格式（Markdown） if (data.columns.length > 0) { reply += '| ' + data.columns.map(c => c).join(' | ') + ' |\n'; reply += '| ' + data.columns.map(() => '---').join(' | ') + ' |\n'; data.data.slice(0, 5).forEach(row => { const cells = data.columns.map(col => String(row[col] || '').replace(/\|/g, '\\|') ); reply += '| ' + cells.join(' | ') + ' |\n'; }); if (data.row_count > 5) { reply += `\n... 还有 ${data.row_count - 5} 条结果`; } } } } else { reply = ` 查询失败：${data.error}`; if (data.sql) { reply += `\n\n**尝试的SQL：**\n\`\`\`sql\n${data.sql}\n\`\`\``; } } // 调用原生回复函数（模拟AI回复） window.addMessageToChat(reply, 'bot'); }) .catch(err => { window.addMessageToChat(` 连接SQL服务失败：${err.message}`, 'bot'); }); return; // 阻止原消息发送 } // 非SQL模式，走原逻辑 if (originalSend) originalSend(message); }; })(); EOF

4.2 将脚本注入DeepChat页面

DeepChat使用Vite构建，我们通过修改Nginx配置，在HTML响应中注入这段JS：

# 备份原始Nginx配置 cp /etc/nginx/conf.d/default.conf /etc/nginx/conf.d/default.conf.bak # 追加JS注入规则（在server块内） sed -i '/location \//a \ \ \ \ sub_filter \'<\/head>\' \'<script src=\"/js/sql_extension.js\"><\\/script><\/head>\';\n\ \ \ \ sub_filter_once on;' /etc/nginx/conf.d/default.conf # 创建JS服务路径 mkdir -p /usr/share/nginx/html/js cp /app/web_extensions/sql_extension.js /usr/share/nginx/html/js/ # 重启Nginx nginx -s reload echo " 前端扩展已注入，刷新页面即可使用"

现在，当你打开DeepChat界面，左下角会出现一个“ SQL模式”按钮。点击开启后，所有输入的问题都会被自动转为SQL查询，并以结构化方式返回结果。

5. 实战效果演示与调优技巧

一切就绪，我们来一场真实的测试。假设你的数据库里有orders和users两张表，结构如前所述。

5.1 经典问题效果对比

5.2 提升准确率的3个实用技巧

1. 给AI“划重点”：在提问时，主动强调关键约束。
   “查一下销售额”
   “查一下2024年Q2、华东区、已完成订单的总销售额”

2. 善用“示例引导”：首次使用时，可以先问一个简单问题建立信任。
   “显示users表的前3条记录” → 看它生成的SQL是否规范，再逐步增加复杂度。

3. 定期更新结构说明书：当你的数据库新增表或字段时，重新运行generate_schema.py。AI的知识库需要同步更新。

5.3 常见问题排查指南

• 问题：“ 错误：Ollama API call failed”
  解决：检查curl http://localhost:11434是否能通，确认Ollama服务正常运行。

• 问题：“ SQL语法错误：Expected end of input”
  解决：通常是AI生成了带中文标点的SQL。在validate_sql函数中加入清理逻辑：sql = sql.replace('，', ',').replace('。', ';')。

• 问题：返回结果为空，但手动执行SQL有数据
  解决：检查时区。在build_prompt中加入提示：“所有时间比较请使用CONVERT_TZ(NOW(), '+00:00', '+08:00')”。

6. 总结：让AI真正成为你的数据搭档

回顾整个过程，我们没有改动DeepChat一行源码，没有升级Ollama，甚至没有重启容器。只是通过三层轻量设计，就赋予了它强大的数据库交互能力：

• 第一层：认知层——用generate_schema.py把数据库结构翻译成AI能理解的自然语言说明书；
• 第二层：决策层——用main.py构建SQL生成、安全验证、执行反馈的闭环；
• 第三层：交互层——用前端JS注入，无缝衔接用户操作，体验零割裂。

这带来的价值是质变的：

• 对开发者：省去重复写CRUD接口的时间，把精力聚焦在核心业务逻辑；
• 对业务人员：不再需要学习SQL，用母语就能探索数据，决策速度提升3倍以上；
• 对安全团队：所有操作在容器内完成，数据不出域，权限最小化，审计日志完整。

下一步，你可以轻松扩展：

• 接入更多数据库（PostgreSQL、SQLite、甚至CSV文件）；
• 增加图表生成功能（用Chart.js把查询结果自动画成柱状图）；
• 设置查询配额，防止恶意高频请求。

技术的终极目的，不是炫技，而是让复杂变简单，让专业变普及。现在，你的DeepChat已经不只是一个对话窗口，而是一个随时待命的数据智能体。

获取更多AI镜像

想探索更多AI镜像和应用场景？访问 CSDN星图镜像广场，提供丰富的预置镜像，覆盖大模型推理、图像生成、视频生成、模型微调等多个领域，支持一键部署。