多平台 AI 回答采集任务如何做调度、归一化和结果入库？

原创

用户12582597

发布于 2026-07-06 10:46:28

向 DeepSeek、Kimi、通义千问等多个 AI 平台提问同一个问题，拿到的回答格式各异——有的返回 Markdown，有的返回纯文本，有的嵌套在 JSON 多层结构里。想从这些异构回答中统一提取“品牌是否被提及”“是否被推荐”等指标，后端需要解决三个核心问题：如何公平调度多平台任务、如何将不同格式的回答归一化、如何将归一化后的结果结构化入库。本文将围绕这三个环节拆解完整的数据链路设计。

一、为什么多平台采集需要归一化

多平台采集与单平台采集的关键区别不在任务数量，而在输出的异构性。同一个问题“有哪些值得推荐的云数据库”，不同平台的返回格式可能完全不同：

平台	返回格式	品牌提及的表现形式
DeepSeek	Markdown 有序列表	1. 品牌A：主打xxx
Kimi	自然段落	在众多方案中，品牌A表现突出...
通义千问	JSON 嵌套结构	{"recommendations": [{"name": "品牌A"}]}

如果不做归一化，后续的提及提取、推荐判断、解释定位等逻辑就需要针对每个平台单独开发适配代码。归一化的目标是把异构输出统一为一个标准结构，让下游的诊断和分析逻辑只依赖这一种结构，从而将平台差异隔离在采集层。

二、整体数据链路

链路分五个核心环节：公平调度、平台适配、原始留档、归一化转换、结构化入库。下文逐一展开。

三、任务调度：跨平台公平采样

多平台采集的首要原则是时间窗口一致性。同一个问题对不同平台的提问应尽量在同一时间段内完成，避免因模型版本更新或联网数据变化导致对比基线漂移。

3.1 任务生成：问题 × 平台矩阵

调度器从问题库中取出一条问题后，按平台配置生成一组“同步任务”：

{
  "sync_group_id": "sync-uuid-xxxx",
  "query_text": "有哪些适合企业的云数据库方案？",
  "scene_type": "RECOMMEND",
  "target_brand": "品牌A",
  "round": 2,
  "platforms": ["deepseek", "kimi", "tongyi"],
  "created_at": "2026-07-06T10:00:00Z"
}

每个平台拆分为独立消息投递到队列，但共享同一个 sync_group_id：

{"sub_task_id": "sub-001", "sync_group_id": "sync-uuid-xxxx", "platform": "deepseek", ...}
{"sub_task_id": "sub-002", "sync_group_id": "sync-uuid-xxxx", "platform": "kimi", ...}
{"sub_task_id": "sub-003", "sync_group_id": "sync-uuid-xxxx", "platform": "tongyi", ...}

3.2 同步组完成追踪

Redis Set 记录每个同步组的完成状态：

def on_sub_task_done(sub_task: dict):
    group_id = sub_task["sync_group_id"]
    redis.sadd(f"sync_done:{group_id}", sub_task["platform"])
    
    expected = set(sub_task["expected_platforms"])
    done = redis.smembers(f"sync_done:{group_id}")
    
    if done == expected:
        redis.setex(f"sync_ready:{group_id}", 3600, "1")
        trigger_normalization(group_id)

当同一同步组的所有平台子任务都完成后，自动触发归一化流程。如果 30 分钟内仍有平台未返回，同步组标记为 partial，已完成的平台数据仍参与后续分析，缺失平台在结果中显式标注。

3.3 失败重试分层

不同平台的失败处理策略独立配置：

PLATFORM_RETRY_CONFIG = {
    "deepseek": {"max_retries": 3, "backoff": "exponential", "base_delay": 60},
    "kimi": {"max_retries": 2, "backoff": "fixed", "base_delay": 30},
    "tongyi": {"max_retries": 3, "backoff": "exponential", "base_delay": 45},
}

重试仍失败的任务进入死信队列，同步组等待超时后以 partial 状态完成，不阻塞其他平台数据的流转。

四、回答归一化：从异构到统一

这是本文的核心环节。归一化层的输入是各平台返回的原始回答（JSON/HTML/Markdown 混合），输出是一个统一的结构体。

4.1 统一回答结构设计

