最近在新闻行业的技术讨论里Mythos 2 是一个频繁出现的名字很多文章直接连用“已接管各大新闻编辑部”来描述它。对后端工程师来说这个消息并不需要用神秘化的方式去理解。围绕 Mythos 2 的讨论通常指向一套面向新闻编辑部的 AI 内容生产系统它把选题、写稿、审核、发布、反馈这些原本分散在不同环节里的动作串成可编排的流程。本文不讨论产品营销也不评价“接管”这种说法是否夸张而是把 Mythos 2 当作一类需要做技术接入、任务调度、状态管理和错误排查的分布式系统来拆解。接下来会给出一个最小可运行的技术示例。这个示例的目标不是复刻某个具体产品而是让你理解如果要接一个类似 Mythos 2 的新闻编辑部系统哪些模块必须存在消息该怎样流转数据该怎样建模出现卡稿、漏稿、误发时又该从哪里查起。1. 先理解核心链路Mythos 2 解决什么问题1.1 编辑部自动化不是“一键写稿”那么简单传统编辑部工作流可以概括为五个环节选题、素材采集、写稿、审稿、发布。过去每个环节都依赖人记者去找信源编辑去改文字排版系统限制样式。AI 系统进入之后最大的变化不是某一个环节被替换而是五个环节之间出现了自动化衔接。用“接管”这个词其实不准确更准确的说法是“流程编排”。以 Mythos 2 这类系统为例它会自动完成素材聚合、初稿生成、多版本标题建议、基础事实标注和分发提醒但最终的来源核实、合规风险判断、事实责任归属仍然要有人来拍板。技术实现上就是在一套已有内容管理系统周围增加一个“AI 生产调度层”。因此做技术接入时先要完成一次职责盘点哪些动作是确定性的可以交给系统哪些动作是有责任的必须加人工确认哪些动作属于重复劳动但一旦自动执行就绝不能丢失流程记录。盘点完之后再设计状态机就有依据了。1.2 把稿件生命周期抽象成状态机无论前端界面多花哨接入 Mythos 2 时先落地的往往是数据模型。稿件会经历以下几个状态creating任务刚创建系统正在生成初稿。pending_review初稿生成完成等待人工复核。approved审核通过等待发布或已进入发布队列。rejected审核不通过退回修改或标记为废弃。published已经发布到渠道。archived已下架或进入归档库。每一个状态都有触发条件和执行者。比如creating - pending_review必须是写稿 worker 成功返回并完成落库而不是前台按钮点了就改pending_review - approved必须记录审核人 ID、审核时间和审核意见published状态必须关联发布渠道和发布 ID否则后续撤回无法定位。这个状态机看起来简单但它决定了任务系统的可靠性。如果直接在数据库里用一列字符串表示状态再在业务代码里到处判断后面会出现三个问题状态被重复修改、异常后无法恢复、人工审核记录无审计线索。推荐的做法是把状态转移做成一张事件表每发生一次变更写入一条记录。即使业务系统只存最新状态审计表也能回答“这篇稿子到底是什么时候从待审核变成已发布的”。1.3 人机边界决定了接口设计Mythos 2 这类系统容易犯的错误是想把“发布”也自动化。但从工程角度看发布是外部行为涉及品牌声誉和合规后果不应该由语言模型的输出直接触发。稳妥的接口设计是在生成结果和发布动作之间加一个人工审核步骤。哪怕审核只是一个点“通过”按钮也必须存在。所以在下面的示例里核心链路会设计成主题进来以后自动生成初稿初稿进入待审核队列人工调用审核接口后稿件才允许进入发布通道。这个设计不是为了拖慢流程而是为了给异常留出拦截时间。2. 环境准备与项目结构2.1 依赖栈学习环境里不需要一开始就上完整的微服务体系。一个 FastAPI 网关、一个 Celery worker、一个 PostgreSQL、一个 Redis再加一个模型服务已经足够跑通整条链路。组件用途建议Python 3.10开发语言建议用 3.11 或 3.12依赖兼容性更好FastAPI对外提供 HTTP API适合快速开发同步/异步接口Celery异步执行写稿任务管理队列和重试Redis任务 broker 和结果存储学习环境单机即可PostgreSQL持久化稿件和审计记录学习环境可用 Docker模型服务生成稿件正文使用兼容 OpenAI API 的本地服务或测试 mock这里要说明代码示例里的模型服务地址要改成你自己的部署地址。如果只是学习用一个简单的 mock 服务返回固定文本也可以。2.2 项目目录这里只建一个后端项目不包含前端。目录结构如下newsroom-mythos/ ├── main.py # FastAPI 网关 ├── worker.py # Celery 异步任务 ├── config.py # 配置加载 ├── schemas.py # 请求/响应模型 ├── db.py # 数据库连接和表结构初始化 ├── prompts.py # 提示词模板 ├── safety.py # 基础内容安全检查 ├── requirements.txt # Python 依赖 └── deploy/ ├── mythos2.yaml # 部署配置 └── init.sql # 建表 SQL模块职责划分要清楚main.py只处理 HTTP 输入校验和调用异步任务worker.py负责和模型交互、写库prompts.py与业务提示词相关safety.py是审核前的基础检查。不要把模型调用逻辑写进路由函数否则后面加队列重试时会发现代码耦合太严重。2.3 配置文件下面这份 YAML 只用于说明参数结构落地时要根据自己的环境、密钥和版本调整。service: name: mythos2-gateway env: staging server: host: 0.0.0.0 port: 8000 database: dsn: postgresql://newsroom:newsroom_passlocalhost:5432/newsroom pool_size: 10 max_overflow: 20 redis: url: redis://localhost:6379/0 queue: broker: redis://localhost:6379/1 backend: redis://localhost:6379/2 task_queue: mythos.draft max_retries: 3 retry_backoff: 45 model: provider: openai-compatible endpoint: http://127.0.0.1:8001/v1 api_key_env: MODEL_API_KEY model_name: news-writer-v1 temperature: 0.3 top_p: 0.9 max_tokens: 1500 request_timeout: 60 review: require_human: true sensitive_check_enabled: true auto_publish: false把配置外置化到 YAML 或环境变量是为了区分开发、测试、生产环境。比如开发环境可以用本机模型服务的 mock 端口生产环境再切真实服务api_key 不写死在代码里通过环境变量注入。3. 最小闭环实现从主题到发布只留一个审核点3.1 建表先保证能追溯再做功能稿件表保存最新状态审计表保存每一次状态变化。只建一张表当然也能跑但出了问题很难追溯所以这里建两张表。CREATE TABLE news_draft ( id bigserial PRIMARY KEY, request_id uuid NOT NULL UNIQUE, topic varchar(200) NOT NULL, category varchar(50) NOT NULL, keywords text[] DEFAULT {}, content text NOT NULL DEFAULT , status varchar(20) NOT NULL DEFAULT creating, model_name varchar(100), temperature numeric(3,2), created_by varchar(100), created_at timestamptz NOT NULL DEFAULT now(), updated_at timestamptz NOT NULL DEFAULT now() ); CREATE TABLE news_draft_event ( id bigserial PRIMARY KEY, draft_id bigint NOT NULL REFERENCES news_draft(id) ON DELETE CASCADE, from_status varchar(20), to_status varchar(20) NOT NULL, operator varchar(100) NOT NULL, comment text, created_at timestamptz NOT NULL DEFAULT now() ); CREATE INDEX idx_draft_status ON news_draft(status); CREATE INDEX idx_draft_created_at ON news_draft(created_at); CREATE INDEX idx_draft_event_draft_id ON news_draft_event(draft_id);参数说明request_id用于幂等。同一篇稿件重复提交时系统能识别出已有的 request_id避免生成两份正文。content默认空字符串而不是 null避免下游拼接时出现 None 异常。status使用短字符串可读性好但要注意在应用层约束合法值不能只依赖数据库。temperature保留小数当天复盘时可以判断某篇稿子是用高随机度还是低随机度生成的。3.2 对外网关接收主题并创建任务FastAPI 网关只做三件事校验请求参数、生成 request_id、把任务塞进队列。# main.py import uuid from fastapi import FastAPI, HTTPException from pydantic import BaseModel, Field from worker import enqueue_draft_task from db import insert_draft app FastAPI(titleMythos 2 Gateway) class DraftCreateRequest(BaseModel): topic: str Field(..., min_length5, max_length200, description新闻主题) category: str Field(tech, pattern^(tech|finance|sports)$, description栏目分类) keywords: list[str] Field(default_factorylist, max_length10) class DraftCreateResponse(BaseModel): request_id: str draft_id: int status: str app.post(/api/v1/drafts, response_modelDraftCreateResponse) def create_draft(req: DraftCreateRequest): request_id str(uuid.uuid4()) draft_id insert_draft( request_idrequest_id, topicreq.topic, categoryreq.category, keywordsreq.keywords, ) enqueue_draft_task.delay(request_id, req.topic, req.category, req.keywords) return DraftCreateResponse( request_idrequest_id, draft_iddraft_id, statuscreating, )这里的检查点是请求进来后立即返回 request_id 和 draft_id而不是等待模型生成完再返回。因为大模型生成一篇长文可能耗时十几秒甚至更久如果 HTTP 请求一直阻塞网关会扛不住并发前端也容易超时。3.3 写稿 worker调用模型并存稿Celery worker 从队列里取出任务调用模型服务生成正文然后更新稿件状态。# worker.py import os import requests from celery import Celery from db import update_draft_content, mark_draft_status, save_event, fetch_draft_by_request_id from prompts import build_news_prompt from safety import run_safety_check celery_app Celery( mythos2, brokeros.getenv(CELERY_BROKER, redis://localhost:6379/1), backendos.getenv(CELERY_BACKEND, redis://localhost:6379/2), ) celery_app.task(bindTrue, max_retries3) def enqueue_draft_task(self, request_id, topic, category, keywords): draft fetch_draft_by_request_id(request_id) if draft is None: return {ok: False, reason: draft_not_found} prompt build_news_prompt(topictopic, categorycategory, keywordskeywords) try: content call_model(prompt) except Exception as exc: mark_draft_status(draft_iddraft[id], to_statusfailed, operatorsystem) raise self.retry(excexc, countdown45) safety_result run_safety_check(content) if safety_result[blocked]: mark_draft_status(draft_iddraft[id], to_statusrejected, operatorsystem, commentsafety_result[reason]) return {ok: False, reason: safety_result[reason]} update_draft_content(draft_iddraft[id], contentcontent, model_namenews-writer-v1) mark_draft_status(draft_iddraft[id], to_statuspending_review, operatorsystem) return {ok: True, request_id: request_id}这里的核心点有两个。一个是重试策略。模型接口超时、网络抖动、服务重启都会导致调用失败不能一失败就退出任务。常见做法是设置max_retries3失败后延迟 45 秒再试而且只在可重试异常时重试明显的参数错误不要重试。另一个是审核前置检查。生成内容之后不要直接落库到发布状态先做一次内容安全检查。如果检查命中状态直接标记为 rejected并记录原因让下一环的人工编辑看到为什么被拦下来。3.4 提示词模板把“新闻写作规范”变成可执行约束模型输出质量很大程度上由提示词决定。下面是一份参考模板用于说明结构实际要按编辑部的栏目规范调整。# prompts.py def build_news_prompt(topic: str, category: str, keywords: list[str]) - str: kw_text 、.join(keywords) if keywords else 无额外关键词 return f你是一名{category}栏目的资深编辑。 请围绕下面的主题写一篇新闻初稿全文不超过 600 字。 主题{topic} 参考关键词{kw_text} 写作要求 1. 第一段必须在 80 字内交代核心事实。 2. 正文分成三个部分每部分使用小标题。 3. 只写基于已给信息和常识推理的句子不要编造具体数据。 4. 不要出现“内部消息”“据说”这类无法核实来源的表达。 5. 结尾写一个需要继续跟进的开放问题。 现在开始写稿。模板里最值得借鉴的是“只写基于已给信息和常识推理的句子”这一条。AI 生成新闻最大的风险不是写不出来而是写得流畅但事实错误。提示词只能约束一部分后续还需要事实核查模块。3.5 人工审核与发布接口审核接口是人工控制和自动化之间的边界。它不生成稿件只改变状态。# main.py class ReviewRequest(BaseModel): draft_id: int decision: str Field(..., pattern^(approve|reject)$) reviewer: str Field(..., min_length2) comment: str app.post(/api/v1/drafts/{draft_id}/review) def review_draft(draft_id: int, req: ReviewRequest): draft fetch_draft_by_id(draft_id) if draft is None: raise HTTPException(status_code404, detaildraft not found) if draft[status] ! pending_review: raise HTTPException(status_code409, detailcurrent status allows no review) if req.decision approve: mark_draft_status(draft_id, to_statusapproved, operatorreq.reviewer, commentreq.comment) return {draft_id: draft_id, status: approved} else: mark_draft_status(draft_id, to_statusrejected, operatorreq.reviewer, commentreq.comment) return {draft_id: draft_id, status: rejected}这个接口最重要的约束是状态校验。不是任何状态下都能审核只有pending_review状态才允许操作。否则稿子已经发布了还能被审核通过状态机就失去意义。4. 关键参数应该怎么调从“能出字”到“能发稿”4.1 模型推理参数参数建议值调小影响调大影响使用场景temperature0.2-0.4内容更保守、句子重复度可能高内容更发散、事实错误概率上升新闻写稿top_p0.8-0.9采样集中、变化少采样分散、控制力弱配合 temperature 使用max_tokens1200-1800容易截断稿子增加等待时间、成本上升长稿生成request_timeout45-90 秒正常模型稍慢就超时报错恢复时间拉长网络不稳定时调大新闻场景最重要的参数是 temperature。新闻要事实优先不能像故事创作那样追求发散。实际项目里建议同一篇文章生成两到三个版本人工对比时不是看哪个更“惊艳”而是看哪个事实陈述更稳。4.2 队列并发与任务优先级接入 Mythos 2 后最常见的资源冲突是写稿任务把模型服务打满导致审核系统调模型做标题建议时也全部超时。所以在队列层要分开处理。queue: task_queue: mythos.draft review_queue: mythos.review draft_concurrency: 4 review_concurrency: 2写稿可以接受排队但人工审核时点击“生成标题建议”这种交互任务不能排在写稿后面。学习环境可以只用一个队列生产环境要考虑用独立队列或者在 Celery 里配置优先级。4.3 审核阈值不是越低越好内容安全检查和敏感词拦截很容易做成“宁可误杀不可漏杀”结果编辑被迫审大量无意义拦截。更好的做法是分成多级硬拦截命中明确的个人电话、金融账号、内部系统路径等直接 rejected。软提醒命中某些不确定表达比如“最高”“独家”“内部”标记为需要人工重点检查。无命中进入正常待审核队列。每一级的处理方式不同。软提醒不是禁止生成而是要让人工看到提示框不能代替人做决定。5. 运行验证接口通了不等于流程对了5.1 启动环境与依赖先安装依赖再启动数据库、Redis最后启动网关和 worker。pip install -r requirements.txt docker run -d --name newsroom-pg \ -e POSTGRES_USERnewsroom \ -e POSTGRES_PASSWORDnewsroom_pass \ -e POSTGRES_DBnewsroom \ -p 5432:5432 postgres:16-alpine docker run -d --name newsroom-redis \ -p 6379:6379 redis:7-alpine python main.py celery -A worker.celery_app worker --loglevelinfo -c 4这里有三个容易出错的地方。第一celery -A worker.celery_app中的worker是文件名celery_app是实例名拼错会提示找不到任务模块。第二数据库连接串里的用户名、密码要和 Docker 启动参数一致。第三模型服务地址如果写成本机 localhostworker 在容器里时就会访问不通。5.2 提交模拟任务curl -X POST http://127.0.0.1:8000/api/v1/drafts \ -H Content-Type: application/json \ -d { topic: 本地社区图书馆周末开放时间调整, category: tech, keywords: [图书馆, 开放时间, 周末] }预期响应{ request_id: 7d85a6a2-1a2c-4c6f-9c92-1cba90e9c015, draft_id: 1, status: creating }然后等几秒查看数据库状态SELECT id, status, model_name, temperature, created_at FROM news_draft WHERE id 1;正常状态应该从 creating 变成 pending_review。如果一直是 creating优先检查 worker 日志和 Redis 队列长度。5.3 日志关键字正常日志应该出现Task mythos2.enqueue_draft_task succeeded异常日志会出现不同关键字Task mythos2.enqueue_draft_task retry Model request timeout after 60 seconds排查时不要只看错误栈第一行要把“任务 ID、请求 ID、模型调用时长、重试次数”拼接起来看。建议在日志中加入 request_id、draft_id 字段否则最终从异常堆栈定位到具体稿子会很难。6. 常见问题与排查链路为什么稿子卡在 creating6.1 排查顺序当稿子一直处于 creating按下面顺序查确认 Redis 是否可用redis-cli ping是否返回 PONG。确认队列里是否有任务celery -A worker.celery_app inspect active。确认 worker 日志是否有模型调用异常。确认模型服务地址是否可达直接 curl 模型接口。确认数据库连接和表结构是否正常。确认是否有任务重试死循环导致状态被反复改写。顺序不能颠倒。很多问题查到最后发现是模型服务地址写成了旧环境但如果前面先翻数据库会浪费很多时间。6.2 错误现象对照表现象可能原因检查方式处理建议提交后一直 creatingworker 未启动或队列未消费celery inspect、Redis 队列长度启动 worker确认注册任务worker 重试三次后失败模型服务地址错误或密钥过期手动 curl 模型接口修正 endpoint重启 worker稿件生成内容为空prompt 没有输出约束查看模型原始响应增加 max_tokens 和兜底判断审核接口返回 409状态不是 pending_review查询 news_draft.status检查前端是否重复提交同一请求生成了多份稿子缺少幂等控制查 request_id 是否唯一在应用中按 request_id 去重更新状态后列表没有变化更新了错误数据库确认连接串和当前库统一使用环境变量管理连接6.3 三个最容易踩的集成坑第一个坑是把模型返回直接当作可发布内容。即使模型服务返回 200文字也可能有事实性错误。正确姿势是生成完成后至少走一次内容安全检查和人工审核。第二个坑是认为提示词写得好就不用加校验。实际情况是提示词只是软约束模型仍然可能输出电话号码、邮箱、内部链接。要在系统层做硬校验把可能泄漏的信息拦下来。第三个坑是网关直接同步调用模型。编辑人员如果点了“生成”以后请求要等几十秒所有体验都会崩。正确做法是异步化先返回任务 ID 和状态前端轮询或通过 WebSocket 接收完成事件。7. 从“能跑”到“敢用”生产环境最佳实践与扩展方向7.1 发版前检查清单检查项说明配置外置化数据库、Redis、模型地址不能写死在代码里密钥管理api_key 使用