这篇讲元数据映射，这篇讲具体怎么融进 Router+Worker 架构。

---

架构：映射层插在哪

用户 → Router Agent → Skill Worker → Metabase MCP
              ↓              ↓
         意图识别        映射执行 + SQL生成

映射层在 Skill Worker 里。Router 只做意图识别，把映射的事交给 Worker。

---

第一步：表管理——划定 LLM 的知识边界

原始问题

LLM 不知道哪些表参与问数、字段叫什么、业务什么意思。

实现

# skill_worker/metadata.py

TABLE_CONFIG = {
    "dw_waybill_detail": {
        "description": "运单明细表，记录每个运单的轨迹和签收状态。",
        "fields": {
            "waybill_no": {
                "alias": "运单号",
                "description": "运单唯一标识，格式：YT开头+12位数字"
            },
            "site_name": {
                "alias": "网点",
                "description": "转运中心/网点名称，如：上海转运中心、北京网点"
            },
            "sign_count": {
                "alias": "签收件数",
                "description": "已签收的包裹数量"
            },
            "total_count": {
                "alias": "总件数",
                "description": "运单总包裹数量"
            },
            "org_code": {
                "alias": "组织编码",
                "description": "归属组织，SH001=上海，BJ001=北京"
            },
            "waybill_time": {
                "alias": "运单时间",
                "description": "运单创建时间，存的是芝加哥时间"
            },
        }
    }
}

def build_table_context(table_name: str) -> str:
    """生成注入给 LLM 的表结构上下文"""
    config = TABLE_CONFIG.get(table_name)
    if not config:
        return ""
    
    lines = [f"表：{table_name}"]
    lines.append(f"说明：{config['description']}")
    lines.append("字段：")
    
    for fname, finfo in config["fields"].items():
        lines.append(f"  - {fname}（{finfo['alias']}）：{finfo['description']}")
    
    return "\n".join(lines)

用法

# skill_worker/router.py

def route_to_worker(user_query: str, context: dict):
    # Router 只判断意图，不做映射
    if "签收" in user_query or "网点" in user_query:
        return "waybill_worker"
    elif "成本" in user_query or "利润" in user_query:
        return "financial_worker"
    return "default_worker"


# skill_worker/workers/waybill_worker.py

class WaybillWorker:
    def __init__(self):
        self.metadata = MetadataService()
    
    def execute(self, user_query: str, context: dict):
        # 1. 注入表结构上下文
        table_context = build_table_context("dw_waybill_detail")
        
        # 2. 拼接 prompt
        prompt = f"""
用户问题：{user_query}

{table_context}

请根据用户问题生成 SQL。
"""
        # 3. LLM 生成 SQL
        sql = self.llm.generate(prompt)
        return sql

---

第二步：术语词典——消除业务歧义

原始问题

用户说"网点"，LLM 不知道不同表里叫法不同。

实现

# skill_worker/terminology.py

TERM_DICTIONARY = {
    "网点": {
        "dw_waybill_detail": "site_name",
        "dw_trajectory": "transfer_center",
        "dw_financial": "branch_name",
    },
    "运单号": {
        "dw_waybill_detail": "waybill_no",
        "dw_trajectory": "ewb_no",
    },
    "签收率": {
        "formula": "SUM(sign_count) / SUM(total_count)",
        "unit": "%",
    },
    "妥投率": {
        "formula": "SUM(delivered_count) / SUM(total_count)",
    },
    "东区": {
        "meaning": "所有 parent_name='东区' 的递归子部门",
        "sql_hint": "需要递归 CTE 查询子部门",
    },
}

def resolve_term(user_term: str, table_name: str = None):
    """解析用户术语，返回数据库字段或计算公式"""
    if user_term not in TERM_DICTIONARY:
        return user_term  # 未收录，返回原词
    
    entry = TERM_DICTIONARY[user_term]
    
    # 优先返回公式（指标类）
    if "formula" in entry:
        return entry["formula"]
    
    # 返回字段映射
    if table_name and table_name in entry:
        return entry[table_name]
    
    # 返回第一个匹配的字段
    for key, value in entry.items():
        if key not in ["formula", "meaning", "sql_hint"]:
            return value
    
    return user_term


