语义层架构：让 AI 真正理解企业数据

在 Text-to-SQL 系统中，元数据质量直接决定了 SQL 生成的准确性。但传统的元数据管理往往只关注技术层面（表名、字段名、数据类型），忽略了业务语义。AskTable 通过 MetaAdmin 语义层，将技术元数据和业务语义有机结合，让 AI 真正理解企业数据。

为什么需要语义层？

传统元数据的局限

技术元数据示例：

CREATE TABLE ord_dtl (
    ord_id VARCHAR(32),
    cust_id VARCHAR(32),
    prod_id VARCHAR(32),
    qty INT,
    amt DECIMAL(10,2),
    sts TINYINT,
    crt_tm DATETIME
);

问题：

•
表名
```
ord_dtl
```
是什么意思？订单详情？
•
字段
```
qty
```
是数量，但是什么的数量？
•
字段
```
amt
```
是金额，但是含税还是不含税？
•
字段
```
sts
```
的值 1、2、3 分别代表什么状态？
•
字段
```
crt_tm
```
是创建时间还是下单时间？

用户提问："上个月的订单金额是多少？"

AI 的困惑：

•
应该查询
```
ord_dtl
```
表还是
```
order_detail
```
表？
•
应该用
```
amt
```
字段还是
```
amount
```
字段？
•
需要过滤
```
sts
```
字段吗？什么值才算有效订单？

语义层的价值

AskTable 的 MetaAdmin 语义层提供了完整的业务语义：

Table(
    name="ord_dtl",
    origin_desc="ord_dtl",  # 原始表名
    curr_desc="订单详情表，记录每笔订单的商品明细信息",  # 业务描述
    fields={
        "ord_id": Field(
            name="ord_id",
            data_type="VARCHAR(32)",
            curr_desc="订单ID，关联订单主表",
        ),
        "amt": Field(
            name="amt",
            data_type="DECIMAL(10,2)",
            curr_desc="订单金额，单位：元，不含税，不含运费",
        ),
        "sts": Field(
            name="sts",
            data_type="TINYINT",
            curr_desc="订单状态：1-待支付，2-已支付，3-已发货，4-已完成，5-已取消",
        ),
    }
)

现在 AI 可以：

•
理解
```
ord_dtl
```
是订单详情表
•
知道
```
amt
```
是不含税不含运费的金额
•
知道只有
```
sts
```
为 2、3、4 的订单才是有效订单

MetaAdmin 分层架构

加载图表中...

上图展示了 MetaAdmin 的三层架构：Schema（模式）→ Table（表）→ Field（字段）。每一层都包含原始描述和业务描述，通过这种分层结构，AI 能够准确理解数据的业务含义和层次关系。

核心架构设计

1. 分层元数据结构

AskTable 采用三层结构：Schema → Table → Field

@dataclasses.dataclass(kw_only=True)
class StructureBase:
    name: str
    origin_desc: str | None = None  # 原始描述（通常是表名/字段名）
    curr_desc: str | None = None    # 当前描述（业务语义）
    curr_desc_stat: str = "origin"  # 描述状态：origin/manual/ai_generated

@dataclasses.dataclass()
class Field(StructureBase):
    schema_name: str
    table_name: str

    sample_data: str | None = None  # 示例数据
    data_type: str | None = None    # 数据类型

    is_index: bool = False          # 是否是索引字段
    index_count: int = 0            # 索引基数

    visibility: bool = True         # 是否可见
    identifiable_type: IdentifiableType = IdentifiableType.plain  # 敏感信息类型

    is_vector_synced: bool = True   # 是否已同步到向量库

    @property
    def full_name(self) -> str:
        return f"{self.schema_name}.{self.table_name}.{self.name}"

@dataclasses.dataclass()
class Table(StructureBase):
    schema_name: str
    fields: dict[str, Field] = dataclasses.field(default_factory=dict)

    @property
    def full_name(self) -> str:
        return f"{self.schema_name}.{self.name}"

@dataclasses.dataclass()
class Schema(StructureBase):
    custom_configs: dict | None = None
    tables: dict[str, Table] = dataclasses.field(default_factory=dict)

    @property
    def full_name(self) -> str:
        return self.name

