大模型+RAG+Agent：AI如何重构新闻编辑部工作流

“Mythos 2 已接管各大新闻编辑部”这个说法，最近在技术社区和媒体圈被频繁转发。如果只看字面意思，很容易产生一种错觉：这是一套能独立写稿、让记者批量失业的AI系统。但如果我们把“Mythos 2”看作一类AI新闻生产系统的代称，而不是某款具体产品的评测对象，就会发现更值得关注的东西——它背后正在发生的变化，是“大模型 + Agent + RAG + 人工审核闸门”这套技术组合，从实验室走进了真正的编辑部日常。

这篇文章不打算猜测某个产品的内部参数，而是把它当作一个典型的技术样本，拆解AI辅助新闻编辑部的整体架构、落地流程和工程风险。读完你会知道：如果想搭建一套覆盖选题、检索、初稿生成、事实核查、多平台分发的辅助系统，需要哪些基础组件；哪些环节自动化收益最大；哪些环节必须留给人来把关。更核心的一句话结论是：所谓“接管”，不是替代记者，而是重构新闻生产流水线。谁先把重复劳动自动化，谁就能把时间留给真正重要的深度调查和内容判断。

1. 别再被“接管”两个字吓到：Mythos 2 到底改变了什么

新闻编辑部的工作流程，长期被一件很现实的事情困扰：大量时间被消耗在重复性劳动里。记者要花很多时间检索历史报道、整理采访录音、把同一篇稿件改写成不同平台的版本；编辑要在多个环节做格式检查、信源核对、发布确认。这些工作不是不重要，但它们会挤占深度思考和内容判断的时间。

Mythos 2这类系统切入的，正是这个结构性问题。它没有真正替代“人”在编辑部里的决策角色，而是把流程中最耗时的环节拆解成可以被模型和脚本自动处理的任务。从实际落地角度看，变化主要集中在四个层面：

• 素材检索从“人工翻资料”变成“RAG辅助召回”。
• 初稿生成从“记者完全从头写”变成“模型基于素材生成结构稿”。
• 事实核查从“全人工交叉验证”变成“抽取观点 + 自动比对 + 人工复核”。
• 多平台发布从“手工改写排版”变成“多模板渲染 + 审核后一键分发”。

用表格看会比较直观：

所以“接管”这个词，准确说是“接管流程中的体力活”。对技术团队来说，这反而是一个利好信号：编辑部终于愿意为AI工程化买单了。对开发者来说，这类系统有非常具体的架构和代码挑战，不是接一个大模型API就结束，而是要把检索、生成、审核、发布串成一条可监控、可回滚的流水线。

2. Mythos 2 背后的核心技术栈：Agent、RAG 与大模型协同

把“Mythos 2 已接管各大新闻编辑部”还原成技术问题，核心是三个概念：大模型、RAG、Agent。这三个概念经常被混着说，但在新闻生产场景里，它们的职责非常清晰。

大模型负责“理解和生成”。编辑部里要给一段素材提炼摘要、把口语化采访转写成书面语、根据要点生成一段初稿，这些任务本质上都需要语言理解和生成能力。常用做法是调用通用大模型API，或者在私有化环境部署开源模型。模型选择不在本文讨论范围，但有一个原则可以提前说：在新闻场景，模型的可解释性和可控性，往往比它“是不是最聪明”更重要。

RAG 负责“检索增强生成”。新闻生产最怕两个问题：模型编造信息，以及引用源头不可追溯。RAG 的做法是先把历史稿件、公开报道、通稿材料、政策文件等内容向量化存入知识库，在生成前先根据用户问题检索出相关片段，再把片段和问题一起交给大模型。模型生成的内容必须基于给定的上下文，而不是凭空发挥。这样就有效降低了“幻觉”风险，也方便编辑回溯来源。

Agent 负责“编排与决策”。编辑部的工作不是一次问答就能结束，而是有一连串子任务：判断选题是否完整、判断素材是否充足、判断生成稿是否符合平台风格、判断哪些事实点需要人工复核。Agent 的意义，就是把这些子任务组织成多步骤流程，每一步的结果都可以被记录、被检查、被人工干预。如果把大模型比作撰稿人，RAG 比作资料室，Agent 就是那个拿着任务清单协调资源、推进进度的执行主编。