def inject_terminology(sql: str, table_name: str) -> str:
    """把 SQL 中的术语替换成实际字段名"""
    for term, mapping in TERM_DICTIONARY.items():
        if term in sql and "formula" not in mapping:
            if table_name in mapping:
                sql = sql.replace(term, mapping[table_name])
    return sql

用法

class WaybillWorker:
    def execute(self, user_query: str, context: dict):
        # 1. 术语预处理
        # 用户说"签收率"，先替换成计算公式
        resolved_query = user_query
        for term in TERM_DICTIONARY:
            if term in user_query and "formula" in TERM_DICTIONARY[term]:
                resolved_query = resolved_query.replace(
                    term,
                    TERM_DICTIONARY[term]["formula"]
                )
        
        # 2. 生成 SQL
        prompt = f"""
用户问题：{resolved_query}

表：dw_waybill_detail
字段：waybill_no, site_name, sign_count, total_count, org_code, waybill_time

生成 SQL。
"""
        sql = self.llm.generate(prompt)
        
        # 3. 字段名后处理
        sql = inject_terminology(sql, "dw_waybill_detail")
        
        return sql

---

第三步：表关联关系——告诉 LLM 怎么 JOIN

原始问题

跨表查询时，LLM 不知道表之间怎么关联。

实现

# skill_worker/relationships.py

TABLE_RELATIONSHIPS = {
    "dw_waybill_detail": {
        "primary_key": "waybill_no",
        "joins": [
            {
                "target": "dim_site",
                "on": "dw_waybill_detail.site_id = dim_site.site_id",
                "description": "网点维度表，提供网点名称和所属区域"
            },
            {
                "target": "dw_organization",
                "on": "dw_waybill_detail.org_code = dw_organization.org_code",
                "description": "组织维度表，提供组织层级关系"
            },
        ]
    }
}

def build_join_context(table_name: str) -> str:
    """生成 JOIN 上下文"""
    rel = TABLE_RELATIONSHIPS.get(table_name, {})
    joins = rel.get("joins", [])
    
    if not joins:
        return ""
    
    lines = ["JOIN 关系："]
    for j in joins:
        lines.append(f"  - {table_name} → {j['target']}")
        lines.append(f"    ON {j['on']}")
        lines.append(f"    说明：{j['description']}")
    
    return "\n".join(lines)

用法

def execute(self, user_query: str, context: dict):
    # 注入表结构
    table_context = build_table_context("dw_waybill_detail")
    
    # 注入 JOIN 关系
    join_context = build_join_context("dw_waybill_detail")
    
    prompt = f"""
用户问题：{user_query}

{table_context}

{join_context}

生成 SQL。
"""
    return self.llm.generate(prompt)

---

第四步：示例 SQL——解决复杂口径

原始问题

复杂查询（如递归部门、复杂筛选），LLM 自己写不出来。

实现

# skill_worker/examples.py

EXAMPLE_SQL = {
    "签收率": [
        {
            "query": "昨天各网点签收率",
            "sql": """
SELECT 
    site_name AS 网点,
    SUM(sign_count) / SUM(total_count) AS 签收率
FROM dw_waybill_detail
WHERE waybill_time >= '2026-06-29 07:00:00'  -- 北京转芝加哥
  AND waybill_time <= '2026-06-30 06:59:59'
GROUP BY site_name
ORDER BY 签收率 DESC
"""
        },
        {
            "query": "上海网点本月妥投率",
            "sql": """
SELECT 
    site_name AS 网点,
    SUM(delivered_count) / SUM(total_count) AS 妥投率
FROM dw_waybill_detail
WHERE org_code = 'SH001'
  AND waybill_time >= DATE_FORMAT(NOW(), '%Y-%m-01')
GROUP BY site_name
"""
        }
    ]
}

