diff --git a/services/ai/src/ai/config.py b/services/ai/src/ai/config.py index b581cac..0c1b30e 100644 --- a/services/ai/src/ai/config.py +++ b/services/ai/src/ai/config.py @@ -60,8 +60,8 @@ class Settings(BaseSettings): # GraphQL Federation 2 子图（v2.1 ADR-036 Router-Authorization 信任凭证） router_auth_secret: str = "" - # 备课工作流 - workflow_ttl_seconds: int = 86400 # 24h + # 备课工作流（v2.1 §4.3：TTL 1h，长期状态由 Temporal 管理） + workflow_ttl_seconds: int = 3600 # 1h workflow_max_retries: int = 3 # 评估 diff --git a/services/ai/src/ai/workflow/state_store.py b/services/ai/src/ai/workflow/state_store.py index 96bf0f8..885e992 100644 --- a/services/ai/src/ai/workflow/state_store.py +++ b/services/ai/src/ai/workflow/state_store.py @@ -1,4 +1,10 @@ -"""工作流状态存储（Redis 持久化，24h TTL）. +"""工作流状态存储（Redis 持久化，v2.1 无状态化改造 ADR-028）. + +v2.1 变更： + - 移除内存降级存储（ai 服务必须无状态，多实例不共享内存） + - Redis key 前缀改为 ai:workflow（与 spec §4.3 一致） + - TTL 默认 1 小时（spec §4.3：TTL 1 小时；长期状态由 Temporal 管理） + - Redis 不可用时抛出异常（不再降级），调用方需处理 存储备课工作流的状态和中间结果，支持： - create: 创建工作流 @@ -6,10 +12,8 @@ - update: 更新工作流状态 - delete: 删除工作流 -Redis key 格式：workflow:{workflow_id} -TTL：24h（86400s，可配置） - -全并行模式：Redis 不可用时降级到内存存储（仅单实例有效）。 +Redis key 格式：ai:workflow:{workflow_id} +TTL：1h（3600s，可配置） """ import json @@ -28,11 +32,11 @@ from ..models.workflow import WorkflowStatus logger = structlog.get_logger() -# Redis key 前缀 -WORKFLOW_KEY_PREFIX = "workflow" +# Redis key 前缀（v2.1 §4.3：ai:workflow） +WORKFLOW_KEY_PREFIX = "ai:workflow" -# 默认 TTL（24h） -DEFAULT_TTL_SECONDS = 86400 +# 默认 TTL（1h，spec §4.3） +DEFAULT_TTL_SECONDS = 3600 @dataclass @@ -69,8 +73,7 @@ class WorkflowState: data = asdict(self) # GeneratedQuestionData 转换为 dict data["questions"] = [ - q.model_dump() if hasattr(q, "model_dump") else asdict(q) - for q in self.questions + q.model_dump() if hasattr(q, "model_dump") else asdict(q) for q in self.questions ] return data @@ -79,8 +82,7 @@ class WorkflowState: """从 dict 反序列化.""" questions_data = data.get("questions", []) questions = [ - GeneratedQuestionData(**q) if isinstance(q, dict) else q - for q in questions_data + GeneratedQuestionData(**q) if isinstance(q, dict) else q for q in questions_data ] return cls( workflow_id=data.get("workflow_id", ""), @@ -106,9 +108,12 @@ class WorkflowState: class WorkflowStateStore: - """工作流状态存储（Redis）. + """工作流状态存储（Redis，v2.1 无状态化）. - 全并行模式：Redis 不可用时降级到内存存储。 + v2.1 ADR-028：ai 服务必须无状态。 + - Redis 是唯一状态存储，多实例共享 + - Redis 不可用时抛出 RedisError，调用方决定降级策略 + - 不再有内存降级存储（内存存储无法跨实例共享，违反无状态约束） """ def __init__( @@ -118,8 +123,6 @@ class WorkflowStateStore: ) -> None: self._redis = redis self._ttl = ttl_seconds - # 内存降级存储（Redis 不可用时使用） - self._memory_store: dict[str, str] = {} async def create(self, state: WorkflowState) -> WorkflowState: """创建工作流状态. @@ -129,28 +132,22 @@ class WorkflowStateStore: Returns: 创建后的 WorkflowState（含生成的 workflow_id） + + Raises: + RedisError: Redis 不可用（v2.1 无状态化，不再降级到内存） """ state.touch() key = self._key(state.workflow_id) value = json.dumps(state.to_dict(), ensure_ascii=False) - if self._redis is not None: - try: - await self._redis.setex(key, self._ttl, value) - logger.info( - "workflow_created", - workflow_id=state.workflow_id, - status=state.status, - ) - except RedisError as exc: - logger.warning( - "workflow_create_redis_failed_using_memory", - error=str(exc), - ) - self._memory_store[key] = value - else: - self._memory_store[key] = value - + if self._redis is None: + raise RedisError("Redis client not initialized (ai stateless mode)") + await self._redis.setex(key, self._ttl, value) + logger.info( + "workflow_created", + workflow_id=state.workflow_id, + status=state.status, + ) return state async def get(self, workflow_id: str) -> WorkflowState: @@ -164,21 +161,13 @@ class WorkflowStateStore: Raises: AIWorkflowNotFoundError: 工作流不存在或已过期 + RedisError: Redis 不可用 """ key = self._key(workflow_id) - value: str | None = None - if self._redis is not None: - try: - value = await self._redis.get(key) - except RedisError as exc: - logger.warning( - "workflow_get_redis_failed_using_memory", - error=str(exc), - ) - value = self._memory_store.get(key) - else: - value = self._memory_store.get(key) + if self._redis is None: + raise RedisError("Redis client not initialized (ai stateless mode)") + value = await self._redis.get(key) if value is None: raise AIWorkflowNotFoundError(workflow_id) @@ -198,6 +187,7 @@ class WorkflowStateStore: Raises: AIWorkflowNotFoundError: 工作流不存在 + RedisError: Redis 不可用 """ state = await self.get(workflow_id) @@ -211,23 +201,14 @@ class WorkflowStateStore: redis_key = self._key(workflow_id) serialized = json.dumps(state.to_dict(), ensure_ascii=False) - if self._redis is not None: - try: - await self._redis.setex(redis_key, self._ttl, serialized) - logger.info( - "workflow_updated", - workflow_id=workflow_id, - status=state.status, - ) - except RedisError as exc: - logger.warning( - "workflow_update_redis_failed_using_memory", - error=str(exc), - ) - self._memory_store[redis_key] = serialized - else: - self._memory_store[redis_key] = serialized - + if self._redis is None: + raise RedisError("Redis client not initialized (ai stateless mode)") + await self._redis.setex(redis_key, self._ttl, serialized) + logger.info( + "workflow_updated", + workflow_id=workflow_id, + status=state.status, + ) return state async def delete(self, workflow_id: str) -> None: @@ -241,10 +222,9 @@ class WorkflowStateStore: try: await self._redis.delete(key) except RedisError as exc: - logger.warning("workflow_delete_redis_failed", error=str(exc)) - self._memory_store.pop(key, None) + logger.warning("workflow_delete_failed", error=str(exc)) @staticmethod def _key(workflow_id: str) -> str: - """构建 Redis key.""" + """构建 Redis key（v2.1 §4.3：ai:workflow 前缀）.""" return f"{WORKFLOW_KEY_PREFIX}:{workflow_id}"