为什么没达标？Snowflake CoWork 生成根因 PDF

分析师过去从多个系统提取数据，制作 PowerPoint 并邮件发送其多个版本。现在，一个问题就能返回一份生产级 RCA PDF，全部在 Snowflake 内完成，无需 SPCS，也无需 Docker。

制造工厂的每次交接班都以同样的问题开始：我们为什么没达标，真正的原因是什么，下一个班次能否补回来？

一条 24/7 运转的产线未达日产量目标，需要的不是一句“设备问题”的耸肩了事——它需要的是一座量化的差异归因桥：主要原因、以件数计的影响、哪些可挽回、首先该采取什么行动。

如今，这需要一名分析师和一两名审核人员花 3–6 小时，从四个源系统提取数据，并组装一份 PowerPoint，通常在中午前后发出前还要经过一两次修订。到那时，下一个班次已经毫无头绪地开始了。

我们构建了不同的东西：一次存储过程调用即可查询数据、计算根因分析、渲染带样式的 HTML 报告、将其转换为精美的 PDF，并返回下载链接——完全在 Snowflake 内完成。

现在，班次主管或工厂经理只需在 CoWork（Snowflake 面向知识工作者的个人智能体，前身为 Snowflake Intelligence）中提问，就能在 5 分钟内拿到一份完成的 PDF，及时在交接班开始前而非开始后向接班人员做简报。这使分析师从手工制作报告中解放出来，专注于更高价值的工作：强化底层数据，并推动该系统在全厂的采用。

以下是我们如何构建它、过程中遇到的死胡同，以及让它奏效的出乎意料的简单修复。

GitHub 仓库：完整代码——引擎、示例数据生成器和 DDL——可在 仓库中获取。克隆它，将其指向你自己的视图，并根据你的领域调整原因分桶。

为什么我们从规范而非代码开始

在写一行代码之前，我们使用结构化需求定义了系统必须做什么。这不是为了仪式而仪式——而是为了让技术选择可测试，而不是寄希望于它们能成功。我在之前的文章 Intent-Driven Development with CoCo Desktop 中详细介绍了这种方法；以下是它如何塑造了本次构建。

以下是最终塑造了架构的需求：

代表性的 EARS 需求

REQ-01：核对产量缺口

类型：事件驱动

陈述：当未达到日目标时，系统应生成一个差异归因桥，其损失条之和恰好等于实际与目标之间的差距，以便报告对产量缺口提供经过核对的解释。

REQ-02：保留诊断窗口和趋势窗口

类型：普遍性

陈述：系统应维护独立的 RCA 窗口和数据窗口，将 RCA 窗口用于诊断日，将数据窗口用于至少 21 天的历史上下文，以便单日报告保留有意义的趋势对比。

REQ-03：生成自包含的 HTML 报告

类型：普遍性

陈述：系统应生成一份自包含的 HTML 报告，内联 CSS 并嵌入图表资源，以便报告在无网络访问、无 JavaScript 渲染或无外部 CDN 资源的情况下仍可查看和打印。

REQ-04：在 PDF 输出中保留报告格式

类型：事件驱动

陈述：当生成 HTML 报告时，系统应生成一份保留报告布局、样式、表格和嵌入图表的 PDF，以便输出适合邮件分发和打印。

该规范让每个技术决策都成为通过/失败测试：这个库在转换为 PDF 时是否保留 CSS？我们可以根据同一需求评估 fpdf2、xhtml2pdf 和 WeasyPrint，而不是猜测哪个可行。

它们如何协同工作

一切都在标准 Snowflake 仓库上的单个存储过程内运行。没有 SPCS 容器，没有外部服务，没有 Docker。

数据长什么样

在讨论表之前，先明确正在诊断什么会有所帮助。问题很窄且反复出现：一条生产线，或整个工厂，未达到日目标。差了多少，为什么？目标是当日计划生产的单位数量，实际是实际生产的数量，两者之间的差距就是报告必须解释的缺口，每一笔损失都可追溯到实际记录的停机。

