企业官网建设流程全解析

做数据分析 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 Assembly

2.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 与组装后上下文的对比

这个对比说明一件事：组装的核心价值不是“多”，而是“准”。让模型在更少但更关键的信息上做推理，效果通常好于一次性塞入大量原始数据。

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 2.x 风格的 API，1.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: postgresql+psycopg2://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(encoding="utf-8", errors="ignore") 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(encoding="utf-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(loader=FileSystemLoader("templates")) template = env.get_template("agent_context.md.j2") return template.render( generated_at=datetime.now().strftime("%Y-%m-%d %H:%M:%S"), tables=tables, table_counts=sorted(table_counts.items(), key=lambda x: x[1], reverse=True), sql_features=sorted(sql_features), doc_sections=doc_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(description="Build context for analytics agents") parser.add_argument("--db-url", default=os.getenv("ANALYTICS_DB_URL"), help="SQLAlchemy 数据库连接串") parser.add_argument("--sql-dir", type=Path, required=True) parser.add_argument("--docs-dir", type=Path, required=True) parser.add_argument("--output", type=Path, default=Path("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, encoding="utf-8") print(f"context 已写入: {args.output}") print(f"context 长度: {len(context)} 字符") if __name__ == "__main__": main()

这个入口会在运行后打印上下文字符数，方便你快速判断 token 规模。后续可以把 build_context.py 接入 CI，每天定时重建上下文文件。

6. 运行结果与效果验证

6.1 运行命令

假设环境变量已经设置好，执行：

export ANALYTICS_DB_URL="postgresql+psycopg2://analytic_ro:your_password@localhost: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. 常见问题与排查思路

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 和业务文档这“三件套”整理得好，模型的真实水平才有机会体现出来。