Python内容任务状态机实现指南：从AI初稿到人工审核全流程解析

用 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: [],
}

其中，pending_review状态的流转规则尤为关键：

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

八、实现人工审核逻辑

审核逻辑分两种情况：通过，或者不通过并退回修改。

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

十、发布后复盘

发布并不是流程的终点。发布后的数据表现和问题反馈，应该被记录回任务中，形成闭环。

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

十一、完整运行示例

把上述所有步骤串联起来，就构成了一个完整的任务生命周期演示。

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才能真正成为一个可维护、可追踪的系统组件，而不再是一个临时生成文本的“黑盒”工具。

来源：互联网