手工完成时，这种解释之所以缓慢，原因与分析关系不大：

• 证据存在于不同地方。产线级状态、设备级状态、利用率 KPI 以及断路器和输送机等配套设备各自位于自己的数据集。分析师必须先提取并对其，然后才能查看任何内容。
• 停机是编码的，而非解释的。原因以 NO MATERIAL FLOW、CONVEYOR FULL 或 HYDRAULIC LEAK 等代码形式出现。必须先将其归入物料搬运、设备可靠性和人员/操作员等原因，然后才能对主管有意义。
• 小时不是件数。损失小时只有在按产线实际运行时所达到的速率计价后，才会变成损失产量，而且各部分必须加总为真实差距。粗略估计很少能做到，而损失条无法核对的报告会失去可信度。
• 一天需要上下文。没有前几周，单日未达标意义不大，因此分析师还需要近期历史来进行趋势对比。

这些都不是困难的分析。这是组装工作，这正是为什么它可以交给一个存储过程。

考虑到这一点，以下是 RCA 引擎所期望的：代表生产运营的六张表：

该引擎与表无关——它使用以逻辑名称为键的 DataFrame 字典 frames，因此适用于任何提供此形态的数据源。

RCA 计算引擎

核心算法归结为四个步骤：

• 计划小时= 运行 + 停机 + 待机 + 延误（来自设备状态）
• 实际速率= 实际件数 / 运行小时
• 各原因原始损失= 损失小时（按原因分桶）× 实际速率
• 核对= 缩放所有原始损失，使其总和恰好等于（实际 − 目标）

# Simplified bridge calculation
run_h = running_hours.sum()
lost = status_df[status_df["STATUS"].isin(["Down", "Standby", "Delay"])]
lost["CATEGORY"] = lost["REASON"].apply(bucket_reason)

r_run = actual_tonnes / run_h  # achieved rate

raw = {}
for cat, hrs in lost.groupby("CATEGORY")["HRS"].sum().items():
    raw[cat] = hrs * r_run  # gross opportunity lost

# Reconcile: scale so bars sum to the net gap
gap = actual_tonnes - target_tonnes
total_raw = sum(raw.values())
bars = {cat: val * (gap / total_raw) for cat, val in raw.items()}
# bars now sums EXACTLY to gap

对账这一步是最关键的部分。如果没有它，你的瀑布图柱形就会偏离实际差距，报告讲述的故事也就对不上账。按比例缩放每个类别，既能让每根柱形保持真实，又能保证它们落在正确的总量上。

图表使用 matplotlib 渲染，并以 base64 形式嵌入，因此报告完全自包含：

def chart_to_base64(fig):
    buf = io.BytesIO()
    fig.savefig(buf, format="png", dpi=110, bbox_inches="tight")
    plt.close(fig)
    return base64.b64encode(buf.getvalue()).decode("ascii")

构建 HTML 报告

报告是一个单一的自包含 HTML 文件：内联 <style> 块、base64 图表图像，无外部依赖。

def build_html(R, cfg):
    # R contains all computed results: bars, stats, charts, actions
    waterfall_b64 = chart_waterfall(R)
    pareto_b64 = chart_pareto(R)

    return f"""<!doctype html><html><head>
    <style>
      body {{ font-family: -apple-system, sans-serif; max-width: 980px; margin: 24px auto; }}
      .kpis {{ display: flex; flex-wrap: wrap; gap: 10px; }}
      .kpi {{ flex: 1 1 150px; border: 1px solid #dfe6ee; border-radius: 6px; padding: 9px; }}
      .kpi .v {{ font-size: 1.3rem; font-weight: 700; }}
      .neg {{ color: #c0504d; }}
      @media print {{ body {{ -webkit-print-color-adjust: exact; }} }}
    </style></head><body>
    <h1>{headline}</h1>
    ...
    <img src='data:image/png;base64,{waterfall_b64}'/>
    ...
    </body></html>"""

