用 Python 实现一个内容任务状态机：从 AI 初稿到人工审核

原创

OPC一人公司-迪迦

修改于 2026-05-25 17:06:26

320

用 Python 实现一个内容任务状态机：从 AI 初稿到人工审核

在内容生产系统里，AI 生成初稿只是其中一步。更关键的问题是：一篇内容从创建、生成、审核、发布到复盘，应该如何被系统追踪？

如果只把 AI 当成一个“写文章接口”，很容易出现几个问题：

生成了多版内容，但不知道哪一版被审核过；
文章还没有人工确认，却被误认为可以发布；
发布后的链接、数据和复盘没有回写；
下次生成时，之前的审核经验无法复用。

所以在接入 AI 之前，我更倾向于先设计一个任务状态机。本文用一个最小可运行的 Python 示例，演示如何管理内容任务的状态流转。

一、状态设计

我们先定义一篇内容的生命周期：

draft -> ai_generated -> pending_review -> approved -> scheduled -> published -> reviewed

每个状态含义如下：

draft：任务刚创建，还没有生成内容
ai_generated：AI 已生成初稿
pending_review：等待人工审核
approved：人工审核通过
scheduled：已进入发布排期
published：已发布并记录链接
reviewed：已完成数据复盘

这里最重要的规则是：

AI 生成完成，不等于可以发布。

内容必须经过人工审核，才能进入发布状态。

二、定义状态枚举

下面用 Python 的 Enum 定义任务状态。

from enum import Enum


class TaskStatus(str, Enum):
    DRAFT = "draft"
    AI_GENERATED = "ai_generated"
    PENDING_REVIEW = "pending_review"
    APPROVED = "approved"
    SCHEDULED = "scheduled"
    PUBLISHED = "published"
    REVIEWED = "reviewed"

用枚举的好处是状态更清晰，也能避免代码里到处写字符串。

三、定义内容任务对象

这里用 dataclass 创建一个简单的内容任务模型。

from dataclasses import dataclass, field
from datetime import datetime
from typing import Optional, List


@dataclass
class ContentTask:
    id: str
    title: str
    topic: str
    platform: str
    status: TaskStatus = TaskStatus.DRAFT

    draft_text: Optional[str] = None
    final_text: Optional[str] = None
    publish_url: Optional[str] = None

    review_passed: bool = False
    review_notes: List[str] = field(default_factory=list)

    created_at: datetime = field(default_factory=datetime.now)
    updated_at: datetime = field(default_factory=datetime.now)

这个模型里保留了几个关键字段：

draft_text：AI 生成的初稿
final_text：人工确认后的最终稿
review_passed：是否通过人工审核
publish_url：发布后的链接
review_notes：审核意见

真实系统中还可以增加作者、审核人、版本号、知识库引用来源等字段。

四、定义合法状态流转

不是所有状态都可以互相跳转。

例如，draft 可以进入 ai_generated，但不能直接进入 published。

我们先定义允许的流转关系：

ALLOWED_TRANSITIONS = {
    TaskStatus.DRAFT: [TaskStatus.AI_GENERATED],
    TaskStatus.AI_GENERATED: [TaskStatus.PENDING_REVIEW],
    TaskStatus.PENDING_REVIEW: [TaskStatus.APPROVED, TaskStatus.AI_GENERATED],
    TaskStatus.APPROVED: [TaskStatus.SCHEDULED],
    TaskStatus.SCHEDULED: [TaskStatus.PUBLISHED],
    TaskStatus.PUBLISHED: [TaskStatus.REVIEWED],
    TaskStatus.REVIEWED: [],
}

其中这一行比较关键：

TaskStatus.PENDING_REVIEW: [TaskStatus.APPROVED, TaskStatus.AI_GENERATED]

意思是：审核通过就进入 approved，审核不通过可以退回到 ai_generated，重新生成或修改初稿。

五、实现状态切换函数

接下来写一个通用的状态切换函数。

def change_status(task: ContentTask, next_status: TaskStatus) -> ContentTask:
    allowed_next = ALLOWED_TRANSITIONS.get(task.status, [])

    if next_status not in allowed_next:
        raise ValueError(
            f"非法状态流转：{task.status.value} -> {next_status.value}"
        )

    task.status = next_status
    task.updated_at = datetime.now()
    return task

