这一页回答什么
本页说明内容理解服务 Grox 的进程构成、任务从 Kafka 到执行的流转过程，以及 Task 与 Plan 的关系。2026-11 版本把业务逻辑移入 grox/flows/，本页也说明一条帖子在此版本中由哪些 Plan 处理。
核心结论
- Grox 由一个主进程和两个子进程组成：
Dispatcher（拉取与调度）与Engine（执行）各自运行在独立的multiprocessing.Process中，两者通过两个进程间队列通信。gRPC 服务和 Kerberos 续期留在主进程（grox/main.py:32-48）。 - 框架代码与业务代码分离：
grox/core/只存放框架代码。Plan、任务生成器和 gRPC servicer 只能写在grox/flows/下，并用@register注册。启动时，load_all()递归导入grox/flows/并执行校验（grox/core/loading.py:58-66）。 - Plan KEY 决定路由：每个生成器声明
PLANS_TO_INJECT，并把 Plan 的KEY写入TaskPayload.plans。Plan 只在自己的KEY出现在plans中时执行（grox/core/plans/plan.py:74-75）。旧版的TaskEligibility路由已删除。 - 当前共有四个 flow、21 个 Plan、26 个生成器：
mm_emb（多模态嵌入，8 个 Plan）、ptos（安全策略，7 个）、reply_spam（回复打分与垃圾回复，4 个）、upa（帖子统一注解，2 个）。 - dispatcher 不再重投失败任务：旧版按
max_attempts把失败任务重新入队。现在任务失败后，dispatcher 调用生成器的on_terminal_failure，随后照常ack（grox/core/dispatcher.py:146-160）。Task 基类仍保留 2 次重试，但多数分类 Task 与写入 Task 主动绕过这项重试。 - Grox 是旁路服务：Grox 不在 For You 请求路径上。它流式消费帖子，产出安全标签、帖子注解、回复分和嵌入。下游服务再读取这些结果。
进程与启动
main.py::serve() 创建共享的 ScheduleContext，其中包含 task_queue、resp_queue 和两个关停事件。随后它按顺序启动 Engine、Dispatcher、GrpcServer（grox/main.py:32-48）：
flowchart LR
K[("Kafka 主题")] --> G["StreamTaskGenerator<br/>写入 PLANS_TO_INJECT"]
G --> P["PriorityTaskGenerator<br/>按 weight 加权抽样"]
P --> D["Dispatcher 子进程"]
D -- "task_queue" --> E["Engine 子进程"]
E -- "resp_queue" --> D
E --> PM["PlanMaster<br/>21 个 Plan 并发"]
PM --> S[("Strato / Manhattan / Kafka 写入")]
M["主进程<br/>GrpcServer + Kerberos"] -.-> D
Engine与Dispatcher都通过multiprocessing.Process启动（grox/core/engine.py:157、grox/core/dispatcher.py:196）。GrpcServer在主进程中注册健康检查，以及grox/flows/下登记的 servicer（grox/core/services/service.py:41-51）。开源代码的grox/flows/中没有 servicer 用@register登记。因此，开源版 gRPC 只暴露健康检查。- 关停时，主进程先触发
queue_connection_shutdown，再等待shutdown_drain_timeout秒，让在途任务执行完毕（grox/main.py:58）。开源默认值为 300 秒（grox/config/config.py:229）。
flows：插件式注册
grox/core/registry.py:8 的 @register 按类型分发：Plan 进入 Plan 注册表，TaskGenerator 进入生成器注册表，GrpcServicer 进入 servicer 注册表。load_all() 执行三个步骤（grox/core/loading.py:58-66）：
_assert_framework_only：grox/core/plans/与grox/core/generators/只允许存放框架文件。出现业务文件时，启动失败（grox/core/loading.py:25-32）。- 递归导入
grox/flows/，触发所有@register。 _validate_plan_keys检查 Plan KEY 不重复，并检查每个生成器注入的 KEY 都有对应 Plan。_validate_generator_types检查配置中声明的生成器类型都已注册（grox/core/loading.py:35-55）。
新增一类内容分析时，开发者只需在 grox/flows/ 下建目录，编写 Plan、Task 和生成器，不修改 grox/core/。
Dispatcher：拉取与确认
Dispatcher（grox/core/dispatcher.py）在子进程中并发运行四个循环（grox/core/dispatcher.py:180-188）：
_fill_loop：从PriorityTaskGenerator.poll()获取TaskPayload。在途任务数达到max_in_flight时，它等待；否则，它把任务推入task_queue（grox/core/dispatcher.py:91-110）。max_in_flight的开源默认值为 3（grox/config/config.py:168）。_result_loop：从resp_queue获取TaskResult。任务成功时，它执行ack。任务失败时，它记录dispatcher.result.failed.final.count，调用on_terminal_failure，并在能识别消息来源时同样执行ack（grox/core/dispatcher.py:127-164）。_inflight_monitor_loop每 15 秒上报一次在途任务数。_wait_for_queue_connection_shutdown在关停信号到达后停止生成器。
PriorityTaskGenerator 封装配置中的全部生成器。每次取任务时，它按 weight 用 random.choices 抽取一个流。某个流在本轮没有数据时，它把该流从本轮候选中移除（grox/core/generators/task_generator.py:100-138）。单个生成器可以配置 max_qps，超过限额时让出本轮（grox/core/generators/task_generator.py:22）。
StreamTaskGenerator 把 Kafka 消息转换成 TaskPayload。task_type 取生成器类型名，plans 取 PLANS_TO_INJECT 的副本（grox/core/generators/stream_generator.py:32-45）。
Engine：执行
Engine（grox/core/engine.py）在子进程中初始化 MediaProcessor、ASRProcessor 和已登记的处理器（grox/core/engine.py:48-54）。随后它对每个任务调用 asyncio.create_task(self._run_task(task))，并发执行（grox/core/engine.py:147）。
_run_task 用 asyncio.wait_for 包装 PlanMaster.exec，超时上限为 task_timeout（grox/core/engine.py:73-92），开源默认值为 900 秒（grox/config/config.py:153）。出现超时、依赖被取消或其他异常时，_run_task 都返回一个 success=False 的 TaskResult。
Task：最小执行单元
Task（grox/core/tasks/task.py:26）是抽象基类：
exec使用tenacity重试，最多 2 次，间隔 1 秒（grox/core/tasks/task.py:30）。should_skip先检查DISABLE_RULES，再检查_should_skip。任一项命中时，它返回SKIPPED。_exec抛出TaskStopExecution时，结果同样记为SKIPPED。其他异常向上抛出。- 便捷子类只剩
TaskWithUser与TaskWithPost（grox/core/tasks/task.py:92-117）。旧版的TaskWithUserContext、TaskWithContentAnalysis已删除。
多数分类 Task 和嵌入写入 Task 覆写 exec，直接调用 Task.exec.__wrapped__，从而绕过 2 次重试（例如 grox/flows/ptos/task_safety_ptos_category.py:83）。嵌入写入 Task 改用自己的 3 次重试，间隔为 1 秒和 2 秒（grox/flows/mm_emb/task_write_mm_embedding_sink.py:32）。
共享 Task
| Task | 作用 | 出处 |
|---|---|---|
TaskTTLDedupeWithPost |
名称是“rate limit”，实际执行 60 秒 TTL 去重：同一帖子在 60 秒内第二次进入同一 Plan 时跳过 | grox/core/tasks/task_rate_limit.py:44-59 |
TaskFilterWithPost |
过滤器基类。_eligible_with_post 返回 False 时，它抛出 TaskStopExecution |
grox/core/tasks/task_filters.py:24 |
TaskMediaHydration |
下载图片、抽取视频帧、缩放尺寸 | grox/core/tasks/task_media.py:8-13 |
TaskASRTranscription |
把视频语音转成文字，供嵌入使用 | grox/core/tasks/task_asr.py |
TaskLoadPost |
只有帖子 ID 时，从 Strato 补全帖子 | grox/core/tasks/task_load_post.py:9-15 |
DisableTaskForNonProd |
在非生产环境禁用写入类 Task | grox/core/tasks/disable_rules.py:19-24 |
媒体水合分档
TaskMediaHydration 按帖子热度决定帧数和分辨率。下表均为开源默认值（grox/config/config.py:104-131）：
| 条件 | 视频最多帧数 | 视频帧边长 | 图片边长 |
|---|---|---|---|
| 非主帖（引用帖、祖先帖等） | video_max_frames_light = 2 |
448 | 448 |
主帖，赞数 < deluxe_fav_count_threshold（64） |
video_max_frames = 20 |
448 | 448 |
| 主帖，赞数 ≥ 64 | deluxe_video_max_frames = 30 |
600 | 600 |
分档逻辑位于 grox/core/data_loaders/media_loader.py:616-634。赞数 ≥ 64 的原帖还会额外拉取回复（descendants），供分类器参考（grox/core/data_loaders/media_loader.py:135-145）。
Plan：Task 的 DAG
Plan（grox/core/plans/plan.py:16）由 TASKS、TASK_DEPENDENCIES 和 KEY 定义：
_eligible检查self.KEY in ctx.plans。不满足时，它返回None（grox/core/plans/plan.py:74-75）。- Plan 为每个被依赖的任务创建一个
asyncio.Future，并用asyncio.gather并发启动所有任务。每个任务先等待其依赖的 future。 - 依赖任务为 SKIPPED 时，本任务也记为 SKIPPED（
grox/core/plans/plan.py:86-89）。因此，过滤器 Task 与去重 Task 能中止整条链路。 - 任一任务抛出异常时，Plan 结果为
success=False。
Task 之间不传递返回值。它们通过 TaskContext.state(cls) 共享按类型存放的状态对象（grox/core/schedules/types.py:50-56）。例如，PTOS 类别检测把结果写入 SafetyPtosState，策略判定再从中读出。
PlanMaster：并行执行所有 Plan
PlanMaster 不再硬编码 Plan 列表。它在首次执行时实例化注册表中的全部 Plan（grox/core/plans/plan_master.py:14-21）。exec 把任务交给所有 Plan 并发执行，只有 KEY 匹配的 Plan 真正运行。merge_results 取最早开始时间和最晚结束时间，success 取全部结果的逻辑与（grox/core/plans/plan_master.py:24-42）。旧版合并 content_categories、multimodal_post_embedding 字段的逻辑已删除。
flows 全景
| flow | Plan KEY | 触发它的生成器（Kafka 主题） | 产出 |
|---|---|---|---|
upa |
banger_initial_screen |
最小热度流、恢复流、优先流 | 帖子统一注解（话题、标签、布尔元数据） |
upa |
post_safety |
热门流 ...-unified-posts-popular |
补写布尔元数据 |
ptos |
safety_ptos |
最小热度、最小曝光、恢复、deluxe、回填共 5 个流 | 安全注解、安全标签、处置 |
ptos |
safety_ptos_realtime_with_signals |
延迟复制流 | 垃圾信息判定 |
ptos |
safety_ptos_live_cluster_anchors |
safety-ptos-live-cluster-anchors |
集群锚点处置决定 |
ptos |
safety_ptos_special_video |
safety-ptos-special-video |
安全注解 |
ptos |
safety_ptos_adult_content_leading_frames |
safety-ptos-adult-content-leading-frames |
安全注解 |
ptos |
safety_ptos_key_frames |
safety-ptos-key-frames |
安全注解 |
ptos |
safety_ptos_media_injected_adult_infrared_video_spam_detection |
同名主题 | 安全注解 |
reply_spam |
reply_ranking、spam_comment、coordinated_spam、multi_step_reply_spam |
content-understanding-realtime-unified-posts-v3（同时注入 4 个 Plan）；恢复流注入除 coordinated_spam 外的 3 个 |
回复分、垃圾回复标签 |
mm_emb |
mm_emb_v5、mm_emb_v8_2、mm_emb_v8_search |
grox-multimodal-embedding-requests-with-summary 及 v5、v8 恢复流 |
原帖嵌入 |
mm_emb |
mm_emb_v5_for_reply、mm_emb_v8_2_for_reply、mm_emb_v8_search_for_reply |
...-min-traction-for-grox-multi-modal |
回复嵌入 |
mm_emb |
mm_emb_v8_search_backfill_batch → mm_emb_v8_search_backfill |
grox-multimodal-embedding-backfill |
为历史帖子补算嵌入 |
生成器定义见 grox/flows/upa/generators.py:18-51、grox/flows/ptos/generators.py:46-144、grox/flows/reply_spam/generators.py:16-40、grox/flows/mm_emb/generators.py:47-158。各 Plan 内部运行的分类器见 Grox 内容分类器；嵌入类 Plan 见 多模态帖子嵌入器。
一条帖子被处理的次数，取决于它出现在哪些 Kafka 主题中。同一条消息只进入其所在生成器注入的 Plan。
设计决策
| 决策 | 理由 |
|---|---|
| 一个主进程加两个子进程 | 拉取调度与重型执行相互隔离；gRPC 健康检查不受执行负载影响 |
业务代码只能写在 grox/flows/ |
_assert_framework_only 锁定框架目录。新增分析不修改框架，依赖方向保持单向 |
| 用 Plan KEY 路由 | 生成器声明“本流运行哪些 Plan”，启动时可静态校验 KEY 是否存在。它替代需要两处同步的 TaskEligibility |
| PlanMaster 读取注册表 | 新增 Plan 时不修改 PlanMaster |
| 取消 dispatcher 重投 | 失败交给生成器的 on_terminal_failure 和各 flow 的恢复主题，避免同一任务在本地反复占用 max_in_flight 名额（推测） |
| 分类 Task 绕过基类重试 | 单次 LLM 调用成本高；失败由 Plan 结果和恢复流兜底 |
| 关停时等待 300 秒 | 让在途任务执行完毕，避免丢失数据 |
FAQ
Q：Grox 的产出写到哪里？ A：Grox 把安全注解结果、统一帖子注解、回复分和多模态嵌入等产出写入 Strato 背后的 Manhattan 存储与 Kafka。Grox 直接调用 Strato 为帖子打上安全标签，可见性过滤 和 home-mixer 的品牌安全判定再读取这些标签。Grox 本身不参与在线打分。
Q：开源代码能直接运行吗？
A：不能。YAML 配置（grox/config/yaml/）、提示词模板和多数 Strato 客户端都不在仓库中（grox/config/config.py:25-37、grox/flows/ptos/prompts.py:11）。本页的常量均为开源默认值，线上值由 YAML 覆盖。
源码锚点
grox/main.py:32-73—— 进程启动与关停grox/core/loading.py:25-66—— flows 导入与启动校验grox/core/registry.py:8-17——@register分发grox/core/dispatcher.py:91-188—— 拉取循环、结果循环grox/core/generators/task_generator.py:70-166—— 加权抽样生成器grox/core/generators/stream_generator.py:12-48——PLANS_TO_INJECT注入grox/core/engine.py:48-147—— 执行循环与 900 秒超时grox/core/tasks/task.py:26-117—— Task 基类grox/core/plans/plan.py:16-103—— Plan 的 DAG 执行grox/core/plans/plan_master.py:10-42—— 读取注册表并并发执行grox/config/config.py:104-169—— 媒体水合、engine、dispatcher 开源默认值
相关页面
- Grox 内容分类器 —— 各 flow 中的分类器、输出与阈值
- 多模态帖子嵌入器 ——
mm_embflow 的三代嵌入器 - 图片与视频模型 —— Grox 之外的媒体模型服务
- 可见性过滤 —— 读取 Grox 写入的安全标签
- 系统架构总览 —— Grox 作为旁路服务的定位
- 端到端数据流：一条帖子从发布到被推荐 —— Grox 旁路管线与 feed 请求路径的汇合点