@media print 规则和 @page 指令确保 HTML 能够干净地打印——而且关键在于，它们告诉 WeasyPrint 如何精确地对 PDF 进行分页。

这并不容易。

失败之处

• fpdf2（Snowflake Anaconda 中可用）有一个 write_html() 方法，但它只能处理基础 HTML。用于 KPI 卡片的 CSS flexbox？被渲染成了竖线字符。嵌套表格？直接崩溃。Base64 图像？能渲染，但表格被破坏了。312 KB 的 HTML 变成了 281 KB 的 PDF，看起来与原版毫无相似之处。
• xhtml2pdf（可通过 PyPI 获取）成功加载并产生了输出——但它只支持 CSS 2.1。KPI flexbox 卡片、彩色背景和现代布局全部退化为纯文本和以竖线分隔的伪表格。323 KB 输入只产出 70 KB 输出，这说明了问题：大部分内容根本没有渲染。
• SPCS + 无头 Chromium可以产生像素级完美的输出（它是真正的浏览器），但它需要计算池、Docker 镜像、镜像仓库、服务生命周期管理——仅仅为了一个 PDF 转换步骤就要投入沉重的基础设施。

成功之处：通过 PyPI 构件仓库使用 WeasyPrint

Snowflake 的 PyPI 构件仓库让你可以在存储过程中安装 任何 PyPI 包——而不只是 Anaconda 频道中已有的包：

CREATE OR REPLACE PROCEDURE MY_PROC(...)
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
ARTIFACT_REPOSITORY = snowflake.snowpark.pypi_shared_repository
PACKAGES = ('weasyprint')

WeasyPrint 是一个完整的 CSS 渲染引擎（支持 flexbox、grid、@media print、颜色、base64 图像）。我们原本确实以为它会失败，因为它需要 pango/cairo 原生库——但 它加载并运行都毫无问题：

from weasyprint import HTML as WeasyHTML

pdf_bytes = WeasyHTML(string=html_content).write_pdf()

输出：从 327 KB 的 HTML 生成 303 KB 的 PDF——图表已嵌入，CSS 格式已保留，表格结构正确。

完整的存储过程

下面是把一切串起来的完整过程——拉取数据、计算 RCA、渲染 HTML、转换为 PDF，并返回预签名下载链接：

CREATE OR REPLACE PROCEDURE WRITE_RCA_REPORT(
    "SCOPE"         STRING,
    "LINE"          STRING,
    "START_DATE"    DATE    DEFAULT NULL,
    "END_DATE"      DATE    DEFAULT NULL,
    "LOOKBACK_DAYS" INTEGER DEFAULT 21
)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = '3.11'
ARTIFACT_REPOSITORY = snowflake.snowpark.pypi_shared_repository
PACKAGES = ('snowflake-snowpark-python', 'pandas', 'pyarrow', 'matplotlib', 'weasyprint')
IMPORTS = ('@MY_STAGE/scripts/rca_engine.py')
HANDLER = 'run'
COMMENT = 'Compute an RCA variance bridge, render HTML + PDF reports, return PDF presigned URL.'
EXECUTE AS OWNER
AS
$$
import io
import os
import datetime as dt
import rca_engine as eng

VIEWS = {
    "line_performance":       "V_LINE_PERFORMANCE",
    "line_status":            "V_LINE_STATUS",
    "equipment_performance":  "V_EQUIPMENT_PERFORMANCE",
    "equipment_status":       "V_EQUIPMENT_STATUS",
}