{
  "platform": "deepseek",
  "query_text": "有哪些适合企业的云数据库方案？",
  "raw_format": "markdown",
  "normalized": {
    "full_text": "根据您的需求，以下是几个值得考虑的方案：\n\n1. **品牌A** ...",
    "paragraphs": [
      {"index": 0, "text": "根据您的需求，以下是几个值得考虑的方案：", "type": "intro"},
      {"index": 1, "text": "1. **品牌A**：主打高性能...", "type": "list_item"},
      {"index": 2, "text": "2. **品牌B**：适合中小企业...", "type": "list_item"}
    ],
    "list_items": [
      {"index": 0, "text": "1. **品牌A**：主打高性能...", "order": 1},
      {"index": 1, "text": "2. **品牌B**：适合中小企业...", "order": 2}
    ],
    "links": [
      {"url": "https://example.com", "text": "品牌A官网", "context": "更多信息参考"}
    ]
  },
  "normalized_at": "2026-07-06T10:00:35Z"
}

字段设计原则：

full_text：始终保留完整纯文本，不做任何裁剪，是后续所有分析的基础。
paragraphs：按双换行或 Markdown 标题切分段落，标注 type（intro/list_item/conclusion/standalone）。
list_items：如果回答包含推荐列表，单独提取为结构化数组，保留原始顺序。
links：提取所有外链及其上下文，用于后续引用率判断。

4.2 分平台归一化适配器

每个平台实现一个适配器，将原始输出转换为统一结构：

class BaseNormalizer:
    def normalize(self, raw_response: dict) -> dict:
        raise NotImplementedError

class DeepSeekNormalizer(BaseNormalizer):
    def normalize(self, raw: dict) -> dict:
        text = raw["choices"][0]["message"]["content"]
        return {
            "full_text": text,
            "paragraphs": self._split_paragraphs(text),
            "list_items": self._extract_list_items(text),
            "links": self._extract_links(text),
        }

class KimiNormalizer(BaseNormalizer):
    def normalize(self, raw: dict) -> dict:
        text = raw["choices"][0]["message"]["content"]
        return {
            "full_text": text,
            "paragraphs": self._split_paragraphs(text),
            "list_items": self._extract_list_items(text),
            "links": self._extract_links(text),
        }

class TongyiNormalizer(BaseNormalizer):
    def normalize(self, raw: dict) -> dict:
        # 通义千问返回结构可能不同，需适配
        text = raw.get("output", {}).get("text", "")
        return {
            "full_text": text,
            "paragraphs": self._split_paragraphs(text),
            "list_items": self._extract_list_items(text),
            "links": self._extract_links(text),
        }

核心归一化函数——段落切分与列表提取：

def split_paragraphs(text: str) -> list:
    """按双换行或Markdown标题切分段落"""
    blocks = re.split(r'\n\n|\n#{1,3}\s', text)
    result = []
    for i, block in enumerate(blocks):
        block = block.strip()
        if not block:
            continue
        para_type = "standalone"
        if re.match(r'^\d+[\.\)]\s', block):
            para_type = "list_item"
        elif i == 0:
            para_type = "intro"
        elif i == len(blocks) - 1:
            para_type = "conclusion"
        result.append({"index": i, "text": block, "type": para_type})
    return result

def extract_list_items(text: str) -> list:
    """提取有序/无序列表项，保留顺序"""
    items = []
    for match in re.finditer(r'(?:^|\n)(?:\d+[\.\)]\s*|[-*]\s+)(.+?)(?=\n(?:\d+[\.\)]\s*|[-*]\s+)|\n\n|$)', text, re.DOTALL):
        items.append({"text": match.group(1).strip(), "order": len(items) + 1})
    return items

4.3 归一化质量校验

归一化完成后，做一次快速质量检查：

def validate_normalized(normalized: dict) -> list:
    issues = []
    if len(normalized["full_text"]) < 20:
        issues.append("text_too_short")
    if not normalized["paragraphs"]:
        issues.append("no_paragraphs_extracted")
    if len(normalized["full_text"]) > 0 and len(normalized["paragraphs"]) == 0:
        issues.append("paragraph_extraction_failed")
    return issues

校验不通过的样本标记为 normalization_failed，保留原始回答进入人工排查队列。

五、结构化入库：归一化结果 + 指标并存储

5.1 归一化结果主表

