问题：传统 RAG 在复杂企业查询中表现不佳，导致答案不完整或不可靠

解决方案：Structured RAG (S-RAG)
  └─ 将非结构化文档转换为结构化、查询感知的表示
  └─ 实现精确、可审计的推理

关键创新：混合架构
  └─ 结合结构化检索 + 嵌入式检索
  └─ 准确率提升高达 60%
  └─ 召回率接近完美

企业级特性：
  ✓ 自动推断或用户自定义 Schema
  ✓ 处理数百万文档
  ✓ 完全透明和可控
  ✓ 适用于合规、报告和关键业务流程

Preparing diagram...

❌ 失败的查询示例:

1. "去年所有子公司中最大的资本支出是多少？"
2. "按准时交付率排名的前五名供应商是谁？"
3. "安全事故如何按季度趋势变化？"

Preparing diagram...

❌ 失败的查询示例:

1. "列出所有 2025 年前到期且罚金条款超过 100 万美元的合同"
2. "哪些员工的认证将在今年到期？"
3. "显示所有监管变化影响报告要求的市场"

问题: 遗漏任何一项不是不便，而是合规风险

传统 RAG 的局限:
  ├─ 本质上是概率性的
  ├─ 优化为找到"最匹配"的块
  ├─ 只获取基于相似度评分的文档子集
  └─ 永远无法保证完全检索

示例: 财务分析查询

问题:
  大多数报告看起来几乎相同
  包含相似术语："总负债"、"股东权益"、"净收入"

对嵌入器来说:
  ├─ 所有文档都"看起来"相关
  ├─ 语义检索表面上相似的文档
  └─ 但实际上不回答问题

结果:
  ❌ 检索变得嘈杂
  ❌ 不相关但"接近"的段落挤掉真正的答案

Preparing diagram...

# 示例: 财务报告

检测到的模式:
{
    "schema_name": "财务报告",
    "attributes": [
        {
            "name": "company_name",
            "type": "string",
            "description": "公司名称"
        },
        {
            "name": "fiscal_year",
            "type": "integer",
            "description": "财年"
        },
        {
            "name": "revenue",
            "type": "decimal",
            "description": "收入（百万美元）",
            "unit": "million_usd"
        },
        {
            "name": "operating_expenses",
            "type": "decimal",
            "description": "运营费用（百万美元）",
            "unit": "million_usd"
        },
        {
            "name": "current_liabilities",
            "type": "decimal",
            "description": "流动负债（百万美元）",
            "unit": "million_usd"
        }
    ]
}

文档中的不同格式:
  - "1,000,000"
  - "1M"
  - "1"
  - "一百万"

统一后的格式:
  → 1000000 (decimal)
  → 单位: million_usd

原始文档 (非结构化):
┌─────────────────────────────────────────┐
│ Netflix Inc. - Annual Report 2017       │
│                                         │
│ Financial Highlights:                   │
│ Total Revenue: $11.7 billion            │
│ Operating Expenses: $9.1B               │
│ Current Liabilities: $4,538 million     │
│ ...                                     │
└─────────────────────────────────────────┘

↓ 结构化提取

结构化记录:
┌─────────────────────────────────────────┐
│ company_name        | Netflix           │
│ fiscal_year         | 2017              │
│ revenue             | 11700.0 (million) │
│ operating_expenses  | 9100.0 (million)  │
│ current_liabilities | 4538.0 (million)  │
│ source_doc_id       | doc_12345         │
│ source_page         | 3                 │
└─────────────────────────────────────────┘

1. 检索包含 "Netflix" 和 "2017" 和 "流动负债" 的文本块
2. 希望 LLM 能从文本中提取正确数字
3. LLM 可能会:
   - 提取错误的数字
   - 混淆不同年份的数据
   - 返回不同单位的值

SELECT
    current_liabilities / 1000000 AS current_liabilities_millions
FROM
    SEC_Report
WHERE
    LOWER(company_name) = 'netflix'
    AND fiscal_year = 2017

精确值: 4,538 百万美元
来源: SEC Report, Page 3, Line 47
置信度: 100%

1. 检索提到 "ARR" 和 "2023" 的文档
2. 传递给 LLM
3. LLM 尝试从文本中找到最大值

结果:
  ❌ 可能遗漏某些子公司
  ❌ 可能混淆不同指标
  ❌ 无法验证答案的完整性

SELECT
    subsidiary_name,
    MAX(arr) AS max_arr
FROM
    Financial_Data
WHERE
    fiscal_year = 2023
