From 47e950c66494f19eacf651441bbd2c20bb598b41 Mon Sep 17 00:00:00 2001 From: SpecialX <47072643+wangxiner55@users.noreply.github.com> Date: Wed, 15 Jul 2026 02:34:34 +0800 Subject: [PATCH] feat(ai): temporal worker for lesson plan workflow - deploy temporal server (postgresql + auto-setup + ui) in docker-compose - new temporal/ module: workflow + activities + worker manager - convert lesson plan 4-step orchestration to temporal workflow - activities wrap existing analyze/recommend/generate/prepare_review steps - worker injects failover_chain/content_client/data_ana_client via module globals - start() uses temporal client.start_workflow, falls back to asyncio in dev - register temporal ports 7233/8085 in port-allocation Implements M6.5 of v2.1 migration plan (ADR-030). --- infra/docker-compose.yml | 58 +++ infra/port-allocation.md | 54 +-- infra/security/secrets.example.env | 5 + services/ai/pyproject.toml | 2 + services/ai/src/ai/config.py | 7 + services/ai/src/ai/main.py | 33 ++ services/ai/src/ai/temporal/__init__.py | 14 + services/ai/src/ai/temporal/activities.py | 365 ++++++++++++++++++ services/ai/src/ai/temporal/worker.py | 80 ++++ services/ai/src/ai/temporal/workflow.py | 69 ++++ .../src/ai/workflow/lesson_plan_workflow.py | 54 ++- 11 files changed, 701 insertions(+), 40 deletions(-) create mode 100644 services/ai/src/ai/temporal/__init__.py create mode 100644 services/ai/src/ai/temporal/activities.py create mode 100644 services/ai/src/ai/temporal/worker.py create mode 100644 services/ai/src/ai/temporal/workflow.py diff --git a/infra/docker-compose.yml b/infra/docker-compose.yml index d8367ea..38b01e1 100644 --- a/infra/docker-compose.yml +++ b/infra/docker-compose.yml @@ -346,6 +346,63 @@ services: interval: 15s timeout: 5s retries: 5 + # ============================================================ + # Temporal Server - AI 工作流引擎（v2.1 §8.2 ADR-030） + # 仅用于 AI 耗时工作流 + Saga，CRUD 短事务禁止 + # ============================================================ + temporal-postgresql: + image: docker.m.daocloud.io/library/postgres:13 + container_name: edu-temporal-postgres + profiles: ["p3", "p4", "p5", "p6"] + restart: unless-stopped + environment: + POSTGRES_USER: temporal + POSTGRES_PASSWORD: ${TEMPORAL_POSTGRES_PASSWORD:-temporal} + POSTGRES_DB: temporal + volumes: + - temporal_pg_data:/var/lib/postgresql/data + healthcheck: + test: ["CMD-SHELL", "pg_isready -U temporal"] + interval: 10s + timeout: 5s + retries: 5 + temporal: + image: temporalio/auto-setup:1.23 + container_name: edu-temporal + profiles: ["p3", "p4", "p5", "p6"] + restart: unless-stopped + depends_on: + temporal-postgresql: + condition: service_healthy + environment: + DBHOST: temporal-postgresql + DBPORT: 5432 + DBUSER: temporal + DBPASSWORD: ${TEMPORAL_POSTGRES_PASSWORD:-temporal} + DBNAME: temporal + DB_PLUGIN: postgres + # 暴露 frontend gRPC 端口 7233 供 Worker 连接 + SERVICES: "frontend,history,matching,worker" + ports: + - "7233:7233" + healthcheck: + test: ["CMD", "tctl", "--address", "localhost:7233", "cluster", "health"] + interval: 15s + timeout: 5s + start_period: 30s + retries: 10 + temporal-ui: + image: temporalio/ui:2.30.0 + container_name: edu-temporal-ui + profiles: ["p3", "p4", "p5", "p6"] + restart: unless-stopped + depends_on: + - temporal + environment: + TEMPORAL_ADDRESS: temporal:7233 + TEMPORAL_CORS_ORIGINS: "http://localhost:4000,http://localhost:4001,http://localhost:4002,http://localhost:4003" + ports: + - "8085:8080" volumes: mysql_data: redis_data: @@ -356,3 +413,4 @@ volumes: prometheus_data: alertmanager_data: loki_data: + temporal_pg_data: diff --git a/infra/port-allocation.md b/infra/port-allocation.md index d772ff3..e692720 100644 --- a/infra/port-allocation.md +++ b/infra/port-allocation.md @@ -93,24 +93,27 @@ ## 6. 基础设施端口（9080-9199） -| 服务 | 端口 | 说明 | -| ---------------- | --------------------------------------------------------------------------- | ------------------------------------- | -| MySQL | 3306 | 业务数据库（external） | -| Redis | 6379 | 缓存 + 会话 + Pub/Sub | -| Kafka | 9092（OUTSIDE）/ 29092（INSIDE） | 双监听器 | -| Kafka UI | 9000 | kafka-ui 容器 | -| Zookeeper | 2181 | Kafka 协调 | -| Debezium Connect | 8083 | CDC connector | -| ClickHouse | 8123（HTTP）/ 9000（TCP） | 数据分析宽表 | -| Neo4j | 7474（HTTP）/ 7687（Bolt） | 知识图谱 | -| Elasticsearch | 9200（HTTP）/ 9300（TCP） | 全文检索 | -| Prometheus | 9090 | 指标采集（--web.enable-lifecycle） | -| Grafana | 3000（容器内，**与 teacher-bff 3003 不冲突，因 teacher-bff 在 host 网络**） | — | 实际部署需注意 | -| Alertmanager | 9093 | 告警 | -| Jaeger | 16686（UI）/ 4317（OTLP gRPC）/ 4318（OTLP HTTP） | 链路追踪 | -| node-exporter | 9100 | 主机指标（host.docker.internal:9100） | -| mysqld-exporter | 9104 | MySQL 指标 | -| redis-exporter | 9121 | Redis 指标 | +| 服务 | 端口 | 说明 | +| ----------------- | --------------------------------------------------------------------------- | ------------------------------------- | +| MySQL | 3306 | 业务数据库（external） | +| Redis | 6379 | 缓存 + 会话 + Pub/Sub | +| Kafka | 9092（OUTSIDE）/ 29092（INSIDE） | 双监听器 | +| Kafka UI | 9000 | kafka-ui 容器 | +| Zookeeper | 2181 | Kafka 协调 | +| Debezium Connect | 8083 | CDC connector | +| ClickHouse | 8123（HTTP）/ 9000（TCP） | 数据分析宽表 | +| Neo4j | 7474（HTTP）/ 7687（Bolt） | 知识图谱 | +| Elasticsearch | 9200（HTTP）/ 9300（TCP） | 全文检索 | +| Prometheus | 9090 | 指标采集（--web.enable-lifecycle） | +| Grafana | 3000（容器内，**与 teacher-bff 3003 不冲突，因 teacher-bff 在 host 网络**） | — | 实际部署需注意 | +| Alertmanager | 9093 | 告警 | +| Jaeger | 16686（UI）/ 4317（OTLP gRPC）/ 4318（OTLP HTTP） | 链路追踪 | +| node-exporter | 9100 | 主机指标（host.docker.internal:9100） | +| mysqld-exporter | 9104 | MySQL 指标 | +| redis-exporter | 9121 | Redis 指标 | +| Temporal | 7233 | gRPC frontend（Worker 连接） | +| Temporal UI | 8085 | Temporal Web UI（容器内 8080） | +| Temporal Postgres | 5433（预留，内部 5432） | Temporal 持久化存储（仅容器内） | > **Grafana 端口冲突风险**：Grafana 默认 3000，与 teacher-bff 3003 不冲突（不同主机层），但开发环境若同主机部署需注意。建议 Grafana 改用 3030 避免混淆。 @@ -118,13 +121,14 @@ ## 7. 变更记录 -| 日期 | 变更 | 决策者 | -| ---------- | --------------------------------------------------------- | ------ | -| 2026-07-09 | 初始创建，登记全部 15 服务 + 基础设施端口 | coord | -| 2026-07-09 | 仲裁 admin-portal 3003 → 4003；MF 配置 3000 → 4000 | coord | -| 2026-07-09 | 仲裁 push-gateway 豁免 gRPC，释放 50057；50058 让给 ai | coord | -| 2026-07-09 | classes 3001 标记为历史（已合并入 core-edu） | coord | -| 2026-07-14 | M3 新增 config-service（3011/50059，ADR-026 从 iam 拆分） | coord | +| 日期 | 变更 | 决策者 | +| ---------- | ----------------------------------------------------------------- | ------ | +| 2026-07-09 | 初始创建，登记全部 15 服务 + 基础设施端口 | coord | +| 2026-07-09 | 仲裁 admin-portal 3003 → 4003；MF 配置 3000 → 4000 | coord | +| 2026-07-09 | 仲裁 push-gateway 豁免 gRPC，释放 50057；50058 让给 ai | coord | +| 2026-07-09 | classes 3001 标记为历史（已合并入 core-edu） | coord | +| 2026-07-14 | M3 新增 config-service（3011/50059，ADR-026 从 iam 拆分） | coord | +| 2026-07-14 | M6.5 新增 Temporal（7233/UI 8085/PG 5433，ADR-030 AI 工作流引擎） | coord | --- diff --git a/infra/security/secrets.example.env b/infra/security/secrets.example.env index 8e7d2db..cb767c0 100644 --- a/infra/security/secrets.example.env +++ b/infra/security/secrets.example.env @@ -55,3 +55,8 @@ ENCRYPTION_KEY= # 生成：openssl rand -hex 32 # 注意：Router 和所有子图必须使用相同的密钥 ROUTER_AUTH_SECRET= + +# ---------- Temporal ---------- +# 用途：Temporal PostgreSQL 存储密码 +# 最小长度：24 字符 +TEMPORAL_POSTGRES_PASSWORD= diff --git a/services/ai/pyproject.toml b/services/ai/pyproject.toml index f541d24..1a9fdfe 100644 --- a/services/ai/pyproject.toml +++ b/services/ai/pyproject.toml @@ -22,6 +22,8 @@ dependencies = [ "tenacity>=9.0.0", # GraphQL Federation 2 子图（v2.1 M1，Apollo Router 组合） "strawberry-graphql[asgi]>=0.257.0", + # Temporal 工作流引擎（v2.1 §8.2 ADR-030，AI 耗时工作流） + "temporalio>=1.7.0", ] [tool.uv.sources] diff --git a/services/ai/src/ai/config.py b/services/ai/src/ai/config.py index 0c1b30e..2d79214 100644 --- a/services/ai/src/ai/config.py +++ b/services/ai/src/ai/config.py @@ -64,6 +64,13 @@ class Settings(BaseSettings): workflow_ttl_seconds: int = 3600 # 1h workflow_max_retries: int = 3 + # Temporal（v2.1 §8.2：AI 耗时工作流引擎） + temporal_host: str = "localhost:7233" + temporal_namespace: str = "default" + temporal_task_queue: str = "ai-lesson-plan" + temporal_workflow_timeout_seconds: int = 3600 # 1h + temporal_activity_timeout_seconds: int = 300 # 5min per activity + # 评估 evaluation_pass_threshold: float = 0.7 evaluation_excellent_threshold: float = 0.85 diff --git a/services/ai/src/ai/main.py b/services/ai/src/ai/main.py index 2115045..6029cf7 100644 --- a/services/ai/src/ai/main.py +++ b/services/ai/src/ai/main.py @@ -74,6 +74,7 @@ from .providers import create_failover_chain from .rate_limiter import RateLimiter from .services import ChatService, ExpressionService, QuestionService, ReportService from .services.evaluation import QualityGate, RuleValidator +from .temporal import TemporalWorkerManager from .usage import KafkaProducer, QuotaEnforcer, UsageRecorder from .workflow import LessonPlanWorkflowService, WorkflowStateStore @@ -178,6 +179,7 @@ _grpc_server = create_grpc_server( ) _redis: Redis | None = None +_temporal_worker: TemporalWorkerManager | None = None @asynccontextmanager @@ -215,6 +217,32 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: error=str(exc), ) + # Temporal Worker（v2.1 M6.5 ADR-030：备课工作流引擎） + # 连接失败时降级（LessonPlanWorkflowService 自动回退 asyncio），不阻断启动 + global _temporal_worker + try: + _temporal_worker = TemporalWorkerManager( + host=settings.temporal_host, + namespace=settings.temporal_namespace, + task_queue=settings.temporal_task_queue, + ) + temporal_client = await _temporal_worker.start( + failover_chain=_failover_chain, + prompt_service=_prompt_service, + quality_gate=_quality_gate, + content_client=_content_client, + data_ana_client=_data_ana_client, + state_store=_state_store, + default_model=settings.default_question_model, + ) + _workflow_service.set_temporal_client( + temporal_client, + settings.temporal_task_queue, + ) + except Exception as exc: # noqa: BLE001 + logger.warning("temporal_worker_start_failed_degraded", error=str(exc)) + _temporal_worker = None + await _kafka_producer.start() await _grpc_server.start() @@ -237,6 +265,11 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: logger.info("ai_service_stopping") await _grpc_server.stop() await _kafka_producer.stop() + if _temporal_worker is not None: + try: + await _temporal_worker.stop() + except Exception as exc: # noqa: BLE001 + logger.warning("temporal_worker_stop_failed", error=str(exc)) for name, client in reversed(downstream_clients): try: await client.close() diff --git a/services/ai/src/ai/temporal/__init__.py b/services/ai/src/ai/temporal/__init__.py new file mode 100644 index 0000000..0d3af91 --- /dev/null +++ b/services/ai/src/ai/temporal/__init__.py @@ -0,0 +1,14 @@ +"""Temporal 工作流引擎模块（v2.1 §8.2 ADR-030）. + +ai 服务作为 Temporal Worker，承载备课工作流（4 步编排）。 +边界：仅 AI 耗时工作流 + Saga，CRUD 短事务禁止走 Temporal。 + +子模块： + - workflow: Temporal Workflow 定义（LessonPlanWorkflow） + - activities: 4 步编排 Activities + - worker: Worker 生命周期管理（TemporalWorkerManager） +""" + +from .worker import TemporalWorkerManager + +__all__ = ["TemporalWorkerManager"] diff --git a/services/ai/src/ai/temporal/activities.py b/services/ai/src/ai/temporal/activities.py new file mode 100644 index 0000000..be68dd8 --- /dev/null +++ b/services/ai/src/ai/temporal/activities.py @@ -0,0 +1,365 @@ +"""Temporal Activities - 备课工作流 4 步编排（v2.1 §8.2 ADR-030）. + +将 lesson_plan_workflow.py 的 4 个 step 方法提取为 Temporal Activities。 +每个 Activity 封装一步业务逻辑，接收/返回可序列化的 dict。 + +Activity 通过模块级 _clients 字典访问依赖（由 worker.py 启动时注入）： + - data_ana_client: 学情查询 + - content_client: 知识点查询 + - failover_chain: LLM 调用 + - prompt_service: Prompt 模板渲染 + - quality_gate: 题目评估三道防线 + - state_store: Redis 工作流状态存储 + - default_model: 默认 LLM 模型 +""" + +from typing import Any + +import structlog +from temporalio import activity + +from ..clients.content_client import ContentClient +from ..clients.data_ana_client import DataAnaClient +from ..errors import AIError +from ..models.question import GeneratedQuestionData +from ..prompt_service import PromptTemplateService +from ..providers import ProviderFailoverChain +from ..services.evaluation import QualityGate +from ..workflow.state_store import WorkflowState, WorkflowStateStore + +logger = structlog.get_logger() + +# 状态常量（与 lesson_plan_workflow.py 保持一致） +STATUS_ANALYZING = "analyzing" +STATUS_GENERATING = "generating" +STATUS_PENDING_REVIEW = "pending_review" + +# 生成重试上限 +MAX_GENERATE_RETRIES = 3 + +# 模块级依赖注册表（由 TemporalWorkerManager.start 注入） +_clients: dict[str, Any] = {} + + +def set_clients(**clients: Any) -> None: + """注入依赖到 Activities（由 worker 启动时调用）.""" + _clients.update(clients) + + +def _get_state_store() -> WorkflowStateStore: + """获取已注入的状态存储.""" + store = _clients.get("state_store") + if store is None: + raise AIError("state_store not injected to temporal activities") + return store # type: ignore[return-value] + + +@activity.defn +async def analyze_activity(state_data: dict) -> dict: + """Step 1: 分析学情. + + 调 data-ana 查询班级学情 + 学生薄弱点。 + data-ana 不可用时降级（返回空分析）。 + 同时更新 Redis 状态：status=analyzing + analysis。 + """ + state = WorkflowState.from_dict(state_data) + store = _get_state_store() + + await store.update(state.workflow_id, status=STATUS_ANALYZING) + + data_ana_client: DataAnaClient | None = _clients.get("data_ana_client") + analysis: dict[str, Any] = {} + + if data_ana_client is not None: + try: + performance = await data_ana_client.get_class_performance( + class_id=state.class_id, + subject_id=state.subject_id, + ) + analysis["class_performance"] = { + "average_score": performance.average_score, + "pass_rate": performance.pass_rate, + "student_count": len(performance.scores), + } + analysis["weak_students"] = [ + {"student_id": s.student_id, "score": s.score} + for s in performance.scores + if s.score < 60 + ] + except Exception as exc: # noqa: BLE001 + logger.warning( + "activity_step1_analyze_degraded", + workflow_id=state.workflow_id, + error=str(exc), + ) + analysis["degraded"] = True + analysis["degraded_reason"] = f"data-ana unavailable: {exc}" + else: + analysis["degraded"] = True + analysis["degraded_reason"] = "data-ana client not configured" + + await store.update(state.workflow_id, analysis=analysis) + + logger.info( + "activity_step1_completed", + workflow_id=state.workflow_id, + degraded=analysis.get("degraded", False), + ) + return analysis + + +@activity.defn +async def recommend_activity(state_data: dict) -> list[dict]: + """Step 2: 推荐知识点. + + 调 content 查询学习路径。content 不可用时降级（基于 topic 推导）。 + """ + state = WorkflowState.from_dict(state_data) + content_client: ContentClient | None = _clients.get("content_client") + knowledge_points: list[dict[str, str]] = [] + + if content_client is not None: + try: + learning_path = await content_client.get_learning_path( + student_id=state.user_id, + subject_id=state.subject_id, + ) + knowledge_points = [{"id": kp.id, "title": kp.title} for kp in learning_path] + except Exception as exc: # noqa: BLE001 + logger.warning( + "activity_step2_recommend_degraded", + workflow_id=state.workflow_id, + error=str(exc), + ) + + # 降级：基于 topic 推导知识点 + if not knowledge_points: + knowledge_points = [ + {"id": "kp_default_1", "title": f"{state.topic} - 基础概念"}, + {"id": "kp_default_2", "title": f"{state.topic} - 进阶应用"}, + {"id": "kp_default_3", "title": f"{state.topic} - 综合题"}, + ] + + logger.info( + "activity_step2_completed", + workflow_id=state.workflow_id, + knowledge_point_count=len(knowledge_points), + ) + return knowledge_points + + +@activity.defn +async def generate_activity( + state_data: dict, + knowledge_points: list[dict], +) -> list[dict]: + """Step 3: 生成题目. + + 使用 LLM 生成题目 + 评估三道防线。 + 评估未通过时重试（最多 MAX_GENERATE_RETRIES 次）。 + 返回 list[dict]（GeneratedQuestionData.model_dump()）。 + """ + state = WorkflowState.from_dict(state_data) + failover_chain: ProviderFailoverChain | None = _clients.get("failover_chain") + prompt_service: PromptTemplateService | None = _clients.get("prompt_service") + quality_gate: QualityGate | None = _clients.get("quality_gate") + default_model: str = _clients.get("default_model", "gpt-4o-mini") + + kp_ids = [kp["id"] for kp in knowledge_points] + + if failover_chain is None or quality_gate is None: + logger.warning( + "activity_step3_degraded_no_llm", + workflow_id=state.workflow_id, + ) + return [ + GeneratedQuestionData( + question=f"[degraded] LLM 未配置，请手动添加题目：{state.topic}", + answer="", + explanation="LLM provider 或 quality_gate 未注入", + question_type="short_answer", + difficulty=state.target_difficulty, + knowledge_point_ids=kp_ids, + evaluation_score=0.0, + degraded=True, + degraded_reason="failover_chain or quality_gate not injected", + ).model_dump(), + ] + + questions: list[GeneratedQuestionData] = [] + for i in range(state.question_count): + question = await _generate_single_question( + state=state, + kp_ids=kp_ids, + question_index=i, + failover_chain=failover_chain, + prompt_service=prompt_service, + quality_gate=quality_gate, + default_model=default_model, + ) + questions.append(question) + + logger.info( + "activity_step3_completed", + workflow_id=state.workflow_id, + question_count=len(questions), + ) + return [q.model_dump() for q in questions] + + +async def _generate_single_question( + state: WorkflowState, + kp_ids: list[str], + question_index: int, + failover_chain: ProviderFailoverChain, + prompt_service: PromptTemplateService | None, + quality_gate: QualityGate, + default_model: str, +) -> GeneratedQuestionData: + """生成单道题目（含重试）.""" + prompt = _render_generate_prompt(state, kp_ids, question_index, prompt_service) + messages = [ + {"role": "system", "content": "你是一个专业的教育题目生成助手。"}, + {"role": "user", "content": prompt}, + ] + + for attempt in range(MAX_GENERATE_RETRIES): + try: + response = await failover_chain.chat(messages, default_model, 0.7) + + evaluation = await quality_gate.evaluate( + llm_output=response.content, + expected_difficulty=state.target_difficulty, + expected_question_type="short_answer", + subject=state.subject_id, + ) + + parsed = evaluation.rule_result.parsed if evaluation.rule_result else None + + if parsed and evaluation.passed: + return GeneratedQuestionData( + question=str(parsed.get("question", "")), + answer=str(parsed.get("answer", "")), + explanation=str(parsed.get("explanation", "")), + question_type=str( + parsed.get("question_type", "short_answer"), + ), + difficulty=str( + parsed.get("difficulty", state.target_difficulty), + ), + knowledge_point_ids=list( + parsed.get("knowledge_point_ids", kp_ids), + ), + evaluation_score=evaluation.score, + ) + + logger.warning( + "activity_generate_retry", + workflow_id=state.workflow_id, + attempt=attempt + 1, + score=evaluation.score, + ) + + except Exception as exc: # noqa: BLE001 + logger.warning( + "activity_generate_error_retry", + workflow_id=state.workflow_id, + attempt=attempt + 1, + error=str(exc), + ) + + logger.warning( + "activity_generate_all_retries_failed", + workflow_id=state.workflow_id, + question_index=question_index, + ) + return GeneratedQuestionData( + question=f"[degraded] 生成失败，请手动添加题目：{state.topic}", + answer="", + explanation="题目生成失败，已达到最大重试次数", + question_type="short_answer", + difficulty=state.target_difficulty, + knowledge_point_ids=kp_ids, + evaluation_score=0.0, + degraded=True, + degraded_reason="max retries exceeded", + ) + + +def _render_generate_prompt( + state: WorkflowState, + kp_ids: list[str], + question_index: int, + prompt_service: PromptTemplateService | None, +) -> str: + """渲染题目生成 prompt.""" + if prompt_service is not None: + try: + return prompt_service.render( + "lesson_plan_generate", + { + "subject": state.subject_id, + "topic": state.topic, + "difficulty": state.target_difficulty, + "knowledge_points": kp_ids, + "knowledge_point_ids": kp_ids, + "question_index": question_index, + "analysis": state.analysis, + }, + ) + except Exception: # noqa: BLE001 + pass + + # 降级 prompt + return ( + f"请为 {state.subject_id} 学科生成一道题目。\n" + f"主题：{state.topic}\n" + f"难度：{state.target_difficulty}\n" + f"知识点：{', '.join(kp_ids)}\n" + "请按 JSON 格式输出：" + '{"question":"...","answer":"...","explanation":"..."}' + ) + + +@activity.defn +async def prepare_review_activity( + state_data: dict, + questions: list[dict], +) -> dict: + """Step 4: 设置为待审核. + + 将生成的题目写入 Redis 状态，状态置为 pending_review。 + """ + state = WorkflowState.from_dict(state_data) + store = _get_state_store() + + questions_objs = [GeneratedQuestionData(**q) for q in questions] + await store.update( + state.workflow_id, + status=STATUS_PENDING_REVIEW, + questions=questions_objs, + ) + + logger.info( + "activity_step4_completed", + workflow_id=state.workflow_id, + question_count=len(questions_objs), + ) + return { + "workflow_id": state.workflow_id, + "status": STATUS_PENDING_REVIEW, + "question_count": len(questions_objs), + } + + +@activity.defn +async def update_status_activity(payload: dict) -> dict: + """更新 Redis 工作流状态（通用）. + + payload: {"workflow_id": str, "status": str} + """ + store = _get_state_store() + workflow_id = payload["workflow_id"] + status = payload["status"] + await store.update(workflow_id, status=status) + return {"workflow_id": workflow_id, "status": status} diff --git a/services/ai/src/ai/temporal/worker.py b/services/ai/src/ai/temporal/worker.py new file mode 100644 index 0000000..86ab9fa --- /dev/null +++ b/services/ai/src/ai/temporal/worker.py @@ -0,0 +1,80 @@ +"""Temporal Worker 管理器（v2.1 §8.2 ADR-030）. + +负责： + - 连接 Temporal Server + - 注入依赖到 Activities（failover_chain / content_client / data_ana_client 等） + - 启动 Worker（注册 Workflow + Activities） + - 优雅关闭 + +ai 服务作为 Temporal Worker，task_queue=ai-lesson-plan。 +""" + +import asyncio +from typing import Any + +import structlog +from temporalio.client import Client +from temporalio.worker import Worker + +from . import activities, workflow + +logger = structlog.get_logger() + + +class TemporalWorkerManager: + """Temporal Worker 生命周期管理.""" + + def __init__(self, host: str, namespace: str, task_queue: str) -> None: + self._host = host + self._namespace = namespace + self._task_queue = task_queue + self._client: Client | None = None + self._worker: Worker | None = None + + async def start(self, **clients: Any) -> Client: + """启动 Worker. + + Args: + **clients: 注入到 Activities 的依赖 + - failover_chain: ProviderFailoverChain + - prompt_service: PromptTemplateService + - quality_gate: QualityGate + - content_client: ContentClient + - data_ana_client: DataAnaClient + - state_store: WorkflowStateStore + - default_model: str + + Returns: + Temporal Client（供 LessonPlanWorkflowService 启动 Workflow） + """ + activities.set_clients(**clients) + + self._client = await Client.connect(self._host, namespace=self._namespace) + self._worker = Worker( + self._client, + task_queue=self._task_queue, + workflows=[workflow.LessonPlanWorkflow], + activities=[ + activities.analyze_activity, + activities.recommend_activity, + activities.generate_activity, + activities.prepare_review_activity, + activities.update_status_activity, + ], + ) + asyncio.create_task(self._worker.run()) + logger.info( + "temporal_worker_started", + host=self._host, + namespace=self._namespace, + task_queue=self._task_queue, + ) + return self._client + + async def stop(self) -> None: + """优雅关闭 Worker.""" + if self._worker is not None: + await self._worker.shutdown() + logger.info("temporal_worker_stopped") + self._worker = None + self._client = None diff --git a/services/ai/src/ai/temporal/workflow.py b/services/ai/src/ai/temporal/workflow.py new file mode 100644 index 0000000..e43108c --- /dev/null +++ b/services/ai/src/ai/temporal/workflow.py @@ -0,0 +1,69 @@ +"""Temporal Workflow - 备课工作流编排（v2.1 §8.2 ADR-030）. + +4 步编排（每步为独立 Activity，可自动重试 + 状态持久化）： + Step 1: 分析学情（analyze_activity） + Step 2: 推荐知识点（recommend_activity） + Step 3: 生成题目（update_status_activity → generate_activity） + Step 4: 设置待审核（prepare_review_activity） + +边界约束（spec §8.2）：仅 AI 耗时工作流走 Temporal，CRUD 短事务禁止。 +""" + +from datetime import timedelta + +from temporalio import workflow + +with workflow.unsafe.imports_passed_through(): + from . import activities + + +@workflow.defn +class LessonPlanWorkflow: + """备课工作流（Temporal 编排）.""" + + @workflow.run + async def run(self, state_data: dict) -> dict: + """执行 4 步编排. + + Args: + state_data: WorkflowState.to_dict() 序列化数据 + + Returns: + {"workflow_id", "status", "question_count"} + """ + # Step 1: 分析学情 + analysis = await workflow.execute_activity( + activities.analyze_activity, + state_data, + start_to_close_timeout=timedelta(seconds=300), + ) + state_data["analysis"] = analysis + + # Step 2: 推荐知识点 + knowledge_points = await workflow.execute_activity( + activities.recommend_activity, + state_data, + start_to_close_timeout=timedelta(seconds=300), + ) + + # Step 3: 生成题目 + await workflow.execute_activity( + activities.update_status_activity, + {"workflow_id": state_data["workflow_id"], "status": "generating"}, + start_to_close_timeout=timedelta(seconds=10), + ) + questions = await workflow.execute_activity( + activities.generate_activity, + state_data, + knowledge_points, + start_to_close_timeout=timedelta(seconds=600), + ) + + # Step 4: 设置待审核 + result = await workflow.execute_activity( + activities.prepare_review_activity, + state_data, + questions, + start_to_close_timeout=timedelta(seconds=30), + ) + return result diff --git a/services/ai/src/ai/workflow/lesson_plan_workflow.py b/services/ai/src/ai/workflow/lesson_plan_workflow.py index 2baa94b..5cd7897 100644 --- a/services/ai/src/ai/workflow/lesson_plan_workflow.py +++ b/services/ai/src/ai/workflow/lesson_plan_workflow.py @@ -10,6 +10,10 @@ P5 实现：FastAPI BackgroundTasks + Redis 状态存储 P6+ 评估：Temporal 工作流引擎 +v2.1 M6.5：迁移到 Temporal Workflow（ADR-030）。 + - temporal_client 可用时：start_workflow 由 Temporal 编排 + - temporal_client 不可用时：降级 asyncio.create_task（开发模式） + 4 步编排： Step 1: 分析学情（调 data-ana.GetClassPerformance + GetStudentWeakness） Step 2: 推荐知识点（调 content.GetPrerequisites + GetLearningPath） @@ -18,7 +22,7 @@ P6+ 评估：Temporal 工作流引擎 """ import asyncio -from typing import Any +from typing import TYPE_CHECKING, Any import structlog @@ -31,6 +35,9 @@ from ..providers import ProviderFailoverChain from ..services.evaluation import QualityGate from .state_store import WorkflowState, WorkflowStateStore +if TYPE_CHECKING: + from temporalio.client import Client + logger = structlog.get_logger() # 状态常量（简化 proto WorkflowStatus，内部用） @@ -51,8 +58,7 @@ ESTIMATED_COMPLETION_SECONDS = 60 class LessonPlanWorkflowService: """备课工作流服务. - P5 使用 asyncio.create_task 在后台执行工作流。 - P6+ 评估迁移到 Temporal。 + v2.1 M6.5：temporal_client 可用时由 Temporal 编排，否则降级 asyncio.create_task。 """ def __init__( @@ -64,6 +70,8 @@ class LessonPlanWorkflowService: content_client: ContentClient | None = None, data_ana_client: DataAnaClient | None = None, default_model: str = "gpt-4o-mini", + temporal_client: "Client | None" = None, + task_queue: str = "ai-lesson-plan", ) -> None: self._store = state_store self._chain = failover_chain @@ -73,6 +81,17 @@ class LessonPlanWorkflowService: self._data_ana_client = data_ana_client self._default_model = default_model self._background_tasks: dict[str, asyncio.Task[Any]] = {} + self._temporal_client = temporal_client + self._task_queue = task_queue + + def set_temporal_client( + self, + client: "Client", + task_queue: str = "ai-lesson-plan", + ) -> None: + """注入 Temporal Client（lifespan 中由 main.py 调用）.""" + self._temporal_client = client + self._task_queue = task_queue async def start( self, @@ -115,15 +134,27 @@ class LessonPlanWorkflowService: ) await self._store.create(state) - # 启动后台任务执行工作流 - task = asyncio.create_task(self._run_workflow(state.workflow_id)) - self._background_tasks[state.workflow_id] = task + if self._temporal_client is not None: + # v2.1 M6.5: 使用 Temporal Workflow 编排（ADR-030） + from ..temporal.workflow import LessonPlanWorkflow + + await self._temporal_client.start_workflow( + LessonPlanWorkflow.run, + args=[state.to_dict()], + id=f"lesson-plan-{state.workflow_id}", + task_queue=self._task_queue, + ) + else: + # 降级：asyncio.create_task（开发模式或 Temporal 不可用） + task = asyncio.create_task(self._run_workflow(state.workflow_id)) + self._background_tasks[state.workflow_id] = task logger.info( "workflow_started", workflow_id=state.workflow_id, user_id=user_id, topic=topic, + engine="temporal" if self._temporal_client is not None else "asyncio", ) return state @@ -361,10 +392,7 @@ class LessonPlanWorkflowService: student_id=state.user_id, subject_id=state.subject_id, ) - knowledge_points = [ - {"id": kp.id, "title": kp.title} - for kp in learning_path - ] + knowledge_points = [{"id": kp.id, "title": kp.title} for kp in learning_path] except Exception as exc: # noqa: BLE001 logger.warning( "workflow_step2_recommend_degraded", @@ -444,11 +472,7 @@ class LessonPlanWorkflowService: subject=state.subject_id, ) - parsed = ( - evaluation.rule_result.parsed - if evaluation.rule_result - else None - ) + parsed = evaluation.rule_result.parsed if evaluation.rule_result else None if parsed and evaluation.passed: return GeneratedQuestionData(