做数据分析 Agent 的团队越来越多但一个很普遍的困惑是明明模型已经很强Text-to-SQL 的效果却总是忽好忽坏。很多团队把问题归结为“模型不行”于是不断换更大参数的模型结果发现提升有限。真正被忽略的往往是一个更基础的工程问题我们把什么样的上下文喂给了模型。最近在 Hacker News 的 Show HN 板块看到一个项目标题很有代表性Assemble context for analytics agents from schemas, SQL and docs。它想表达的核心思想很朴素却直击痛点——analytics agents 能不能用不取决于你堆了多少复杂提示词而取决于你是否能把数据库 schema、历史 SQL 和业务文档这三类信息系统性地组装成模型真正能理解的高质量上下文。这篇文章不打算替某个项目做宣传而是把这个思路拆开讲透为什么上下文组装会成为 analytics agents 的胜负手如果我们要在自己的项目里实现类似能力应该怎么设计采集、解析、组装和注入流程。我会直接给出可复制的 Python 示例覆盖 schema 采集、SQL 文件解析、文档解析和上下文模板渲染最后再聊一聊常见问题和工程化建议。读完你至少能搭出一个最小可用的 context builder而不是继续在提示词里打补丁。1. 为什么 analytics agents 真正卡在上下文上1.1 表面上是模型问题实际是上下文问题过去一年关于自然语言转 SQL 的讨论热度一直很高相关搜索里也充斥着“agent 实现把自然语言转换成 SQL”“sql 语句大全及用法”“慢 sql 优化”这类内容。很多团队抱着“大模型能写 SQL”的期待开始做数据分析助手但一上线就发现模型确实会写 SQL但经常写错表、搞错字段、忽略业务口径甚至生成的 SQL 在生产库上跑出全表扫描。问题通常不在模型的理解能力而在上下文。数据库 schema 本身是非常稀疏的表名可能是 order_info、字段可能是 cust_id如果没有注释模型根本无法知道 cust_id 是客户 ID、order_status 的 0 和 1 分别代表什么。更棘手的是同一个指标在不同部门的文档里可能有完全不同的口径比如“销售额”到底是订单金额、实付金额还是确认收货后的金额。这些信息不可能靠模型“猜”出来只能靠我们主动喂给它。1.2 三个典型失败场景我见过很多项目在初期踩同样的坑归纳起来是三类场景。第一类是 schema 直接裸奔。开发同学把 information_schema 里的表结构全部导出拼成一段超长文本塞进提示词结果 token 严重超限模型抓到一些表就开始自由发挥。第二类是历史 SQL 没有利用。团队里明明积累了成百上千条经过业务验证的查询语句但完全没有沉淀成“经验”每次 agent 都是从零开始推理自然容易犯错。第三类是业务文档断层。指标定义散落在各个文档平台里格式不统一有的叫“GMV”有的叫“成交总额”agent 根本不知道它们是同一个东西。这些问题的本质是我们想让 agent 替代数据分析师的一部分工作却没有把数据分析师日常依赖的“公司知识”交接给它。数据分析师会看数据字典、翻历史报表、问业务方口径agent 也需要一套类似的输入这就是 context assembly 要做的事。1.3 一个明确的判断所以这篇文章的核心判断是analytics agents 的工程瓶颈已经从模型推理能力转移到了上下文组装能力。谁先把 schema、SQL、docs 这三个信息源整合成稳定、可更新、可度量的上下文谁的 agent 就能率先在真实业务场景里跑通。2. 核心概念什么是 Context Assembly2.1 定义Context Assembly直译是“上下文组装”。它不是一个新的算法而是一套工程流程把多个异构数据源数据库结构、历史查询语句、业务文档经过采集、解析、过滤、结构化之后编排成 LLM 可以直接消费的上下文。这个流程的输出通常是一段结构良好的 Markdown 或 JSON最终被注入到 system prompt 里或者落到向量数据库中供检索。关键点在于“组装”而不是“拼接”。简单的字符串拼接只会堆出一堆噪声真正的上下文组装需要做取舍、去重、分层还要控制 token 预算。2.2 三类输入源Schema、SQL、Docs项目标题里把信息源列得很清楚schemas、SQL 和 docs。Schema 是数据库的骨架包括表名、字段名、字段类型、是否为空、主外键关系、索引信息。它回答的是“数据库里有什么”。但仅有 schema模型只看到了结构没看到语义。SQL 是团队沉淀下来的行为数据。历史 SQL 里藏着真实业务怎么查数据哪些表经常 join 在一起常用的过滤条件是什么时间字段习惯用什么函数截断。这些信息不会写在 schema 里却是模型写对 SQL 的关键先验。Docs 是业务语义的来源包括数据字典、指标口径说明、埋点文档、数仓规范。它回答的是“这个数字到底是什么意思”。这恰恰是纯 schema 永远无法覆盖的部分。2.3 它和 RAG 有什么区别很多人会问这和 RAG检索增强生成难道不是一回事吗其实不一样。RAG 的核心是先检索后生成适合知识分散、无法全部塞进上下文的大规模文档场景。而 Context Assembly 更接近“编译打包”它面向的是元数据规模相对可控、但结构要求极高的场景。数据库 schema 可能只有几百张表docs 可能只有几十个指标口径我们可以把它们解析成结构化对象再通过模板生成一份确定性的上下文。这份上下文是全量的、预先编译好的而不是在每次请求时临时检索。两者可以结合Context Assembly 负责构建高质量的“基础上下文包”RAG 负责在会话中动态检索更细的业务文档。但第一步还是先把基础包做好。2.4 原始 schema 与组装后上下文的对比维度直接导出 schemaContext Assembly 后表结构全部表堆在一起过滤系统表按主题归类字段语义只有字段名和类型补充注释、枚举值、示例表关系靠模型自行推断显式标注外键和常用 JOIN 路径业务口径完全没有补齐指标定义和计算逻辑历史经验没有融入高频表、常用函数、过滤习惯Token 控制容易超长按主题裁剪分层注入这个对比说明一件事组装的核心价值不是“多”而是“准”。让模型在更少但更关键的信息上做推理效果通常好于一次性塞入大量原始数据。3. 整体架构与数据流要把 Context Assembly 落地可以按四层来设计采集层、解析层、组装层、注入层。每一层解耦后续替换数据源或调整输出格式都会更容易。3.1 采集层采集层负责从不同来源拉取原始数据。从数据库读取 information_schema 或通过 SQLAlchemy inspect 获取表结构、字段、外键。从代码仓库扫描历史 SQL 文件常见目录包括sql/、queries/、analytics/。从文档目录扫描 Markdown 或数据字典文件后续也可以扩展接入 Wiki API。采集层只做一件事把原始数据拿回来不做太多加工。因为不同数据库方言差异很大采集层最好对上层隐藏差异统一输出结构化的中间对象。3.2 解析层解析层负责把原始数据变成结构化对象。Schema 解析把表、字段、类型、注释、外键映射为TableSchema对象。SQL 解析从历史 SQL 中提取表引用、JOIN 关系、常用函数、过滤模式。轻量场景可以用正则复杂场景建议集成sqlparse。Docs 解析把 Markdown 标题映射为文档章节提取指标名称和定义甚至可以按约定格式解析出目录树。解析层是上下文质量的分水岭。如果 schema 清洗不干净上下文里全是pg_开头的系统表如果 SQL 解析只做简单正则容易漏掉 CTE 里的临时表引用。所以这一层值得花时间打磨。3.3 组装层组装层把结构化对象按照模板拼接成最终上下文。这里通常使用 Jinja2 或类似的模板引擎。组装时要考虑三件事内容组织顺序一般先总览、再表结构、再业务口径、长度控制超出 token 预算时裁剪低频表、版本信息在上下文里标注生成时间和数据来源方便排查。3.4 注入层注入层决定上下文怎么传递给模型。最简单的方式是拼进 system prompt如果需要支持多业务域可以按“公共上下文 业务上下文 会话检索片段”三层注入。这一层不用做太复杂只要保证每次请求使用的上下文是可追踪的即可。4. 环境准备与前置条件在写代码之前先准备一个最小可运行的环境。本文示例以 PostgreSQL 作为演示数据库MySQL 的思路完全一致只是 information_schema 的字段名略有差异。4.1 运行环境Python 3.9 及以上建议 3.11。可访问的 PostgreSQL 实例并准备一个只读账号。一个存放历史 SQL 的目录例如sql/。一个存放业务文档的目录例如docs/。这里强调一个安全前提连接数据库的账号必须是只读权限绝不能给 agent 直接写库的权限。生产环境建议单独建一个专用账号只授予SELECT权限。4.2 安装依赖建议先创建一个虚拟环境再安装依赖。python -m venv .venv source .venv/bin/activate pip install sqlalchemy psycopg2-binary pydantic jinja2 pyyaml各依赖的作用如下依赖作用sqlalchemy统一数据库连接与 schema 检查psycopg2-binaryPostgreSQL 驱动pydantic定义结构化数据模型便于解析和校验jinja2渲染上下文模板pyyaml解析配置文件版本以你项目实际安装为准本文示例使用的是 SQLAlchemy 2.x 风格的 API1.4 版本同样兼容 inspect 接口。4.3 目录结构建议按下面的方式组织项目analytics-agent-context/ ├── build_context/ │ ├── __init__.py │ ├── schema_collector.py │ ├── sql_collector.py │ ├── docs_collector.py │ └── assembler.py ├── templates/ │ └── agent_context.md.j2 ├── sql/ │ ├── daily_revenue.sql │ └── user_retention.sql ├── docs/ │ ├── metrics.md │ └── data_dictionary.md ├── build_context.py └── config.yaml这个结构清晰区分了构建逻辑、模板、输入源和入口脚本后续做增量更新或接入 CI 都比较方便。4.4 数据库连接配置为了让代码不硬编码密码连接信息建议通过环境变量或配置文件传入。# config.yaml db_url: postgresqlpsycopg2://analytic_ro:${DB_PASSWORD}localhost:5432/analytics sql_dir: ./sql docs_dir: ./docs output: ./agent_context.md注意DB_PASSWORD这里只是示意实际读取时需要用环境变量替换建议先用 Python 的os.getenv读取不要直接把密码写进文件推到 Git 里。5. 从 Schema、SQL、Docs 构建上下文完整示例下面我们按前面设计的四层架构实现一个最小可用的 context builder。代码会拆成几个模块每个模块聚焦一类信息源。5.1 采集数据库 Schema第一个模块负责连接数据库并采集表结构。我们用 SQLAlchemy 的inspect接口这样可以屏蔽不同数据库方言的差异。# build_context/schema_collector.py from sqlalchemy import create_engine, inspect def collect_schema(engine) - list[dict]: 返回结构化表信息列表。 insp inspect(engine) tables [] for table_name in insp.get_table_names(): # 过滤系统表、临时表 if table_name.startswith((pg_, _, tmp, temp, bak)): continue columns [] for col in insp.get_columns(table_name): col_type str(col.get(type)) comment col.get(comment, ) or columns.append({ name: col[name], type: col_type, nullable: col.get(nullable, True), comment: comment, }) foreign_keys [] for fk in insp.get_foreign_keys(table_name): local_cols fk.get(constrained_columns) or [] ref_table fk.get(referred_table) or for local_col in local_cols: foreign_keys.append(f{local_col} - {ref_table}) tables.append({ name: table_name, columns: columns, foreign_keys: foreign_keys, }) return tables这段代码的核心作用是过滤掉pg_开头的系统表和临时表同时把字段类型、注释、外键关系整理成字典。实际项目中注释是最容易被忽略但最有价值的信息如果你的数据库字段本来就写了 comment一定要保留。5.2 解析历史 SQL 文件第二个模块扫描 SQL 文件目录提取对模型有用的模式。这里先用正则做轻量解析覆盖最常见的场景如果要处理非常复杂的 SQL建议在第二步接入sqlparse。# build_context/sql_collector.py import re from pathlib import Path TABLE_RE re.compile(r\b(?:from|join)\s([a-zA-Z_][\w.]*), re.IGNORECASE) FEATURES_RE { ILIKE 模糊查询: re.compile(r\bILIKE\b, re.IGNORECASE), DATE_TRUNC 时间聚合: re.compile(r\bDATE_TRUNC\b, re.IGNORECASE), 窗口函数: re.compile(r\bROW_NUMBER|RANK|DENSE_RANK|LAG|LEAD\b, re.IGNORECASE), WITH CTE: re.compile(r\bWITH\b[\s\w,]AS\s*\(, re.IGNORECASE), } def collect_sql_patterns(sql_dir: Path): 扫描 SQL 目录返回表引用频率和常用特征。 table_count {} features set() for sql_file in sorted(sql_dir.glob(*.sql)): content sql_file.read_text(encodingutf-8, errorsignore) for table in TABLE_RE.findall(content): # 去掉可能的多级 schema 前缀只保留表名 table_short table.split(.)[-1] table_count[table_short] table_count.get(table_short, 0) 1 for feature, pattern in FEATURES_RE.items(): if pattern.search(content): features.add(feature) return table_count, features这里统计了历史 SQL 中哪些表被高频引用、哪些写法在本项目里很常见。高频表的作用很大当模型面对几十张表时优先考虑高频引用的表生成的 SQL 往往更符合团队实际业务。5.3 解析业务文档第三个模块负责解析 Markdown 文档。假设团队的指标口径都写在 Markdown 文件里我们用标题层级切分文档生成“标题 正文”的结构化条目。# build_context/docs_collector.py import re from pathlib import Path def extract_sections(md_text: str, max_level: int 2): 按 Markdown 标题切分文档提取标题正文对。 sections [] current_title 未分类 current_lines [] for line in md_text.splitlines(): heading re.match(r^(#{1,%d})\s(.*)$ % max_level, line) if heading: if current_lines: sections.append((current_title, \n.join(current_lines).strip())) current_title heading.group(2).strip() current_lines [] else: current_lines.append(line) if current_lines: sections.append((current_title, \n.join(current_lines).strip())) return sections def collect_docs(docs_dir: Path) - list[tuple[str, str]]: 读取 docs 目录下所有 Markdown 文件提取章节。 all_sections [] for md_file in sorted(docs_dir.glob(*.md)): text md_file.read_text(encodingutf-8) sections extract_sections(text) all_sections.extend(sections) return all_sections如果团队文档格式不统一这个模块会是最需要扩展的部分。比如有的团队用 Excel 维护数据字典那就需要接入pandas读取有的团队用 Wiki那就需要调用 API 拉取页面内容。5.4 用模板组装最终上下文信息源都解析成结构化对象之后最后一步是用 Jinja2 渲染一份上下文文档。先定义模板{# templates/agent_context.md.j2 #} # 数据分析 Agent 上下文 生成时间{{ generated_at }} 数据来源schema / sql / docs ## 数据库表结构 {% for table in tables %} ### 表{{ table.name }} {% for col in table.columns %} - {{ col.name }} {{ col.type }} nullable{{ col.nullable }} {% if col.comment %} // {{ col.comment }}{% endif %} {% endfor %} {% if table.foreign_keys %} **外键关系** {% for fk in table.foreign_keys %} - {{ fk }} {% endfor %} {% endif %} {% endfor %} ## 历史 SQL 高频表与特征 {% for table, count in table_counts %} - {{ table }}引用 {{ count }} 次 {% endfor %} ## 常用 SQL 特征 {% for feature in sql_features %} - {{ feature }} {% endfor %} ## 指标口径与业务文档 {% for title, body in doc_sections %} ### {{ title }} {{ body }} {% endfor %}然后写组装逻辑# build_context/assembler.py from datetime import datetime from pathlib import Path from jinja2 import Environment, FileSystemLoader from sqlalchemy import create_engine from .docs_collector import collect_docs from .schema_collector import collect_schema from .sql_collector import collect_sql_patterns def build_context(db_url: str, sql_dir: Path, docs_dir: Path) - str: engine create_engine(db_url) tables collect_schema(engine) table_counts, sql_features collect_sql_patterns(sql_dir) doc_sections collect_docs(docs_dir) env Environment(loaderFileSystemLoader(templates)) template env.get_template(agent_context.md.j2) return template.render( generated_atdatetime.now().strftime(%Y-%m-%d %H:%M:%S), tablestables, table_countssorted(table_counts.items(), keylambda x: x[1], reverseTrue), sql_featuressorted(sql_features), doc_sectionsdoc_sections, )模板负责呈现解析模块负责取数组装逻辑只做编排。这样后续想换输出格式比如输出 JSON 给其他 Agent 框架用只需要改模板或增加一个 renderer。5.5 一键构建入口最后写一个命令行入口把整个流程串起来。# build_context.py import argparse import os from pathlib import Path from build_context.assembler import build_context def main(): parser argparse.ArgumentParser(descriptionBuild context for analytics agents) parser.add_argument(--db-url, defaultos.getenv(ANALYTICS_DB_URL), helpSQLAlchemy 数据库连接串) parser.add_argument(--sql-dir, typePath, requiredTrue) parser.add_argument(--docs-dir, typePath, requiredTrue) parser.add_argument(--output, typePath, defaultPath(agent_context.md)) args parser.parse_args() if not args.db_url: raise SystemExit(请设置 ANALYTICS_DB_URL 环境变量或传入 --db-url) context build_context(args.db_url, args.sql_dir, args.docs_dir) args.output.write_text(context, encodingutf-8) print(fcontext 已写入: {args.output}) print(fcontext 长度: {len(context)} 字符) if __name__ __main__: main()这个入口会在运行后打印上下文字符数方便你快速判断 token 规模。后续可以把 build_context.py 接入 CI每天定时重建上下文文件。6. 运行结果与效果验证6.1 运行命令假设环境变量已经设置好执行export ANALYTICS_DB_URLpostgresqlpsycopg2://analytic_ro:your_passwordlocalhost:5432/analytics python build_context.py --sql-dir ./sql --docs-dir ./docs --output ./agent_context.md如果一切正常会看到类似输出context 已写入: ./agent_context.md context 长度: 8642 字符6.2 预期输出agent_context.md 应该包含四大部分数据库表结构、历史 SQL 高频表、常用 SQL 特征、指标口径与业务文档。表结构部分应该已经滤掉了系统表高频表列表应该和业务直觉一致比如orders、users、payments出现在前面。6.3 如何评估上下文是否合格上下文不是生成完就算成功建议用下面几个检查点验证核心业务表是否都出现了外键关系是否完整文档中的关键指标是否都有对应章节高频表的统计结果是否符合经验判断token 长度是否在模型上下文预算的可控范围内如果答案都是肯定的再拿几个典型问题去问 agent比如“最近 7 天各渠道销售额趋势”观察它是否会用对表和字段。6.4 失败时的第一排查点如果生成的上下文里很多表缺失第一步先检查数据库账号权限。很多只读账号默认只能看到自己 schema 下的表需要在连接串中指定options-csearch_path%3Dpublic或给账号授予对应 schema 的 USAGE 权限。如果 SQL 解析结果为空先确认 SQL 文件的扩展名是不是.sql并且目录路径是否正确。7. 常见问题与排查思路问题现象可能原因排查方式解决方案Schema 采集结果为空数据库账号没有权限读取 information_schema用 psql 命令行手工查询information_schema.tables验证给账号授予对应 schema 的 USAGE/SELECT 权限SQL 解析漏掉 CTE 表历史 SQL 大量使用 WITH 语句正则只匹配 from/join抽查几个复杂 SQL 文件观察解析结果接入 sqlparse 或自定义 CTE 解析逻辑上下文 Token 超限系统表未过滤干净或文档全文被塞入打印各模块生成的字符数定位超限来源加过滤规则docs 只保留摘要或指定章节指标口径文档无法切分Markdown 标题不规范比如标题和文字不换行打开原文检查标题格式统一文档规范或针对现有格式写专用解析器Agent 仍然写错表名上下文里有同名表或字段命名不规范查看高频表统计确认重点表是否被突出在高频表章节补充“推荐使用的表”提示敏感字段被带入上下文采集逻辑未过滤字段级敏感信息检查 schema 输出中是否包含手机号、身份证字段在字段解析中加入脱敏黑名单过滤连接数据库超时网络不通或数据库连接池参数未配置用 psql 手工连接测试在 create_engine 中增加 connect_args 超时配置8. 最佳实践与工程建议8.1 权限与安全边界Context Assembly 涉及数据库元数据和业务文档必须有清晰的安全边界。第一数据库账号坚持最小权限原则只给 SELECT不给写权限第二字段级脱敏要在采集层完成比如把phone、id_card、email这类字段从上下文中剔除或者只输出哈希后的示例值第三如果 agent 最终要直接执行 SQL必须在应用层做查询白名单、超时控制和返回行数限制避免生成一条超大查询拖垮数据库。另外要特别提醒用户输入永远不能直接拼接进 SQL。agent 生成的 SQL 也要先经过只读账号、超时限制和审计才能送到真实库执行。这与我们前面强调的只读账号是一套组合拳。8.2 上下文分层与裁剪不要把所有上下文一股脑塞给模型。更合理的做法是分层全局公共层表结构概览、指标口径、常用 JOIN 路径每次请求都带上。业务域层按用户提问的主题动态注入对应业务域的详细表结构。会话层RAG 检索到的临时文档片段只在当前会话使用。这样做的目的是控制 prompt 长度同时确保高优先级信息始终存在。token 预算有限时优先裁剪低频表和长文档而不是裁掉高频表和外键关系。8.3 增量更新与缓存数据库 schema 不是每天变历史 SQL 也相对稳定没必要每次请求都重新构建上下文。更合适的做法是做成离线任务表结构发生变化时触发重建或者在 CI 里每天定时重建一次并把生成结果缓存起来。重建过程加上生成时间和schema 版本号方便出问题时对齐上下文和实际库结构是否一致。8.4 质量评估上下文改得好不好不能光靠感觉。建议准备一组 benchmark 查询比如 20 到 50 个典型业务问题覆盖单表查询、多表 JOIN 和复杂时间聚合。每次调整上下文组装逻辑后跑一遍 benchmark记录“表选对率”“口径正确率”“SQL 可执行率”这几个指标。长期维护这套评估集比口头争论“换大模型还是小模型”有用得多。8.5 团队协作规范上下文组装要发挥作用必须有团队配合。SQL 仓库化是第一步所有分析查询都进 Git按目录组织避免散落在个人聊天记录里。文档也要规范指标口径统一用 Markdown 维护标题结构固定方便自动解析。这两件事做得越早context builder 的收益越大。9. 总结与后续学习方向回到开头那个问题analytics agents 的效果为什么忽好忽坏很大一部分答案不在模型参数里而在上下文组装里。通过 schema、SQL、docs 三条信息源我们可以把数据库结构、团队查询经验和业务口径系统性地编译成模型真正能用的上下文。这不是什么神秘技术而是一套工程规范。这篇文章里实现的最小示例已经可以让你跑通从采集到渲染的完整链路。下一步值得继续深入的方向有三个一是把 SQL 解析从正则升级为 sqlparse 或 AST 级别解析提升复杂语句的识别能力二是把输出格式从 Markdown 扩展为 JSON方便接入 LangChain 或其他 Agent 框架三是建立团队自己的 benchmark 评估集让上下文优化变成一个可以持续迭代的工程指标。如果你正在做数据分析 Agent建议把 context assembly 当成一等公民来设计而不是临上线前再补。数据库 schema 历史 SQL 和业务文档这“三件套”整理得好模型的真实水平才有机会体现出来。 数据分析Agent的胜负手：Context Assembly组装Schema、SQL与Docs
做数据分析 Agent 的团队越来越多但一个很普遍的困惑是明明模型已经很强Text-to-SQL 的效果却总是忽好忽坏。很多团队把问题归结为“模型不行”于是不断换更大参数的模型结果发现提升有限。真正被忽略的往往是一个更基础的工程问题我们把什么样的上下文喂给了模型。最近在 Hacker News 的 Show HN 板块看到一个项目标题很有代表性Assemble context for analytics agents from schemas, SQL and docs。它想表达的核心思想很朴素却直击痛点——analytics agents 能不能用不取决于你堆了多少复杂提示词而取决于你是否能把数据库 schema、历史 SQL 和业务文档这三类信息系统性地组装成模型真正能理解的高质量上下文。这篇文章不打算替某个项目做宣传而是把这个思路拆开讲透为什么上下文组装会成为 analytics agents 的胜负手如果我们要在自己的项目里实现类似能力应该怎么设计采集、解析、组装和注入流程。我会直接给出可复制的 Python 示例覆盖 schema 采集、SQL 文件解析、文档解析和上下文模板渲染最后再聊一聊常见问题和工程化建议。读完你至少能搭出一个最小可用的 context builder而不是继续在提示词里打补丁。1. 为什么 analytics agents 真正卡在上下文上1.1 表面上是模型问题实际是上下文问题过去一年关于自然语言转 SQL 的讨论热度一直很高相关搜索里也充斥着“agent 实现把自然语言转换成 SQL”“sql 语句大全及用法”“慢 sql 优化”这类内容。很多团队抱着“大模型能写 SQL”的期待开始做数据分析助手但一上线就发现模型确实会写 SQL但经常写错表、搞错字段、忽略业务口径甚至生成的 SQL 在生产库上跑出全表扫描。问题通常不在模型的理解能力而在上下文。数据库 schema 本身是非常稀疏的表名可能是 order_info、字段可能是 cust_id如果没有注释模型根本无法知道 cust_id 是客户 ID、order_status 的 0 和 1 分别代表什么。更棘手的是同一个指标在不同部门的文档里可能有完全不同的口径比如“销售额”到底是订单金额、实付金额还是确认收货后的金额。这些信息不可能靠模型“猜”出来只能靠我们主动喂给它。1.2 三个典型失败场景我见过很多项目在初期踩同样的坑归纳起来是三类场景。第一类是 schema 直接裸奔。开发同学把 information_schema 里的表结构全部导出拼成一段超长文本塞进提示词结果 token 严重超限模型抓到一些表就开始自由发挥。第二类是历史 SQL 没有利用。团队里明明积累了成百上千条经过业务验证的查询语句但完全没有沉淀成“经验”每次 agent 都是从零开始推理自然容易犯错。第三类是业务文档断层。指标定义散落在各个文档平台里格式不统一有的叫“GMV”有的叫“成交总额”agent 根本不知道它们是同一个东西。这些问题的本质是我们想让 agent 替代数据分析师的一部分工作却没有把数据分析师日常依赖的“公司知识”交接给它。数据分析师会看数据字典、翻历史报表、问业务方口径agent 也需要一套类似的输入这就是 context assembly 要做的事。1.3 一个明确的判断所以这篇文章的核心判断是analytics agents 的工程瓶颈已经从模型推理能力转移到了上下文组装能力。谁先把 schema、SQL、docs 这三个信息源整合成稳定、可更新、可度量的上下文谁的 agent 就能率先在真实业务场景里跑通。2. 核心概念什么是 Context Assembly2.1 定义Context Assembly直译是“上下文组装”。它不是一个新的算法而是一套工程流程把多个异构数据源数据库结构、历史查询语句、业务文档经过采集、解析、过滤、结构化之后编排成 LLM 可以直接消费的上下文。这个流程的输出通常是一段结构良好的 Markdown 或 JSON最终被注入到 system prompt 里或者落到向量数据库中供检索。关键点在于“组装”而不是“拼接”。简单的字符串拼接只会堆出一堆噪声真正的上下文组装需要做取舍、去重、分层还要控制 token 预算。2.2 三类输入源Schema、SQL、Docs项目标题里把信息源列得很清楚schemas、SQL 和 docs。Schema 是数据库的骨架包括表名、字段名、字段类型、是否为空、主外键关系、索引信息。它回答的是“数据库里有什么”。但仅有 schema模型只看到了结构没看到语义。SQL 是团队沉淀下来的行为数据。历史 SQL 里藏着真实业务怎么查数据哪些表经常 join 在一起常用的过滤条件是什么时间字段习惯用什么函数截断。这些信息不会写在 schema 里却是模型写对 SQL 的关键先验。Docs 是业务语义的来源包括数据字典、指标口径说明、埋点文档、数仓规范。它回答的是“这个数字到底是什么意思”。这恰恰是纯 schema 永远无法覆盖的部分。2.3 它和 RAG 有什么区别很多人会问这和 RAG检索增强生成难道不是一回事吗其实不一样。RAG 的核心是先检索后生成适合知识分散、无法全部塞进上下文的大规模文档场景。而 Context Assembly 更接近“编译打包”它面向的是元数据规模相对可控、但结构要求极高的场景。数据库 schema 可能只有几百张表docs 可能只有几十个指标口径我们可以把它们解析成结构化对象再通过模板生成一份确定性的上下文。这份上下文是全量的、预先编译好的而不是在每次请求时临时检索。两者可以结合Context Assembly 负责构建高质量的“基础上下文包”RAG 负责在会话中动态检索更细的业务文档。但第一步还是先把基础包做好。2.4 原始 schema 与组装后上下文的对比维度直接导出 schemaContext Assembly 后表结构全部表堆在一起过滤系统表按主题归类字段语义只有字段名和类型补充注释、枚举值、示例表关系靠模型自行推断显式标注外键和常用 JOIN 路径业务口径完全没有补齐指标定义和计算逻辑历史经验没有融入高频表、常用函数、过滤习惯Token 控制容易超长按主题裁剪分层注入这个对比说明一件事组装的核心价值不是“多”而是“准”。让模型在更少但更关键的信息上做推理效果通常好于一次性塞入大量原始数据。3. 整体架构与数据流要把 Context Assembly 落地可以按四层来设计采集层、解析层、组装层、注入层。每一层解耦后续替换数据源或调整输出格式都会更容易。3.1 采集层采集层负责从不同来源拉取原始数据。从数据库读取 information_schema 或通过 SQLAlchemy inspect 获取表结构、字段、外键。从代码仓库扫描历史 SQL 文件常见目录包括sql/、queries/、analytics/。从文档目录扫描 Markdown 或数据字典文件后续也可以扩展接入 Wiki API。采集层只做一件事把原始数据拿回来不做太多加工。因为不同数据库方言差异很大采集层最好对上层隐藏差异统一输出结构化的中间对象。3.2 解析层解析层负责把原始数据变成结构化对象。Schema 解析把表、字段、类型、注释、外键映射为TableSchema对象。SQL 解析从历史 SQL 中提取表引用、JOIN 关系、常用函数、过滤模式。轻量场景可以用正则复杂场景建议集成sqlparse。Docs 解析把 Markdown 标题映射为文档章节提取指标名称和定义甚至可以按约定格式解析出目录树。解析层是上下文质量的分水岭。如果 schema 清洗不干净上下文里全是pg_开头的系统表如果 SQL 解析只做简单正则容易漏掉 CTE 里的临时表引用。所以这一层值得花时间打磨。3.3 组装层组装层把结构化对象按照模板拼接成最终上下文。这里通常使用 Jinja2 或类似的模板引擎。组装时要考虑三件事内容组织顺序一般先总览、再表结构、再业务口径、长度控制超出 token 预算时裁剪低频表、版本信息在上下文里标注生成时间和数据来源方便排查。3.4 注入层注入层决定上下文怎么传递给模型。最简单的方式是拼进 system prompt如果需要支持多业务域可以按“公共上下文 业务上下文 会话检索片段”三层注入。这一层不用做太复杂只要保证每次请求使用的上下文是可追踪的即可。4. 环境准备与前置条件在写代码之前先准备一个最小可运行的环境。本文示例以 PostgreSQL 作为演示数据库MySQL 的思路完全一致只是 information_schema 的字段名略有差异。4.1 运行环境Python 3.9 及以上建议 3.11。可访问的 PostgreSQL 实例并准备一个只读账号。一个存放历史 SQL 的目录例如sql/。一个存放业务文档的目录例如docs/。这里强调一个安全前提连接数据库的账号必须是只读权限绝不能给 agent 直接写库的权限。生产环境建议单独建一个专用账号只授予SELECT权限。4.2 安装依赖建议先创建一个虚拟环境再安装依赖。python -m venv .venv source .venv/bin/activate pip install sqlalchemy psycopg2-binary pydantic jinja2 pyyaml各依赖的作用如下依赖作用sqlalchemy统一数据库连接与 schema 检查psycopg2-binaryPostgreSQL 驱动pydantic定义结构化数据模型便于解析和校验jinja2渲染上下文模板pyyaml解析配置文件版本以你项目实际安装为准本文示例使用的是 SQLAlchemy 2.x 风格的 API1.4 版本同样兼容 inspect 接口。4.3 目录结构建议按下面的方式组织项目analytics-agent-context/ ├── build_context/ │ ├── __init__.py │ ├── schema_collector.py │ ├── sql_collector.py │ ├── docs_collector.py │ └── assembler.py ├── templates/ │ └── agent_context.md.j2 ├── sql/ │ ├── daily_revenue.sql │ └── user_retention.sql ├── docs/ │ ├── metrics.md │ └── data_dictionary.md ├── build_context.py └── config.yaml这个结构清晰区分了构建逻辑、模板、输入源和入口脚本后续做增量更新或接入 CI 都比较方便。4.4 数据库连接配置为了让代码不硬编码密码连接信息建议通过环境变量或配置文件传入。# config.yaml db_url: postgresqlpsycopg2://analytic_ro:${DB_PASSWORD}localhost:5432/analytics sql_dir: ./sql docs_dir: ./docs output: ./agent_context.md注意DB_PASSWORD这里只是示意实际读取时需要用环境变量替换建议先用 Python 的os.getenv读取不要直接把密码写进文件推到 Git 里。5. 从 Schema、SQL、Docs 构建上下文完整示例下面我们按前面设计的四层架构实现一个最小可用的 context builder。代码会拆成几个模块每个模块聚焦一类信息源。5.1 采集数据库 Schema第一个模块负责连接数据库并采集表结构。我们用 SQLAlchemy 的inspect接口这样可以屏蔽不同数据库方言的差异。# build_context/schema_collector.py from sqlalchemy import create_engine, inspect def collect_schema(engine) - list[dict]: 返回结构化表信息列表。 insp inspect(engine) tables [] for table_name in insp.get_table_names(): # 过滤系统表、临时表 if table_name.startswith((pg_, _, tmp, temp, bak)): continue columns [] for col in insp.get_columns(table_name): col_type str(col.get(type)) comment col.get(comment, ) or columns.append({ name: col[name], type: col_type, nullable: col.get(nullable, True), comment: comment, }) foreign_keys [] for fk in insp.get_foreign_keys(table_name): local_cols fk.get(constrained_columns) or [] ref_table fk.get(referred_table) or for local_col in local_cols: foreign_keys.append(f{local_col} - {ref_table}) tables.append({ name: table_name, columns: columns, foreign_keys: foreign_keys, }) return tables这段代码的核心作用是过滤掉pg_开头的系统表和临时表同时把字段类型、注释、外键关系整理成字典。实际项目中注释是最容易被忽略但最有价值的信息如果你的数据库字段本来就写了 comment一定要保留。5.2 解析历史 SQL 文件第二个模块扫描 SQL 文件目录提取对模型有用的模式。这里先用正则做轻量解析覆盖最常见的场景如果要处理非常复杂的 SQL建议在第二步接入sqlparse。# build_context/sql_collector.py import re from pathlib import Path TABLE_RE re.compile(r\b(?:from|join)\s([a-zA-Z_][\w.]*), re.IGNORECASE) FEATURES_RE { ILIKE 模糊查询: re.compile(r\bILIKE\b, re.IGNORECASE), DATE_TRUNC 时间聚合: re.compile(r\bDATE_TRUNC\b, re.IGNORECASE), 窗口函数: re.compile(r\bROW_NUMBER|RANK|DENSE_RANK|LAG|LEAD\b, re.IGNORECASE), WITH CTE: re.compile(r\bWITH\b[\s\w,]AS\s*\(, re.IGNORECASE), } def collect_sql_patterns(sql_dir: Path): 扫描 SQL 目录返回表引用频率和常用特征。 table_count {} features set() for sql_file in sorted(sql_dir.glob(*.sql)): content sql_file.read_text(encodingutf-8, errorsignore) for table in TABLE_RE.findall(content): # 去掉可能的多级 schema 前缀只保留表名 table_short table.split(.)[-1] table_count[table_short] table_count.get(table_short, 0) 1 for feature, pattern in FEATURES_RE.items(): if pattern.search(content): features.add(feature) return table_count, features这里统计了历史 SQL 中哪些表被高频引用、哪些写法在本项目里很常见。高频表的作用很大当模型面对几十张表时优先考虑高频引用的表生成的 SQL 往往更符合团队实际业务。5.3 解析业务文档第三个模块负责解析 Markdown 文档。假设团队的指标口径都写在 Markdown 文件里我们用标题层级切分文档生成“标题 正文”的结构化条目。# build_context/docs_collector.py import re from pathlib import Path def extract_sections(md_text: str, max_level: int 2): 按 Markdown 标题切分文档提取标题正文对。 sections [] current_title 未分类 current_lines [] for line in md_text.splitlines(): heading re.match(r^(#{1,%d})\s(.*)$ % max_level, line) if heading: if current_lines: sections.append((current_title, \n.join(current_lines).strip())) current_title heading.group(2).strip() current_lines [] else: current_lines.append(line) if current_lines: sections.append((current_title, \n.join(current_lines).strip())) return sections def collect_docs(docs_dir: Path) - list[tuple[str, str]]: 读取 docs 目录下所有 Markdown 文件提取章节。 all_sections [] for md_file in sorted(docs_dir.glob(*.md)): text md_file.read_text(encodingutf-8) sections extract_sections(text) all_sections.extend(sections) return all_sections如果团队文档格式不统一这个模块会是最需要扩展的部分。比如有的团队用 Excel 维护数据字典那就需要接入pandas读取有的团队用 Wiki那就需要调用 API 拉取页面内容。5.4 用模板组装最终上下文信息源都解析成结构化对象之后最后一步是用 Jinja2 渲染一份上下文文档。先定义模板{# templates/agent_context.md.j2 #} # 数据分析 Agent 上下文 生成时间{{ generated_at }} 数据来源schema / sql / docs ## 数据库表结构 {% for table in tables %} ### 表{{ table.name }} {% for col in table.columns %} - {{ col.name }} {{ col.type }} nullable{{ col.nullable }} {% if col.comment %} // {{ col.comment }}{% endif %} {% endfor %} {% if table.foreign_keys %} **外键关系** {% for fk in table.foreign_keys %} - {{ fk }} {% endfor %} {% endif %} {% endfor %} ## 历史 SQL 高频表与特征 {% for table, count in table_counts %} - {{ table }}引用 {{ count }} 次 {% endfor %} ## 常用 SQL 特征 {% for feature in sql_features %} - {{ feature }} {% endfor %} ## 指标口径与业务文档 {% for title, body in doc_sections %} ### {{ title }} {{ body }} {% endfor %}然后写组装逻辑# build_context/assembler.py from datetime import datetime from pathlib import Path from jinja2 import Environment, FileSystemLoader from sqlalchemy import create_engine from .docs_collector import collect_docs from .schema_collector import collect_schema from .sql_collector import collect_sql_patterns def build_context(db_url: str, sql_dir: Path, docs_dir: Path) - str: engine create_engine(db_url) tables collect_schema(engine) table_counts, sql_features collect_sql_patterns(sql_dir) doc_sections collect_docs(docs_dir) env Environment(loaderFileSystemLoader(templates)) template env.get_template(agent_context.md.j2) return template.render( generated_atdatetime.now().strftime(%Y-%m-%d %H:%M:%S), tablestables, table_countssorted(table_counts.items(), keylambda x: x[1], reverseTrue), sql_featuressorted(sql_features), doc_sectionsdoc_sections, )模板负责呈现解析模块负责取数组装逻辑只做编排。这样后续想换输出格式比如输出 JSON 给其他 Agent 框架用只需要改模板或增加一个 renderer。5.5 一键构建入口最后写一个命令行入口把整个流程串起来。# build_context.py import argparse import os from pathlib import Path from build_context.assembler import build_context def main(): parser argparse.ArgumentParser(descriptionBuild context for analytics agents) parser.add_argument(--db-url, defaultos.getenv(ANALYTICS_DB_URL), helpSQLAlchemy 数据库连接串) parser.add_argument(--sql-dir, typePath, requiredTrue) parser.add_argument(--docs-dir, typePath, requiredTrue) parser.add_argument(--output, typePath, defaultPath(agent_context.md)) args parser.parse_args() if not args.db_url: raise SystemExit(请设置 ANALYTICS_DB_URL 环境变量或传入 --db-url) context build_context(args.db_url, args.sql_dir, args.docs_dir) args.output.write_text(context, encodingutf-8) print(fcontext 已写入: {args.output}) print(fcontext 长度: {len(context)} 字符) if __name__ __main__: main()这个入口会在运行后打印上下文字符数方便你快速判断 token 规模。后续可以把 build_context.py 接入 CI每天定时重建上下文文件。6. 运行结果与效果验证6.1 运行命令假设环境变量已经设置好执行export ANALYTICS_DB_URLpostgresqlpsycopg2://analytic_ro:your_passwordlocalhost:5432/analytics python build_context.py --sql-dir ./sql --docs-dir ./docs --output ./agent_context.md如果一切正常会看到类似输出context 已写入: ./agent_context.md context 长度: 8642 字符6.2 预期输出agent_context.md 应该包含四大部分数据库表结构、历史 SQL 高频表、常用 SQL 特征、指标口径与业务文档。表结构部分应该已经滤掉了系统表高频表列表应该和业务直觉一致比如orders、users、payments出现在前面。6.3 如何评估上下文是否合格上下文不是生成完就算成功建议用下面几个检查点验证核心业务表是否都出现了外键关系是否完整文档中的关键指标是否都有对应章节高频表的统计结果是否符合经验判断token 长度是否在模型上下文预算的可控范围内如果答案都是肯定的再拿几个典型问题去问 agent比如“最近 7 天各渠道销售额趋势”观察它是否会用对表和字段。6.4 失败时的第一排查点如果生成的上下文里很多表缺失第一步先检查数据库账号权限。很多只读账号默认只能看到自己 schema 下的表需要在连接串中指定options-csearch_path%3Dpublic或给账号授予对应 schema 的 USAGE 权限。如果 SQL 解析结果为空先确认 SQL 文件的扩展名是不是.sql并且目录路径是否正确。7. 常见问题与排查思路问题现象可能原因排查方式解决方案Schema 采集结果为空数据库账号没有权限读取 information_schema用 psql 命令行手工查询information_schema.tables验证给账号授予对应 schema 的 USAGE/SELECT 权限SQL 解析漏掉 CTE 表历史 SQL 大量使用 WITH 语句正则只匹配 from/join抽查几个复杂 SQL 文件观察解析结果接入 sqlparse 或自定义 CTE 解析逻辑上下文 Token 超限系统表未过滤干净或文档全文被塞入打印各模块生成的字符数定位超限来源加过滤规则docs 只保留摘要或指定章节指标口径文档无法切分Markdown 标题不规范比如标题和文字不换行打开原文检查标题格式统一文档规范或针对现有格式写专用解析器Agent 仍然写错表名上下文里有同名表或字段命名不规范查看高频表统计确认重点表是否被突出在高频表章节补充“推荐使用的表”提示敏感字段被带入上下文采集逻辑未过滤字段级敏感信息检查 schema 输出中是否包含手机号、身份证字段在字段解析中加入脱敏黑名单过滤连接数据库超时网络不通或数据库连接池参数未配置用 psql 手工连接测试在 create_engine 中增加 connect_args 超时配置8. 最佳实践与工程建议8.1 权限与安全边界Context Assembly 涉及数据库元数据和业务文档必须有清晰的安全边界。第一数据库账号坚持最小权限原则只给 SELECT不给写权限第二字段级脱敏要在采集层完成比如把phone、id_card、email这类字段从上下文中剔除或者只输出哈希后的示例值第三如果 agent 最终要直接执行 SQL必须在应用层做查询白名单、超时控制和返回行数限制避免生成一条超大查询拖垮数据库。另外要特别提醒用户输入永远不能直接拼接进 SQL。agent 生成的 SQL 也要先经过只读账号、超时限制和审计才能送到真实库执行。这与我们前面强调的只读账号是一套组合拳。8.2 上下文分层与裁剪不要把所有上下文一股脑塞给模型。更合理的做法是分层全局公共层表结构概览、指标口径、常用 JOIN 路径每次请求都带上。业务域层按用户提问的主题动态注入对应业务域的详细表结构。会话层RAG 检索到的临时文档片段只在当前会话使用。这样做的目的是控制 prompt 长度同时确保高优先级信息始终存在。token 预算有限时优先裁剪低频表和长文档而不是裁掉高频表和外键关系。8.3 增量更新与缓存数据库 schema 不是每天变历史 SQL 也相对稳定没必要每次请求都重新构建上下文。更合适的做法是做成离线任务表结构发生变化时触发重建或者在 CI 里每天定时重建一次并把生成结果缓存起来。重建过程加上生成时间和schema 版本号方便出问题时对齐上下文和实际库结构是否一致。8.4 质量评估上下文改得好不好不能光靠感觉。建议准备一组 benchmark 查询比如 20 到 50 个典型业务问题覆盖单表查询、多表 JOIN 和复杂时间聚合。每次调整上下文组装逻辑后跑一遍 benchmark记录“表选对率”“口径正确率”“SQL 可执行率”这几个指标。长期维护这套评估集比口头争论“换大模型还是小模型”有用得多。8.5 团队协作规范上下文组装要发挥作用必须有团队配合。SQL 仓库化是第一步所有分析查询都进 Git按目录组织避免散落在个人聊天记录里。文档也要规范指标口径统一用 Markdown 维护标题结构固定方便自动解析。这两件事做得越早context builder 的收益越大。9. 总结与后续学习方向回到开头那个问题analytics agents 的效果为什么忽好忽坏很大一部分答案不在模型参数里而在上下文组装里。通过 schema、SQL、docs 三条信息源我们可以把数据库结构、团队查询经验和业务口径系统性地编译成模型真正能用的上下文。这不是什么神秘技术而是一套工程规范。这篇文章里实现的最小示例已经可以让你跑通从采集到渲染的完整链路。下一步值得继续深入的方向有三个一是把 SQL 解析从正则升级为 sqlparse 或 AST 级别解析提升复杂语句的识别能力二是把输出格式从 Markdown 扩展为 JSON方便接入 LangChain 或其他 Agent 框架三是建立团队自己的 benchmark 评估集让上下文优化变成一个可以持续迭代的工程指标。如果你正在做数据分析 Agent建议把 context assembly 当成一等公民来设计而不是临上线前再补。数据库 schema 历史 SQL 和业务文档这“三件套”整理得好模型的真实水平才有机会体现出来。