GROUP BY
    subsidiary_name
ORDER BY
    max_arr DESC
LIMIT 1

结果:
  ✅ 精确值: $125.3M (Subsidiary Alpha)
  ✅ 保证检查了所有子公司
  ✅ 可审计: SQL 查询可追溯

检索: 相似度最高的 10 个合同
问题: 如果有 15 个符合条件的合同怎么办？

结果:
  ❌ 遗漏 5 个合同
  ❌ 合规风险
  ❌ 无法保证完整性

SELECT
    contract_id,
    contract_name,
    signing_date,
    arbitration_clause
FROM
    Contracts
WHERE
    signing_date > '2020-01-01'
    AND arbitration_clause = TRUE
ORDER BY
    signing_date DESC

结果:
  ✅ 返回所有 15 个符合条件的合同
  ✅ 保证 100% 召回率
  ✅ 合规安全

并非所有信息都适合结构化

场景 1: 结构化属性
  → "2023 年的总收入是多少？"
  → 完美适合 SQL

场景 2: 叙述性文本
  → "公司对 ESG 战略的愿景是什么？"
  → 需要语义理解

场景 3: 混合查询
  → "2023 年收入超过 1 亿的公司中，哪些强调可持续性？"
  → 需要结构化过滤 + 语义检索

Preparing diagram...

-- 先用 SQL 缩小范围
SELECT
    company_id,
    company_name,
    revenue
FROM
    Financial_Data
WHERE
    fiscal_year = 2023
    AND revenue > 50000000

# 在这 45 家公司的报告中进行语义搜索
filtered_docs = get_documents(company_ids=sql_results)

# 向量检索
relevant_mentions = semantic_search(
    query="AI strategy artificial intelligence",
    documents=filtered_docs,
    top_k=10
)

测试集: Aggregative Questions Dataset

对比系统:
  - Embedder-based RAG + GPT-o3
  - OpenAI Responses API
  - Structured RAG (AI21 Maestro)

结果:
  传统 RAG:     40% 准确率
  OpenAI API:   52% 准确率
  S-RAG:        ⭐ 95% 准确率

提升: +60% vs 传统 RAG

准确率对比 (Aggregative Questions)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

传统 RAG        ████████              40%
OpenAI API      █████████████         52%
S-RAG           ███████████████████   95%
                ↑
                接近完美

测试: 穷尽覆盖查询

传统 RAG 召回率:
  - 平均: 65%
  - 问题: 遗漏 35% 的相关数据

Structured RAG 召回率:
  - 平均: 98%
  - 给定正确 Schema: 接近 100%

召回率 (Exhaustive Coverage)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━

传统 RAG        █████████████         65%
                ↓ 
                遗漏 35% 数据 (高风险)

S-RAG           ███████████████████   98%
                ↑
                几乎完美召回

数据集: FinanceBench (大型财务分析基准)

系统对比:
  1. Classic Embedder-based RAG
  2. OpenAI Responses API
  3. Maestro Hybrid RAG (S-RAG + Embedder)

结果:
  Traditional:    58%
  OpenAI:         71%
  Maestro Hybrid: ⭐ 89%

# 支持的文档格式
SUPPORTED_FORMATS = [
    "PDF",
    "DOCX", 
    "XLSX",
    "CSV",
    "JSON",
    "HTML"
]

# 最佳实践
documents = {
    "financial_reports": [
        "Q1_2023_Report.pdf",
        "Q2_2023_Report.pdf",
        "Annual_Report_2023.pdf"
    ],
    "contracts": [
        "Supplier_Contract_001.pdf",
        "Service_Agreement_XYZ.docx"
    ],
    "hr_records": [
        "Employee_Data.xlsx",
        "Certifications.csv"
    ]
}

from maestro import StructuredRAG

# 自动 Schema 推断
srag = StructuredRAG()

inferred_schema = srag.infer_schema(
    documents=documents["financial_reports"],
    domain="financial_reporting"  # 可选: 提供领域提示
)

print(inferred_schema)

{
    "schema_name": "FinancialReport",
    "confidence": 0.92,
    "attributes": [
        {
            "name": "company_name",
            "type": "string",
            "nullable": false,
            "examples": ["Netflix", "Apple Inc.", "Microsoft"]
        },
        {
            "name": "fiscal_year",
            "type": "integer",
            "nullable": false,
            "range": [2015, 2023]
        },
        {
            "name": "revenue",
            "type": "decimal",
            "unit": "million_usd",
            "nullable": false
        }
    ]
}