这三者叠加后的效果，是一套“半自动车床”式的系统。它不会独立完成从选题到发布的全部工作，但它能把已经规范化、流程化的部分做得极其稳定。比如“根据今日通告，生成一条300字快讯初稿并附上来源”这样的任务，完全可以在几十秒内完成。

3. 环境准备与最小系统设计

在写代码之前，先把环境准备清楚。Mythos 2 相关的系统实现并不需要特别复杂的服务器配置，但组件比较多，最好用 Docker 统一管理基础中间件。以下版本建议是一个通用组合，实际项目请以团队技术栈为准，本文重点是演示通用思路。

我建议的最小系统包含四部分：

• 运行环境：Python 3.10 及以上，Node.js 可选（如果前端用）
• 数据存储：PostgreSQL + pgvector 扩展，用来存结构化稿件和向量化素材
• 对象存储：MinIO 或兼容 S3 的服务，用来存图片、附件、历史稿件源文件
• 异步任务：Redis + Celery 或 RQ，用来处理比较耗时的生成任务
• 模型服务：一个 OpenAI 协议兼容的大模型接口，可以调用云端API，也可以部署本地模型

先给出一个docker-compose.yml示例。它只负责把基础设施拉起来，业务代码我们会在后面自己写。

# 文件路径：docker-compose.yml version: "3.9" services: postgres: image: pgvector/pgvector:pg16 container_name: mythos-postgres restart: unless-stopped environment: POSTGRES_USER: mythos POSTGRES_PASSWORD: mythos_password POSTGRES_DB: newsroom ports: - "5432:5432" volumes: - pg_data:/var/lib/postgresql/data redis: image: redis:7.2-alpine container_name: mythos-redis restart: unless-stopped ports: - "6379:6379" minio: image: minio/minio:latest container_name: mythos-minio restart: unless-stopped command: server /data --console-address ":9001" environment: MINIO_ROOT_USER: mythos_admin MINIO_ROOT_PASSWORD: mythos_admin_password ports: - "9000:9000" - "9001:9001" volumes: - minio_data:/data volumes: pg_data: minio_data:

启动命令很简单：

docker-compose up -d

这个步骤会帮你把 PostgreSQL、Redis、MinIO 拉起来。如果是开发环境，这样已经够用；如果是生产环境，建议把数据库单独托管、对象存储走云厂商，同时加上网络白名单和私有网络配置。

Python 依赖建议保持精简，核心库是 FastAPI、SQLAlchemy、pgvector、openai 客户端、langchain 可选，还有 pydantic。下面是一个可以用的requirements.txt：

fastapi==0.115.0 uvicorn[standard]==0.30.6 openai==1.47.1 pydantic==2.9.2 sqlalchemy==2.0.35 pgvector==0.3.6 psycopg2-binary==2.9.9 redis==5.0.8 httpx==0.27.2 python-dotenv==1.0.1

版本号只是一个示例，不同环境里可以用较新的兼容版本。安装命令：

pip install -r requirements.txt

有一点要提前提醒：不要把模型API密钥写进代码里，更不要提交到Git仓库。开发时用.env文件，生产环境用密钥管理服务。

4. 把“新闻编辑部”拆成一条 Agent 流水线

现在进入最核心的一步：把编辑部的工作流拆成可编排的Agent流水线。这里的关键不是每个人都写一堆复杂代码，而是先定义清楚每一步的输入、输出和失败处理方式。

一条典型的AI新闻生产流水线可以这样拆分：

1. 选题校验：输入是一句话或一个关键词，输出是结构化的选题描述，包括报道角度、预计覆盖范围、目标栏目。
2. 素材采集与召回：输入是选题描述，输出是相关信息列表，每条必须带来源URL或文档ID。
3. 初稿生成：输入是选题描述 + 召回素材，输出是带引用标记的初稿内容。
4. 事实核查：输入是初稿，输出是待核查事实点列表，以及每个事实点与素材源的匹配度判断。
5. 多平台改写：输入是初稿，输出是不同平台的版本，比如APP版、微信公众号版、微博版。
6. 人工审核与发布：所有AI生成的版本都必须经过编辑确认，确认后才能写入发布队列。