def find_similar_example(user_query: str, metric: str) -> str:
    """找相似示例 SQL"""
    examples = EXAMPLE_SQL.get(metric, [])
    for ex in examples:
        # 简单关键词匹配
        if any(kw in user_query for kw in ex["query"].split()):
            return ex["sql"]
    return ""

用法

def execute(self, user_query: str, context: dict):
    # 1. 找相似示例
    example_sql = find_similar_example(user_query, "签收率")
    
    # 2. 拼接 prompt
    prompt = f"""
用户问题：{user_query}

参考示例：
{example_sql}

表：dw_waybill_detail
字段：waybill_no, site_name, sign_count, total_count, delivered_count, org_code, waybill_time

生成 SQL。
"""
    return self.llm.generate(prompt)

---

第五步：时区处理——这是自己踩过的坑

原始问题

用户说"昨天"，数据库存的是芝加哥时间。

实现

# skill_worker/timezone.py

TIMEZONE_CONFIG = {
    "数据库": "America/Chicago",
    "业务": "Asia/Shanghai",
}

def parse_relative_date(user_query: str) -> dict:
    """解析相对时间，返回数据库时区的时间范围"""
    import re
    from datetime import datetime, timedelta
    import pytz
    
    beijing_tz = pytz.timezone(TIMEZONE_CONFIG["业务"])
    chicago_tz = pytz.timezone(TIMEZONE_CONFIG["数据库"])
    
    now_beijing = datetime.now(beijing_tz)
    
    if "昨天" in user_query:
        yesterday_beijing = now_beijing - timedelta(days=1)
        start_chicago = yesterday_beijing.astimezone(chicago_tz)
        end_chicago = start_chicago + timedelta(days=1) - timedelta(seconds=1)
        return {
            "start": start_chicago.strftime("%Y-%m-%d %H:%M:%S"),
            "end": end_chicago.strftime("%Y-%m-%d %H:%M:%S"),
        }
    
    # 可以扩展：今天、上周、上月...
    return None

用法

def execute(self, user_query: str, context: dict):
    # 1. 解析时间
    time_range = parse_relative_date(user_query)
    
    # 2. 生成 SQL
    if time_range:
        prompt = f"""
用户问题：{user_query}

表：dw_waybill_detail
字段：waybill_no, site_name, sign_count, total_count, org_code, waybill_time

注意：waybill_time 字段存的是芝加哥时间。
用户说的"昨天"指北京时间。
北京时间 {time_range['start']} 对应芝加哥时间。
北京时间 {time_range['end']} 对应芝加哥时间。

生成 SQL，使用芝加哥时间过滤。
"""
    else:
        prompt = f"用户问题：{user_query}\n表：dw_waybill_detail"
    
    return self.llm.generate(prompt)

---

完整 Worker 组合

class WaybillWorker:
    def execute(self, user_query: str, context: dict):
        # 1. 术语解析
        resolved_query = self.terminology.resolve(user_query)
        
        # 2. 时间解析
        time_range = self.timezone.parse(user_query)
        
        # 3. 找相似示例
        example_sql = self.examples.find(resolved_query)
        
        # 4. 拼接 prompt
        prompt = self.build_prompt(resolved_query, time_range, example_sql)
        
        # 5. LLM 生成
        sql = self.llm.generate(prompt)
        
        # 6. 后处理
        sql = self.terminology.inject(sql)
        
        # 7. org_code 注入（来自 x-user-info）
        sql = self.inject_org_code(sql, context)
        
        return sql

---

写在最后

五步按顺序来：

1. 术语词典——用户说什么，映射成什么字段/公式
2. 表管理——哪些表参与，字段叫什么、什么意思
3. 表关联——表之间怎么 JOIN
4. 示例 SQL——复杂口径直接给答案
5. 时区处理——相对时间转成数据库时区

不是一次性全塞进去，是按优先级一步步加。术语词典改动最小，从它开始。

---