6.8 KiB
6.8 KiB
Integration（FastAPI API + Celery Worker）｜Plan
对应规范：
spec_kit/Personalized Reco/modules/integration-api-worker/spec.md依赖（已实现）：
server/app/features/personalized_reco/reco_engine/：统一引擎入口recommend(...)server/app/features/personalized_reco/content_repository/：SqlAlchemyContentRepositoryserver/app/db/session.py：get_db/AsyncSessionLocalserver/app/worker.py：celery_app本计划已按确认项固化：
- API 路由：按场景拆分（
/v1/reco/feed、/v1/reco/push、/v1/reco/widget）- locale：从
Accept-Language解析并映射到en/tc，缺省en- 限流：按客户端 IP，1 分钟 10 次
- Celery：实现
tasks.reco.generate+tasks.reco.push_once- Celery 内调用 async：使用
asyncio.run(...)- now 注入：支持 Header
X-Now（方案 B），并保留请求体now（spec 已定义）
1. 目标与交付物
1.1 目标
- 对外提供推荐能力：
- FastAPI：客户端同步获取推荐结果。
- Celery：后台任务式生成推荐（Push/Widget 的定时/批处理）。
- 不复制推荐逻辑：API 与任务均只调用同一
Reco Engine。 - 提供基础可用的 IP 限流（1 分钟 10 次）。
- 支持 now 注入 以实现确定性回归测试。
- 多语言仅支持 EN/TC，不允许语言回退。
1.2 交付物（tasks 阶段落地）
- 新增 API 路由文件（建议）：
server/app/api/v1/reco.py
main.py注册路由：app.include_router(reco_router)
- 新增 Celery 任务：
server/app/tasks/reco.py（包含tasks.reco.generate、tasks.reco.push_once）
- 新增限流中间件/依赖：
server/app/api/limits.py（或server/app/core/ratelimit.py）
- 单元/集成测试（至少）：
- API schema 校验（请求/响应模型）
- 限流行为（同 IP 超过阈值返回 429）
- Celery 任务能跑通
ping -> reco.generate
2. FastAPI 设计
2.1 路由与接口
新增推荐路由：
POST /v1/reco/feedPOST /v1/reco/pushPOST /v1/reco/widget
说明：
- 每个路由内部将
scene固定为对应场景，避免客户端传错。 k若未传：按引擎默认（建议：feed=30，push/widget=1；此默认可在 API 层写死或由调用方显式传入）。
2.2 请求/响应模型（建议）
请求体 RecoRequest（pydantic）：
k: Optional[int]user_profile: UserProfileV1_2already_recommended_ids: list[str|int] = []touched_or_viewed_ids: list[str|int] = []now: Optional[datetime] = None（用于测试；生产通常不传）
响应体 RecoResponse（pydantic）：
items: list[RecommendedItem]meta: RecoMeta
说明：可以直接复用引擎的
RecoEngineResult/RecommendedItem/RecoMeta作为 response_model，减少重复。
2.3 now 注入优先级（确定性）
支持两种注入：
- Header：
X-Now（ISO8601 字符串，如2026-02-02T12:00:00Z） - Body：
now
建议优先级：
- 若
X-Now存在且可解析 → 使用 header 的时间 - 否则若 body.now 存在 → 使用 body.now
- 否则 → 使用服务端
datetime.now(timezone.utc)
2.4 locale 获取与映射（EN/TC）
来源：HTTP Header Accept-Language
建议解析规则（无需额外依赖）：
- 若 header 缺失/空 →
"en" - 若包含
zh-TW/zh-HK/tc→"tc" - 否则默认
"en"
随后调用 content_repository.types.normalize_locale(locale) 做严格校验（保证只出 en/tc）。
2.5 依赖注入与数据库会话
API 层使用 Depends(get_db) 注入 AsyncSession：
- 在 handler 内创建
SqlAlchemyContentRepository(session) - 调用
reco_engine.recommend(repo=..., scene=..., ...)
2.6 限流（按 IP：1 分钟 10 次）
实现方式（V1 推荐：无外部依赖、内存版）：
- 在 FastAPI 层添加一个依赖或中间件：
- 从
Request.client.host取 IP（若有反代需后续支持X-Forwarded-For，V1 先不做） - 使用滑动窗口或固定窗口计数（推荐固定窗口：按分钟 bucket）
- 超过阈值：返回
HTTP 429，响应体包含detail="rate_limited"
- 从
注意与取舍：
- 内存限流在多进程/多实例下不共享（V1 可接受）；后续可升级为 Redis 限流。
3. Celery Worker 设计
3.1 任务列表
tasks.reco.generate- 输入：与 API 等价，但建议 payload 小（user_profile + ids + scene + 可选 now/locale）
- 输出：默认忽略结果（worker 已配置
task_ignore_result），但函数可返回items/meta用于调试
tasks.reco.push_once- 输入：尽量只包含 push 需要字段（user_profile + ids + 可选 now/locale）
- 行为：内部调用
tasks.reco.generate(scene="push")，并预留“写入下游”的占位函数（V1 不接真实推送系统）
3.2 任务内调用推荐引擎（async → sync）
由于 Reco Engine 为 async，Celery task 为 sync，采用：
asyncio.run(_run_reco_async(...))
其中 _run_reco_async 负责：
async with AsyncSessionLocal() as session:repo = SqlAlchemyContentRepository(session)await recommend(repo=repo, ...)
说明：
- Celery 环境通常没有运行中的事件循环，
asyncio.run可用。 - 若未来引入 async worker/或在已有 loop 环境中调用，再考虑改为“可复用事件循环工具”。
3.3 locale 与 now
- locale：
- Celery 输入可直接传
"en"/"tc"，缺省"en" - 仍通过
normalize_locale严格校验
- Celery 输入可直接传
- now：
- 任务输入支持传入
now（用于回归测试/离线批处理），否则用服务端当前时间
- 任务输入支持传入
4. 一致性策略（API vs Celery）
必须保证：
- API 与任务都只调用
reco_engine.recommend - 对同一份输入（含固定 now/locale），输出 items 结果一致（允许浮点微差）
建议做一个对比测试：
- 在测试中构造固定
now，用同样的 repo/同样的输入分别走 API handler 与 Celery 的_run_reco_async，断言content_id列表一致。
5. 错误处理与返回规范
5.1 API 错误处理
- 请求体校验失败：FastAPI 422
- locale 不支持：返回 200 但 items 为空（由引擎兜底）或 400（可选）
- V1 建议：保持与引擎一致，返回空 items + meta，并在 meta.config_snapshot 记录错误 stage
- 限流触发：429
5.2 Celery 错误处理
- 任务内部捕获异常并记录日志
- 默认不回写结果，避免 Redis 占用
- 必要时将错误信息写入任务日志或后续的可观测系统（V2）
6. 安全与性能（V1）
- 限流：按 IP 10/min，保护服务与数据库
- Payload 控制：
- Celery 输入避免传大数组；历史集合若过大，后续演进为“传引用 ID”
- 多语言：只支持 EN/TC，不做语言回退（与 repository 口径一致）