在这个流程里，Agent编排层就像一个负责人，把“任务状态”从一个节点传到下一个节点。为了让人工审核可追溯，每一步产生的中间结果都应该落库保存，不能只保留最终发布稿。

下面用一段简化代码演示这个编排逻辑。先把主要的Agent节点定义出来，然后按顺序执行。

# 文件路径：app/news_pipeline.py from dataclasses import dataclass, field from typing import List import json import httpx @dataclass class NewsTopic: title: str angle: str = "" sources: List[str] = field(default_factory=list) @dataclass class DraftResult: content: str claims: List[dict] = field(default_factory=list) class MythosNewsPipeline: def __init__(self, llm_endpoint: str, api_key: str): self.llm_endpoint = llm_endpoint self.api_key = api_key def _call_llm(self, prompt: str, system_prompt: str = "") -> str: headers = {"Authorization": f"Bearer {self.api_key}"} payload = { "model": "mythos-2-demo", "messages": [ {"role": "system", "content": system_prompt}, {"role": "user", "content": prompt}, ], "temperature": 0.3, } resp = httpx.post( f"{self.llm_endpoint}/chat/completions", headers=headers, json=payload, timeout=60, ) resp.raise_for_status() return resp.json()["choices"][0]["message"]["content"] def validate_topic(self, raw_text: str) -> NewsTopic: prompt = f"请将下面的输入拆解成一个新闻选题对象，输出JSON格式，包含title和angle字段。\n输入：{raw_text}" # 实际项目中建议用JSON Mode或结构化输出，这里为了演示直接解析 output = self._call_llm(prompt, system_prompt="你是一个新闻选题编辑。") data = json.loads(output) return NewsTopic(title=data["title"], angle=data.get("angle", "")) def generate_draft(self, topic: NewsTopic, context_chunks: List[str]) -> DraftResult: context = "\n\n".join( f"[{i+1}] {chunk}" for i, chunk in enumerate(context_chunks) ) prompt = f""" 请根据下面的素材写一篇新闻初稿。 要求： 1. 开头写明导语，概括核心事实。 2. 正文引用素材中的关键数字和信息，用 [1] [2] 标注来源序号。 3. 不要编造素材中不存在的事实。 4. 结尾留一个“有待核实”清单。 选题：{topic.title} 角度：{topic.angle} 素材： {context} """ content = self._call_llm(prompt, system_prompt="你是一名严谨的新闻记者。") return DraftResult(content=content) def run(self, raw_text: str, retriever=None): topic = self.validate_topic(raw_text) # 假设retriever是一个可调用对象，输入topic，返回素材列表 chunks = retriever(topic.title, topic.angle) if retriever else [] draft = self.generate_draft(topic, chunks) return { "topic": topic, "draft": draft, "status": "waiting_for_review", }

这份代码的重点不是让你直接用，而是理解编排层应该做什么：选题解析、调用RAG、生成初稿、返回可审查结构。一个实际的Mythos类系统，通常会在这一步引入异步任务队列，否则生成一篇长稿可能要等十几秒甚至几十秒，HTTP请求容易超时。

5. 完整示例：从一句话选题到多平台初稿

为了让你能真正跑通一个最小闭环，这里给出一个可运行的FastAPI服务。它不依赖复杂框架，核心就是三个文件：一个入口，一个流水线逻辑，一个事实核查工具。

第一步，先创建FastAPI入口。我们暴露两个接口：一个是异步创建任务，另一个是根据任务ID查询状态。这里为了简洁，用内存字典存储任务状态，可以理解为最简版的“任务队列”。

