AI 回答监测报告生成任务如何做调度、快照和结果查询？

原创

用户12582597

发布于 2026-07-10 11:35:18

数据采集完了，指标也算出来了，但品牌方最终看到的是一份报告——它可能是一个 PDF 文件、一张趋势图、一份竞品对比表。报告生成看似是“把数据填进模板”，实际上涉及任务编排、数据快照锁定、诊断标签汇总和版本管理等多个工程问题。本文将围绕报告生成的调度触发、指标快照关联、诊断标签汇总、模板渲染和结果查询五个环节，拆解从聚合指标到可交付报告的完整数据链路。

一、报告生成在整体架构中的位置

在 AI 回答监测系统中，报告生成是数据链路的最后一个环节，也是品牌方直接接触的“产品表面”。它需要解决三个核心问题：

数据锚定：报告引用的指标必须对应一个确定的时间点和样本集，不能随时间或后续采集而变化。
信息完整：报告不仅要有数字，还要有诊断结论——哪些场景表现好、哪些平台存在风险。
版本可追溯：同一份报告的多个版本（初稿、复核后、最终版）需要独立存储，任何版本都能回溯。

这三个问题决定了报告生成不能是“查数据库 + 渲染模板”的简单操作，而需要独立的任务编排和数据快照锁定机制。

二、整体数据链路

链路分五个环节：任务触发、快照加载、诊断汇总、模板渲染、元数据入库。

三、报告任务生成：事件驱动 + 手动触发

3.1 触发模式

触发模式	场景	说明
自动触发	指标快照生成后自动出报告	快照完成后发送消息到报告生成队列
手动触发	运营人员选择特定快照生成报告	通过管理后台 API 提交生成请求
定时触发	周报/月报定期生成	定时器扫描最近完成的快照，批量生成

3.2 报告任务消息体

{
  "report_task_id": "rpt-uuid-xxxx",
  "snapshot_id": "SNAP_20260710_001",
  "brand_name": "品牌A",
  "report_type": "brand_weekly",
  "include_competitors": true,
  "competitor_list": ["品牌B", "品牌C"],
  "template_version": "v2.1",
  "output_format": ["pdf", "html"],
  "created_at": "2026-07-10T10:30:00Z"
}

关键字段说明：

snapshot_id：锁定数据版本，报告中的所有指标均来源于此快照及同期竞品快照。
template_version：报告模板版本号，模板迭代后旧快照仍可用旧模板重新渲染。
output_format：支持多格式输出，PDF 用于交付，HTML 用于在线预览。

四、数据锚定：快照加载与竞品关联

报告生成的第一步是加载快照数据。通过 snapshot_id 拉取主品牌的指标明细，同时查找同一时间段内竞品的最新快照：

def load_report_data(snapshot_id: str, competitor_list: list) -> dict:
    # 加载主品牌快照
    main_snapshot = db.query(
        "SELECT * FROM metric_snapshot WHERE snapshot_id = %s", snapshot_id
    )
    main_details = db.query(
        "SELECT * FROM metric_snapshot_detail WHERE snapshot_id = %s",
        snapshot_id
    )
    
    # 查找同期竞品快照（前后 24 小时内）
    snapshot_time = main_snapshot["calculated_at"]
    competitor_data = {}
    for competitor in competitor_list:
        comp_snapshot = db.query(
            "SELECT snapshot_id FROM metric_snapshot "
            "WHERE brand_name = %s AND is_partial = 0 "
            "AND calculated_at BETWEEN %s AND %s "
            "ORDER BY ABS(TIMESTAMPDIFF(SECOND, calculated_at, %s)) LIMIT 1",
            competitor,
            snapshot_time - timedelta(hours=24),
            snapshot_time + timedelta(hours=24),
            snapshot_time
        )
        if comp_snapshot:
            competitor_data[competitor] = db.query(
                "SELECT * FROM metric_snapshot_detail WHERE snapshot_id = %s",
                comp_snapshot["snapshot_id"]
            )
    
    return {
        "main": {"snapshot": main_snapshot, "details": main_details},
        "competitors": competitor_data,
        "data_lock_time": snapshot_time,
    }