测试一下：

task = ContentTask(
    id="CNT-001",
    title="用 Python 实现一个内容任务状态机",
    topic="AI 工作流",
    platform="tencent_cloud"
)

change_status(task, TaskStatus.AI_GENERATED)
print(task.status)

输出结果：

TaskStatus.AI_GENERATED

如果直接从 draft 跳到 published，就会报错：

task = ContentTask(
    id="CNT-002",
    title="错误示例",
    topic="AI 工作流",
    platform="tencent_cloud"
)

change_status(task, TaskStatus.PUBLISHED)

会抛出异常：

ValueError: 非法状态流转：draft -> published

这就是状态机的价值：系统不会允许内容绕过审核直接发布。

六、模拟 AI 生成初稿

下面写一个简单函数，模拟 AI 生成初稿。

实际项目里，这一步可以替换成大模型 API 调用。

def generate_draft(task: ContentTask, knowledge_context: str) -> ContentTask:
    if task.status != TaskStatus.DRAFT:
        raise ValueError("只有 draft 状态的任务可以生成初稿")

    task.draft_text = f"""
# {task.title}

本文主题：{task.topic}

参考知识：
{knowledge_context}

这里是 AI 生成的文章初稿，需要人工审核后才能发布。
"""

    change_status(task, TaskStatus.AI_GENERATED)
    return task

调用示例：

task = ContentTask(
    id="CNT-003",
    title="用 Python 实现一个内容任务状态机",
    topic="AI 内容生产工作流",
    platform="tencent_cloud"
)

task = generate_draft(
    task,
    knowledge_context="内容发布前必须经过人工审核，不能直接发布 AI 初稿。"
)

print(task.status)
print(task.draft_text)

生成后，任务状态会从：

draft

变成：

ai_generated

七、进入人工审核

AI 生成完成后，任务不能直接发布，而是要进入待审核状态。

def submit_for_review(task: ContentTask) -> ContentTask:
    if task.status != TaskStatus.AI_GENERATED:
        raise ValueError("只有 AI 已生成初稿的任务才能提交审核")

    change_status(task, TaskStatus.PENDING_REVIEW)
    return task

调用：

task = submit_for_review(task)
print(task.status)

输出：

TaskStatus.PENDING_REVIEW

八、实现人工审核逻辑

审核分两种情况：

通过审核；
不通过，退回修改。

def review_task(task: ContentTask, passed: bool, notes: str) -> ContentTask:
    if task.status != TaskStatus.PENDING_REVIEW:
        raise ValueError("只有 pending_review 状态的任务可以审核")

    task.review_notes.append(notes)

    if passed:
        task.review_passed = True
        task.final_text = task.draft_text
        change_status(task, TaskStatus.APPROVED)
    else:
        task.review_passed = False
        change_status(task, TaskStatus.AI_GENERATED)

    return task

审核通过示例：

task = review_task(
    task,
    passed=True,
    notes="内容事实无明显问题，表达可以发布。"
)

print(task.status)

输出：

TaskStatus.APPROVED

审核不通过示例：

task = review_task(
    task,
    passed=False,
    notes="初稿中存在未经确认的结论，需要重新生成。"
)

这时任务会回到：

ai_generated

然后可以重新修改或生成。

九、排期与发布

只有审核通过的内容，才能进入排期。

def schedule_task(task: ContentTask) -> ContentTask:
    if task.status != TaskStatus.APPROVED:
        raise ValueError("只有 approved 状态的任务可以排期")

    change_status(task, TaskStatus.SCHEDULED)
    return task

发布时必须写入发布链接：

def publish_task(task: ContentTask, url: str) -> ContentTask:
    if task.status != TaskStatus.SCHEDULED:
        raise ValueError("只有 scheduled 状态的任务可以发布")

    if not url:
        raise ValueError("发布链接不能为空")

    task.publish_url = url
    change_status(task, TaskStatus.PUBLISHED)
    return task

调用示例：

task = schedule_task(task)

task = publish_task(
    task,
    url="https://cloud.tencent.com/developer/article/example"
)

print(task.status)
print(task.publish_url)

输出：

TaskStatus.PUBLISHED
https://cloud.tencent.com/developer/article/example

十、发布后复盘

发布不是流程终点。发布后的数据和问题应该回到任务记录中。