# 文件路径：app/main.py import uuid from fastapi import FastAPI, HTTPException from pydantic import BaseModel from app.news_pipeline import MythosNewsPipeline app = FastAPI(title="Mythos News Service") tasks: dict[str, dict] = {} class TopicRequest(BaseModel): raw_text: str def get_retriever(): # 这里可以用pgvector或外部搜索服务实现 # 本文为了演示，直接返回两条固定素材 def _fake_retriever(title: str, angle: str): return [ "某公司在12月1日发布了2025年三季度财报，营收同比增长12%。", "该公司发言人称，增长主要来自海外业务，预计明年将继续扩大投入。", ] return _fake_retriever @app.post("/api/v1/tasks") def create_task(req: TopicRequest): task_id = str(uuid.uuid4()) tasks[task_id] = {"status": "pending", "result": None} # 生产环境应该使用Celery/RQ异步执行，这里直接同步调用演示 pipeline = MythosNewsPipeline( llm_endpoint="http://your-llm-endpoint:8000/v1", api_key="your-api-key", ) try: result = pipeline.run(req.raw_text, retriever=get_retriever()) tasks[task_id] = {"status": "success", "result": result} except Exception as e: tasks[task_id] = {"status": "failed", "error": str(e)} return {"task_id": task_id} @app.get("/api/v1/tasks/{task_id}") def get_task(task_id: str): task = tasks.get(task_id) if not task: raise HTTPException(status_code=404, detail="task not found") return task

第二步，加上事实核查模块。这里先用规则实现一个最小版本：从初稿里找出所有带数字的句子，然后判断该数字是否出现在素材里。生产环境会更复杂，会涉及实体识别、属性对齐、模糊匹配等，但这个小例子已经能说明问题。

# 文件路径：app/claim_check.py import re from typing import List def extract_numeric_claims(text: str) -> List[str]: sentences = re.split(r"[。\n]", text) claims = [] for sentence in sentences: if re.search(r"\d+", sentence): claims.append(sentence.strip()) return claims def check_claims_against_sources(claims: List[str], sources: List[str]) -> List[dict]: checks = [] for claim in claims: matched = False for source in sources: # 简单判断：关键数字都出现在source中，且人名/公司名也尽量一致 numbers = re.findall(r"\d+(?:\.\d+)?%?", claim) if numbers: matched = all(num in source for num in numbers) if matched: break checks.append({ "claim": claim, "matched": matched, "need_review": not matched, "suggestion": "证据充分" if matched else "请人工核对数字、时间、主体", }) return checks if __name__ == "__main__": source = [ "某公司在12月1日发布了2025年三季度财报，营收同比增长12%。", "该公司发言人称，增长主要来自海外业务。", ] claims = extract_numeric_claims("该公司12月1日发布财报，营收同比增长12%。增长主要来自海外业务。") result = check_claims_against_sources(claims, source) for r in result: print(r)

第三步，运行服务并调用接口。启动命令如下：

cd your_project uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload

然后打开另一个终端，发送一个最简单的请求：

curl -X POST http://localhost:8000/api/v1/tasks \ -H "Content-Type: application/json" \ -d '{"raw_text": "某公司发布了2025年三季度财报，营收增长明显"}'

如果代码正常，你会收到一个类似下面的响应：

{ "task_id": "0b07a80d-4f40-4e10-9c85-1cd2f7e0ae1f" }

再用这个ID查询结果：

curl http://localhost:8000/api/v1/tasks/0b07a80d-4f40-4e10-9c85-1cd2f7e0ae1f

返回结果里会包含选题解析结果、初稿内容和待核查清单。到这里，一个最小AI新闻辅助系统就跑通了。

6. 运行结果与效果验证

上面这个示例，核心是为了验证一个事：AI能不能把“一句话被拆成结构化任务、生成初稿、给出核查点”这个闭环跑通。运行后，我们要重点看几个地方。

第一，看选题解析是否正确。如果输入的原始描述很含糊，模型是否能把“报道角度”提取出来？如果提取出来的角度和真正新闻价值点不符，后面生成的初稿也会偏。第二，看初稿是否严格引用了素材。一个有效的信号是，初稿里必须出现类似[1][2]的引用标记，且这些标记对应的内容能在素材里找到。第三，看事实核查模块有没有标出“待人工复核”的内容。如果一条带数字的新闻没有触发任何复核建议，说明要么素材和稿件高度一致，要么核查模块太薄弱。

一个比较理想的输出看起来是这样的：

{ "status": "success", "result": { "topic": { "title": "某公司发布2025年三季度财报", "angle": "营收同比增长12%，海外业务为主要增长引擎" }, "draft": { "content": "某公司12月1日发布财报，营收同比增长12% [1]。公司发言人称…… [2]。有待核实：海外业务具体占比。", "claims": [ { "claim": "营收同比增长12%", "matched": true, "need_review": false } ] } } }