# 自定义 Schema
custom_schema = {
    "schema_name": "EmployeeCertification",
    "attributes": [
        {
            "name": "employee_id",
            "type": "string",
            "primary_key": True
        },
        {
            "name": "certification_name",
            "type": "string",
            "examples": ["PMP", "AWS Certified", "CPA"]
        },
        {
            "name": "issue_date",
            "type": "date",
            "format": "YYYY-MM-DD"
        },
        {
            "name": "expiry_date",
            "type": "date",
            "format": "YYYY-MM-DD",
            "validation": "expiry_date > issue_date"
        },
        {
            "name": "status",
            "type": "enum",
            "values": ["active", "expired", "revoked"]
        }
    ],
    "indexes": ["employee_id", "expiry_date"]
}

srag.define_schema(custom_schema)

# 批量摄入文档
ingestion_result = srag.ingest(
    documents=documents["financial_reports"],
    schema=inferred_schema,
    
    # 配置选项
    options={
        "batch_size": 100,
        "validation": True,  # 验证提取的数据
        "error_handling": "skip",  # "skip" 或 "fail"
        "preserve_links": True,  # 保留到原始文档的链接
        "progress_callback": log_progress
    }
)

print(f"成功摄入: {ingestion_result['success_count']} 文档")
print(f"失败: {ingestion_result['failed_count']} 文档")
print(f"提取的记录: {ingestion_result['total_records']}")

Preparing diagram...

# 示例 1: 精确数值查询
result = srag.query(
    question="Netflix 在 2023 年的总收入是多少？"
)

print(result)

{
    "answer": "Netflix 在 2023 年的总收入为 316.9 亿美元。",
    "confidence": 0.98,
    "sql_query": "SELECT revenue FROM FinancialReport WHERE company_name = 'Netflix' AND fiscal_year = 2023",
    "result_data": [{"revenue": 31690.0}],
    "source": {
        "document": "Netflix_Annual_Report_2023.pdf",
        "page": 12,
        "line": 45
    }
}

# 示例 2: 聚合计算
result = srag.query(
    question="2023 年所有科技公司的平均收入是多少？"
)

print(result)

SELECT 
    AVG(revenue) AS avg_revenue,
    COUNT(*) AS company_count
FROM 
    FinancialReport
WHERE 
    fiscal_year = 2023
    AND industry = 'Technology'

{
    "answer": "2023 年科技行业共有 47 家公司，平均收入为 85.3 亿美元。",
    "confidence": 0.95,
    "result_data": [
        {
            "avg_revenue": 8530.0,
            "company_count": 47
        }
    ],
    "sources": [
        "Multiple documents (47 companies)"
    ]
}

# 示例 3: 结构化 + 语义
result = srag.query(
    question="2023 年收入超过 100 亿的公司中，哪些在年报中强调了可持续发展战略？",
    mode="hybrid"  # 启用混合检索
)

print(result)

# 步骤 1: 结构化过滤
sql_filter = """
SELECT company_id, company_name, revenue
FROM FinancialReport
WHERE fiscal_year = 2023 AND revenue > 10000
"""
# 结果: 12 家公司

# 步骤 2: 语义检索
semantic_results = search_in_documents(
    company_ids=[...],
    query="sustainability strategy ESG environmental",
    top_k=5
)
# 结果: 7 家公司提到可持续发展

# 步骤 3: 组合结果
final_answer = format_answer(
    structured_data=sql_results,
    semantic_data=semantic_results
)

# 启用详细审计日志
result = srag.query(
    question="...",
    options={
        "explain": True,  # 返回执行计划
        "audit_trail": True,  # 完整审计追踪
        "source_links": True  # 链接到原始文档
    }
)

# 审计信息
print(result["audit_trail"])

{
    "query_id": "q_12345",
    "timestamp": "2025-11-15T10:30:00Z",
    "execution_plan": {
        "mode": "structured",
        "sql_query": "SELECT ...",
        "execution_time_ms": 23,
        "rows_scanned": 1,
        "index_used": "idx_company_year"
    },
    "sources": [
        {
            "document_id": "doc_789",
            "filename": "Netflix_2023.pdf",
            "page": 12,
            "extraction_confidence": 0.98
        }
    ],
    "data_lineage": {
        "raw_value": "$31,690 million",
        "normalized_value": 31690.0,
        "unit_conversion": "million_usd",
        "validation_passed": true
    }
}

