Dropbox 的 Riviera 并不是为生成式 AI 临时搭建的一条新管道，而是一个经过近十年迭代的通用内容处理平台。它长期承担产品中的内容转换工作，如今又需要面对 AI 带来的新需求：解析文档结构、切分上下文、生成向量表示，并持续追踪模型与提示词版本。
来源摘要没有披露 Riviera 的内部 API 或实现细节，因此下面不复刻其架构，而是分析这类平台为何能够跨越多个技术周期，以及团队可以怎样实践类似的设计。
真正稳定的抽象是“内容变换”
内容处理系统很容易从几个孤立任务起步：生成缩略图、提取文本、转码视频或者读取文档元数据。每个任务单独实现并不困难，困难在于规模扩大后仍然保持一致的执行语义。
一个通用平台可以把处理过程抽象为：
输入内容 + 转换器版本 + 参数 -> 输出制品 + 元数据
这个模型不关心转换器运行的是传统解析器还是 AI 模型。以下操作都可以落在同一套生命周期中：
- PDF 转纯文本
- 图片转缩略图
- 视频转预览片段
- 文档转结构化段落
- 段落转向量表示
- 内容转摘要或分类标签
平台真正需要统一的是任务调度、重试、缓存、版本管理、权限校验、资源限制和可观测性。具体算法则应该留在可独立演进的转换器中。
AI 改变了制品，却没有推翻处理管线
传统转换通常相对确定：同一文件交给同一版本的解析器，理应得到相同输出。AI 处理增加了更多变量，包括模型版本、提示词、采样参数、分块策略和外部服务状态。
因此，AI 制品不能只记录一段结果文本。至少还应保留：
- 源内容的不可变标识或哈希
- 转换器、模型和提示词版本
- 分块参数及父子块关系
- 输入与输出的内容类型
- 安全策略和访问范围
- 执行时间、费用、失败原因与重试次数
这让平台可以回答几个关键问题：某个摘要由什么版本生成，模型升级后哪些制品需要重算，以及用户失去源文件权限后哪些派生数据必须同步失效。
缓存键也应覆盖所有影响结果的因素。只用文件 ID 缓存 AI 结果，会让模型升级、提示词调整或文件覆盖后的旧数据悄悄混入新流程。
可以这样实践：构建可版本化的最小转换器
下面是一个可直接运行的 Python 示例。它没有使用 Riviera 的内部接口，而是演示通用内容平台需要的几个基本语义：转换器注册、内容寻址缓存、显式版本和链式执行。
将代码保存为 pipeline.py，使用 Python 3.10 或更高版本运行。示例只依赖标准库。
from __future__ import annotations
import hashlib
import json
from dataclasses import dataclass
from pathlib import Path
from typing import Callable
CACHE_DIR = Path(".content-cache")
CACHE_DIR.mkdir(exist_ok=True)
@dataclass(frozen=True)
class Artifact:
media_type: str
data: bytes
@dataclass(frozen=True)
class Transformer:
name: str
version: str
input_type: str
output_type: str
run: Callable[[bytes, dict], bytes]
def cache_key(source: Artifact, transformer: Transformer, params: dict) -> str:
descriptor = json.dumps(
{
"source_sha256": hashlib.sha256(source.data).hexdigest(),
"transformer": transformer.name,
"version": transformer.version,
"params": params,
},
sort_keys=True,
separators=(",", ":"),
).encode()
return hashlib.sha256(descriptor).hexdigest()
def execute(source: Artifact, transformer: Transformer, **params) -> Artifact:
if source.media_type != transformer.input_type:
raise ValueError(
f"expected {transformer.input_type}, got {source.media_type}"
)
key = cache_key(source, transformer, params)
output_path = CACHE_DIR / key
if output_path.exists():
output = output_path.read_bytes()
print(f"cache hit: {transformer.name}@{transformer.version}")
else:
output = transformer.run(source.data, params)
output_path.write_bytes(output)
print(f"executed: {transformer.name}@{transformer.version}")
return Artifact(transformer.output_type, output)
def normalize_text(data: bytes, params: dict) -> bytes:
text = data.decode("utf-8")
normalized = " ".join(text.split())
return normalized.encode("utf-8")
def chunk_text(data: bytes, params: dict) -> bytes:
words = data.decode("utf-8").split()
size = int(params.get("size", 8))
chunks = [" ".join(words[i:i + size]) for i in range(0, len(words), size)]
return json.dumps({"chunks": chunks}, ensure_ascii=False).encode("utf-8")
NORMALIZE = Transformer(
name="normalize-text",
version="1.0.0",
input_type="text/plain",
output_type="text/plain",
run=normalize_text,
)
CHUNK = Transformer(
name="chunk-for-ai",
version="1.0.0",
input_type="text/plain",
output_type="application/json",
run=chunk_text,
)
if __name__ == "__main__":
source = Artifact(
"text/plain",
"Riviera turns content into reusable product artifacts.".encode(),
)
normalized = execute(source, NORMALIZE)
chunks = execute(normalized, CHUNK, size=5)
print(chunks.data.decode("utf-8"))
运行：
python pipeline.py
python pipeline.py
第二次执行会命中缓存。把 CHUNK.version 改为 1.1.0，任务会重新计算；调整 size 参数也会生成新的制品。这种行为正是模型、提示词或分块算法升级时需要的基础能力。
生产环境还应把本地目录替换为对象存储和元数据数据库，把同步调用替换为队列任务，并为每个转换器设置 CPU、内存、超时与并发上限。
采用时优先检查边界
建设通用平台不等于把所有处理逻辑塞进同一个服务。更稳妥的边界是统一控制平面和制品契约，同时允许转换器采用不同运行时与资源规格。
落地前可以检查以下事项：
- 转换器是否声明输入类型、输出类型、版本和资源需求
- 缓存键是否包含源内容哈希及全部有效参数
- 派生制品能否追溯到源内容和执行版本
- 权限撤销与数据删除是否会传播到摘要、分块和向量
- 非确定性 AI 输出是否记录模型、提示词与采样配置
- 重试是否具备幂等性，是否会重复计费或重复写入
- 指标能否区分排队时间、执行时间、失败率和单次成本
Riviera 的长期价值所揭示的重点，不只是支持了多少文件格式，而是把不断变化的内容能力放进稳定的工程模型中。AI 扩大了内容转换的种类和成本，也进一步证明：当版本、溯源、权限与执行语义已经统一，新能力更适合作为新的转换器接入，而不是再建设一套孤立管线。