小模型也能做数据分析智能体：把 harness 收窄做小-育师

小模型也能做数据分析智能体：把 harness 收窄做小

原文：NVIDIA Technical Blog - 《Building Reliable Data Analytics Agents: Lessons from the KDD Cup》（https://developer.nvidia.com/blog/building-reliable-data-analytics-agents-lessons-from-the-kdd-cup/）

一、先看清难点：数据分析智能体难在哪

如果只把问题理解成"连上数据库、把自然语言转成 SQL、把结果念出来"，数据分析智能体看起来早就被解决了。但 KDD Cup 2026 的 Data Agents 赛道把真实难度摆了出来：

每个任务给的数据源是异构的：SQL 数据库、CSV、JSON 文件、散文文档、PDF，甚至还有简报视频；
问题用自然语言提出，但不是检索题——智能体要自己去看数据、挑工具、跨数据源推理，最后按固定格式产出答案文件；
题目里还埋了分析工作流中常见的陷阱。

更关键的是那条约束：比赛要求所有队伍使用一个小而固定的 LLM。模型不能换、不能堆算力，那么留给参赛队的优化面只剩一个——harness，也就是围绕模型搭的那层脚手架。

NVIDIA 的 KGMON 团队靠这个思路拿下第二名。他们自己的表述很克制：这不是通用配方，但有一条更普适的结论——可靠性往往不来自把模型放开，而来自把 harness 搭对。这篇文章就按他们的复盘，把可迁移的设计拆出来。

二、两个原则：约束动作空间，让每次尝试可检视

整个系统由两个原则塑形。

第一个原则是约束动作空间。查看数据的方式太多、调用工具的入口太多、写文件与纠错的分支太多，都会变成失败点。KGMON 的做法是：统一数据访问、只暴露一小撮工具、最终答案只允许走一条输出路径。

第二个原则是让每一次尝试都可检视。一次运行可能因为工具调用写错、join 写错、漏掉文档里的规则、答案格式不对而失败。所以执行痕迹（trace）、重复尝试和轨迹检查不是调试工具，而是系统的组成部分。

后面的九条设计，都是在回答同一个问题：怎么让一个能力有限的小模型，在一个窄而清楚的环境里稳定干完活。

三、九条可以照着做的设计

1. 把结构化数据统一到一个查询面

每个任务都可能同时出现 SQL 数据库、CSV 和 JSON 文件。KGMON 把 CSV 与 JSON 转成 SQLite 里的表，让智能体对所有结构化数据只有一个SQL 接口。

他们在自定义的持久化 Python 环境里内置了两个函数，用于探查和查询这个统一数据面：

schema()：查看表和列；
sql(query)：查询 context.db。

注意这两个函数是内置在环境里的，不是由 Python 或 SQLite 提供的。为什么要费这个劲？因为单接口显著减少了"路由失败"和浪费的轮次，把省下来的额度留给模型去推理数据本身。

落地建议：在智能体启动前就把结构化数据访问归一化。一个临时 SQLite 库、一层虚拟化查询接口，或者一个受治理的数仓接口，都能提供一个稳定、窄、有文档的查询面。

2. 开跑前先做一次 schema 侦察

KGMON 在主推理循环之前加了一个 schema 侦察步骤：系统预先检查表、列、可能的 join key、重名列、长得像的字段、单位、null 分布，以及行粒度问题。

侦察结果在任务开始时一次性交给智能体，省掉了一个"探索"轮次，更重要的是减少了三类高频错误：拿错列、漏 join、搞不清答案行到底代表什么。

落地建议：加一个只读的预检步骤，提前告诉智能体有哪些表、可能的 join、关键列、单位、可疑字段和已知歧义。这件事的价值往往比再加一个检索工具更高。

3. 工具集要小而"有主见"

KGMON 把智能体的工具限制在schema 探查、SQL 查询、文档查找、散文抽取、答案写入这几件事上，环境暴露的函数大致是：

检查表与列；
查询 context.db；
原子地写出最终答案文件；
从散文里取答案，或把散文抽成 SQL 表。

他们还加了中间件来修复畸形的工具调用，让一次坏调用不至于直接终结整次尝试。同时环境是有状态的 Python 环境，变量在工具调用之间保留，中间结果可以被复用。预定义的schema()、sql(query)、write_answer(df)这类函数减少了样板代码、语法错误和文件处理失误。

回报是间接但实在的：一次尝试更短、更少报错，就留下更多轮次去做额外运行、评估和答案集成。

落地建议：工具按工作流需求设计，把重复路径删掉。只留一条被批准的 SQL 执行方式和一条答案写入方式，就少了污染状态、产出非法结果的机会。

4. 把散文当一等输入，但别让原文进主上下文

分析类任务经常夹带 PDF、Markdown、政策文本、说明和报告，里面可能有阈值、规则、定义和类表格式的记录。

大文档会吃掉上下文窗口。KGMON 直接封禁了通过 Python 的 open() 或 .read() 读取整个文件的做法，改为提供按字符数或正则匹配来做受限预览与检索的工具。

找到相关段落后，智能体可以调用一个自定义工具，把文档片段交给另一次LLM 调用：温度设为 0、关闭推理，只返回答案或抽出的表。这样原始文档内容不会进入主智能体的上下文。

该工具有两种模式：

answer 模式：抽出规则、阈值或简短答案；
table 模式：把重复记录抽成一张 SQL 表。

把规则从散文里抽出来之后，智能体就能在 SQL 分析里应用它们，同时保持工作上下文聚焦。