监管要求: 生成跨 50 个子公司的季度合规报告
挑战:
  - 数据分散在数百份文档中
  - 需要精确的数值聚合
  - 必须可审计
  - 零容错

人工流程:
  1. 手动打开每份报告
  2. 复制粘贴数据到 Excel
  3. 手动计算汇总
  4. 多次审查验证

问题:
  - 耗时: 2-3 天
  - 易出错
  - 不可扩展

# 一次性查询
compliance_report = srag.query("""
生成 Q3 2023 合规报告:
- 所有子公司的总收入
- 按地区的运营费用明细
- 超过监管阈值的任何负债
- 所有待决诉讼的清单
""")

# 结果:
#   - 耗时: 5 秒
#   - 准确率: 100%
#   - 完全可审计
#   - 自动链接到源文档

任务: 识别高风险供应商
标准:
  - 准时交付率 < 85%
  - 过去 12 个月有质量问题
  - 合同即将到期 (6 个月内)

# 混合查询
risky_suppliers = srag.query("""
识别满足以下条件的供应商:
1. 准时交付率低于 85%
2. 过去 12 个月有记录的质量问题
3. 合同在 2025 年 6 月前到期
4. 在最近的审计报告中提到风险因素

按风险等级排序并提供建议。
""", mode="hybrid")

# 结果包括:
#   - 结构化数据: 交付率、合同日期
#   - 语义分析: 质量问题描述、风险因素
#   - 建议: 基于多维度分析

合规要求: 确保所有员工的关键认证有效
问题:
  - 5000+ 员工
  - 数百种不同认证
  - 需要提前通知到期

# 穷尽覆盖查询
expiring_certs = srag.query("""
列出所有在未来 90 天内到期的员工认证:
- 员工姓名和 ID
- 认证名称
- 到期日期
- 部门
- 经理联系信息

按到期日期排序。
""")

# 关键优势:
#   ✓ 保证 100% 召回 (不会遗漏任何人)
#   ✓ 精确的日期计算
#   ✓ 可用于自动发送提醒邮件

任务: 尽职调查 - 分析 200+ 份合同
查找:
  - 特定条款 (仲裁、不竞争)
  - 财务承诺总额
  - 潜在风险条款

# 多步骤分析
contract_analysis = srag.multi_query([
    # 查询 1: 结构化筛选
    """找出所有包含仲裁条款且价值超过 50 万美元的合同""",
    
    # 查询 2: 语义分析
    """在这些合同中，识别任何提到 '不可抗力' 或 '终止权' 的条款""",
    
    # 查询 3: 风险评估
    """汇总所有潜在的财务风险和建议的缓解措施"""
])

# 输出:
#   - 符合条件的合同: 23 份
#   - 关键条款提取: 自动高亮
#   - 风险评分: 基于多维度分析
#   - 可追溯: 每个发现都链接到原始合同

✅ DO:
  - 从自动推断开始，然后迭代优化
  - 包含单位和格式信息
  - 添加验证规则
  - 使用描述性属性名称
  - 创建适当的索引

❌ DON'T:
  - 过度规范化 (太多小表)
  - 忽略数据质量验证
  - 硬编码特定值
  - 遗漏必填字段标记

{
    "schema_name": "SupplierPerformance",
    "attributes": [
        {
            "name": "supplier_id",
            "type": "string",
            "primary_key": True,
            "description": "Unique supplier identifier"
        },
        {
            "name": "on_time_delivery_rate",
            "type": "decimal",
            "unit": "percentage",
            "range": [0, 100],
            "description": "Percentage of deliveries on time"
        },
        {
            "name": "contract_value",
            "type": "decimal",
            "unit": "usd",
            "nullable": False,
            "description": "Total contract value in USD"
        },
        {
            "name": "last_audit_date",
            "type": "date",
            "validation": "last_audit_date <= CURRENT_DATE",
            "description": "Date of most recent audit"
        }
    ],
    "indexes": ["supplier_id", "last_audit_date"],
    "validation_rules": [
        "on_time_delivery_rate BETWEEN 0 AND 100",
        "contract_value > 0"
    ]
}

# ❌ 模糊查询
"告诉我关于收入的信息"

# ✅ 清晰查询
"2023 年 Q4 Netflix 的总收入是多少（以百万美元计）？"

# ❌ 过于复杂
"分析所有公司的财务表现并与行业平均值比较，同时考虑宏观经济因素和市场趋势"

# ✅ 分解为多个查询
queries = [
    "2023 年科技行业所有公司的平均收入增长率是多少？",
    "哪些公司的增长率高于行业平均水平？",
    "这些高增长公司的共同特征是什么？"
]