def run(session, scope, line, start_date, end_date, lookback_days):
    # Resolve end date
    if end_date is None:
        pp0 = session.table(VIEWS["line_performance"]).to_pandas()
        end_date = max(eng._to_date(pp0["SHIFT_DATE"]).dropna())

    lb = max(int(lookback_days) if lookback_days is not None else 21, 21)
    data_start = end_date - dt.timedelta(days=lb)
    rca_start = start_date if start_date is not None else end_date

    # Build output path
    tag = line if (scope == "line" and line) else "plant"
    base_name = f"rca_{tag}_{end_date}"

    # Pull data (full history window for trend context)
    frames = {}
    for name, view in VIEWS.items():
        df = session.table(view).to_pandas()
        if "SHIFT_DATE" in df.columns:
            d = eng._to_date(df["SHIFT_DATE"])
            df = df[(d >= data_start) & (d <= end_date)]
        frames[name] = df

    # Compute RCA + render HTML
    html, stats = eng.build_report_html(frames, scope, line, rca_start, end_date)

    # Write HTML to stage
    html_path = f"reports/{base_name}/{base_name}.html"
    session.file.put_stream(
        io.BytesIO(html.encode("utf-8")),
        f"@MY_STAGE/{html_path}",
        auto_compress=False, overwrite=True
    )

    # --- PDF generation (WeasyPrint via PyPI Artifact Repository) ---
    from weasyprint import HTML as WeasyHTML

    pdf_bytes = WeasyHTML(string=html).write_pdf()
    if not pdf_bytes or len(pdf_bytes) < 100:
        raise RuntimeError(f"WeasyPrint produced empty/invalid PDF ({len(pdf_bytes) if pdf_bytes else 0} bytes)")

    pdf_path = f"reports/{base_name}/{base_name}.pdf"

    # Use put_stream to upload. auto_compress=False prevents gzip wrapping.
    # If the stage has AUTO_COMPRESS=TRUE, the .pdf extension is excluded from
    # automatic compression by Snowflake (binary formats are skipped).
    session.file.put_stream(
        io.BytesIO(pdf_bytes),
        f"@MY_STAGE/{pdf_path}",
        auto_compress=False,
        overwrite=True
    )

    # Refresh directory so GET_PRESIGNED_URL can resolve the new file
    session.sql(
        "ALTER STAGE MY_STAGE REFRESH"
        f" SUBPATH = 'reports/{base_name}'"
    ).collect()

    # Presigned URL for the PDF (valid for 1 hour)
    pdf_url = session.sql(
        f"SELECT GET_PRESIGNED_URL('@MY_STAGE', '{pdf_path}', 3600)"
    ).collect()[0][0]

    return {
        "presigned_url": pdf_url,
        "path": f"@MY_STAGE/{pdf_path}",
        "pdf_bytes": len(pdf_bytes),
        "html_bytes": stats["bytes"],
        "charts": stats["charts"],
        "mode": stats["mode"],
        "target": stats["target"],
        "actual": stats["actual"],
        "gap": stats["gap"],
        "rca_start": str(rca_start),
        "rca_end": str(end_date),
    }
$$;

关键陷阱： 从 Anaconda 切换到 PyPI 时，你必须在 packages 列表中显式包含 pyarrow——它是 .to_pandas() 所需的，但不像在 Anaconda 频道中那样会被自动解析。

将其包装进 Cortex Agent

存储过程很有用，但自然语言界面更好。借助 Snowflake Cortex Agents，值班主管可以直接问 “我们昨天为什么没达成目标？” 就能拿回 PDF 链接，完全不需要了解过程调用或 SQL。

该代理只需要内联 INSTRUCTIONS，告诉它何时以及如何调用该过程：

CREATE OR REPLACE AGENT PRODUCTION_RCA_ANALYST
  COMMENT = 'Diagnoses missed production targets and delivers a formatted PDF RCA report.'
  PROFILE = '{"display_name": "Production RCA Analyst"}'
  FROM SPECIFICATION $$
models:
  orchestration: auto
instructions:
  system: >
    You are a Production RCA Analyst. When a user asks why production missed target,
    call WRITE_RCA_REPORT with the appropriate scope (plant or line), line name, and date.
    The procedure returns a presigned URL to a PDF report. Present the URL as a download
    link and tell the user to download and open it in their browser or PDF viewer.
    Do NOT attempt to summarize the report content — the PDF is the deliverable.
    If the user does not specify a date, use yesterday. If they name a specific line,
    set scope to line and pass the line name; otherwise use scope plant with line NULL.
