在经历了无数次 Prompt 优化和三层架构（解析-校验-纠错）的失败后，我们不得不承认一个残酷的产品现实：Qwen2.5-7B 无法在 3 表及以上的多表连查任务中，提供稳定、可信的 SQL 输出。 它的常识偏见（幻觉）是神经网络级的，靠外部修补无法根除。

作为产品经理，在追求“全能智能”与保障“产品确定性”之间，我选择了后者。我们停止了对模型能力的无谓榨取，转而在 Dify 工作流中构建了一个物理级“熔断闸机。

1. 产品策略的根本转变：从“生成”转向“控制风险”

早期的产品逻辑是：用户提问 -> 模型生成 -> 尽最大努力修正 -> 输出。

• 后果：即便修正了一个错误，模型还可能在 Join 路径或聚合函数上犯错。在 3 表以上的复杂逻辑中，这种概率性的正确是不可接受的。

新的产品逻辑：

1. 意图识别：模型先输出语义 JSON。

2. 熔断检查（The Hard Gate）：Python 节点进行硬核审计。

3. 服务降级：一旦发现3表及以上连查或已知幻觉路径，立即拦截，不再生成 SQL。

4. 人工介入：给用户一个确定性的答复，并将需求抛给真正懂数据的人。

2. 核心功能：Agent中的“物理熔断器”

我们在 Agent 的 Python 节点中部署了最严格的审计逻辑。它不再试图“纠正”模型，而是直接扮演“审核员”。

import json

def main(mapping_json_raw: str):
    # --- 1. 深度清洗逻辑 ---
    mapping_json = None
    try:
        if isinstance(mapping_json_raw, str):
            # 去除 Markdown 标签（如 ```json ... ```）
            clean_str = mapping_json_raw.strip()
            if clean_str.startswith("```"):
                # 处理可能存在的多种 Markdown 格式
                lines = clean_str.splitlines()
                if lines[0].startswith("```"):
                    lines = lines[1:]
                if lines[-1].startswith("```"):
                    lines = lines[:-1]
                clean_str = "\n".join(lines).strip()
            
            mapping_json = json.loads(clean_str)
        else:
            mapping_json = mapping_json_raw
    except Exception as e:
        return {
            "is_valid": False,
            "error_msg": f"解析失败：无法将输入转化为 JSON 对象。请检查格式。内容：{str(mapping_json_raw)[:50]}...",
            "cleaned_json": {}
        }

    # --- 2. 数据库白名单 (与你的 ER 图保持一致) ---
    db_schema = {
        "tracks": ["TrackId", "Name", "AlbumId", "MediaTypeId", "GenreId", "Composer", "Milliseconds", "Bytes", "UnitPrice"],
        "artists": ["ArtistId", "Name"],
        "albums": ["AlbumId", "Title", "ArtistId"],
        "genres": ["GenreId", "Name"],
        "media_types": ["MediaTypeId", "Name"],
        "invoices": ["InvoiceId", "CustomerId", "InvoiceDate", "BillingAddress", "BillingCity", "BillingState", "BillingCountry", "BillingPostalCode", "Total"],
        "invoice_items": ["InvoiceLineId", "InvoiceId", "TrackId", "UnitPrice", "Quantity"],
        "customers": ["CustomerId", "FirstName", "LastName", "Company", "Address", "City", "State", "Country", "PostalCode", "Phone", "Fax", "Email", "SupportRepId"],
        "employees": ["EmployeeId", "LastName", "FirstName", "Title", "ReportsTo", "BirthDate", "HireDate", "Address", "City", "State", "Country", "PostalCode", "Phone", "Fax", "Email"],
        "playlists": ["PlaylistId", "Name"],
        "playlist_track": ["PlaylistId", "TrackId"]
    }
    
    errors = []
    
    # --- 3. 核心校验逻辑 ---
    # 校验 columns 里的字段
    for col in mapping_json.get("mapping_analysis", {}).get("columns", []):
        if '.' not in col:
            errors.append(f"格式错误：'{col}' 必须包含表名（如 artists.Name）")
            continue
        table, field = col.split('.')
        if table not in db_schema or field not in db_schema[table]:
            errors.append(f"幻觉错误：表 '{table}' 中不存在字段 '{field}'")

    # 校验 joins 里的关联路径
    for join in mapping_json.get("mapping_analysis", {}).get("joins", []):
        for side in ["left", "right"]:
            col_full = join.get(side, "")
            if '.' not in col_full:
                errors.append(f"关联路径格式错误：'{col_full}' 缺少表名前缀")
                continue
            table, field = col_full.split('.')
            if table not in db_schema or field not in db_schema[table]:
                errors.append(f"幻觉错误：在关联逻辑中使用了不存在的字段 '{col_full}'")

    # --- 4. 返回结果 ---
    return {
        "is_valid": len(errors) == 0,
        "error_msg": "; ".join(errors) if errors else "",
        "cleaned_json": mapping_json  # 将解析成功的字典对象传给下游
    }

3. 用户体验（UX）的降维设计

当熔断触发时，用户看到的不再是一个不知所云的错误 SQL，而是一个透明、坦诚的服务降级通知：

⚠️ 温馨提示： 您的问题涉及多张表的复杂关联。受限于本地小模型的能力上限，为了确保数据准确性，系统已自动拦截生成。 该需求已同步转发给数据分析师进行人工处理，预计 15 分钟内给您答复。

PM 视角： 这虽然不是“全自动”的 AI，但它是“全透明”的工具。通过引入人工兜底，我们守住了产品的可靠性底线。

4. 总结：Qwen-7B 本地化部署的终极形态

【实战 04】让我们明白：不要让一个 7B 的模型去干 70B 的活。

在数据隐私必须本地化的前提下，我们的产品路径必须进化：

1. 承认模型边界：Qwen2.5-7B 的“舒适区”是简单报表和双表关联。

2. 场景收敛：通过 Python 节点强制限制使用范围。

3. 熔断与降级：幻觉消除不了，就拦截需求，转人工处理。

终极产品架构总图：基于 Qwen2.5-7B的Text2SQL Agent

这就是我们最终形成的，能够在不完美组件下交付确定性结果的工业级架构。

1. 物理隔离：智能解析层（LLM）与数据执行层（DB）之间，横亘着一层不依赖 AI 的物理风控层（Python）。

2. 不信任原则：系统默认不信任 LLM 的多表推理能力，凡是 3 表以上的复杂需求，一律在“物理层面”熔断。

3. 确定性路径：只有经过 Python 校验通过的简单（1-2表）JSON，才会被喂给下一个 LLM 节点生成 SQL。这保证了进入 DB 的每一个 SQL 都是结构严谨、无幻觉的。

4. 业务闭环：通过熔断后的“人工接管”，我们守住了 100% 交付需求的商业承诺。