result = srag.query(
    question="列出前 10 名供应商",
    output_format={
        "type": "table",
        "columns": ["supplier_name", "delivery_rate", "contract_value"],
        "sort_by": "delivery_rate",
        "sort_order": "desc"
    }
)

# 为常用查询创建索引
srag.create_index(
    schema="FinancialReport",
    columns=["company_name", "fiscal_year"],
    index_type="composite"
)

# 性能提升: 10x-100x

# ❌ 逐个查询
for company in companies:
    result = srag.query(f"{company} 的 2023 年收入")

# ✅ 批量查询
result = srag.query(
    "所有公司的 2023 年收入",
    filters={"company_name": companies}
)

# 启用查询缓存
srag.configure(
    cache={
        "enabled": True,
        "ttl": 3600,  # 1 小时
        "max_size": 1000  # 最多 1000 个查询
    }
)

# 重复查询从缓存返回 (< 1ms)

摄入前:
  ✓ 文档格式检查
  ✓ 必填字段验证
  ✓ 数据类型验证
  ✓ 范围检查

摄入中:
  ✓ 实时错误检测
  ✓ 异常值标记
  ✓ 重复数据检测

摄入后:
  ✓ 完整性检查
  ✓ 一致性验证
  ✓ 审计日志记录

# 生成数据质量报告
quality_report = srag.data_quality_report(
    schema="FinancialReport"
)

print(quality_report)

{
    "total_records": 5000,
    "completeness": {
        "company_name": 100.0,
        "fiscal_year": 100.0,
        "revenue": 98.5,
        "operating_expenses": 97.2
    },
    "accuracy_issues": [
        {
            "field": "revenue",
            "issue": "Outlier detected",
            "record_id": "rec_12345",
            "suggested_action": "Manual review"
        }
    ],
    "consistency": {
        "date_formats": "pass",
        "unit_consistency": "pass",
        "cross_field_validation": "pass"
    }
}

# 基于角色的访问控制
srag.configure_access(
    role="financial_analyst",
    permissions={
        "schemas": ["FinancialReport", "Budget"],
        "operations": ["read", "query"],
        "data_filters": {
            "fiscal_year": ">=2020",  # 只能访问 2020 年后的数据
            "department": "finance"
        }
    }
)

# 启用完整审计
srag.enable_audit_logging(
    log_level="detailed",
    include=[
        "queries",
        "results",
        "data_access",
        "schema_changes"
    ],
    retention_days=365
)

# 查询审计日志
audit_logs = srag.get_audit_logs(
    start_date="2023-01-01",
    end_date="2023-12-31",
    user="analyst@company.com"
)

Preparing diagram...

# 分析现有查询日志
query_analysis = analyze_query_log(
    log_file="rag_queries_2024.log"
)

print(f"""
查询类型分布:
  - 简单事实查询: {query_analysis['simple']}%
  - 聚合查询: {query_analysis['aggregation']}%  ← 高收益
  - 对比查询: {query_analysis['comparison']}%  ← 高收益
  - 穷尽列表: {query_analysis['exhaustive']}%  ← 高收益
  
建议: {query_analysis['aggregation'] + query_analysis['exhaustive']}% 的查询
      将从 Structured RAG 获得显著改进
""")

# 选择一个高价值用例
pilot = StructuredRAGPilot(
    use_case="quarterly_financial_reporting",
    documents=finance_docs[:100],  # 从小规模开始
    target_queries=high_value_queries
)

pilot.run()
pilot.compare_with_baseline()

# 逐步扩展到更多文档和用例
rollout_plan = [
    {"phase": 1, "docs": 1000, "schemas": 2},
    {"phase": 2, "docs": 10000, "schemas": 5},
    {"phase": 3, "docs": 100000, "schemas": 10}
]

for phase in rollout_plan:
    deploy_phase(phase)
    monitor_performance()
    collect_feedback()

1. 传统 RAG 的局限性
   └─ 聚合查询失败
   └─ 无法保证穷尽覆盖
   └─ 密集语料库表现差

2. Structured RAG 的突破
   └─ 结构化表示 + SQL 查询
   └─ 准确率提升 60%
   └─ 召回率接近 100%

3. 混合架构的优势
   └─ 结合两种方法的优点
   └─ 适应不同查询类型
   └─ 企业级可靠性

4. 企业价值
   └─ 可信赖的答案
   └─ 完全可审计
   └─ 合规安全
   └─ 规模化运营