@dataclasses.dataclass(kw_only=True)
class MetaAdmin:
    datasource_id: str | None = None
    schemas: dict[str, Schema] = dataclasses.field(default_factory=dict)

    @property
    def schema_count(self):
        return len(self.schemas.values())

    @property
    def table_count(self):
        return sum([len(schema.tables.values()) for schema in self.schemas.values()])

    @property
    def field_count(self):
        return sum([
            len(table.fields.values())
            for schema in self.schemas.values()
            for table in schema.tables.values()
        ])

    def __repr__(self):
        return f"[{self.datasource_id}] {self.schema_count} S, {self.table_count} T, {self.field_count} F"

设计亮点：

1.1 双重描述机制

origin_desc: str | None = None  # 原始描述
curr_desc: str | None = None    # 当前描述
curr_desc_stat: str = "origin"  # 描述来源

为什么需要两个描述？

•origin_desc：保留原始信息，便于追溯
•
curr_desc：当前使用的业务描述，可以是：
- •手动编写（manual）
- •AI 生成（ai_generated）
- •原始值（origin）

应用场景：

# 场景 1：手动优化描述
field.origin_desc = "amt"
field.curr_desc = "订单金额，单位：元，不含税"
field.curr_desc_stat = "manual"

# 场景 2：AI 生成描述
field.origin_desc = "amt"
field.curr_desc = "订单金额（根据历史数据分析得出）"
field.curr_desc_stat = "ai_generated"

# 场景 3：回退到原始描述
field.curr_desc = field.origin_desc
field.curr_desc_stat = "origin"

1.2 丰富的字段属性

sample_data: str | None = None  # 示例数据
is_index: bool = False          # 是否是索引
index_count: int = 0            # 索引基数
visibility: bool = True         # 是否可见
identifiable_type: IdentifiableType = IdentifiableType.plain  # 敏感类型

用途：

示例数据：帮助 AI 理解字段内容

field.sample_data = "['北京', '上海', '广州', '深圳']"
# AI 可以推断这是地区字段

索引信息：优化查询性能

field.is_index = True
field.index_count = 1000000  # 高基数，适合作为过滤条件
# AI 生成 SQL 时优先使用索引字段

可见性控制：隐藏敏感字段

field.visibility = False  # 用户无法查询此字段

敏感信息标记：数据脱敏

field.identifiable_type = IdentifiableType.phone  # 手机号
# 查询结果自动脱敏：138****5678

2. 动态过滤能力

MetaAdmin 提供了强大的过滤功能，支持按名称、正则表达式过滤：

2.1 按字段名过滤

def filter_fields_by_names(
    self, field_full_names: list[tuple[str, str, str]]
) -> "MetaAdmin":
    """
    按字段全名过滤
    field_full_names: [(schema_name, table_name, field_name), ...]
    """
    filtered_meta = MetaAdmin(datasource_id=self.datasource_id)

    for schema_name, table_name, field_name in field_full_names:
        if schema := self.schemas.get(schema_name):
            if table := schema.tables.get(table_name):
                if field := table.fields.get(field_name):
                    # 创建 schema（如果不存在）
                    if schema_name not in filtered_meta.schemas:
                        filtered_schema = dataclasses.replace(schema, tables={})
                        filtered_meta.schemas[schema_name] = filtered_schema
                    else:
                        filtered_schema = filtered_meta.schemas[schema_name]

                    # 创建 table（如果不存在）
                    if table_name not in filtered_schema.tables:
                        filtered_table = dataclasses.replace(table, fields={})
                        filtered_schema.tables[table_name] = filtered_table

                    # 添加 field
                    filtered_schema.tables[table_name].fields[field_name] = field

    return filtered_meta

应用场景：

# 向量检索返回相关字段
relevant_fields = [
    ("public", "orders", "order_id"),
    ("public", "orders", "amount"),
    ("public", "customers", "customer_name"),
]

# 只保留相关字段
filtered_meta = meta.filter_fields_by_names(relevant_fields)

# 传给 LLM 的元数据大幅减少
print(filtered_meta)  # [ds_001] 1 S, 2 T, 3 F

2.2 按表名过滤

