AI数据采集与任务调度的系统实现
从批次拆分、状态机和租约，到证据归档与计费对账。

AI数据采集是GEO监测系统的数据入口。一次采集可能跨多个平台和终端，返回时间不同，部分任务成功、部分超时，截图也可能晚于正文到达。如果只有“成功”和“失败”两个状态，就很难解释页面为什么已有回答却没有截图，或已经提交却长时间没有完成。本篇以智叟云集系统的问题采集流程为基础，设计可恢复、可追溯的执行链路。

实现范围：配置与进度截图来自已留存界面；调度中心配图和执行框架属于建议方案。本文以PostgreSQL记录业务状态、Redis Streams传递事件、独立Worker执行连接器，并使用S3兼容存储归档证据。各平台的授权接口和计费契约需要分别确认，本文不假定现网已经采用这些组件。

一 将问题配置转换为不可变采样计划

图1 已填写的采集配置，包含品牌、竞品、问题、模型与预估费用。
计划展开前先规范化输入：去除问题首尾空白、识别完全重复的问题、校验长度和数量，再将品牌别名、问题版本、平台、终端、地区、模式与截图策略固化到config_snapshot。不要在执行过程中读取正在被编辑的表单，否则同一批次的前半段和后半段可能使用不同设置。

基础任务量可按问题数Q、平台终端组合数T、重复次数R计算为Q×T×R。20个问题、4个入口、2次独立采样对应160个事件；若单次提交上限为100，需要分为两个受控批次。深度思考和截图是否额外计价，应依据平台返回的报价计算；费用不能只用“任务数量×固定单价”推定。

采样身份建议由run_id、question_revision、platform、device、region、mode和repeat_index组成。重试属于同一采样事件，增加attempt_no；下一天的复测则生成新的run_id。这样既能避免重复回调虚增回答数，也不会把不同日期恰好相同的答案误删。

二 使用分层状态机解释任务进度

图2 采集执行链路，业务状态由数据库持久化，队列负责通知执行器。
状态层 推荐状态 业务含义
提交层 planned / submitting / accepted / uncertain 是否已被上游受理
采集层 queued / running / completed / failed 正文采集是否完成
证据层 pending / stored / partial / failed 原文、引用和截图是否齐全
分析层 pending / accepted / review / failed 品牌分析是否可用于统计
“已受理”不等于“已完成”，“正文完成”也不等于“所有证据完成”。如果平台返回正文但截图失败，可以让采集层完成、证据层部分完成；业务报告按正文生成，但展示截图缺失原因。如果品牌识别还在运行，页面应显示分析中，而不是把临时空标签显示为最终未提及。

在已留存的进度截图中，问题列表显示3次提问、0条回答，详情卡片仍在等待平台返回。因此聚合状态应由子任务状态推导：只有所有必需采样事件进入终态，才判断批次是否完成；全量成功、部分成功和全部失败分别展示。列表卡片、详情页和导出报告必须读取同一个状态聚合服务，避免不同区域使用不同判断条件。

图3 进行中的问题详情，平台回答尚未返回，不能提前形成最终提及结论。
三 队列投递与Worker消费
Redis Streams提供追加日志式的数据结构和消费者组，适合把“某个事件需要处理”交给Worker。消息只携带event_id、sample_id和必要的路由字段，完整配置从数据库读取。这样可以控制消息大小，并避免敏感原始回答和平台凭据出现在队列中。消息送达后先检查是否已处理，再执行持久化状态转换。

消费者处理成功后确认消息；中途退出的消息会保留待确认记录，可由恢复流程重新认领。恢复时不能仅凭消息还未确认就立即再次向外部提交：先检查数据库中的受理标识、当前租约和上游状态。队列至少一次投递意味着系统必须容忍重复，不意味着业务可以随意重复计费。

队列积压较大时，可按平台或任务类别分流。平台A变慢不应阻塞平台B，截图生成也不应占满正文采集的全部并发。公平调度按租户限制并发和速率，避免单个大客户耗尽所有Worker。任务优先级需要明确上限，防止低优先级项目长期得不到执行。

四 用租约和版本令牌防止迟到覆盖

图4 采集调度中心设计，展示任务队列、执行阶段和证据状态。
任务被Worker领取后写入lease_until和递增的fencing_token。Worker定期续租；如果进程崩溃，租约到期后其他Worker可接管。旧Worker恢复时即使返回了结果，也必须携带旧令牌提交，数据库通过条件更新拒绝迟到写入。仅使用分布式锁而没有版本令牌，容易在锁过期后出现两个执行器同时写结果。

"""运行：python 02_lease.py；验证过期租约与迟到写入保护。"""
import sqlite3

def connect():
db = sqlite3.connect(":memory:")
db.execute("""CREATE TABLE jobs (
id TEXT PRIMARY KEY, status TEXT NOT NULL,
token INTEGER NOT NULL DEFAULT 0, lease_until INTEGER,
result TEXT)""")
db.execute("INSERT INTO jobs(id,status) VALUES('j1','queued')")
db.commit()
return db

def claim(db, job_id, now, ttl=30):
if ttl <= 0:
raise ValueError("ttl必须为正")
with db:
cur = db.execute("""UPDATE jobs SET status='running',
token=token+1, lease_until=? WHERE id=? AND
(status='queued' OR
(status='running' AND lease_until<=?))""",
(now+ttl, job_id, now))
if cur.rowcount != 1:
return None
return db.execute("SELECT token FROM jobs WHERE id=?",
(job_id,)).fetchone()[0]

def complete(db, job_id, token, now, result):
with db:
cur = db.execute("""UPDATE jobs SET status='completed',result=?
WHERE id=? AND token=? AND status='running'
AND lease_until>?""", (result, job_id, token, now))
return cur.rowcount == 1