CREATE TABLE normalized_answer (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    sync_group_id VARCHAR(64) NOT NULL,
    sub_task_id VARCHAR(64) NOT NULL UNIQUE,
    platform VARCHAR(30) NOT NULL,
    query_text TEXT NOT NULL,
    target_brand VARCHAR(100) NOT NULL,
    
    full_text LONGTEXT COMMENT '归一化后的纯文本全文',
    paragraphs_json JSON COMMENT '段落数组',
    list_items_json JSON COMMENT '列表项数组',
    links_json JSON COMMENT '链接数组',
    
    raw_format VARCHAR(20) COMMENT '原始格式: markdown/json/html',
    normalization_status VARCHAR(20) DEFAULT 'success' COMMENT 'success/failed',
    raw_data_url VARCHAR(512) COMMENT 'COS原始回答链接',
    
    collected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    INDEX idx_sync_group (sync_group_id),
    INDEX idx_brand_platform (target_brand, platform)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

5.2 指标提取（基于归一化后的统一结构）

归一化之后，提及判断和推荐提取只需要处理统一格式的 full_text 和 list_items，不再感知平台差异：

def extract_metrics(normalized: dict, brand_aliases: list) -> dict:
    text = normalized["full_text"]
    
    # 提及判断
    mentioned = any(alias in text for alias in brand_aliases)
    
    # 推荐判断：品牌出现在列表项中，且排名靠前
    recommended = False
    rank = 0
    for item in normalized.get("list_items", []):
        if any(alias in item["text"] for alias in brand_aliases):
            recommended = True
            rank = item["order"]
            break
    
    return {
        "is_mentioned": mentioned,
        "is_recommended": recommended,
        "recommendation_rank": rank
    }

提取结果写入指标表：

CREATE TABLE platform_metric (
    id BIGINT AUTO_INCREMENT PRIMARY KEY,
    sync_group_id VARCHAR(64) NOT NULL,
    sub_task_id VARCHAR(64) NOT NULL UNIQUE,
    platform VARCHAR(30) NOT NULL,
    target_brand VARCHAR(100) NOT NULL,
    
    is_mentioned TINYINT(1) DEFAULT 0,
    is_recommended TINYINT(1) DEFAULT 0,
    recommendation_rank INT DEFAULT 0,
    
    normalization_status VARCHAR(20),
    collected_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    
    INDEX idx_sync_brand (sync_group_id, target_brand)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4；

六、一致性查询：跨平台指标对比

归一化入库后，跨平台一致性查询变得简单：

SELECT 
    p.target_brand,
    p.platform,
    p.is_mentioned,
    p.is_recommended,
    p.recommendation_rank,
    n.paragraphs_json
FROM platform_metric p
JOIN normalized_answer n ON p.sub_task_id = n.sub_task_id
WHERE p.sync_group_id = 'sync-uuid-xxxx'
ORDER BY p.platform, p.recommendation_rank；

通过同步组 ID，可以在一行查询中看到同一个问题在 DeepSeek、Kimi、通义千问上的归一化回答和指标结果。业务侧可以进一步计算“多平台提及一致性”——品牌在所有平台上都被提及，还是仅在部分平台出现。

七、工程实践要点

1. 归一化是隔离点，不是性能瓶颈

归一化操作（段落切分 + 正则匹配）的计算开销很小，单条处理在毫秒级。不建议为了“节省计算”而跳过归一化直接在原始回答上做分析——这会把平台差异扩散到所有下游模块，后期维护成本远高于归一化的一次性投入。

2. 列表提取不要过度依赖正则

AI 输出的列表格式可能不规范——有时用 1、 有时用 1. 有时用 -。正则表达式需要覆盖这些变体，并在提取失败时降级为全文搜索，避免漏掉推荐信息。

3. 原始留档与归一化结果并存

COS 中的原始回答是“证据”，数据库中的 full_text 是“工作副本”。两者都要保留。当归一化逻辑更新后（例如改进了列表提取正则），可以从 COS 回溯原始数据重新归一化，而不需要重新调用 AI 平台。

八、结语

多平台 AI 回答采集的工程复杂度，本质上来自“异构输出”与“统一分析”之间的矛盾。本文提出的归一化层设计，通过一个统一回答结构体和分平台适配器，将平台差异隔离在采集层，让下游的指标提取、一致性分析、跨平台对比等逻辑可以基于同一套数据结构运行。

这套方案已在消费品牌、企业服务等行业的多平台 AI 可见度监测中实际使用，支持 DeepSeek、Kimi、通义千问、腾讯元宝等主流平台的回答归一化处理。开发者可在此基础上扩展新的平台适配器，或丰富归一化结构的字段维度（如语义倾向标签、引用标注）。

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

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

人工智能

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

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

人工智能

登录后参与评论

0 条评论

热度