def filter_tables_by_names(
    self, table_full_names: list[tuple[str, str]]
) -> "MetaAdmin":
    """
    按表全名过滤
    table_full_names: [(schema_name, table_name), ...]
    """
    filtered_meta = MetaAdmin(datasource_id=self.datasource_id)

    for schema_name, table_name in table_full_names:
        if schema := self.schemas.get(schema_name):
            if table := schema.tables.get(table_name):
                if schema_name not in filtered_meta.schemas:
                    filtered_schema = dataclasses.replace(schema, tables={})
                    filtered_meta.schemas[schema_name] = filtered_schema
                else:
                    filtered_schema = filtered_meta.schemas[schema_name]

                filtered_schema.tables[table_name] = table

    return filtered_meta

应用场景：

# Agent 需要查看特定表的详细信息
tables_to_show = [
    ("public", "orders"),
    ("public", "order_items"),
]

filtered_meta = meta.filter_tables_by_names(tables_to_show)

# 返回完整的表结构（包含所有字段）
return filtered_meta.to_markdown()

2.3 按正则表达式过滤

def filter_by_regex(
    self, schema_pattern=None, table_pattern=None, field_pattern=None
) -> "MetaAdmin":
    """
    按正则表达式过滤
    """
    filtered_meta = MetaAdmin(datasource_id=self.datasource_id)

    schema_regex = re.compile(schema_pattern) if schema_pattern else None
    table_regex = re.compile(table_pattern) if table_pattern else None
    field_regex = re.compile(field_pattern) if field_pattern else None

    for schema in self.schemas.values():
        if schema_regex and not schema_regex.match(schema.name):
            continue

        filtered_schema = dataclasses.replace(schema, tables={})
        schema_added = False

        for table in schema.tables.values():
            if table_regex and not table_regex.match(table.name):
                continue

            filtered_table = dataclasses.replace(table, fields={})
            table_added = False

            for field in table.fields.values():
                if field_regex and not field_regex.match(field.name):
                    continue

                filtered_table.fields[field.name] = field
                table_added = True

            if table_added:
                filtered_schema.tables[table.name] = filtered_table
                schema_added = True

        if schema_added:
            filtered_meta.schemas[schema.name] = filtered_schema

    return filtered_meta

应用场景：

# 场景 1：只查询订单相关的表
order_meta = meta.filter_by_regex(table_pattern=r"^ord.*")

# 场景 2：只查询金额相关的字段
amount_meta = meta.filter_by_regex(field_pattern=r".*amt.*|.*amount.*|.*price.*")

# 场景 3：排除测试表
prod_meta = meta.filter_by_regex(table_pattern=r"^(?!test_).*")

3. 元数据合并

支持多个 MetaAdmin 的合并和差集操作：

def merge_metas(allow_metas: list[MetaAdmin], deny_metas: list[MetaAdmin]) -> MetaAdmin:
    """
    合并多个 allow_metas，并从结果中移除 deny_metas 中的字段
    """
    result_meta = MetaAdmin()

    # 合并所有 allow_metas
    for allow_meta in allow_metas:
        for schema_name, schema in allow_meta.schemas.items():
            for table_name, table in schema.tables.items():
                for field_name, field in table.fields.items():
                    add_field(result_meta, schema_name, table_name, field_name,
                             schema, table, field)

    # 移除 deny_metas 中的字段
    for deny_meta in deny_metas:
        for schema_name, schema in deny_meta.schemas.items():
            for table_name, table in schema.tables.items():
                for field_name, field in table.fields.items():
                    remove_field_if_exists(result_meta, schema_name, table_name, field_name)

    return result_meta

应用场景：

# 场景 1：权限控制
user_allowed_meta = get_user_allowed_meta(user_id)  # 用户有权限的字段
user_denied_meta = get_user_denied_meta(user_id)    # 用户被禁止的字段

final_meta = merge_metas([user_allowed_meta], [user_denied_meta])

# 场景 2：多角色合并
role1_meta = get_role_meta("sales")      # 销售角色可见的字段
role2_meta = get_role_meta("marketing")  # 市场角色可见的字段

combined_meta = merge_metas([role1_meta, role2_meta], [])

4. Markdown 导出

MetaAdmin 可以导出为 Markdown 格式，便于传给 LLM：

