这阵子在折腾基于大模型的分析型代理Analytics Agent时我发现一个规律最难的不是写模型调用代码也不是调提示词模板而是给代理准备好一份“高质量上下文”。代理能不能从自然语言问题正确生成 SQL很大程度上取决于它有没有看明白表结构、业务口径和团队已有的查询习惯。恰好最近看到“Show HN: Assemble context for analytics agents from schemas, SQL and docs”这个项目思路它把上下文组装这件事从“凭经验往里塞”变成了“有体系地构建”。本文就来完整拆解这套思路并用一个可运行的 Python 示例带大家亲手实现自己的上下文组装模块。无论你是正在做 Natural Language to SQL、BI 对话式分析还是给内部数据平台接入 AI 助手这篇实战笔记都能给你一份可以直接落地的参考方案。1. 背景分析代理为什么需要“组装上下文”1.1 分析代理的真实困境先看一个很常见的需求让 AI 回答“上个月华东区的销售额环比变化了多少”。如果是人来做我们会先去数据库里找订单表、看地区字段、确认“销售额”的口径再写一条 SQL。但大模型没有这些先验信息它只知道语言模式和泛化知识。如果直接给它一个普通提示词你是一个数据分析师请根据数据库回答用户问题。它大概率会猜一个不存在的表名把“销售额”默认理解成 sum(amount)分不清created_at和paid_at哪个是下单时间、哪个是支付时间忽略订单状态里pending、paid、cancelled的含义。这就是分析代理和普通聊天机器人的最大区别聊天机器人只需要“会说”分析代理必须“说得对”。而“说得对”的前提是先把数据世界的结构、规则和经验组织成模型能读懂的文本上下文再交给它推理。1.2 什么是上下文组装上下文组装Context Assembly就是从数据库的多个信息源中抽取、格式化并聚合出代理执行任务所需的背景信息。这些信息通常包括三类Schema数据库的表、字段、类型、主外键、索引。SQL团队常用的查询、聚合逻辑、Join 写法、优化过的样例行。Docs业务术语、指标口径、命名规范、踩坑记录。组装的目标是在有限的 token 预算内尽可能让模型“看见”最相关、最关键的信息。这跟在 RAG 里做知识检索、重排是一个道理只是这里的知识来源更加结构化也更加可控。1.3 Schema、SQL、Docs 的分工可以这样理解三者的关系Schema 是“地图”告诉代理有哪些实体、字段、关系。SQL 是“样例路线”告诉代理前人是怎么走的哪些查询被验证过。Docs 是“地标说明”告诉代理每个字段背后真正的业务含义。光有地图代理容易迷路光有例子代理只会照抄光有说明代理缺少落地的抓手。只有三者配合才能形成一个可靠的决策基础。2. 核心组成拆解Schema、SQL、Docs2.1 Schema数据库结构的“地图”Schema 是指数据库的元数据包括表名、注释字段名、类型、长度、默认值主键、外键、唯一约束索引定义。示例格式如下表名: orders 字段: - id (INTEGER) [主键] - user_id (INTEGER) [非空] - status (TEXT) [非空] - total_amount (REAL) [非空] - paid_at (DATETIME) - created_at (DATETIME) 索引: - idx_orders_user ON (user_id) - idx_orders_status ON (status)这部分内容价值很高但也容易把上下文撑爆。如果一个库有 100 张表、几千个字段把全部 Schema 塞进提示词里显然不现实。所以要做的是按业务域拆 Schema把通用表、低频表排除压缩字段注释避免废话。2.2 SQL团队公共知识的“样本”SQL 示例的价值在于“可参考性”。代理从几十个高质量示例中学到的不仅仅是语法更是团队对数据的理解方式。好的 SQL 示例通常包括常用指标计算例如销售额、复购率典型 Join 关系如订单与用户、订单与商品复杂窗口函数、CTE被优化过的慢查询改写如避免SELECT *、正确走索引的写法。这里推荐将 SQL 示例按“查询意图”打标签。例如## 意图: 统计每日销售额 SELECT DATE(paid_at) AS day, SUM(total_amount) AS sales FROM orders WHERE paid_at datetime(now, -30 days) AND status paid GROUP BY DATE(paid_at) ORDER BY day;代理在看到示例后再遇到“统计周销售额”这一类请求时就更容易按同样的口径扩展。2.3 Docs业务口径的“说明书”Docs 不一定需要很长的文档大部分情况下几页 Markdown 就够了。关键是写清楚指标怎么定义例如“销售额只统计已支付订单”字段含义例如status字段有哪些枚举值时区、币种、小数位等公共约定特殊数据规则例如“删除采用物理删除无软删除字段”。举个例子如果业务团队内部规定“销售额按支付时间归期而不是下单时间”那么文档里必须写清楚。否则模型很可能按created_at聚合导致报表结果和业务对不上。2.4 组合示例一个分析问题的完整上下文清单假设用户问“本月销售额和上个月相比变化了多少”在组装上下文时代理至少需要看到信息类型具体内容作用Schemaorders、users 表结构和索引知道有哪些可用字段SQL 示例一个按月统计销售额的 SQL知道统计口径与写法Docs“销售额”和“订单状态”定义避免用错时间字段和状态安全规则禁止更新、禁止删除、必须加 LIMIT限制代理行为边界这就是“组装上下文”的含义不是把所有东西都塞进去而是把这条查询链路需要的信息组织成最优组合。3. 环境准备与项目结构3.1 演示环境本文使用 Python 3.9 和 SQLite 作为演示数据库不需要安装额外第三方库只需要标准库sqlite3、os、re。实际项目中你可以将同样的思路迁移到 MySQL、PostgreSQL 等数据库上。3.2 项目目录与演示数据我们先创建一个项目目录命名为context_buildercontext_builder/ ├── main.py # 入口脚本串联完整流程 ├── schema_extractor.py # 提取数据库 Schema ├── sql_examples.py # 加载 SQL 示例 ├── docs_loader.py # 加载业务文档 ├── assembler.py # 组装上下文控制 token 预算 └── test_data/ ├── app.db # SQLite 演示数据库由 init_db.py 生成 ├── sql_samples.sql # 示例 SQL 集合 └── docs/ ├── business_glossary.md └── data_rules.md为了方便复现我们再准备一个初始化数据库的脚本init_db.py放在项目根目录。# init_db.py import sqlite3 import os DB_PATH test_data/app.db def init_db(): if os.path.exists(DB_PATH): os.remove(DB_PATH) conn sqlite3.connect(DB_PATH) c conn.cursor() # 用户表 c.execute( CREATE TABLE users ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE NOT NULL, created_at DATETIME DEFAULT CURRENT_TIMESTAMP ) ) # 商品表 c.execute( CREATE TABLE products ( id INTEGER PRIMARY KEY, name TEXT NOT NULL, category TEXT NOT NULL, price REAL NOT NULL, stock INTEGER DEFAULT 0 ) ) # 订单表 c.execute( CREATE TABLE orders ( id INTEGER PRIMARY KEY, user_id INTEGER NOT NULL, status TEXT NOT NULL, total_amount REAL NOT NULL, paid_at DATETIME, created_at DATETIME DEFAULT CURRENT_TIMESTAMP, FOREIGN KEY (user_id) REFERENCES users(id) ) ) # 订单明细表 c.execute( CREATE TABLE order_items ( id INTEGER PRIMARY KEY, order_id INTEGER NOT NULL, product_id INTEGER NOT NULL, quantity INTEGER NOT NULL, unit_price REAL NOT NULL, FOREIGN KEY (order_id) REFERENCES orders(id), FOREIGN KEY (product_id) REFERENCES products(id) ) ) # 索引 c.execute(CREATE INDEX idx_orders_user ON orders(user_id)) c.execute(CREATE INDEX idx_orders_status ON orders(status)) c.execute(CREATE INDEX idx_order_items_order ON order_items(order_id)) # 测试数据 c.executemany( INSERT INTO users(name, email, created_at) VALUES (?, ?, ?), [ (Alice, aliceexample.com, 2024-01-05 10:00:00), (Bob, bobexample.com, 2024-02-18 12:30:00), (Carol, carolexample.com, 2024-03-22 09:15:00), ], ) c.executemany( INSERT INTO products(name, category, price, stock) VALUES (?, ?, ?, ?), [ (手机, 数码, 3999.0, 100), (耳机, 数码, 499.0, 500), (T恤, 服装, 129.0, 800), (球鞋, 服装, 599.0, 300), ], ) c.executemany( INSERT INTO orders(user_id, status, total_amount, paid_at, created_at) VALUES (?, ?, ?, ?, ?) , [ (1, paid, 3999.0, 2024-04-01 10:00:00, 2024-04-01 09:30:00), (1, paid, 998.0, 2024-04-05 15:00:00, 2024-04-05 14:40:00), (2, pending, 599.0, None, 2024-04-10 08:20:00), (2, paid, 129.0, 2024-04-12 11:00:00, 2024-04-12 10:55:00), (3, paid, 499.0, 2024-04-20 20:00:00, 2024-04-20 19:50:00), ], ) c.executemany( INSERT INTO order_items(order_id, product_id, quantity, unit_price) VALUES (?, ?, ?, ?) , [ (1, 1, 1, 3999.0), (2, 2, 2, 499.0), (3, 3, 1, 599.0), (4, 3, 1, 129.0), (5, 2, 1, 499.0), ], ) conn.commit() conn.close() print(数据库初始化完成:, DB_PATH) if __name__ __main__: init_db()运行这个脚本python init_db.py执行后会在test_data目录下生成app.db包含四张表和三个索引。4. 完整实战为分析代理组装上下文下面我们逐步实现一个精简但完整的上下文组装模块。4.1 提取 Database Schema创建schema_extractor.py负责从 SQLite 数据库读取表、字段、索引# schema_extractor.py import sqlite3 from typing import Dict, Any def extract_schema(db_path: str) - Dict[str, Any]: 提取数据库模式信息. conn sqlite3.connect(db_path) cursor conn.cursor() # 获取所有用户表 cursor.execute( SELECT name, sql FROM sqlite_master WHERE typetable ORDER BY name ) tables cursor.fetchall() schema: Dict[str, Any] {} for table_name, create_sql in tables: if table_name.startswith(sqlite_): continue # 字段信息 cursor.execute(fPRAGMA table_info(\{table_name}\)) columns cursor.fetchall() # 每行结构: (cid, name, type, notnull, dflt_value, pk) column_list [] for col in columns: column_list.append({ name: col[1], type: col[2], notnull: bool(col[3]), primary_key: bool(col[5]), }) # 索引信息 cursor.execute(fPRAGMA index_list(\{table_name}\)) indexes [] for idx in cursor.fetchall(): # 每行结构: (seq, name, unique, origin, partial) index_name idx[1] cursor.execute(fPRAGMA index_info(\{index_name}\)) index_columns [item[2] for item in cursor.fetchall()] indexes.append({ name: index_name, unique: bool(idx[2]), columns: index_columns, }) schema[table_name] { create_sql: create_sql, columns: column_list, indexes: indexes, } conn.close() return schema这里需要注意在真实项目中表名可能来自用户输入拼接 SQL 时不能直接把不可信内容拼进去。本例中表名来自sqlite_master属于可信来源但依然建议养成参数化查询的习惯。4.2 加载 SQL 参考示例创建sql_examples.py从 SQL 文件中读取示例。这里按分号拆分语句并用正则去掉--注释# sql_examples.py from typing import List, Dict import re def load_sql_examples(file_path: str) - List[Dict[str, str]]: 从 SQL 文件中加载示例查询. with open(file_path, r, encodingutf-8) as f: content f.read() raw_statements [s.strip() for s in content.split(;) if s.strip()] examples [] for stmt in raw_statements: # 去掉单行注释 clean_stmt re.sub(r--.*$, , stmt, flagsre.MULTILINE).strip() if not clean_stmt: continue examples.append({ raw: stmt, clean: clean_stmt, }) return examples这个方法能覆盖大多数情况但如果 SQL 里包含字符串字面量中的分号就会切分错误。真实项目中建议使用数据库 SQL parser 来处理。准备test_data/sql_samples.sql-- 近30天销售额按支付时间口径 SELECT DATE(paid_at) AS day, SUM(total_amount) AS sales FROM orders WHERE paid_at datetime(now, -30 days) AND status paid GROUP BY DATE(paid_at) ORDER BY day; -- 用 WITH 计算用户复购率 WITH user_orders AS ( SELECT user_id, COUNT(DISTINCT id) AS order_cnt FROM orders WHERE status paid GROUP BY user_id ) SELECT SUM(CASE WHEN order_cnt 2 THEN 1 ELSE 0 END) * 1.0 / COUNT(*) AS repurchase_rate FROM user_orders; -- 按月份统计订单量使用 BETWEEN 限定时间范围 SELECT strftime(%Y-%m, created_at) AS month, COUNT(*) AS order_cnt, SUM(total_amount) AS total_sales FROM orders WHERE created_at BETWEEN 2024-01-01 AND 2024-12-31 23:59:59 GROUP BY strftime(%Y-%m, created_at) ORDER BY month;这里故意放入了WITH AS和BETWEEN AND两个典型用法因为它们在真实分析场景里出现频率很高而且模型能从中学习到业务团队偏好的写法。4.3 加载业务文档创建docs_loader.py将 Markdown 文档读取为文本块# docs_loader.py import os from typing import List, Dict def load_docs(doc_dir: str) - List[Dict[str, str]]: 从目录加载 .md 和 .txt 文档. docs [] if not os.path.isdir(doc_dir): return docs for filename in sorted(os.listdir(doc_dir)): if not filename.endswith((.md, .txt)): continue file_path os.path.join(doc_dir, filename) with open(file_path, r, encodingutf-8) as f: content f.read() docs.append({ file: filename, content: content[:2000], }) return docs文档写入test_data/docs/business_glossary.md# 业务指标字典 ## 销售额Sales 指用户已经完成支付的订单金额合计统计时只取状态为 paid 的订单按支付时间paid_at归期。 ## 订单Order 用户下单生成的一条记录。状态包括 - pending待支付订单刚创建 - paid已支付 - cancelled已取消 ## 复购率Repurchase Rate 在一定周期内产生至少 2 笔已支付订单的用户数占该周期产生已支付订单用户总数的比例。再写入test_data/docs/data_rules.md# 数据口径与规则 1. 时间维度日/周/月等时间聚合默认按订单创建时间created_at统计但“销售”相关指标必须按支付时间paid_at统计。 2. 软删除当前所有表均为物理删除无软删除字段。 3. 金额字段total_amount 和 unit_price 单位为“元”保留两位小数。 4. 时区数据库存储 UTC 时间前端展示转换为东八区。 5. 订单状态只有 paid 状态的订单才计入销售指标cancelled 订单不计入任何营收统计。4.4 按 token 预算组装上下文核心模块是assembler.py。我们用一个简易 token 估算函数模拟 LLM 上下文窗口的限制再按优先级组装# assembler.py from typing import Dict, List, Any def estimate_tokens(text: str) - int: 粗略估算 token 数中文按 1.5 字/token英文按 4 字符/token. chinese_chars sum(1 for c in text if \u4e00 c \u9fff) other_chars len(text) - chinese_chars return int(chinese_chars * 1.5 other_chars / 4) def format_schema(schema: Dict[str, Any]) - str: 将 Schema 格式化为文本. lines [] for table_name, info in schema.items(): lines.append(f表名: {table_name}) lines.append(f建表 SQL: {info[create_sql]}) lines.append(字段:) for col in info[columns]: constraints [] if col[primary_key]: constraints.append(主键) if col[notnull]: constraints.append(非空) constraint_text f [{, .join(constraints)}] if constraints else lines.append(f - {col[name]} ({col[type]}){constraint_text}) if info[indexes]: lines.append(索引:) for idx in info[indexes]: unique_text UNIQUE if idx[unique] else lines.append( f - {idx[name]} ON ({, .join(idx[columns])}){unique_text} ) lines.append() return \n.join(lines) def assemble_context( schema: Dict[str, Any], sql_examples: List[Dict[str, str]], docs: List[Dict[str, str]], max_tokens: int 8000, ) - Dict[str, Any]: 组装上下文按优先级控制 token 预算. schema_text format_schema(schema) doc_text \n\n.join( f### 文档[{d[file]}]\n{d[content]} for d in docs ) sql_text \n\n.join( f### SQL 示例 {i 1}\n{ex[clean]} for i, ex in enumerate(sql_examples) ) sections [ (## 数据库 Schema, schema_text), (## 业务文档与指标定义, doc_text), (## SQL 参考示例, sql_text), ] final_text [] included [] token_used 0 for title, content in sections: if not content.strip(): continue section_full f{title}\n{content} token_est estimate_tokens(section_full) if token_used token_est max_tokens: # 如果连 Schema 都放不下只能强制截断 if not final_text: section_full section_full[: int(max_tokens * 4)] token_est estimate_tokens(section_full) else: break final_text.append(section_full) token_used token_est included.append(title) return { context: \n\n.join(final_text), token_count: token_used, max_tokens: max_tokens, included_sections: included, }这段代码的逻辑核心是“渐进式装配”Schema 永远是第一优先级文档第二SQL 示例第三。当上下文窗口紧张时优先保留表结构再逐层裁掉可选内容。真实项目可以进一步把“文档”和“SQL 示例”拆成更细的分块并通过向量检索选择最相关的部分。4.5 运行与验证最后创建main.py把整个流程串起来# main.py from schema_extractor import extract_schema from sql_examples import load_sql_examples from docs_loader import load_docs from assembler import assemble_context DB_PATH test_data/app.db SQL_SAMPLES_PATH test_data/sql_samples.sql DOCS_DIR test_data/docs def main(): print( * 60) print(① 提取数据库 Schema) schema extract_schema(DB_PATH) print(f 发现表: {list(schema.keys())}) print(② 加载 SQL 示例) sql_examples load_sql_examples(SQL_SAMPLES_PATH) print(f 加载示例: {len(sql_examples)} 条) print(③ 加载业务文档) docs load_docs(DOCS_DIR) print(f 加载文档: {len(docs)} 篇) print(④ 组装上下文) result assemble_context( schemaschema, sql_examplessql_examples, docsdocs, max_tokens8000, ) print(f 上下文 token 数: {result[token_count]}) print(f 包含区块: {result[included_sections]}) print(\n * 60) print(最终上下文预览前 1200 字符:) print(- * 60) print(result[context][:1200]) if __name__ __main__: main()运行python main.py预期输出如下 ① 提取数据库 Schema 发现表: [order_items, orders, products, users] ② 加载 SQL 示例 加载示例: 3 条 ③ 加载业务文档 加载文档: 2 篇 ④ 组装上下文 上下文 token 数: 约 343 包含区块: [## 数据库 Schema, ## 业务文档与指标定义, ## SQL 参考示例]你会得到一个结构清晰的上下文文本之后可以把它作为 system message 或 few-shot 示例传给任意 LLM 接口。这样代理再回答问题“上个月销售额是多少”时就能知道应该用orders表按paid_at归期状态要过滤为paid可以参考“近30天销售额”的 SQL 写法。5. 常见问题与排查思路5.1 常见问题速查表问题现象常见原因解决思路代理使用了不存在的表或字段Schema 缺失、过期每次会话前重新提取 Schema并校验表名指标口径错误例如销售额统计错文档缺失模型按常识理解在指标字典中写清楚口径并附正反示例报错context length exceeded上下文超过模型限制分层裁剪、按需检索、降低max_tokens代理生成的 SQL 执行太慢忽略索引产生大表扫描Schema 中展示索引并提供优化后的 SQL 样例代理重复生成相似 SQL缺少公共 SQL 参考维护意图标签库让 SQL 示例更易检索文档更新后代理行为漂移上下文没有版本管理文档纳入 Git 管理变更后刷新缓存代理执行了删除、更新等危险操作缺少安全规则在上下文中加入行为边界数据库最小权限5.2 排查清单当你发现代理效果不稳定可以按以下顺序排查先看 Schema 是否完整表名、字段名、字段类型是否都提取到了。再看文档是否覆盖指标定义、枚举值、时间口径是否写清楚。再看 SQL 示例是否相关是否有与问题相近的查询样本。查 token 预算上下文是否被截断哪些部分被丢掉了。查数据库权限代理连接的是只读账号还是写权限账号。查日志把组装后的完整上下文打印出来直接判断是否“像人话”。很多时候问题并不在模型而在上下文本身。6. 最佳实践与工程建议6.1 用三层结构治理上下文建议把上下文组装模块设计成三层基础层Schema 信息全量且稳定按表缓存。业务层指标字典、数据规则变更频率低适合人工维护。样例层SQL 示例随查询场景增长适合做向量索引和检索。三层分开维护有利于减少重复构建成本。比如 Schema 每天刷新一次文档每周人工 reviewSQL 示例持续补充。6.2 上下文安全与权限边界这是最容易踩坑的地方也是最重要的部分。生产环境中必须做到分析代理连接数据库使用只读账号禁用DELETE、UPDATE、DROP等操作不要在文档和 Schema 中暴露敏感字段值如果上下文来自不可信文本需要过滤 prompt injection 内容所有代理执行的 SQL 都要记录日志便于审计。安全边界不是靠提示词保证的而是靠数据库权限和网关层。提示词只能作为辅助约束。6.3 缓存、刷新与审计Schema 和文档都是相对静态的信息没必要每次请求都重新组装。建议采用以下策略Schema 缓存 5 到 10 分钟并在表结构变更时主动失效文档缓存后做内容 hash变更时自动更新每次请求记录上下文版本号、token 数、被裁掉的部分定期统计哪些查询最频繁把高频模式沉淀为新的 SQL