---
id: scenario.report-agent
title: 报表生成 Agent：自然语言查数、安全的 Text-to-SQL、结果校验、图表和成本控制
type: scenario
domain: cross
stack: [agent-design, function-calling, postgresql, data-pipeline, web-security, react, agent-ops]
tags: [Text-to-SQL, 报表, 只读账号, SQL 校验, 语义层, 行级安全, 评测]
level: advanced
prerequisites: [ai.agent-design.index, backend.postgresql.index]
est_hours: 3
related: [ai.agent-design.scenario.report-agent, cs.web-security.scenario.ai-app-security, backend.data-pipeline.scenario.third-party-sync]
week: [10, 13, 14]
---

# 报表生成 Agent：自然语言查数、安全的 Text-to-SQL、结果校验、图表和成本控制

## 业务背景与问题

运营和商家希望用一句话查数据（“上个月各渠道的退款率，按周看趋势”）。早期原型直接让模型写 SQL 执行：

1. 生成的 SQL 有时口径错误，结果看起来合理但是错的；
2. 有一次生成了没有租户条件的查询，看到了其他商家的数据；
3. 复杂查询拖慢了业务主库；
4. 一个问题调用模型多次，成本不可控。

## 约束

- 数据来自同步过来的订单库（见[数据同步](data-sync.md)）。
- 只读；结果要能解释（口径、SQL）。

## 方案设计

1. **语义层**：数据团队以 YAML 定义指标（退款率 = 退款订单数 / 支付订单数）和维度（渠道、周）；模型优先选择“指标 + 维度 + 过滤条件”，由代码生成 SQL；覆盖不了的才让模型写 SQL。
2. **SQL 安全**（见 [AI 应用安全](../topics/cs/web-security/scenarios/ai-app-security.md)）：只读账号、只授权报表视图、连接只读副本；SQL 解析后只允许单条 SELECT 和白名单表；**租户隔离由数据库行级安全（RLS）保证**，会话变量由代码设置，模型无法绕过；强制 LIMIT、语句超时。
3. **Agent 流程**：理解 → 澄清（时间范围不明确时先问）→ 选指标 / 生成 SQL → 校验 → 执行 → 结果检查（空结果、异常值）→ 图表配置和文字结论。见[报表 Agent](../topics/ai/agent-design/scenarios/report-agent.md)。
4. **可解释**：附口径说明和 SQL（可折叠），用户可“报告错误”进入 Bad Case 库。
5. **成本**：模型调用次数上限（中间件）、常见问题缓存、简单问题路由到小模型（见[多模型网关](../topics/ai/llm-serving/scenarios/multi-provider-gateway.md)）。
6. **评测**：问题 → 期望结果（不是期望 SQL），比较执行结果；CI 中运行。

## 关键代码

数据库层的租户隔离（PostgreSQL RLS）：

```sql
ALTER TABLE rpt_orders ENABLE ROW LEVEL SECURITY;
CREATE POLICY tenant_isolation ON rpt_orders
    USING (tenant_id = current_setting('app.tenant_id'));
GRANT SELECT ON rpt_orders TO report_reader;        -- 报表账号只读，且不是表的所有者（所有者默认不受 RLS 限制）
```

执行模型生成的 SQL（示意）：

```python
def run_report_sql(sql: str, tenant_id: str) -> list[tuple]:
    check_select_only(sql, allowed_tables={"rpt_orders", "rpt_refunds"})   # 用 SQL 解析库（如 sqlglot）做结构校验，不用正则
    with readonly_pool.connection() as conn, conn.transaction():
        conn.execute("SET LOCAL statement_timeout = '5s'")
        conn.execute("SELECT set_config('app.tenant_id', %s, true)", (tenant_id,))   # 事务内有效，由代码设置
        return conn.execute(f"SELECT * FROM ({sql}) AS q LIMIT 1000").fetchall()
```

## 取舍

| 决策 | 选择 | 原因 |
| --- | --- | --- |
| 生成方式 | 语义层优先，SQL 兜底 | 口径一致、可控 |
| 租户隔离 | RLS + 会话变量 | 不依赖模型写对条件，子查询 / CTE 也覆盖 |
| 执行位置 | 只读副本 | 不影响主库 |
| 评测方式 | 比较执行结果 | 同一问题有多种正确 SQL |

## 易错点

- 让模型自己写租户条件。
- 用正则判断 SQL 是否只读（注释、多语句可绕过）。
- 报表账号是表的所有者，RLS 不生效（需要 `FORCE ROW LEVEL SECURITY` 或换账号）。
- 结果为空时模型“编”一个结论。
- 只评测 SQL 能否执行，不评测结果是否正确。

## 面试怎么讲

> “报表 Agent 的核心是不信任模型生成的 SQL。常见指标走语义层由代码生成 SQL；必须让模型写时，解析后只允许单条 SELECT 和白名单表，租户隔离交给 PostgreSQL 的行级安全，会话变量由代码设置，模型绕不过去；只读账号连只读副本并设超时。评测集比较执行结果而不是 SQL 文本，每个回答附口径和 SQL，用户一键报错进 Bad Case 库。”