def to_markdown(
    self,
    level: Literal["table", "field"] = "field",
    include: set[str] = set(),
    exclude: set[str] = set(),
) -> str:
    """
    转换为 Markdown 格式
    level: "table" 只显示表名，"field" 显示字段详情
    include: 额外包含的字段属性
    exclude: 排除的字段属性
    """
    # 构建数据结构
    result = {
        "schemas": [schema_to_dict(schema) for schema in self.schemas.values()],
    }

    # 转换为 Markdown
    return dict_to_markdown(result, table_format_keys=("fields",))

输出示例：

## Schemas

### public

**Description:** 公共模式

#### Tables

##### orders

**Full Name:** public.orders
**Description:** 订单表，记录所有客户订单信息

**Fields:**

| name | desc | data_type |
|------|------|-----------|
| order_id | 订单ID，主键 | VARCHAR(32) |
| customer_id | 客户ID，关联客户表 | VARCHAR(32) |
| amount | 订单金额，单位：元，不含税 | DECIMAL(10,2) |
| status | 订单状态：1-待支付，2-已支付，3-已完成 | TINYINT |
| created_at | 订单创建时间 | DATETIME |

##### order_items

**Full Name:** public.order_items
**Description:** 订单明细表，记录订单中的商品信息

**Fields:**

| name | desc | data_type |
|------|------|-----------|
| item_id | 明细ID，主键 | VARCHAR(32) |
| order_id | 订单ID，关联订单表 | VARCHAR(32) |
| product_id | 商品ID，关联商品表 | VARCHAR(32) |
| quantity | 商品数量 | INT |
| price | 商品单价，单位：元 | DECIMAL(10,2) |

优势：

•结构清晰，易于 LLM 理解
•包含完整的业务语义
•支持自定义显示内容

实际应用场景

场景 1：权限控制

需求：销售部门只能查询自己负责地区的订单数据

实现：

# 1. 获取用户角色的元数据
role = get_user_role(user_id)
accessible_meta = role.get_accessible_meta(datasource)

# 2. 过滤可见字段
for schema in accessible_meta.schemas.values():
    for table in schema.tables.values():
        # 移除不可见字段
        fields_to_remove = [
            field_name
            for field_name, field in table.fields.items()
            if not field.visibility
        ]
        for field_name in fields_to_remove:
            table.fields.pop(field_name)

# 3. 传给 Agent
agent = DBAgent(datasource=datasource, meta=accessible_meta, assumed_role=role)

效果：

•用户只能看到有权限的表和字段
•即使知道字段名也无法查询
•查询结果自动过滤

场景 2：敏感信息脱敏

需求：客服人员可以查询用户信息，但手机号和身份证号需要脱敏

实现：

# 1. 标记敏感字段
user_table.fields["phone"].identifiable_type = IdentifiableType.phone
user_table.fields["id_card"].identifiable_type = IdentifiableType.id_card

# 2. 查询时自动脱敏
result = datasource.accessor.query("SELECT * FROM users LIMIT 10")

for row in result:
    for field_name, field in user_table.fields.items():
        if field.identifiable_type == IdentifiableType.phone:
            row[field_name] = mask_phone(row[field_name])  # 138****5678
        elif field.identifiable_type == IdentifiableType.id_card:
            row[field_name] = mask_id_card(row[field_name])  # 110***********1234

场景 3：元数据质量提升

需求：自动生成字段描述，提升元数据质量

实现：

async def enhance_metadata(meta: MetaAdmin) -> MetaAdmin:
    """使用 AI 增强元数据描述"""
    for schema in meta.schemas.values():
        for table in schema.tables.values():
            # 生成表描述
            if not table.curr_desc or table.curr_desc == table.name:
                table.curr_desc = await generate_table_description(table)
                table.curr_desc_stat = "ai_generated"

            for field in table.fields.values():
                # 生成字段描述
                if not field.curr_desc or field.curr_desc == field.name:
                    field.curr_desc = await generate_field_description(
                        table, field, field.sample_data
                    )
                    field.curr_desc_stat = "ai_generated"

    return meta

async def generate_field_description(table: Table, field: Field, sample_data: str) -> str:
    """使用 LLM 生成字段描述"""
    prompt = f"""
    表名：{table.name}
    表描述：{table.curr_desc}
    字段名：{field.name}
    数据类型：{field.data_type}
    示例数据：{sample_data}

    请生成一个简洁的字段描述（不超过 50 字），说明：
    1. 字段的业务含义
    2. 数据的单位（如果有）
    3. 特殊说明（如枚举值含义）
    """

    response = await llm_client.create_completion(
        messages=[{"role": "user", "content": prompt}],
        model="gpt-4o-mini"
    )

    return response.choices[0].message.content.strip()