时间窗口匹配（前后 24 小时）保证了竞品对比的公平性——同一时间段内的采集数据才具有可比性。

五、诊断标签汇总：从离散标签到报告结论

报告不仅展示指标数字，还需要一段可读的诊断结论，例如：“品牌在推荐场景表现优秀，但在风险判断场景存在可见度缺口，建议加强第三方评测内容建设。”

诊断结论的生成依赖诊断链路中产生的异常标签。报告生成时汇总这些标签：

-- 汇总指定快照关联的诊断标签
SELECT 
    d.target_brand,
    r.tag_type,
    r.tag_detail,
    COUNT(*) AS tag_count
FROM diagnosis_risk_tag r
JOIN diagnosis_result d ON r.task_id = d.task_id
WHERE d.snapshot_id IN (
    -- 快照关联的所有采集任务ID
    SELECT source_task_id FROM clean_result WHERE batch_id = (
        SELECT batch_id FROM metric_snapshot WHERE snapshot_id = %s
    )
)
GROUP BY d.target_brand, r.tag_type
ORDER BY tag_count DESC；

汇总结果按严重程度排序，生成报告中的“风险提示”章节：

def generate_diagnosis_summary(snapshot_id: str) -> dict:
    tags = load_diagnosis_tags(snapshot_id)
    
    summary = {
        "high_risk": [],
        "medium_risk": [],
        "low_risk": [],
        "overall_assessment": "",
    }
    
    for tag in tags:
        item = {
            "type": tag["tag_type"],
            "count": tag["tag_count"],
            "description": TAG_DESCRIPTIONS.get(tag["tag_type"], ""),
        }
        if tag["tag_type"] in ("confusion", "fact_error"):
            summary["high_risk"].append(item)
        elif tag["tag_type"] in ("omission", "outdated"):
            summary["medium_risk"].append(item)
        else:
            summary["low_risk"].append(item)
    
    # 自动生成总评
    if len(summary["high_risk"]) > 0:
        summary["overall_assessment"] = "存在高风险项，建议优先处理品牌混淆和事实错误问题。"
    elif len(summary["medium_risk"]) > 0:
        summary["overall_assessment"] = "整体表现良好，部分场景存在信息缺失，建议针对性补充。"
    else:
        summary["overall_assessment"] = "各项指标正常，品牌在AI回答中呈现状态稳定。"
    
    return summary

TAG_DESCRIPTIONS 是标签类型到可读描述的映射表，配置在模板变量中，运营人员可随时调整措辞。

六、模板渲染：数据与展示分离

6.1 模板管理表

不同报告类型使用不同的模板，模板版本化管理：