这里先用一个简单函数模拟复盘。

def review_result(task: ContentTask, summary: str) -> ContentTask:
    if task.status != TaskStatus.PUBLISHED:
        raise ValueError("只有 published 状态的任务可以复盘")

    task.review_notes.append(f"复盘：{summary}")
    change_status(task, TaskStatus.REVIEWED)
    return task

调用：

task = review_result(
    task,
    summary="文章阅读表现一般，但状态机示例部分被读者重点关注。"
)

print(task.status)
print(task.review_notes)

输出：

TaskStatus.REVIEWED

十一、完整运行示例

把上面的流程串起来，可以得到一个完整的任务生命周期。

def run_demo():
    task = ContentTask(
        id="CNT-004",
        title="用 Python 实现一个内容任务状态机",
        topic="AI 内容生产工作流",
        platform="tencent_cloud"
    )

    task = generate_draft(
        task,
        knowledge_context="AI 生成内容必须经过人工审核后才能发布。"
    )

    task = submit_for_review(task)

    task = review_task(
        task,
        passed=True,
        notes="审核通过，内容可以进入排期。"
    )

    task = schedule_task(task)

    task = publish_task(
        task,
        url="https://cloud.tencent.com/developer/article/example"
    )

    task = review_result(
        task,
        summary="发布后记录数据，并将审核经验回写到知识库。"
    )

    return task


if __name__ == "__main__":
    result = run_demo()
    print(result)

这个示例虽然简单，但已经覆盖了一个内容生产系统最基本的状态流转。

十二、后续可以怎么扩展？

实际项目中，可以继续扩展几个方向。

1. 增加数据库

可以把 ContentTask 存入 SQLite、MySQL 或 PostgreSQL，而不是只保存在内存中。

2. 增加版本管理

每次 AI 重新生成或人工修改，都生成一个新的版本号：

v1: AI 初稿
v2: 人工修改稿
v3: 最终发布稿

这样发布后可以追踪具体使用的是哪一版内容。

3. 增加操作日志

每次状态变化都记录：

谁在什么时间，把任务从哪个状态改到了哪个状态。

这对多人协作和审核追踪很重要。

4. 接入知识库检索

在生成初稿前，可以先从知识库中检索资料，再把相关资料传给模型。

流程可以变成：

任务创建
-> 知识库检索
-> 构造 Prompt
-> AI 生成初稿
-> 人工审核

5. 接入真实模型 API

generate_draft 函数里目前是模拟文本。真实项目中可以替换成大模型 API，例如：

prompt = 任务信息 + 知识库片段 + 输出要求
model.generate(prompt)

但无论使用哪个模型，都不建议让模型输出后直接发布。

十三、总结

AI 内容生产系统不应该只关注“能不能生成文章”，更应该关注生成之后如何被审核、追踪和复盘。

本文用 Python 实现了一个最小状态机，覆盖了以下流程：

创建任务
-> AI 生成初稿
-> 提交人工审核
-> 审核通过
-> 排期发布
-> 写入发布链接
-> 发布后复盘

这个流程的核心不是让 AI 替代人，而是让 AI 参与可控的内容生产链路。

在技术实现上，状态机可以避免内容绕过审核直接发布；任务对象可以保留上下文；审核记录可以沉淀经验；发布链接和复盘结果可以为后续内容提供依据。

当这些基础结构建立起来之后，AI 才更像系统中的一个可维护组件，而不是一个临时生成文本的工具。

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

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

工作流

python

AIGC

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

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

工作流

python

AIGC

登录后参与评论

0 条评论

热度

用 Python 实现一个内容任务状态机：从 AI 初稿到人工审核

用 Python 实现一个内容任务状态机：从 AI 初稿到人工审核

用 Python 实现一个内容任务状态机：从 AI 初稿到人工审核

一、状态设计

二、定义状态枚举

三、定义内容任务对象

四、定义合法状态流转

五、实现状态切换函数

六、模拟 AI 生成初稿

七、进入人工审核

八、实现人工审核逻辑

九、排期与发布

十、发布后复盘

十一、完整运行示例

十二、后续可以怎么扩展？

1. 增加数据库

2. 增加版本管理

3. 增加操作日志

4. 接入知识库检索

5. 接入真实模型 API

十三、总结

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