效果：

•自动生成高质量的字段描述
•大幅减少人工标注工作量
•持续提升元数据质量

性能优化实践

1. 延迟加载

class MetaAdmin:
    _lazy_loaded: bool = False

    def ensure_loaded(self):
        """延迟加载元数据"""
        if not self._lazy_loaded:
            self._load_from_database()
            self._lazy_loaded = True

    def _load_from_database(self):
        """从数据库加载元数据"""
        # 只加载必要的信息
        # 详细信息按需加载
        pass

2. 增量更新

def update_field_description(
    self,
    schema_name: str,
    table_name: str,
    field_name: str,
    new_desc: str
):
    """增量更新字段描述"""
    field = self.schemas[schema_name].tables[table_name].fields[field_name]
    field.curr_desc = new_desc
    field.curr_desc_stat = "manual"
    field.is_vector_synced = False  # 标记需要重新同步到向量库

    # 只更新这一个字段，不重建整个元数据
    await update_field_vector(field)

3. 缓存策略

class MetaCache:
    _cache: dict[str, MetaAdmin] = {}
    _ttl: int = 3600  # 1 小时

    @classmethod
    def get(cls, datasource_id: str) -> MetaAdmin | None:
        cache_key = f"meta:{datasource_id}"
        if cache_key in cls._cache:
            meta, timestamp = cls._cache[cache_key]
            if time.time() - timestamp < cls._ttl:
                return meta
        return None

    @classmethod
    def set(cls, datasource_id: str, meta: MetaAdmin):
        cache_key = f"meta:{datasource_id}"
        cls._cache[cache_key] = (meta, time.time())

最佳实践建议

1. 元数据描述规范

表描述：

# ✅ 好的描述
table.curr_desc = "订单表，记录所有客户订单信息，包括订单金额、状态、创建时间等"

# ❌ 差的描述
table.curr_desc = "orders"  # 没有业务含义

字段描述：

# ✅ 好的描述
field.curr_desc = "订单金额，单位：元，不含税，不含运费"

# ❌ 差的描述
field.curr_desc = "金额"  # 信息不足

2. 敏感信息标记

# 手机号
field.identifiable_type = IdentifiableType.phone

# 身份证号
field.identifiable_type = IdentifiableType.id_card

# 邮箱
field.identifiable_type = IdentifiableType.email

# 银行卡号
field.identifiable_type = IdentifiableType.bank_card

3. 可见性控制

# 隐藏内部字段
field.visibility = False

# 隐藏测试表
table.visibility = False

# 隐藏废弃字段
field.visibility = False
field.curr_desc = f"{field.curr_desc}（已废弃）"

总结

AskTable 的 MetaAdmin 语义层通过分层元数据管理、动态过滤和权限控制，让 AI 真正理解企业数据：

•双重描述：保留原始信息，支持业务语义
•丰富属性：示例数据、索引信息、敏感标记
•动态过滤：按名称、正则表达式、权限过滤
•元数据合并：支持多角色、多数据源
•Markdown 导出：结构清晰，易于 LLM 理解

语义层是 Text-to-SQL 系统的基础，高质量的元数据直接决定了 SQL 生成的准确性。通过 MetaAdmin，AskTable 实现了技术元数据和业务语义的完美结合。

加入 AskTable 社群

咨询我们

语义层架构：让 AI 真正理解企业数据

为什么需要语义层？

传统元数据的局限

语义层的价值

MetaAdmin 分层架构

核心架构设计

1. 分层元数据结构

1.1 双重描述机制

1.2 丰富的字段属性

2. 动态过滤能力

2.1 按字段名过滤

2.2 按表名过滤

2.3 按正则表达式过滤

3. 元数据合并

4. Markdown 导出

实际应用场景

场景 1：权限控制

场景 2：敏感信息脱敏

场景 3：元数据质量提升

性能优化实践

1. 延迟加载

2. 增量更新

3. 缓存策略

最佳实践建议

1. 元数据描述规范

2. 敏感信息标记

3. 可见性控制

总结