Nanbeige 4.1-3B与MySQL数据库联动智能数据查询与报告生成你有没有过这样的经历面对一个装满数据的数据库想快速了解业务情况却不知道怎么写SQL查询。或者产品经理、运营同学拿着一堆数据想让你帮忙做个分析报告但你手头还有一堆开发任务。数据就在那里但获取洞察的门槛却像一堵墙。现在情况可能不一样了。通过将像Nanbeige 4.1-3B这样的轻量级大语言模型与MySQL数据库结合起来我们可以搭建一个“会说话”的数据助手。你只需要用日常语言问它比如“上个月销售额最高的产品是什么”它就能帮你把问题变成SQL从数据库里找到答案甚至还能把结果整理成一段清晰的分析摘要。这篇文章我就来和你聊聊怎么把这件事落地。我们不讲复杂的理论就从一个具体的场景出发看看怎么一步步搭建这个系统让它真正能用起来。1. 场景与价值当数据库“听懂”人话在开始动手之前我们先看看这个组合拳到底能打在哪些地方解决什么实际问题。1.1 谁需要这个能力首先最直接的受益者是非技术背景的业务人员。想象一下市场部的同事他们不熟悉SQL语法但每天都想知道“哪个渠道带来的用户最活跃”、“新功能上线后用户留存率变化如何”。以前他们需要提需求给数据分析师或工程师排队等待。现在他们可以在一个简单的聊天框里输入问题立刻获得数据和初步解读。其次是开发者和数据分析师自己。我们经常需要做重复性的数据探查和例行报告。写类似的SQL、整理格式、做简单的趋势描述这些工作耗时且枯燥。一个智能查询助手可以承担这部分基础工作让我们能更专注于复杂的模型构建和深度分析。1.2 它能做什么具体来说这个联动系统主要能实现三个核心功能自然语言转SQLText-to-SQL这是核心。你把“帮我找出最近一周下单但未付款的用户按注册时间排序”这样的中文描述丢给它它能理解你的意图并生成对应的SELECT语句。这大大降低了查询数据库的门槛。数据报告自动摘要执行查询后面对返回的表格数据系统能自动生成一段文字总结。比如它不会只给你一堆销售数字而是告诉你“第二季度销售额环比增长15%其中华东地区贡献了主要增长产品A销量突出。”这让数据变得可读、可理解。基于数据库的智能问答结合前两者形成一个闭环的问答系统。你可以连续追问“为什么华东地区增长快”“那么产品A在华东区的具体销售数据呢”。系统能联系上下文结合数据库中的新老数据给出连贯的回答。1.3 为什么选Nanbeige 4.1-3B你可能会问现在大模型那么多为什么用这个Nanbeige 4.1-3B是一个30亿参数的中英文模型它的优势正好契合这个场景轻量高效3B的参数量对计算资源要求相对友好无论是部署在云端还是本地服务器成本都更低响应速度也更快。中文能力强针对中文进行了优化在理解中文查询指令、生成中文报告摘要方面表现更自然。指令遵循性好经过精心微调它能比较好地理解并执行“将以下问题转为SQL”或“总结以下表格数据”这类指令。当然它可能没有一些百亿、千亿参数模型那样“博学”但对于垂直的、结构化的数据库查询和总结任务它的能力是足够且高效的。2. 系统搭建思路与核心组件明白了要做什么我们来看看整个系统是怎么串起来的。整个过程可以分成几个清晰的步骤就像一条流水线。整个流程大致是这样的用户输入一个问题 - 模型理解问题并生成SQL - 系统安全审查并执行SQL - 从数据库拿到结果数据 - 模型分析数据并生成报告 - 把报告返回给用户。为了实现这个流程我们需要几个关键部分模型服务提供Nanbeige 4.1-3B模型推理能力的API。你可以用一些开源框架来部署比如FastChat、vLLM或者Xinference它们能帮你轻松启动一个模型服务并通过HTTP接口调用。应用后端这是系统的大脑负责处理业务逻辑。它接收用户请求调用模型API连接数据库处理数据。可以用Python的Web框架如FastAPI、Flask快速搭建。MySQL数据库你的数据仓库存放所有业务数据。简单的用户界面一个让用户输入问题和查看结果的网页或聊天窗口。前端可以用任何你熟悉的技术比如简单的HTML/JS或者Streamlit这种快速构建数据应用的工具。下面这张图展示了各组件如何协作用户 | | (自然语言问题) v [Web前端/界面] | | (HTTP请求包含问题) v [应用后端 (FastAPI/Flask)] | | | (调用) | (安全审查 执行) v v [模型服务] [MySQL数据库] (Nanbeige 4.1-3B) (执行SQL查询) | | | (生成的SQL/报告) | (查询结果) |________________________| | v [应用后端] (整合结果) | | (HTTP响应包含数据与报告) v [Web前端/界面] | v 用户 (看到答案)3. 分步实现指南理论说完了我们进入实战环节。我会用一个模拟的电商数据库场景来演示假设我们有一个orders订单表和一个products产品表。3.1 第一步准备环境与数据库连接首先确保你的Python环境已经就绪然后安装必要的库。pip install fastapi uvicorn sqlalchemy pymysql openai这里我们用FastAPI做后端SQLAlchemy来操作数据库pymysql是MySQL驱动。openai库的格式可以用来调用兼容OpenAI API的模型服务。接着建立数据库连接。假设你的MySQL数据库信息如下# database.py from sqlalchemy import create_engine, MetaData from sqlalchemy.orm import sessionmaker # 替换为你的实际数据库信息 DATABASE_URL mysqlpymysql://username:passwordlocalhost:3306/your_database engine create_engine(DATABASE_URL) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) metadata MetaData() # 获取数据库会话的函数 def get_db(): db SessionLocal() try: yield db finally: db.close()3.2 第二步部署与连接Nanbeige 4.1-3B模型你需要先部署好Nanbeige 4.1-3B模型服务。假设你使用Xinference部署在本地的8000端口并提供了兼容OpenAI的API。那么在后端代码中我们可以这样设置模型客户端# model_client.py from openai import OpenAI # 指向你部署的模型服务地址 client OpenAI( base_urlhttp://localhost:8000/v1, # Xinference的OpenAI兼容端点 api_keyyour-api-key-if-any, # 如果设置了密钥 ) def ask_model(prompt, max_tokens500): 向模型发送提示并获取回复 try: response client.chat.completions.create( modelnanbeige-4.1-3b, # 模型名称根据实际部署调整 messages[{role: user, content: prompt}], max_tokensmax_tokens, temperature0.1, # 温度调低让生成更确定、更稳定 ) return response.choices[0].message.content.strip() except Exception as e: return f模型调用出错: {e}3.3 第三步实现自然语言转SQL功能这是最核心的一步。我们需要精心设计一个“提示词”Prompt引导模型根据我们数据库的表结构把中文问题转换成正确的SQL。首先我们需要获取数据库的“模式信息”Schema也就是表名、字段名、字段类型等作为模型的上下文。# schema_loader.py from sqlalchemy import inspect def get_db_schema(engine): 获取数据库的schema描述信息 inspector inspect(engine) schema_info [] for table_name in inspector.get_table_names(): columns inspector.get_columns(table_name) column_desc [] for col in columns: # 简单描述列例如user_id (INT), username (VARCHAR) column_desc.append(f{col[name]} ({col[type]})) schema_info.append(f表名: {table_name}\n列: {, .join(column_desc)}) return \n\n.join(schema_info)假设我们的数据库有这些表get_db_schema函数会返回类似这样的字符串表名: orders 列: order_id (INT), user_id (INT), product_id (INT), quantity (INT), amount (DECIMAL), status (VARCHAR), created_at (DATETIME) 表名: products 列: product_id (INT), product_name (VARCHAR), category (VARCHAR), price (DECIMAL)接下来构建提示词并调用模型# text_to_sql.py from model_client import ask_model from schema_loader import get_db_schema def natural_language_to_sql(question, engine): 将自然语言问题转换为SQL查询 # 1. 获取数据库schema schema get_db_schema(engine) # 2. 构建提示词 prompt f 你是一个专业的SQL专家。请根据以下数据库表结构将用户的问题转换为一条准确、安全、只涉及查询(SELECT)的MySQL SQL语句。 数据库表结构 {schema} 请遵守以下规则 1. 只生成SELECT语句不要生成INSERT、UPDATE、DELETE等。 2. 使用易于理解的别名。 3. 如果问题中涉及“最近一周”、“上个月”等时间请使用CURDATE()、DATE_SUB等MySQL日期函数进行推算。 4. 只查询问题明确要求的数据。 用户问题{question} 请直接输出SQL语句不要有任何额外的解释、标记或代码块。 # 3. 调用模型 sql_response ask_model(prompt, max_tokens300) # 4. 简单清理响应确保只获取SQL语句 # 模型有时会在SQL外加sql 我们需要提取纯SQL if sql_response.startswith(sql): sql_response sql_response.split()[1][3:] # 去掉sql和 elif sql_response.startswith(): sql_response sql_response.split()[1] return sql_response.strip()3.4 第四步执行SQL与安全考量拿到模型生成的SQL后绝对不能直接执行必须进行安全检查防止恶意或错误的查询。# sql_executor.py import re from sqlalchemy import text def is_safe_sql(sql_statement): 进行基础的SQL安全校验 sql_upper sql_statement.upper() # 禁止非SELECT语句 if not re.match(r^\s*SELECT, sql_upper, re.IGNORECASE): return False, 只允许执行SELECT查询语句。 # 检查是否有危险操作这里只是简单示例生产环境需要更严格 dangerous_keywords [DROP, DELETE, INSERT, UPDATE, ALTER, TRUNCATE, EXEC, UNION ALL SELECT] for keyword in dangerous_keywords: if keyword in sql_upper: return False, f语句中包含被禁止的关键字: {keyword} # 可以添加更多检查如子查询深度、没有LIMIT的查询等 # 对于可能返回大量数据的查询强制添加LIMIT在提示词中已要求这里做二次保障 if LIMIT not in sql_upper: # 这是一个非常简单的处理实际应根据情况智能添加 # 例如判断是否有聚合函数没有则加LIMIT 100 if not re.search(r(COUNT|SUM|AVG|MIN|MAX)\s*\(, sql_upper): sql_statement sql_statement.rstrip(;) LIMIT 100; return True, sql_statement def execute_safe_sql(sql_statement, db_session): 执行经过安全检查的SQL is_safe, result is_safe_sql(sql_statement) if not is_safe: return None, result # result是错误信息 safe_sql result try: # 使用SQLAlchemy的text()执行原生SQL result_proxy db_session.execute(text(safe_sql)) # 获取所有结果 rows result_proxy.fetchall() # 获取列名 columns result_proxy.keys() # 将结果转为字典列表方便后续处理 data [dict(zip(columns, row)) for row in rows] return data, None except Exception as e: return None, fSQL执行错误: {e}3.5 第五步生成数据报告摘要查询到数据后我们再次请模型出马把冰冷的表格变成有温度的文字报告。# report_generator.py from model_client import ask_model def generate_data_summary(question, sql_statement, query_result): 根据查询结果生成文字摘要报告 # 将结果数据格式化为字符串方便模型阅读 # 只取前10行展示避免上下文过长 sample_data_str str(query_result[:10]) prompt f 你是一个数据分析师。用户最初的问题是“{question}” 为了回答这个问题我们执行了以下SQL查询 {sql_statement} 查询返回了{len(query_result)}条记录。以下是数据样例前10条 {sample_data_str} 请根据这些数据撰写一段简洁、专业的数据分析摘要。要求如下 1. 直接回答用户问题。 2. 指出数据中的关键发现、趋势或异常点。 3. 用口语化的中文表述避免罗列所有数据。 4. 如果数据量很大说明主要趋势即可。 5. 不要编造数据中不存在的信息。 请直接输出分析摘要 summary ask_model(prompt, max_tokens400) return summary3.6 第六步整合成API服务最后我们用FastAPI把上面的功能组装起来提供一个完整的HTTP接口。# main.py from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from pydantic import BaseModel from typing import Optional from database import get_db, engine from text_to_sql import natural_language_to_sql from sql_executor import execute_safe_sql from report_generator import generate_data_summary app FastAPI(title智能数据查询助手) class QueryRequest(BaseModel): question: str max_rows: Optional[int] 100 # 用户可指定最大返回行数 class QueryResponse(BaseModel): success: bool sql_statement: Optional[str] None data: Optional[list] None summary: Optional[str] None error: Optional[str] None app.post(/query, response_modelQueryResponse) async def intelligent_query(request: QueryRequest, db: Session Depends(get_db)): 智能数据查询端点。 1. 接收自然语言问题。 2. 转换为SQL。 3. 安全执行SQL。 4. 生成数据摘要。 user_question request.question # 1. 自然语言转SQL try: sql natural_language_to_sql(user_question, engine) if not sql: return QueryResponse(successFalse, error模型未能生成有效的SQL语句。) except Exception as e: return QueryResponse(successFalse, errorfSQL生成阶段出错: {e}) # 2. 执行SQL data, exec_error execute_safe_sql(sql, db) if exec_error: return QueryResponse(successFalse, sql_statementsql, errorexec_error) # 3. 生成报告摘要 summary if data: try: summary generate_data_summary(user_question, sql, data) except Exception as e: summary f报告生成遇到问题: {e}。以下是原始数据。 # 4. 返回结果 return QueryResponse( successTrue, sql_statementsql, datadata[:request.max_rows], # 限制返回数据量 summarysummary ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port7860)运行这个应用你就拥有了一个智能数据查询的API。你可以用Postman测试或者快速写一个前端页面来调用它。4. 实际效果与应用建议搭建好之后效果怎么样我们来模拟几个查询。用户问题“上个月销量前十的产品是哪些”生成SQLSELECT p.product_name, SUM(o.quantity) as total_quantity FROM orders o JOIN products p ON o.product_id p.product_id WHERE o.created_at DATE_SUB(CURDATE(), INTERVAL 1 MONTH) GROUP BY p.product_name ORDER BY total_quantity DESC LIMIT 10;返回摘要“根据查询上个月销量排名第一的产品是‘无线蓝牙耳机’总计售出1250件第二名是‘智能手机支架’售出980件。前十名产品均来自数码配件类别显示该品类上月需求旺盛。”可以看到系统不仅生成了正确的SQL关联了两张表处理了时间条件进行了聚合和排序还给出了有洞察的总结。在实际使用中有几点建议可以帮助你获得更好的效果提示词工程这是决定效果的关键。你需要根据自己数据库的特点不断优化给模型的指令。比如明确表之间的关联关系定义好业务术语如“活跃用户”指最近7天有登录的用户。Schema描述质量提供给模型的表结构信息越清晰、越有业务含义越好。可以在提示词中加入字段的注释说明。错误处理与反馈模型生成的SQL可能出错。系统应该能捕获数据库执行错误并尝试给用户友好的提示或者记录下这些错误案例用于后续优化提示词。分步复杂查询对于非常复杂的问题可以尝试让模型先输出一个查询计划或者分解成多个子查询分步执行。权限控制在实际业务系统中一定要结合数据库的账户权限管理确保模型生成的查询只能在授权范围内访问数据。5. 总结把Nanbeige 4.1-3B这样的轻量模型和MySQL结合起来构建一个智能数据查询系统听起来有点技术含量但拆解开来每一步都是可以实现的。它的核心价值在于降低数据获取的门槛让业务想法能快速通过数据验证。这个方案目前可能还无法100%替代专业的数据分析师因为它对复杂逻辑和深层业务背景的理解仍有局限。但它绝对是一个强大的“副驾驶”能处理大量常规、重复的查询和报告需求把人从繁琐的“取数”工作中解放出来。如果你正在为团队的数据查询效率发愁或者想给产品增加一个智能数据分析的亮点不妨从一个小而具体的场景开始尝试。比如先针对一两个核心业务表搭建原型让一两个业务同学试用收集反馈。技术总是在解决具体问题的过程中迭代成熟的。也许用不了多久你团队里的每个人都能拥有一个随叫随到的“数据助手”了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
Nanbeige 4.1-3B与MySQL数据库联动：智能数据查询与报告生成
Nanbeige 4.1-3B与MySQL数据库联动智能数据查询与报告生成你有没有过这样的经历面对一个装满数据的数据库想快速了解业务情况却不知道怎么写SQL查询。或者产品经理、运营同学拿着一堆数据想让你帮忙做个分析报告但你手头还有一堆开发任务。数据就在那里但获取洞察的门槛却像一堵墙。现在情况可能不一样了。通过将像Nanbeige 4.1-3B这样的轻量级大语言模型与MySQL数据库结合起来我们可以搭建一个“会说话”的数据助手。你只需要用日常语言问它比如“上个月销售额最高的产品是什么”它就能帮你把问题变成SQL从数据库里找到答案甚至还能把结果整理成一段清晰的分析摘要。这篇文章我就来和你聊聊怎么把这件事落地。我们不讲复杂的理论就从一个具体的场景出发看看怎么一步步搭建这个系统让它真正能用起来。1. 场景与价值当数据库“听懂”人话在开始动手之前我们先看看这个组合拳到底能打在哪些地方解决什么实际问题。1.1 谁需要这个能力首先最直接的受益者是非技术背景的业务人员。想象一下市场部的同事他们不熟悉SQL语法但每天都想知道“哪个渠道带来的用户最活跃”、“新功能上线后用户留存率变化如何”。以前他们需要提需求给数据分析师或工程师排队等待。现在他们可以在一个简单的聊天框里输入问题立刻获得数据和初步解读。其次是开发者和数据分析师自己。我们经常需要做重复性的数据探查和例行报告。写类似的SQL、整理格式、做简单的趋势描述这些工作耗时且枯燥。一个智能查询助手可以承担这部分基础工作让我们能更专注于复杂的模型构建和深度分析。1.2 它能做什么具体来说这个联动系统主要能实现三个核心功能自然语言转SQLText-to-SQL这是核心。你把“帮我找出最近一周下单但未付款的用户按注册时间排序”这样的中文描述丢给它它能理解你的意图并生成对应的SELECT语句。这大大降低了查询数据库的门槛。数据报告自动摘要执行查询后面对返回的表格数据系统能自动生成一段文字总结。比如它不会只给你一堆销售数字而是告诉你“第二季度销售额环比增长15%其中华东地区贡献了主要增长产品A销量突出。”这让数据变得可读、可理解。基于数据库的智能问答结合前两者形成一个闭环的问答系统。你可以连续追问“为什么华东地区增长快”“那么产品A在华东区的具体销售数据呢”。系统能联系上下文结合数据库中的新老数据给出连贯的回答。1.3 为什么选Nanbeige 4.1-3B你可能会问现在大模型那么多为什么用这个Nanbeige 4.1-3B是一个30亿参数的中英文模型它的优势正好契合这个场景轻量高效3B的参数量对计算资源要求相对友好无论是部署在云端还是本地服务器成本都更低响应速度也更快。中文能力强针对中文进行了优化在理解中文查询指令、生成中文报告摘要方面表现更自然。指令遵循性好经过精心微调它能比较好地理解并执行“将以下问题转为SQL”或“总结以下表格数据”这类指令。当然它可能没有一些百亿、千亿参数模型那样“博学”但对于垂直的、结构化的数据库查询和总结任务它的能力是足够且高效的。2. 系统搭建思路与核心组件明白了要做什么我们来看看整个系统是怎么串起来的。整个过程可以分成几个清晰的步骤就像一条流水线。整个流程大致是这样的用户输入一个问题 - 模型理解问题并生成SQL - 系统安全审查并执行SQL - 从数据库拿到结果数据 - 模型分析数据并生成报告 - 把报告返回给用户。为了实现这个流程我们需要几个关键部分模型服务提供Nanbeige 4.1-3B模型推理能力的API。你可以用一些开源框架来部署比如FastChat、vLLM或者Xinference它们能帮你轻松启动一个模型服务并通过HTTP接口调用。应用后端这是系统的大脑负责处理业务逻辑。它接收用户请求调用模型API连接数据库处理数据。可以用Python的Web框架如FastAPI、Flask快速搭建。MySQL数据库你的数据仓库存放所有业务数据。简单的用户界面一个让用户输入问题和查看结果的网页或聊天窗口。前端可以用任何你熟悉的技术比如简单的HTML/JS或者Streamlit这种快速构建数据应用的工具。下面这张图展示了各组件如何协作用户 | | (自然语言问题) v [Web前端/界面] | | (HTTP请求包含问题) v [应用后端 (FastAPI/Flask)] | | | (调用) | (安全审查 执行) v v [模型服务] [MySQL数据库] (Nanbeige 4.1-3B) (执行SQL查询) | | | (生成的SQL/报告) | (查询结果) |________________________| | v [应用后端] (整合结果) | | (HTTP响应包含数据与报告) v [Web前端/界面] | v 用户 (看到答案)3. 分步实现指南理论说完了我们进入实战环节。我会用一个模拟的电商数据库场景来演示假设我们有一个orders订单表和一个products产品表。3.1 第一步准备环境与数据库连接首先确保你的Python环境已经就绪然后安装必要的库。pip install fastapi uvicorn sqlalchemy pymysql openai这里我们用FastAPI做后端SQLAlchemy来操作数据库pymysql是MySQL驱动。openai库的格式可以用来调用兼容OpenAI API的模型服务。接着建立数据库连接。假设你的MySQL数据库信息如下# database.py from sqlalchemy import create_engine, MetaData from sqlalchemy.orm import sessionmaker # 替换为你的实际数据库信息 DATABASE_URL mysqlpymysql://username:passwordlocalhost:3306/your_database engine create_engine(DATABASE_URL) SessionLocal sessionmaker(autocommitFalse, autoflushFalse, bindengine) metadata MetaData() # 获取数据库会话的函数 def get_db(): db SessionLocal() try: yield db finally: db.close()3.2 第二步部署与连接Nanbeige 4.1-3B模型你需要先部署好Nanbeige 4.1-3B模型服务。假设你使用Xinference部署在本地的8000端口并提供了兼容OpenAI的API。那么在后端代码中我们可以这样设置模型客户端# model_client.py from openai import OpenAI # 指向你部署的模型服务地址 client OpenAI( base_urlhttp://localhost:8000/v1, # Xinference的OpenAI兼容端点 api_keyyour-api-key-if-any, # 如果设置了密钥 ) def ask_model(prompt, max_tokens500): 向模型发送提示并获取回复 try: response client.chat.completions.create( modelnanbeige-4.1-3b, # 模型名称根据实际部署调整 messages[{role: user, content: prompt}], max_tokensmax_tokens, temperature0.1, # 温度调低让生成更确定、更稳定 ) return response.choices[0].message.content.strip() except Exception as e: return f模型调用出错: {e}3.3 第三步实现自然语言转SQL功能这是最核心的一步。我们需要精心设计一个“提示词”Prompt引导模型根据我们数据库的表结构把中文问题转换成正确的SQL。首先我们需要获取数据库的“模式信息”Schema也就是表名、字段名、字段类型等作为模型的上下文。# schema_loader.py from sqlalchemy import inspect def get_db_schema(engine): 获取数据库的schema描述信息 inspector inspect(engine) schema_info [] for table_name in inspector.get_table_names(): columns inspector.get_columns(table_name) column_desc [] for col in columns: # 简单描述列例如user_id (INT), username (VARCHAR) column_desc.append(f{col[name]} ({col[type]})) schema_info.append(f表名: {table_name}\n列: {, .join(column_desc)}) return \n\n.join(schema_info)假设我们的数据库有这些表get_db_schema函数会返回类似这样的字符串表名: orders 列: order_id (INT), user_id (INT), product_id (INT), quantity (INT), amount (DECIMAL), status (VARCHAR), created_at (DATETIME) 表名: products 列: product_id (INT), product_name (VARCHAR), category (VARCHAR), price (DECIMAL)接下来构建提示词并调用模型# text_to_sql.py from model_client import ask_model from schema_loader import get_db_schema def natural_language_to_sql(question, engine): 将自然语言问题转换为SQL查询 # 1. 获取数据库schema schema get_db_schema(engine) # 2. 构建提示词 prompt f 你是一个专业的SQL专家。请根据以下数据库表结构将用户的问题转换为一条准确、安全、只涉及查询(SELECT)的MySQL SQL语句。 数据库表结构 {schema} 请遵守以下规则 1. 只生成SELECT语句不要生成INSERT、UPDATE、DELETE等。 2. 使用易于理解的别名。 3. 如果问题中涉及“最近一周”、“上个月”等时间请使用CURDATE()、DATE_SUB等MySQL日期函数进行推算。 4. 只查询问题明确要求的数据。 用户问题{question} 请直接输出SQL语句不要有任何额外的解释、标记或代码块。 # 3. 调用模型 sql_response ask_model(prompt, max_tokens300) # 4. 简单清理响应确保只获取SQL语句 # 模型有时会在SQL外加sql 我们需要提取纯SQL if sql_response.startswith(sql): sql_response sql_response.split()[1][3:] # 去掉sql和 elif sql_response.startswith(): sql_response sql_response.split()[1] return sql_response.strip()3.4 第四步执行SQL与安全考量拿到模型生成的SQL后绝对不能直接执行必须进行安全检查防止恶意或错误的查询。# sql_executor.py import re from sqlalchemy import text def is_safe_sql(sql_statement): 进行基础的SQL安全校验 sql_upper sql_statement.upper() # 禁止非SELECT语句 if not re.match(r^\s*SELECT, sql_upper, re.IGNORECASE): return False, 只允许执行SELECT查询语句。 # 检查是否有危险操作这里只是简单示例生产环境需要更严格 dangerous_keywords [DROP, DELETE, INSERT, UPDATE, ALTER, TRUNCATE, EXEC, UNION ALL SELECT] for keyword in dangerous_keywords: if keyword in sql_upper: return False, f语句中包含被禁止的关键字: {keyword} # 可以添加更多检查如子查询深度、没有LIMIT的查询等 # 对于可能返回大量数据的查询强制添加LIMIT在提示词中已要求这里做二次保障 if LIMIT not in sql_upper: # 这是一个非常简单的处理实际应根据情况智能添加 # 例如判断是否有聚合函数没有则加LIMIT 100 if not re.search(r(COUNT|SUM|AVG|MIN|MAX)\s*\(, sql_upper): sql_statement sql_statement.rstrip(;) LIMIT 100; return True, sql_statement def execute_safe_sql(sql_statement, db_session): 执行经过安全检查的SQL is_safe, result is_safe_sql(sql_statement) if not is_safe: return None, result # result是错误信息 safe_sql result try: # 使用SQLAlchemy的text()执行原生SQL result_proxy db_session.execute(text(safe_sql)) # 获取所有结果 rows result_proxy.fetchall() # 获取列名 columns result_proxy.keys() # 将结果转为字典列表方便后续处理 data [dict(zip(columns, row)) for row in rows] return data, None except Exception as e: return None, fSQL执行错误: {e}3.5 第五步生成数据报告摘要查询到数据后我们再次请模型出马把冰冷的表格变成有温度的文字报告。# report_generator.py from model_client import ask_model def generate_data_summary(question, sql_statement, query_result): 根据查询结果生成文字摘要报告 # 将结果数据格式化为字符串方便模型阅读 # 只取前10行展示避免上下文过长 sample_data_str str(query_result[:10]) prompt f 你是一个数据分析师。用户最初的问题是“{question}” 为了回答这个问题我们执行了以下SQL查询 {sql_statement} 查询返回了{len(query_result)}条记录。以下是数据样例前10条 {sample_data_str} 请根据这些数据撰写一段简洁、专业的数据分析摘要。要求如下 1. 直接回答用户问题。 2. 指出数据中的关键发现、趋势或异常点。 3. 用口语化的中文表述避免罗列所有数据。 4. 如果数据量很大说明主要趋势即可。 5. 不要编造数据中不存在的信息。 请直接输出分析摘要 summary ask_model(prompt, max_tokens400) return summary3.6 第六步整合成API服务最后我们用FastAPI把上面的功能组装起来提供一个完整的HTTP接口。# main.py from fastapi import FastAPI, Depends, HTTPException from sqlalchemy.orm import Session from pydantic import BaseModel from typing import Optional from database import get_db, engine from text_to_sql import natural_language_to_sql from sql_executor import execute_safe_sql from report_generator import generate_data_summary app FastAPI(title智能数据查询助手) class QueryRequest(BaseModel): question: str max_rows: Optional[int] 100 # 用户可指定最大返回行数 class QueryResponse(BaseModel): success: bool sql_statement: Optional[str] None data: Optional[list] None summary: Optional[str] None error: Optional[str] None app.post(/query, response_modelQueryResponse) async def intelligent_query(request: QueryRequest, db: Session Depends(get_db)): 智能数据查询端点。 1. 接收自然语言问题。 2. 转换为SQL。 3. 安全执行SQL。 4. 生成数据摘要。 user_question request.question # 1. 自然语言转SQL try: sql natural_language_to_sql(user_question, engine) if not sql: return QueryResponse(successFalse, error模型未能生成有效的SQL语句。) except Exception as e: return QueryResponse(successFalse, errorfSQL生成阶段出错: {e}) # 2. 执行SQL data, exec_error execute_safe_sql(sql, db) if exec_error: return QueryResponse(successFalse, sql_statementsql, errorexec_error) # 3. 生成报告摘要 summary if data: try: summary generate_data_summary(user_question, sql, data) except Exception as e: summary f报告生成遇到问题: {e}。以下是原始数据。 # 4. 返回结果 return QueryResponse( successTrue, sql_statementsql, datadata[:request.max_rows], # 限制返回数据量 summarysummary ) if __name__ __main__: import uvicorn uvicorn.run(app, host0.0.0.0, port7860)运行这个应用你就拥有了一个智能数据查询的API。你可以用Postman测试或者快速写一个前端页面来调用它。4. 实际效果与应用建议搭建好之后效果怎么样我们来模拟几个查询。用户问题“上个月销量前十的产品是哪些”生成SQLSELECT p.product_name, SUM(o.quantity) as total_quantity FROM orders o JOIN products p ON o.product_id p.product_id WHERE o.created_at DATE_SUB(CURDATE(), INTERVAL 1 MONTH) GROUP BY p.product_name ORDER BY total_quantity DESC LIMIT 10;返回摘要“根据查询上个月销量排名第一的产品是‘无线蓝牙耳机’总计售出1250件第二名是‘智能手机支架’售出980件。前十名产品均来自数码配件类别显示该品类上月需求旺盛。”可以看到系统不仅生成了正确的SQL关联了两张表处理了时间条件进行了聚合和排序还给出了有洞察的总结。在实际使用中有几点建议可以帮助你获得更好的效果提示词工程这是决定效果的关键。你需要根据自己数据库的特点不断优化给模型的指令。比如明确表之间的关联关系定义好业务术语如“活跃用户”指最近7天有登录的用户。Schema描述质量提供给模型的表结构信息越清晰、越有业务含义越好。可以在提示词中加入字段的注释说明。错误处理与反馈模型生成的SQL可能出错。系统应该能捕获数据库执行错误并尝试给用户友好的提示或者记录下这些错误案例用于后续优化提示词。分步复杂查询对于非常复杂的问题可以尝试让模型先输出一个查询计划或者分解成多个子查询分步执行。权限控制在实际业务系统中一定要结合数据库的账户权限管理确保模型生成的查询只能在授权范围内访问数据。5. 总结把Nanbeige 4.1-3B这样的轻量模型和MySQL结合起来构建一个智能数据查询系统听起来有点技术含量但拆解开来每一步都是可以实现的。它的核心价值在于降低数据获取的门槛让业务想法能快速通过数据验证。这个方案目前可能还无法100%替代专业的数据分析师因为它对复杂逻辑和深层业务背景的理解仍有局限。但它绝对是一个强大的“副驾驶”能处理大量常规、重复的查询和报告需求把人从繁琐的“取数”工作中解放出来。如果你正在为团队的数据查询效率发愁或者想给产品增加一个智能数据分析的亮点不妨从一个小而具体的场景开始尝试。比如先针对一两个核心业务表搭建原型让一两个业务同学试用收集反馈。技术总是在解决具体问题的过程中迭代成熟的。也许用不了多久你团队里的每个人都能拥有一个随叫随到的“数据助手”了。获取更多AI镜像想探索更多AI镜像和应用场景访问 CSDN星图镜像广场提供丰富的预置镜像覆盖大模型推理、图像生成、视频生成、模型微调等多个领域支持一键部署。
