读完这篇，你能做到这样一件事：对着 OceanBase 里的业务库，用中文问一句「上个月华东区销售额前五的门店是哪些，环比怎么样」，几秒后拿到一张结果表、一张柱状图，以及三段基于数据写的结论，全程不手写 SQL。
下面用一个「门店 + 订单明细」的小数据集把它跑通。整套链路分三块：数据源（OceanBase）→ 语义层（让模型看懂你的表）→ 执行与报表（SQL 校验 + 查询 + 出结果）。OceanBase 生态里的 Data Agent 能力在不同版本、不同平台上的入口名称和界面会有差异，具体开通方式和菜单路径以官方文档当前版本为准；本文重点讲你能自己掌控的那部分——数据怎么准备、问题怎么问、SQL 怎么防错。
前置条件清单
- 一个可连接的 OceanBase 实例，MySQL 模式（本地 Docker、自建集群、云上租户都可以）。
- 客户端工具任选其一：
obclient、mysql命令行，或 Python。 - Python 3.x 及
pymysql、pandas、openai三个包（后两个用于出报表和调模型）。 - 一个大模型服务的调用凭据（OpenAI 兼容接口的
base_url+api_key+ 模型名，模型名以你所用平台的官方文档为准）。 - 一个只有
SELECT权限的数据库账号。这一条不是可选项，后面会讲原因。
安装依赖：
```bash
pip install pymysql pandas openai
```
第 1 步：把 OceanBase 连上
本地起一个实例的方式，镜像名和 tag 请以 OceanBase 官方文档当前版本为准：
```bash
docker run -d --name ob -p 2881:2881 <OCEANBASE_IMAGE>
```
容器起来后，用命令行连进去。注意 OceanBase 的用户名格式，直连 observer 和经过 OBProxy 的端口不一样：
```bash
直连 observer 默认 2881；经过 OBProxy 一般是 2883
用户格式：用户名@租户名#集群名，具体以你的部署方式为准
obclient -h 127.0.0.1 -P 2881 -u root@sys -p
```
连上以后先确认能查：
```sql
SHOW DATABASES;
SELECT VERSION();
```
如果这里就连不上，先别往下走，排查看后面的「常见坑」。
第 2 步：建演示库表和一批数据
在 MySQL 模式下执行下面这段，建两张维表/事实表级别的表并塞入数据：
```sql
CREATE DATABASE IF NOT EXISTS demo_retail DEFAULT CHARACTER SET utf8mb4;
USE demo_retail;
CREATE TABLE dim_store (
store_id INT PRIMARY KEY,
store_name VARCHAR(64) COMMENT '门店名称',
region VARCHAR(32) COMMENT '大区，如华东/华南/华北',
city VARCHAR(32) COMMENT '城市'
);
CREATE TABLE fact_order (
order_id BIGINT PRIMARY KEY,
store_id INT,
order_date DATE,
amount DECIMAL(12,2) COMMENT '订单实收金额，含税，已剔除取消订单',
member_id BIGINT,
channel VARCHAR(16) COMMENT '下单渠道：app/小程序/门店'
);
INSERT INTO dim_store VALUES
(1,'上海南京西路店','华东','上海'),
(2,'杭州湖滨店','华东','杭州'),
(3,'广州天河店','华南','广州'),
(4,'北京朝阳店','华北','北京');
INSERT INTO fact_order VALUES
(1001,1,'2024-05-03',268.00,9001,'app'),
(1002,1,'2024-05-11',512.50,9002,'小程序'),
(1003,2,'2024-05-12',199.00,9003,'门店'),
(1004,3,'2024-05-15',880.00,9004,'app'),
(1005,1,'2024-06-02',301.00,9002,'app'),
(1006,2,'2024-06-05',420.00,9005,'小程序'),
(1007,4,'2024-06-09',150.00,9006,'门店'),
(1008,3,'2024-06-18',960.00,9004,'app');
```
两个细节值得注意：amount 的注释里写清了「含税、已剔除取消订单」，channel 里列出了全部合法取值。这些注释后面会被直接喂给模型。
第 3 步：开一个只读账号
自然语言查询意味着 SQL 是模型生成的，模型会写错，也可能被提问诱导写出危险语句。用权限兜底比用提示词兜底可靠。
```sql
CREATE USER 'ai_reader'@'%' IDENTIFIED BY '换成你的强密码';
GRANT SELECT ON demo_retail.* TO 'ai_reader'@'%';
FLUSH PRIVILEGES;
```
验证一下这个账号确实写不进去：
```bash
obclient -h 127.0.0.1 -P 2881 -u ai_reader -p -D demo_retail
```
```sql
SELECT COUNT(*) FROM fact_order; -- 正常
DELETE FROM fact_order; -- 应当被拒绝
```
第 4 步：写「语义层」，这是效果的分水岭
Data Agent 类工具的效果差异，八成来自这一步，而不是模型换了哪家。模型看不到你的业务口径，只能看到列名。所以你要准备一份结构说明，内容包括三部分：表是干什么的、字段是什么含义、指标怎么算。
用 Python 常量维护一份就够用，也可以落成一张元数据表：
```python
SCHEMA = """
【表 dim_store】门店维表，约 4 行，每行一个门店。
store_id 门店ID，主键
store_name 门店名称
region 大区，取值：华东、华南、华北
city 城市
【表 fact_order】订单明细事实表，每行一笔订单。
order_id 订单ID，主键
store_id 门店ID，关联 dim_store.store_id
order_date 下单日期，DATE 类型
amount 订单实收金额，单位元，含税，已剔除已取消订单
member_id 会员ID，可为空表示非会员
channel 下单渠道，取值：app、小程序、门店
【指标口径】
销售额 = SUM(fact_order.amount)，不要用 order_id 计数代替
「上个月」指相对于当前日期的上一个自然月，跨月请用日期范围条件
环比 = (本期 - 上期) / 上期
【命名约定】表名和字段名均为小写下划线，查询时不需要加引号。
"""
```
如果只读账号在 information_schema 里看不到全部对象，可以让 DBA 用高权限账号导出一份结构快照，定期更新给 Agent 用，这样运行时不依赖元数据表的可见性。
第 5 步：接上 Data Agent，两条路线
路线 A：用平台自带的 Data Agent 入口。 在 OceanBase 的控制台或云平台里按官方文档开通对应功能，填数据源连接信息（主机、端口、租户、账号）、选择模型、把上面的语义层内容录进去。界面字段名称各版本不同，以官方文档当前版本为准。这条路省事，适合先跑通感觉。
路线 B：自己搭一条最小链路。 好处是语义层、安全策略、报表格式都捏在自己手里。下面这套代码就是这条路线，可以直接照抄改配置。
先写数据库连接和 SQL 安全闸门：
```python
agent_core.py
import re
import pymysql
DB = dict(
host="127.0.0.1", port=2881,
user="ai_reader", password="你的密码",
database="demo_retail", charset="utf8mb4",
cursorclass=pymysql.cursors.DictCursor,
connect_timeout=5, read_timeout=30,
)
FORBIDDEN = re.compile(
r"\b(insert|update|delete|drop|truncate|alter|create|grant|revoke|replace)\b",
re.IGNORECASE,
)
def validate_sql(sql: str) -> str:
sql = sql.strip().rstrip(";").strip()
if not sql:
raise ValueError("空 SQL")
if not re.match(r"^(select|with)\b", sql, re.IGNORECASE):
raise ValueError("只允许 SELECT / WITH 开头的查询")
if FORBIDDEN.search(sql):
raise ValueError("包含被禁止的写操作关键字")
if " limit " not in sql.lower():
sql += " LIMIT 200" # 兜底行数上限
return sql
def run_query(sql: str):
sql = validate_sql(sql)
conn = pymysql.connect(**DB)
try:
with conn.cursor() as cur:
cur.execute(sql)
return cur.fetchall()
finally:
conn.close()
```
再写调模型的函数。SDK 的调用方式和参数以你所用 SDK 的官方文档为准：
```python
llm.py
import os
from openai import OpenAI
client = OpenAI(
api_key=os.environ["LLM_API_KEY"],
base_url=os.environ.get("LLM_BASE_URL"), # 按你的平台文档填写
)
SYSTEM = """你是一名严谨的数据分析师。根据给定的表结构，把用户的中文问题翻译成一条 OceanBase(MySQL 模式) 可执行的 SQL。
硬性要求：
1. 只输出 SQL 本身，不要解释，不要 markdown 代码块标记。
2. 只能使用表结构里出现过的表和字段，禁止臆造。
3. 必须带 LIMIT，聚合查询除外。
4. 涉及金额时使用 SUM(amount)，不要用 COUNT(*) 代替销售额。
5. 日期条件请写成显式的日期范围，避免依赖数据库时区。"""
def text_to_sql(question: str, schema: str, history: str = "") -> str:
resp = client.chat.completions.create(
model=os.environ["LLM_MODEL"],
temperature=0,
messages=[
{"role": "system", "content": SYSTEM},
{"role": "user", "content": f"表结构：\n{schema}\n\n历史对话：\n{history}\n\n问题：{question}"},
],
)
return resp.choices[0].message.content.strip()
```
temperature=0 是为了让同样的提问尽量产出同样的 SQL，便于排查问题。
第 6 步：问第一个问题，把链路跑通
把两段拼起来试一句：
```python
ask.py
from agent_core import run_query
from llm import text_to_sql, SCHEMA_PLACEHOLDER # 换成你在第 4 步定义的 SCHEMA
q = "上个月华东区哪些门店销售额最高？给我门店名和销售额，按销售额从高到低排"
sql = text_to_sql(q, SCHEMA_PLACEHOLDER)
print("生成的 SQL：\n", sql)
rows = run_query(sql)
for r in rows:
print(r)
```
生成的 SQL 大致长这样：
```sql
SELECT s.store_name, SUM(o.amount) AS sales
FROM fact_order o
JOIN dim_store s ON s.store_id = o.store_id
WHERE s.region = '华东'
AND o.order_date >= '2024-06-01' AND o.order_date < '2024-07-01'
GROUP BY s.store_name
ORDER BY sales DESC
```
跑出结果后，务必人工对一遍数字。拿这段 SQL 直接在 obclient 里手工执行，确认两边一致，再进入下一步。前几次的「对答案」是建立信任的必要成本。
第 7 步：把单次问答变成报表
报表的价值在于「一次问一批问题，出一份能发出去的东西」。把问题列成清单，循环执行，结果落成 Markdown，再顺手画张图：
```python
report.py
import pandas as pd
import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt
from agent_core import run_query
from llm import text_to_sql, SCHEMA_PLACEHOLDER
QUESTIONS = [
"各大区上个月的销售额分别是多少？按销售额降序",
"上个月销售额前五的门店及其销售额",
"上个月 app、小程序、门店三个渠道的销售额占比",
]
lines = ["# 门店经营周报\n"]
for q in QUESTIONS:
sql = text_to_sql(q, SCHEMA_PLACEHOLDER)
rows = run_query(sql)
df = pd.DataFrame(rows)
lines.append(f"## {q}\n")
lines.append(df.to_markdown(index=False) + "\n")
if len(df) >= 2 and df.shape[1] >= 2:
df.plot(x=df.columns[0], y=df.columns[1], kind="bar", legend=False)
plt.tight_layout()
fn = f"chart_{len(lines)}.png"
plt.savefig(fn)
plt.close()
lines.append(f"!图表\n")
open("report.md", "w", encoding="utf-8").write("\n".join(lines))
print("已生成 report.md")
```
需要 Excel 的话，把每个问题写到一个 sheet 里：
```python
with pd.ExcelWriter("report.xlsx") as writer:
for i, q in enumerate(QUESTIONS, 1):
df = pd.DataFrame(run_query(text_to_sql(q, SCHEMA_PLACEHOLDER)))
df.to_excel(writer, sheet_name=f"Q{i}", index=False)
```
第 8 步：让模型把结论也写出来
有了数据，再让模型基于数据写结论——但要把原始结果喂给它，并明确约束：
```python
SUMMARY_SYSTEM = """你是一名数据分析师。只根据用户提供的数据表写结论。
要求：不超过三条；每条必须引用具体数字；数据不足以支撑的判断不要写；不要提建议以外的事实。"""
def summarize(question, rows):
payload = f"问题：{question}\n数据：{rows}"
resp = client.chat.completions.create(
model=os.environ["LLM_MODEL"],
temperature=0,
messages=[
{"role": "system", "content": SUMMARY_SYSTEM},
{"role": "user", "content": payload},
],
)
return resp.choices[0].message.content
```
注意 rows 只传你查询返回的那部分数据，不要传整表。
到这里，一条「中文提问 → SQL → 执行 → 表格 + 图表 + 结论」的完整链路就跑通了。
常见坑与排错
连不上数据库。 先分清端口：直连 observer 通常是 2881，走 OBProxy 通常是 2883。用户名格式在不同部署下不同，自建集群多为 用户名@租户名 或 用户名@租户名#集群名。报「Access denied」时，先确认账号和租户是否匹配，再看密码。具体格式以官方文档当前版本为准。
模型老是写错字段名。 九成是语义层没喂全，或者字段名不够直观。把列名和注释一并写进 SCHEMA，并在系统提示里明确「只能使用表结构里出现过的表和字段」。另加一道防线：拿到 SQL 后，用正则从里面提取表名，和自己维护的白名单比对，不在白名单里的直接拒绝执行。
同一个问题每次生成的 SQL 不一样。 把 temperature 设为 0；把历史对话里的 SQL 一并带进去，让模型做增量修改而不是重新生成。
数字对不上。 通常是口径问题：含税还是不含税、退款算不算、测试门店有没有排除。解决办法不是改提示词，而是把口径写进第 4 步的语义层，并且让业务方确认一次。
查询很慢。 生成的 SQL 容易漏掉时间范围而全表扫。在提示词里明确「必须用日期范围收窄」，同时给数据库账号配上资源限制，再在 validate_sql 里对没有 LIMIT 的语句强制追加行数上限。
权限报错说找不到表。 只读账号只被授权了部分库表，模型可能引用到它看不到的对象。要么补授权，要么用静态 schema 快照替代运行时读元数据。
中文出现乱码。 连接串里显式指定 charset="utf8mb4"，建库时也用 DEFAULT CHARACTER SET utf8mb4。
模型生成的 SQL 在 OceanBase 上执行失败。 MySQL 模式下大部分语法兼容，但日期函数、窗口函数等细节建议对着官方文档确认一遍，把不支持的写法写进系统提示的禁用清单。
下一步建议
1. 沉淀问题库。 把常用问题整理成清单，配上定时任务，每天或每周自动跑一遍出日报，人只需要看结论。
2. 补一层缓存。 相同问题在短时间内返回缓存结果，省调用费也省数据库压力。
3. 接进 BI 或办公工具。 report.md 和 report.xlsx 可以直接推到报表系统或群机器人，让不看 SQL 的同事也能用。
4. 做效果评估。 抽一批问题，人工标注正确 SQL，定期统计「生成 SQL 可直接执行的比例」和「数字正确率」，模型换了或语义层改了都能立刻看出影响。
5. 区分环境。 生产库给只读账号，测试库放开权限做实验，别在生产上试新提示词。
真正决定这套东西好不好用的，不是模型参数有多大，而是语义层写得够不够清楚、安全边界设得够不够死。把这两件事做扎实，自然语言数据分析就能从演示变成日常工具。