原文也标了一个边界：表抽取适合"结构化信息被塞进文档"的比赛任务，生产系统可能只需要定向的散文查找，表抽取可以保持可选。

5. 视频先离线预处理

有些任务带简报视频。为了不在智能体循环里承担视频处理的开销，KGMON 先离线抽关键帧、转写音频，再把转写片段与关键帧对齐，最后把证据喂给智能体。

每个任务最多一个视频，通常是幻灯片式的约束或干扰值。把转写和关键帧对齐，做的是"把说出来的话和画面对上"这件事。

边界同样要看清：这种预处理适配的是比赛设置。视频量大的应用更适合做一个按需的视频检索/查看工具，思路和散文处理一致。

6. trace 是为"失败"准备的

每一次尝试都会记录：prompt、工具调用、SQL 语句、中间结果、错误、修复动作、文档查找和最终答案。

有了这些，一个专门的审查智能体就可以回看失败轨迹、给错误分类，并把反复出现的失败类型汇总出来，帮团队排优先级。

trace 能直接回答一个很实用的问题：这个错答案，究竟来自 schema 理解错、join 写错、漏了散文证据、输出格式不对，还是 prompt 规则太脆？

落地建议：把 trace 检查做进开发流程。一个子智能体或评估脚本就能把最近运行的失败归类，并给出 harness 的修改建议。

7. 多次尝试要有选择，也要算成本

KGMON 用重复尝试加答案选择来提升覆盖率和稳定性。他们有个细节值得记：聚合尝试时是按答案值分组，而不是按列名，对有争议的任务可以追加更多次运行。在排行榜按值精确计分的规则下，多次尝试能区分"稳定答案"和"一次性失误"。

代价也很直接：token、延迟和算力都会涨。生产环境应该把集成（ensembling）留给那些价值、不确定性或风险足以支撑成本的场景。

落地建议：先做单次运行加 trace 检查。等出现置信度低、多答案分歧或校验失败时，再判断是否值得追加尝试。

8. 改进循环需要一个闸门

在自主改进循环里，智能体用评估反馈去改 prompt、工具、后处理和评估逻辑。这些改动能提分，也能过拟合到基准上。KGMON 点出了几种具体风险：

把训练样例硬编码进 prompt；
指令越堆越多而且互相矛盾；
加进脆弱的后处理规则；
提升了某一个切分，却损害了泛化。

他们要的是"快速改进"，不是"背下基准的 harness"。所以改进循环里要求：留出集（held-out）、prompt 审计、trace 复查，以及人类批准之后才能把改动提升为正式配置。

9. 人放在正确的位置上

这个系统是人加规模化实验的组合。人负责定义任务需求、审计 trace、引导智能体早期行为、否掉脆弱的改动、决定哪些改进值得留在 harness 里。

值得注意的是它对人的定位：定向介入，而不是逐次工具调用都盯着。人管任务设计、评估标准、失败分析和拟复用的技能，智能体负责执行与探索。

四、一个最小的 harness 骨架

下面这段是自拟示意代码（非官方实现），把上面几条设计拼成一个可读的最小结构，帮你对照自己的项目看缺口在哪。

# 自拟示意：把"窄工具面 + 持久状态 + 结构化痕迹"落到一个骨架里classAnalyticsHarness:def__init__(self,ctx_db,trace_path):self.state={}# 持久状态：跨工具调用保留变量self.trace=[]# 每次尝试的执行痕迹self.ctx_db=ctx_db# 统一后的结构化查询面self.trace_path=trace_path# ---- 工具层：只暴露四件事 ----defschema(self):"""返回统一查询面的表与列，供智能体一次性拿到结构信息"""result=self._query("SELECT name, sql FROM sqlite_master")self._log("schema",result)returnresultdefsql(self,query):"""唯一的结构化查询入口；失败时返回结构化错误而不是抛栈"""try:rows=self._query(query)self._log("sql",{"query":query,"rows":len(rows)})returnrowsexceptExceptionase:self._log("sql_error",{"query":query,"err":str(e)})return{"error":str(e)}# 让模型自己看到错误并改写defwrite_answer(self,df):"""原子写出最终答案；只保留一条输出路径"""self._atomic_dump(df,self.answer_path)self._log("write_answer",{"shape":getattr(df,"shape",None)})defprose_helper(self,chunk,mode="answer"):"""把文档片段交给独立的低温 LLM 调用，原文不进主上下文"""assertmodein("answer","table")out=self._sub_llm(chunk,temperature=0,reasoning=False,mode=mode)self._log("prose_helper",{"mode":mode,"chars":len(chunk)})returnout# answer -> 文本；table -> 可写入 ctx_db 的结构# ---- 中间件层：畸形调用就地修复，不终止整次尝试 ----defcall(self,name,**kwargs):ifnamenotin("schema","sql","write_answer","prose_helper"):return{"error":f"unknown tool:{name}"}# 约束动作空间try:returngetattr(self,name)(**kwargs)exceptTypeErrorase:returnself._repair_and_retry(name,kwargs,e)defdump_trace(self):"""把痕迹交给审查智能体做失败分类"""returnself._write_json(self.trace_path,self.trace)

调用链可以这样理解：智能体先在任务开始时调一次 schema 拿到结构信息（第 2 条设计），之后所有结构化查询都走 sql 这一条路（第 1、3 条设计）；遇到文档时不对原文直接 open，而是先做定向预览或正则搜索，再交给 prose_helper（第 4 条设计）；最终答案只通过 write_answer 原子落盘，避免半成品文件；整个过程的每一步都进 trace，交给独立的审查智能体分类失败（第 6 条设计）。