tools:
  - tool_spec:
      type: generic
      name: WRITE_RCA_REPORT
      description: "Generates a PDF Root Cause Analysis report for production shortfalls. Returns a presigned URL to the PDF."
      input_schema:
        type: object
        properties:
          SCOPE:
            type: string
            description: "Analysis scope: 'plant' for entire plant or 'line' for a specific line"
          LINE:
            type: string
            description: "Line name (e.g. LINE_A). Required when scope is 'line', NULL when scope is 'plant'"
          START_DATE:
            type: string
            description: "Start date in YYYY-MM-DD format. Defaults to yesterday if omitted"
          END_DATE:
            type: string
            description: "End date in YYYY-MM-DD format. Defaults to START_DATE if omitted"
          LOOKBACK_DAYS:
            type: integer
            description: "Number of days of historical context for trending. Default 21"
        required:
          - SCOPE
          - LINE
tool_resources:
  WRITE_RCA_REPORT:
    type: procedure
    identifier: WRITE_RCA_REPORT
    execution_environment:
      type: warehouse
      warehouse: MY_WAREHOUSE
$$;

现在用户对话看起来是这样的：

你将得到

一次过程调用：

CALL WRITE_RCA_REPORT(‘plant’, NULL, ‘2026-07-22’, ‘2026-07-22’);

产出：

reports/rca_plant_2026–07–22/
rca_plant_2026–07–22.html (327 KB - interactive, viewable in browser)
rca_plant_2026–07–22.pdf (303 KB - printable, emailable)

PDF 保留了 HTML 源文件中的一切：

• 带彩色数值和基准区间的 Flexbox KPI 卡片
• 瀑布图和帕累托图（base64 PNG 图像）
• 带交替行颜色的样式化表格
• 带优先级徽章和行动文本的根本原因卡片
• 通过 @media print CSS 在章节边界处分页

PDF 的示例输出

提示：为 2026 年 7 月 22 日生成 RCA

我们学到了什么

PyPI 构件仓库改变了游戏规则。 以前在 Snowflake 中“不可能”的包（WeasyPrint 以及许多需要原生库的其他包）现在可以在标准存储过程中使用。无需 SPCS，无需 Docker。

规格驱动开发让 PDF 需求变得可测试。 我们可以针对同一个验收标准——“保留 CSS flexbox 和 base64 图像”——评估三个库（fpdf2、xhtml2pdf、WeasyPrint），并得到明确的通过/失败结果。

诚实的路径很重要。 我们先尝试了低成本选项（fpdf2、xhtml2pdf），记录了它们失败的原因，然后才转向 WeasyPrint。这些失败为需求提供了信息：“CSS 2.1 不够”成为了一条具体的约束。

一切都停留在 schema 层面。 无需计算池，无需镜像仓库，无需服务生命周期。一个 DDL 脚本就能将整个解决方案与现有的存储过程一起部署。

这个模式可复用。 任何生成样式化 HTML 的工作流，现在只需将 ‘weasyprint’ 添加到 packages 列表并加上三行 Python，就能获得 PDF 输出。困难的部分在于发现它可行——现在重复起来就轻而易举了。

完整源代码见 Github cortex-agent-orchestration-patterns。将视图和原因分桶调整到你的领域；方差桥接数学和 PDF 渲染适用于任何跟踪实际值与目标值并带有编码停机原因的生产运营。

本文与 [email protected] 共同撰写。本文代表作者个人观点和经验，不代表任何雇主。

👏 如果有价值就鼓掌 | 🔗 分享给你的团队 | ➕ 关注获取更多

“我们为什么没达成目标？”——问 Snowflake CoWork，获取根本原因 PDF最初发表于 Snowflake Builders Blog: Data Engineers, App Developers, AI, & Data Science 在 Medium 上，人们通过高亮和回应这个故事来继续对话。