CREATE TABLE report_template (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    template_code VARCHAR(50) NOT NULL COMMENT '模板代码: brand_weekly/competition_monthly',
    version VARCHAR(10) NOT NULL COMMENT '版本号: v2.1',
    template_content LONGTEXT COMMENT '模板内容(HTML/Jinja2)',
    css_content TEXT COMMENT '样式表',
    is_active TINYINT(1) DEFAULT 1,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_code_version (template_code, version)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

6.2 渲染引擎

使用 Jinja2 模板引擎将数据填入模板：

from jinja2 import Template

def render_report(report_data: dict, template_content: str, css_content: str) -> str:
    template = Template(template_content)
    html = template.render(
        brand=report_data["main"],
        competitors=report_data["competitors"],
        diagnosis=report_data["diagnosis"],
        generated_at=report_data["generated_at"],
        snapshot_id=report_data["snapshot_id"],
    )
    # 注入 CSS
    full_html = f"<html><head><style>{css_content}</style></head><body>{html}</body></html>"
    return full_html

模板中通过变量控制展示逻辑，示例片段：

<!-- 场景指标表格 -->
<table>
  <tr><th>场景</th><th>提及率</th><th>推荐率</th><th>趋势</th></tr>
  {% for detail in brand.details %}
  <tr>
    <td>{{ detail.scene_type }}</td>
    <td>{{ detail.mention_rate }}%</td>
    <td>{{ detail.recommend_rate }}%</td>
    <td class="{% if detail.recommend_rate > 50 %}trend-up{% else %}trend-down{% endif %}">
      {{ detail.recommend_rate }}%
    </td>
  </tr>
  {% endfor %}
</table>

<!-- 诊断结论 -->
<div class="diagnosis">
  <h3>诊断结论</h3>
  <p>{{ diagnosis.overall_assessment }}</p>
  {% if diagnosis.high_risk %}
    <div class="alert-high">
      <strong>高风险项：</strong>
      {% for risk in diagnosis.high_risk %}
        <li>{{ risk.description }} (出现 {{ risk.count }} 次)</li>
      {% endfor %}
    </div>
  {% endif %}
</div>

6.3 PDF 生成

HTML 渲染完成后，通过无头浏览器生成 PDF：

from playwright.async_api import async_playwright

async def generate_pdf(html_content: str, output_path: str):
    async with async_playwright() as p:
        browser = await p.chromium.launch()
        page = await browser.new_page()
        await page.set_content(html_content)
        await page.pdf(path=output_path, format="A4", print_background=True)
        await browser.close()

PDF 生成后上传到 COS，获得永久访问链接。

七、报告元数据与版本管理

每次生成都写入报告记录表，支持版本追溯：

CREATE TABLE report_record (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    report_id VARCHAR(64) NOT NULL UNIQUE COMMENT '报告ID: RPT_20260710_001',
    snapshot_id VARCHAR(64) NOT NULL COMMENT '关联的指标快照',
    brand_name VARCHAR(100) NOT NULL,
    report_type VARCHAR(30) NOT NULL,
    template_version VARCHAR(10),
    
    report_status VARCHAR(20) DEFAULT 'draft' COMMENT 'draft/published/archived',
    pdf_url VARCHAR(512) COMMENT 'PDF文件COS链接',
    html_url VARCHAR(512) COMMENT 'HTML预览链接',
    
    data_lock_time TIMESTAMP COMMENT '数据锚定时间',
    generated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    published_at TIMESTAMP NULL,
    
    INDEX idx_brand_type (brand_name, report_type),
    INDEX idx_snapshot (snapshot_id)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

report_status 流转路径：

八、查询接口设计

8.1 品牌最新报告

SELECT report_id, report_type, report_status, pdf_url, html_url, generated_at
FROM report_record
WHERE brand_name = '品牌A'
  AND report_status = 'published'
ORDER BY generated_at DESC
LIMIT 1；

8.2 报告历史列表

SELECT report_id, report_type, report_status, snapshot_id, generated_at
FROM report_record
WHERE brand_name = '品牌A'
ORDER BY generated_at DESC
LIMIT 20；

8.3 通过快照反查报告

SELECT report_id, report_type, report_status, pdf_url
FROM report_record
WHERE snapshot_id = 'SNAP_20260710_001'；

九、工程实践要点

1. 快照关联保证数据一致性

报告中的所有数据必须来源于同一个快照。不要在渲染时实时查询样本库——样本库可能在报告生成过程中被新数据更新，导致报告中不同页面的指标口径不一致。

2. 模板版本与快照解耦

模板会迭代（调整布局、增加新图表），但旧快照的数据结构不变。模板版本号记录在报告元数据中，重新渲染时可以选择“用新版模板渲染旧快照”或“用旧版模板复现历史报告”。

3. PDF 生成要做超时保护

无头浏览器渲染复杂图表时可能超时。建议在云函数中设置 PDF 生成超时时间（如 30 秒），超时后降级为纯 HTML 输出，并标记报告状态为 partial，后续异步补生成 PDF。

十、结语

报告生成是 AI 回答监测系统面向品牌方的最终交付环节。快照锁定保证了数据一致性，诊断标签汇总让报告不仅有数字还有结论，模板与数据分离让报告样式可独立迭代，版本管理让每一次交付都有据可查。

这套方案已在多个消费品牌和企业服务的定期报告自动生成中实际运行，支持 PDF/HTML 双格式输出和竞品对比报告自动编排。开发者可在此基础上扩展图表组件（如雷达图、趋势折线图）、增加邮件自动分发或接入企业微信推送。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

人工智能

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

人工智能

登录后参与评论

0 条评论

热度