---
title: "Grox 服务架构"
description: "内容理解服务 Grox：一个主进程加 dispatcher 与 engine 两个子进程，flows 插件式注册，按 Plan KEY 路由，共 21 个 Plan"
---

> Documentation Index
> Fetch the complete documentation index at: https://x-algorithm.vergego.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Grox 服务架构

## 这一页回答什么

本页说明内容理解服务 Grox 的进程构成、任务从 Kafka 到执行的流转过程，以及 Task 与 Plan 的关系。2026-11 版本把业务逻辑移入 `grox/flows/`，本页也说明一条帖子在此版本中由哪些 Plan 处理。

## 核心结论

1. **Grox 由一个主进程和两个子进程组成**：`Dispatcher`（拉取与调度）与 `Engine`（执行）各自运行在独立的 `multiprocessing.Process` 中，两者通过两个进程间队列通信。gRPC 服务和 Kerberos 续期留在主进程（`grox/main.py:32-48`）。
2. **框架代码与业务代码分离**：`grox/core/` 只存放框架代码。Plan、任务生成器和 gRPC servicer 只能写在 `grox/flows/` 下，并用 `@register` 注册。启动时，`load_all()` 递归导入 `grox/flows/` 并执行校验（`grox/core/loading.py:58-66`）。
3. **Plan KEY 决定路由**：每个生成器声明 `PLANS_TO_INJECT`，并把 Plan 的 `KEY` 写入 `TaskPayload.plans`。Plan 只在自己的 `KEY` 出现在 `plans` 中时执行（`grox/core/plans/plan.py:74-75`）。旧版的 `TaskEligibility` 路由已删除。
4. **当前共有四个 flow、21 个 Plan、26 个生成器**：`mm_emb`（多模态嵌入，8 个 Plan）、`ptos`（安全策略，7 个）、`reply_spam`（回复打分与垃圾回复，4 个）、`upa`（帖子统一注解，2 个）。
5. **dispatcher 不再重投失败任务**：旧版按 `max_attempts` 把失败任务重新入队。现在任务失败后，dispatcher 调用生成器的 `on_terminal_failure`，随后照常 `ack`（`grox/core/dispatcher.py:146-160`）。Task 基类仍保留 2 次重试，但多数分类 Task 与写入 Task 主动绕过这项重试。
6. **Grox 是旁路服务**：Grox 不在 For You 请求路径上。它流式消费帖子，产出安全标签、帖子注解、回复分和嵌入。下游服务再读取这些结果。

## 进程与启动

`main.py::serve()` 创建共享的 `ScheduleContext`，其中包含 `task_queue`、`resp_queue` 和两个关停事件。随后它按顺序启动 `Engine`、`Dispatcher`、`GrpcServer`（`grox/main.py:32-48`）：

```mermaid
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`）：

1. `_assert_framework_only`：`grox/core/plans/` 与 `grox/core/generators/` 只允许存放框架文件。出现业务文件时，启动失败（`grox/core/loading.py:25-32`）。
2. 递归导入 `grox/flows/`，触发所有 `@register`。
3. `_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` 定义：

1. `_eligible` 检查 `self.KEY in ctx.plans`。不满足时，它返回 `None`（`grox/core/plans/plan.py:74-75`）。
2. Plan 为每个被依赖的任务创建一个 `asyncio.Future`，并用 `asyncio.gather` 并发启动所有任务。每个任务先等待其依赖的 future。
3. **依赖任务为 SKIPPED 时，本任务也记为 SKIPPED**（`grox/core/plans/plan.py:86-89`）。因此，过滤器 Task 与去重 Task 能中止整条链路。
4. 任一任务抛出异常时，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 内容分类器](https://x-algorithm.vergego.ai/concepts/grox-classifiers/)；嵌入类 Plan 见 [多模态帖子嵌入器](https://x-algorithm.vergego.ai/concepts/multimodal-embedders/)。

一条帖子被处理的次数，取决于它出现在哪些 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 为帖子打上安全标签，[可见性过滤](https://x-algorithm.vergego.ai/concepts/visibility-filtering/) 和 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 内容分类器](https://x-algorithm.vergego.ai/concepts/grox-classifiers/) —— 各 flow 中的分类器、输出与阈值
- [多模态帖子嵌入器](https://x-algorithm.vergego.ai/concepts/multimodal-embedders/) —— `mm_emb` flow 的三代嵌入器
- [图片与视频模型](https://x-algorithm.vergego.ai/concepts/media-models/) —— Grox 之外的媒体模型服务
- [可见性过滤](https://x-algorithm.vergego.ai/concepts/visibility-filtering/) —— 读取 Grox 写入的安全标签
- [系统架构总览](https://x-algorithm.vergego.ai/concepts/system-architecture/) —— Grox 作为旁路服务的定位
- [端到端数据流：一条帖子从发布到被推荐](https://x-algorithm.vergego.ai/concepts/end-to-end-dataflow/) —— Grox 旁路管线与 feed 请求路径的汇合点

Source: https://x-algorithm.vergego.ai/concepts/grox-architecture/index.mdx