如果运行失败，不要急着看复杂中间层，先按顺序检查以下几处：

1. 模型API地址是否通。用curl直接访问/chat/completions接口验证。
2. .env里是否配置了正确的 API Key，且没有提交到仓库。
3. RAG 检索结果是否为空。如果素材为空，初稿质量会急剧下降，甚至报错。
4. JSON解析是否失败。很多模型在没有结构化输出约束时，会在返回内容里夹带额外文字，导致json.loads失败。

7. 常见问题与排查思路

在编辑部场景落地时，常见问题往往不在算法层面，而在工程链路和人工协作边界上。我整理了几个高频问题：

排查的重点是“先看链路，再调模型”。很多问题在工程层就能解决，不要一开始就反复改prompt。

8. 最佳实践与工程建议

AI新闻生产系统看起来是算法问题，但真正拉开差距的是工程规范。以下几条建议，来自常见落地项目中的共性经验。

第一，人审闸门不能省。哪怕模型已经具备较高的事实一致性，编辑部仍需要保留最终审核权限。系统设计上应该把“AI生成”和“人工审核通过”分成两个明确状态，未审核的稿件不能进入发布通道。这个流程看起来多了一步，但它能解决大部分由此带来的风险和响应速度问题。

第二，所有素材必须有来源。很多模型输出看起来“通顺合理”，但无法回答“这个数据从哪里来”。规范做法是：每条生成内容都携带引用ID，引用ID必须指向知识库中的具体文档或外部信源。如果RAG检索不到可信来源，系统应该提示“素材不足”，而不是强行生成。

第三，建立本地评测集。不要只靠人工感觉判断系统好坏。建议收集团队过往的重点稿件，整理成测试集，标注出“正确事实点”和“错误事实点”。每次修改prompt或替换模型，都先跑一遍评测集，用“事实不一致率”“人工修改率”等指标量化效果。

第四，配置和模板要分离。选题提示词、稿件风格、平台模板、发布渠道，不要硬编码在代码里。最好像这样：

# 文件路径：config/platforms.yml platforms: app: max_length: 600 style: "信息密度优先，避免长句" need_image: true wechat: max_length: 1200 style: "导语感强，段落短，保留叙事节奏" need_summary: "一句话摘要" twitter_cn: max_length: 280 style: "口语化，直接给结论"

这样产品运营可以自行维护风格，开发者不需要频繁改代码。

第五，安全和合规底线要前置。新闻生产的数据可能涉及个人隐私、未公开信息或敏感数据。系统设计时要做到最小权限：素材检索系统只能访问经过授权的语料；日志里不记录完整用户文本；对外发布前必须通过敏感词过滤和合规审核。这不是“以后再说”的事，而是一开始就要画出的边界。

第六，任务要支持幂等和重试。生成任务可能因为API抖动而失败，如果重试时重复生成，会造成稿件冲突和素材浪费。建议为每个任务维护唯一ID，并把中间状态写入数据库。重试时先查状态，避免重复计算。

9. 总结与后续学习方向

“Mythos 2 已接管各大新闻编辑部”这类标题，背后真正想表达的是：AI辅助内容生产已经进入流程化阶段。它不再是一个单独的“写稿机器人”，而是由大模型、RAG、Agent、审核系统、发布系统共同组成的复杂工作流。对开发者来说，这是一次很典型的工程化机会。

如果你所在团队正准备尝试做类似系统，我建议从最小闭环开始：先解决“一句话选题 -> 素材召回 -> 初稿生成 -> 人工审核”这个主链路，不要一开始就做多平台发布和复杂编排。先让编辑用起来，收集他们的修改意见，比先追求技术大而全更有价值。之后可以继续深入三个方向：一是RAG检索质量控制，二是Agent流程可视化与可观测性，三是内容安全的自动审核模型。这几个方向每一个都足够做深，也会直接决定系统能否从一个Demo变成编辑部真正每天在用的基础设施。

需要特别提醒的是，技术能代替流程里的重复劳动，但代替不了编辑的判断力。任何接入编辑部的AI系统，都应该把“人”放在审核和决策的关键节点上。这样既能让AI发挥效率优势，也能守住信息内容的底线。