if name == "main":
db = connect()
first = claim(db, "j1", now=100)
assert claim(db, "j1", now=110) is None
second = claim(db, "j1", now=131)
assert second > first
assert not complete(db, "j1", first, 132, "old")
assert complete(db, "j1", second, 133, "new")
assert not complete(db, "j1", second, 134, "duplicate")
assert db.execute("SELECT result FROM jobs").fetchone()[0] == "new"
print("PASS: lease ownership, takeover, fencing, duplicate rejection")
该代码验证领取、过期接管、旧令牌拒绝和重复完成拒绝。生产实现应使用数据库时间降低机器时钟偏差，并把结果记录和任务完成状态放在同一事务。PostgreSQL可用条件UPDATE RETURNING原子领取，或使用FOR UPDATE SKIP LOCKED批量挑选待执行任务。SQLite片段用于验证状态规则，不承担多机器并发性能验证。

租约只保护本地状态，无法撤销已经提交到外部平台的请求。若旧Worker在超时前已经成功创建任务，新Worker必须查询上游request_id或使用上游支持的幂等键。没有可查询标识时，应进入uncertain状态等待核查，不能把“没有收到响应”当成“对方没有执行”。

五 重试策略与限流控制
错误分类 建议动作 是否自动重试
参数无效或权限不足 修正配置、记录错误 否
明确限流 遵循Retry-After并延后任务 有次数与总时长上限
连接超时且受理未知 查询上游受理状态 先核查再决定
受理后查询暂时失败 保留任务ID并继续轮询 有预算
正文解析失败 保存原文并转人工复核 不重做原始提问
指数退避可以采用delay=min(max_delay, base×2^attempt)再添加随机抖动，避免大量任务在同一时间重试。限流器不仅限制请求数，还要分别限制提交、轮询和截图下载。轮询间隔随任务年龄调整，较长任务不应每秒重复查询；服务明确返回完成时应立即停止轮询。

重复提问次数属于业务采样计划，自动重试次数属于执行恢复策略。两者必须使用不同字段和统计口径。费用流水应关联上游任务ID与采样事件，重试是否产生费用按上游规则核对；不能因为本地认为是重试，就自动假定免费。

六 连接器接口与结果标准化
每个平台连接器建议实现capabilities、submit、poll、fetch_evidence和cancel这五类方法。capabilities声明终端、地区、思考模式、截图支持和限制；不支持的选项在提交前拒绝。submit返回受理标识而不是伪造成功回答，poll返回明确状态，fetch_evidence负责保留原始内容和可用附件。cancel只能承诺本地取消意图，是否能终止外部请求取决于平台契约。

连接器契约的结果结构，字段由适配层统一。

{
"sample_id": "sample-101",
"upstream_id": "provider-task-9",
"collection_state": "completed",
"evidence_state": "partial",
"answer_text": "原始回答正文",
"citations": [{"id": "c1", "url": "https://example.com/a"}],
"screenshot_key": null,
"captured_at": "2026-11-11T02:00:00Z",
"warnings": ["SCREENSHOT_UNAVAILABLE"]
}
引用链接必须保留原始值与规范化值，清洗时只删除确定不影响内容的追踪参数。不同查询参数可能对应不同页面，不能一律删除。平台返回的Markdown或HTML作为不可信内容处理，展示时做转义或采用经过配置的清洗器；原文存储与安全展示版本分别保存，保证后续还能核查采集内容。

七 证据归档与数据质量
对象键可按tenant/run/sample/revision组织，但路径命名不是访问权限。数据库记录正文摘要、对象大小、MIME类型和时间，应用服务鉴权后才允许下载。截图使用原始分辨率存储，报告中的缩略图另行生成。附件传输失败不能覆盖已成功归档的正文；修复附件时保留修订记录。

数据质量检查分为字段完整性、采样唯一性、正文有效性和引用可追溯性。空字符串、错误提示页、登录页和截断响应不应当成有效回答。一个已完成状态的上游任务，仍可能返回无法用于品牌分析的正文，因此需要单独的validity字段和原因，不应直接等价为有效分析样本。

八 报价 快照与结算对账
任务创建时保存price_version、unit_prices、estimated_max和currency，用户确认的是当时的报价快照。执行结束后根据实际完成事件与上游费用对账；金额用整数最小货币单位或定点数，不使用二进制浮点累计。预估、预留、实际扣费和退款是不同流水类型，余额不能只靠一个可随意修改的数字维护。

若一个批次部分成功，应按规则区分完成费用、失败费用和仍待核对费用。取消本地任务后，已受理的外部任务可能继续消耗资源，因此退款动作不能只根据本地cancelled状态自动触发。账务接口同样需要幂等键，重复回调不得产生第二笔扣费。

九 从故障注入到可观测性验收
建议测试四个崩溃点：业务提交后消息尚未发布；上游已受理但本地未记录；正文已入库但任务未完成；任务已完成但消息未确认。每个故障恢复后，都应满足“没有丢失受理记录、没有重复统计、没有旧结果覆盖新结果、没有重复记账”。这些条件比正常路径的一次成功更能说明GEO优化监测系统的可靠性。

监控队列等待时间、上游受理延迟、执行耗时、待确认消息数量、uncertain事件数量、证据缺失率和对账差异。AI品牌可见度监测系统只有在有效采样窗口清楚时才能比较结果；AI诊断、舆情监控与舆情分析也都依赖这条证据链。对智叟云集系统而言，稳定的采集调度是所有上层分析可信的基础。

参考资料
Redis Streams 官方文档

FastAPI 后台任务说明

智叟云集产品入口