让业务同事用说话查数据：一个取数 Agent 的四层架构与核心代码

把 Text2SQL 接到聊天框里第一版通常跑不顺。原因不是 SQL 生成不准而是真实对话里的问题从来不是单轮的「那华东呢」——上一轮的指标和时间段要继承下来「按这个口径再算一遍去年」——口径指的是哪一条得记住「这两个数怎么差这么多」——要能自己回去查口径定义区域经理和总部问同一个问题看到的行范围必须不同单轮 Text2SQL 只解决一句话翻译成 SQL上面四件事一件都不管。要撑住业务场景得把它放进 Agent 架构里。一、四层架构用户提问 → 意图路由 判定这是问数 / 问口径 / 改条件 / 闲聊 → 槽位与上下文 补齐指标、时间、维度、过滤条件处理指代 → 工具编排 选工具查指标字典 / 生成并执行 SQL / 画图 → 答案生成 数字 口径说明 可展开的 SQL缺信息就反问四层各司其职任何一层可独立替换。逐层给代码。二、意图路由先分清他在问什么把问口径改条件混进问数流程是系统答非所问的主要来源。先做分类ROUTER 判断用户意图只输出一个标签 - QUERY_METRIC 要一个数字含追问、换维度、换时间 - ASK_DEFINITION 问某个指标怎么算、口径是什么 - REVISE 修改上一轮的条件 - CHITCHAT 与取数无关 对话历史 {history} 用户{utterance} ASK_DEFINITION直接走字典检索不生成 SQLREVISE走增量改槽位不重新解析整句。分类对了后面才简单。三、槽位与上下文让「那华东呢」能成立每一轮把解析结果沉淀成一个槽位对象下一轮先继承再覆盖class Slots(dict): 指标 / 时间 / 维度 / 过滤条件 / 口径版本 def merge(self, new: dict): for k, v in new.items(): if v __INHERIT__: # 本轮没说沿用上一轮 continue if v __DROP__: # 明确去掉某个条件 self.pop(k, None) continue self[k] v return self def missing(self) - list: return [k for k in (metric, time_range) if k not in self]指代消解放在解析之前做把「那华东呢」先改写成一个完整问句再交给解析。改写比在 Prompt 里塞历史会话更稳也能避免上下文越滚越长。上下文只保留最近三轮的槽位快照不要塞整段对话原文——长上下文既贵又干扰模型判断。四、工具编排三个工具够用TOOLS [ {name: lookup_metric, desc: 查指标字典返回指标名、计算表达式、过滤条件、口径版本, args: {metric_name: string}}, {name: run_sql, desc: 执行只读 SQL返回结果集最多 500 行, args: {sql: string}}, {name: render_chart, desc: 把结果集渲染成折线图或柱状图, args: {rows: array, chart_type: string}}, ]主循环用 ReAct但限制轮次建议 3 轮超了就转人工def agent_loop(utterance, slots, ctx, max_steps3): intent classify(ROUTER, ctx.history, utterance) if intent ASK_DEFINITION: return lookup_metric(extract_metric(utterance)) for step in range(max_steps): slots slots.merge(parse(utterance, slots)) if slots.missing(): # 缺关键信息反问而不是猜 return ask_back(slots.missing()) metric lookup_metric(slots[metric]) # 先查口径再生成 SQL sql generate_sql(slots, metric) rows run_with_repair(sql, ctx.conn) # 校验与自纠错见上一篇 if rows is not None: return answer(rows, metric, sql) # 数字 口径 SQL 一起给 return handoff_to_human(utterance, slots)lookup_metric必须在generate_sql之前调用——先确认口径再谈取数。顺序反了就会出现SQL 能跑、数字是错的这种最难发现的问题。五、权限下推过滤条件不能靠模型自觉行级权限绝对不能指望模型在 Prompt 里记得加。正确做法是在 SQL 生成之后、执行之前由中间层强制注入def apply_rls(sql: str, user: User) - str: scope user.data_scope # 如 {region: [华东], bu: [零售]} return inject_filters(sql, scope, allowlistUSER_TABLES[user.role])配合只读数据库账号、单次查询超时、返回结果行数上限、全量审计日志。这四件是安全底线任何一件缺失都不该上线。六、人机边界三种情况必须交回给人槽位补齐后仍缺关键信息、自纠错超过两轮、结果为空但问句不该为空。宁可反问一句不要给一个编出来的数字——业务一旦信错一次这套系统就没人再用了。七、两个经验回归集要按对话建不按单句建。真实场景里 60% 以上是追问只测单句会高估能力。建议建 30 组多轮对话每组三到五轮覆盖换维度、换时间、改口径、反问澄清。成本可控。路由和改写用小模型SQL 生成和答案生成用大模型整体开销能降一半以上准确率基本不掉。把意图路由、槽位继承、工具编排、权限下推串起来之后业务同事在聊天框里问一句就能拿到带口径说明的数字。广瞬科技的 AI 客户经营系统中「对话问数」即按这套架构实现支持多轮追问与按角色的数据范围控制。本文方案由广瞬科技数据智能团队整理官网 www.ginstant.cn。