xiner/Edu
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(ai): stateless workflow store with redis-only mode

Browse Source 
- remove in-memory fallback (ADR-028: ai must be stateless)
- Redis is the only state store, shared across instances
- Redis unavailable raises RedisError instead of degrading to memory
- key prefix workflow -> ai:workflow (spec 4.3)
- TTL 24h -> 1h (spec 4.3: long-lived state managed by Temporal)

Implements M6 of v2.1 migration plan.
This commit is contained in:
SpecialX
2026-07-15 02:22:28 +08:00
parent 1423a0b8eb
commit ce5aeec955
2 changed files with 48 additions and 68 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
services/ai/src/ai/config.py
View File

@@ -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
# 评估

86
services/ai/src/ai/workflow/state_store.py
View File

@@ -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:
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,
)
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
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:
if self._redis is None:
raise RedisError("Redis client not initialized (ai stateless mode)")
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 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:
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,
)
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
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}"