xiner/Edu

Files

SpecialX fd5b6e19ae docs(data-ana): ai06 补充 AI Agent 身份标注（§14.7）

2026-07-09 17:30:49 +08:00

48 KiB

Raw Blame History

ai06 阶段 2 交付物：模块架构设计文档

AI 标识：ai06 负责模块：data-ana（P4）、ai（P5）阶段：架构设计外包 · 阶段 2（模块架构设计）日期：2026-07-09 关联文档：ai06 阶段 1 确认书、ai-allocation.md、004 架构影响地图、pending-features.md 审查请求：本设计文档待 coord 按 ai-allocation.md §8 交叉审查（接口一致性 / 端口冲突 / Topic 重复 / 错误码重叠 / 黄金模板对齐）

设计原则与全局约束

本设计遵循以下强制约束（来自 project_rules.md + coding-standards.md + 004）：

契约先行：proto 已定义（analytics.proto / ai.proto），实现前不修改 proto，如需修改走 coord 流程
CQRS 读写分离：data-ana 是纯读模型服务（无 MySQL 写），ClickHouse 宽表由 CDC 投影构建
事件驱动：data-ana 消费 CDC + 领域事件；ai 不参与事件流（无状态）
gRPC 优先：004 §4.1 明确 BFF → 业务服务走 gRPC，两服务需实现 gRPC server（HTTP 保留作 Gateway 直连降级）
DataScope 过滤：004 §5.3 DataScope 6 级在查询层注入 WHERE
三支柱可观测：structlog + prometheus-client + OpenTelemetry（已具备，需补业务指标）
降级模式：外部依赖（ClickHouse / LLM / 下游 gRPC）不可用时返回骨架数据 + degraded: true
Python 规范：pydantic-settings 配置 / Pydantic 模型校验 / async 优先 / 类型注解强制 / ruff 零错误

模块架构设计文档 — data-ana

1. 模块内部分层图

flowchart TB
    subgraph Entry["入口层"]
        HTTP[FastAPI HTTP Router<br/>/analytics/* + /healthz + /readyz]
        GRPC[grpc.aio Server<br/>AnalyticsService]
    end

    subgraph Middleware["中间件层"]
        AUTH[AuthDepends<br/>校验 x-user-id / x-user-roles]
        SCOPE[DataScopeDepends<br/>注入 data_scope 元数据]
        TRACE[OTel FastAPIInstrumentor<br/>+ grpc.aio server interceptor]
    end

    subgraph Service["应用服务层 Application Service"]
        S1[AnalyticsService<br/>班级/学生/趋势查询编排]
        S2[MasteryService<br/>掌握度计算 + 事件发布]
        S3[ErrorBookService<br/>错题本查询]
    end

    subgraph Repo["数据访问层 Repository"]
        R1[ClickHouseRepository<br/>宽表查询 + DataScope WHERE 注入]
        R2[KafkaProducer<br/>mastery.updated 事件发布]
        R3[IamClient<br/>gRPC 调 iam.getEffectiveDataScope]
    end

    subgraph Consumer["CDC 消费者（后台任务）"]
        C1[CdcConsumer<br/>aiokafka AIOKafkaConsumer]
        C2[ExamCache<br/>exam_id→class_id 内存映射]
        C3[EventHandler<br/>grades/exams/homework/classes 路由]
    end

    subgraph Storage["存储 / 总线"]
        CH[(ClickHouse<br/>edu_analytics 库)]
        KAFKA[(Kafka<br/>edu-cdc.* + edu.insight.mastery.updated)]
        IAM[iam:3002 gRPC]
    end

    HTTP --> AUTH --> SCOPE --> S1
    HTTP --> S3
    GRPC --> S1
    S1 --> R1
    S3 --> R1
    S2 --> R1
    S2 --> R2
    SCOPE --> R3
    R1 --> CH
    R2 --> KAFKA
    R3 --> IAM

    C1 --> C3
    C3 --> C2
    C3 --> R1
    C1 --> KAFKA

分层规则：

入口层：HTTP（保留作 Gateway 直连降级）+ gRPC（主入口，BFF 调用）。两入口共享同一 Application Service
中间件层：FastAPI Depends 链（AuthDepends → DataScopeDepends）；gRPC 用 server interceptor 注入身份元数据
应用服务层：编排查询 / 计算掌握度 / 发布事件，不直接访问存储
数据访问层：ClickHouse 查询封装 + Kafka producer + gRPC client（调 iam）
CDC 消费者：独立后台任务（lifespan 启动），与 HTTP/gRPC 入口解耦

2. 领域模型

data-ana 是纯读模型服务，不持有写聚合根。领域模型为视图聚合（ClickHouse 物化）：

聚合根（视图型）

聚合根	含义	物化载体	不变式
`StudentDashboard`	学生学情宽表	ClickHouse `student_dashboard_view`	同一 (student_id, exam_id, knowledge_point_id) 仅保留最新版本
`ClassPerformance`	班级成绩聚合	ClickHouse 即时聚合（不物化）	聚合维度为 class_id + 时间窗
`StudentErrorBook`	学生错题本	ClickHouse `student_errors`	同一 (student_id, question_id) 累计 error_count
`MasterySnapshot`	知识点掌握度快照	ClickHouse `mastery_snapshot`（新增）	同一 (student_id, knowledge_point_id) 保留历史版本

值对象

WeakPoint：knowledge_point_id + title + mastery_level
TrendPoint：date + score
DataScope：level (SELF/CLASS/GRADE/SCHOOL/DISTRICT/ALL) + scope_ids（具体可见的 class_id / grade_id 列表）

聚合间通信

同服务内：直接函数调用（Application Service → Repository）
跨服务：仅通过 Kafka 事件（发布 mastery.updated）+ gRPC（调 iam 查 DataScope）

3. 数据模型（ClickHouse DDL）

DDL 文件由 coord 统一管理在 infra/clickhouse/ddl/（待 coord 建立），data-ana 提供内容。

3.1 宽表 `student_dashboard_view`

-- 学生学情宽表：每次成绩写入产生一行，按 ORDER BY 去重保留最新版本
CREATE TABLE IF NOT EXISTS student_dashboard_view
(
    student_id        String,
    class_id          String,
    exam_id           String,
    subject_id        String,
    score             Float64,
    rank_in_class     UInt32,
    knowledge_point_id String,
    mastery_level     Float32,       -- 0.0-1.0
    error_count       UInt32,
    last_updated      DateTime64(3, 'UTC')
)
ENGINE = ReplacingMergeTree(last_updated)
PARTITION BY toYYYYMM(last_updated)
ORDER BY (student_id, exam_id, knowledge_point_id)
SETTINGS index_granularity = 8192;

索引策略：

ORDER BY (student_id, exam_id, knowledge_point_id)：主键索引，支持按学生查学情、按考试查成绩、按知识点查掌握度
PARTITION BY toYYYYMM(last_updated)：按月分区，支持历史数据归档
ReplacingMergeTree(last_updated)：同 ORDER BY 自动去重，保留 last_updated 最大版本（幂等消费保证）

3.2 错题本 `student_errors`

CREATE TABLE IF NOT EXISTS student_errors
(
    student_id          String,
    question_id         String,
    knowledge_point_id  String,
    error_count         UInt32,
    last_error_time     DateTime64(3, 'UTC'),
    content             String
)
ENGINE = ReplacingMergeTree(last_error_time)
PARTITION BY toYYYYMM(last_error_time)
ORDER BY (student_id, question_id);

3.3 掌握度快照 `mastery_snapshot`（新增）

-- 知识点掌握度历史快照：每次掌握度计算产生新版本，支持趋势查询
CREATE TABLE IF NOT EXISTS mastery_snapshot
(
    student_id          String,
    knowledge_point_id  String,
    subject_id          String,
    mastery_level       Float32,
    calculated_at       DateTime64(3, 'UTC'),
    calculation_method  LowCardinality(String)  -- 'weighted_moving_avg' / 'simple_avg'
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(calculated_at)
ORDER BY (student_id, knowledge_point_id, calculated_at);

3.4 用量计费 `ai_usage_log`（新增，供 ai 服务写入）

-- AI 用量记录：ai 服务通过 Kafka 事件投递，data-ana 消费落库
CREATE TABLE IF NOT EXISTS ai_usage_log
(
    request_id          String,
    user_id             String,
    provider            LowCardinality(String),  -- 'openai' / 'anthropic' / 'baichuan' / 'local'
    model               LowCardinality(String),
    prompt_tokens       UInt32,
    completion_tokens   UInt32,
    total_tokens        UInt32,
    latency_ms          UInt32,
    success             Boolean,
    occurred_at         DateTime64(3, 'UTC')
)
ENGINE = MergeTree
PARTITION BY toYYYYMM(occurred_at)
ORDER BY (user_id, occurred_at);

3.5 读写分离策略

操作	路径	说明
学情查询	ClickHouse 宽表	实时聚合，亚秒级响应
错题本查询	ClickHouse student_errors	实时查询
掌握度趋势	ClickHouse mastery_snapshot	历史快照
掌握度计算	CDC 触发 → 内存计算 → 写 ClickHouse	派生数据，非事务写
DataScope 解析	gRPC 调 iam	实时查询，结果 Redis 缓存 5min（004 §6.3）

4. API 设计

4.1 HTTP 端点（保留作 Gateway 直连降级）

method	path	权限	请求	响应	说明
GET	`/healthz`	—	—	`{status, service}`	liveness
GET	`/readyz`	—	—	`{status, ready, degraded, clickhouse, cdc_consumer}`	readiness
GET	`/metrics`	—	—	Prometheus 格式	指标
GET	`/analytics/class/{class_id}/performance`	`ANALYTICS_CLASS_READ`	query: `subject_id?`, `start_date?`, `end_date?`	`ClassPerformanceResponse`	班级成绩分析
GET	`/analytics/student/{student_id}/weakness`	`ANALYTICS_STUDENT_READ`	query: `subject_id?`	`StudentWeaknessResponse`	学生薄弱知识点（DataScope 过滤）
GET	`/analytics/student/{student_id}/errorbook`	`ANALYTICS_STUDENT_READ`	query: `page?`, `size?`	`StudentErrorBookResponse`	学生错题本（DataScope 过滤）
GET	`/analytics/student/{student_id}/trend`	`ANALYTICS_STUDENT_READ`	query: `start_date`, `end_date`, `subject_id?`	`LearningTrendResponse`	学习趋势（新增）
GET	`/analytics/dashboard/teacher/{user_id}`	`ANALYTICS_TEACHER_DASHBOARD`	query: `class_id?`	`TeacherDashboardResponse`	教师仪表盘聚合（新增）

4.2 gRPC 契约（analytics.proto，待实现 server）

RPC	请求	响应	权限
`GetClassPerformance`	`GetClassPerformanceRequest{class_id, subject_id, start_date, end_date}`	`ClassPerformance`	`ANALYTICS_CLASS_READ`
`GetStudentWeakness`	`GetStudentWeaknessRequest{student_id, subject_id}`	`StudentWeakness`	`ANALYTICS_STUDENT_READ`
`GetLearningTrend`	`GetLearningTrendRequest{student_id, start_date, end_date}`	`LearningTrend`	`ANALYTICS_STUDENT_READ`

权限校验：gRPC server interceptor 从 metadata 提取 x-user-id / x-user-roles / x-data-scope，调用 AuthDepends 等价逻辑。

4.3 Pydantic 请求/响应模型

# 示例：班级成绩分析响应
class ClassPerformanceResponse(BaseModel):
    success: bool
    data: ClassPerformanceData
    degraded: bool = False

class ClassPerformanceData(BaseModel):
    class_id: str
    average_score: float
    pass_rate: float
    total_students: int
    scores: list[StudentScore] = []  # 详细成绩列表（受 DataScope 过滤）

class StudentScore(BaseModel):
    student_id: str
    score: float
    grade: str

5. 事件设计

5.1 消费的事件

Topic	来源	消息格式	消费动作
`edu-cdc.next_edu_cloud.core_edu_grades`	Debezium CDC（core-edu MySQL）	Debezium JSON（before/after/source/op/ts_ms）	解析 → 查 ExamCache 填 class_id → 计算掌握度 → upsert student_dashboard_view
`edu-cdc.next_edu_cloud.core_edu_exams`	Debezium CDC	Debezium JSON	upsert ExamCache（exam_id → class_id, subject_id）
`edu-cdc.next_edu_cloud.core_edu_homework_submissions`	Debezium CDC（新增订阅）	Debezium JSON	记录作业提交行为 → 更新 student_dashboard_view
`edu-cdc.next_edu_cloud.classes`	Debezium CDC	Debezium JSON	同步班级维度（head_teacher_id）用于教师 DataScope
`edu-cdc.next_edu_cloud.iam_users`	Debezium CDC（新增订阅）	Debezium JSON	同步用户 dataScope 用于查询过滤（避免每次查 iam）
`edu.insight.ai.usage`	ai 服务 Kafka producer（待 coord 新增 topic）	JSON（UsageRecord）	落 ai_usage_log 表

幂等性：

CDC 事件：依赖 ClickHouse ReplacingMergeTree(last_updated) 引擎按 ORDER BY 去重
领域事件（若消费）：基于 event_id 去重（Redis SETNX，TTL 7 天）

5.2 发布的事件

事件	Topic	触发时机	消费者	Payload
`MasteryUpdated`	`edu.insight.mastery.updated`	掌握度计算完成（CDC grades 事件触发后异步计算）	core-edu（推荐个性化练习）、msg（掌握度预警）	`{event_id, student_id, knowledge_point_id, mastery_level, calculated_at}`

发布实现（Python 无 Outbox 模式）：

掌握度计算是派生数据（非业务事务写），不需要 Outbox 保证事务一致
直接用 aiokafka.AIOKafkaProducer 发布，idempotent=true + 事务性 producer
失败重试 3 次，仍失败记录日志 + 落 mastery_publish_failed 本地表（待 P6 评估是否引入 Outbox）

6. 横切关注点对齐清单

6.1 权限装饰器等价物（FastAPI Depends）

# 权限点常量（与 iam 权限点对齐）
class Permissions:
    ANALYTICS_CLASS_READ = "analytics:class:read"
    ANALYTICS_STUDENT_READ = "analytics:student:read"
    ANALYTICS_TEACHER_DASHBOARD = "analytics:teacher:dashboard"

async def require_permission(permission: str) -> UserContext:
    """FastAPI Depends 权限校验.

    从 x-user-id / x-user-roles 头提取身份，校验角色是否含 permission。
    """
    ...

async def inject_data_scope(ctx: UserContext = Depends(require_permission(...)))-> DataScope:
    """注入 DataScope（从 iam.getEffectiveDataScope 查询，Redis 缓存 5min）."""
    ...

6.2 错误码清单（前缀 `DATA_ANA_*`）

错误码	触发条件	HTTP	gRPC status
`DATA_ANA_UNAUTHORIZED`	缺失 x-user-id 头或 token 无效	401	UNAUTHENTICATED
`DATA_ANA_FORBIDDEN`	角色无对应权限	403	PERMISSION_DENIED
`DATA_ANA_DATASCOPE_VIOLATION`	查询目标超出 DataScope 范围	403	PERMISSION_DENIED
`DATA_ANA_CLICKHOUSE_UNAVAILABLE`	ClickHouse 不可达（降级模式仍返回骨架）	200 + degraded:true	OK + degraded flag
`DATA_ANA_INVALID_DATE_RANGE`	start_date > end_date	400	INVALID_ARGUMENT
`DATA_ANA_STUDENT_NOT_FOUND`	student_id 不存在	404	NOT_FOUND
`DATA_ANA_CLASS_NOT_FOUND`	class_id 不存在	404	NOT_FOUND
`DATA_ANA_INTERNAL_ERROR`	未捕获异常	500	INTERNAL

6.3 Logger 初始化

位置：main.py init_logger()（已具备）
配置：structlog.make_filtering_bound_logger(level) + TimeStamper(fmt="iso") + ConsoleRenderer
改进：生产环境改用 structlog.processors.JSONRenderer()（当前 ConsoleRenderer 适合开发）

6.4 Metrics 指标清单

指标名	类型	标签	描述
`data_ana_http_requests_total`	Counter	method, path, status	HTTP 请求总数
`data_ana_http_request_duration_seconds`	Histogram	method, path	HTTP 请求延迟
`data_ana_clickhouse_query_duration_seconds`	Histogram	query_type	ClickHouse 查询延迟
`data_ana_clickhouse_query_total`	Counter	query_type, status	ClickHouse 查询总数
`data_ana_cdc_events_consumed_total`	Counter	table, op	CDC 事件消费总数
`data_ana_cdc_event_process_duration_seconds`	Histogram	table	CDC 事件处理延迟
`data_ana_cdc_consumer_lag`	Gauge	topic, partition	CDC 消费者 lag
`data_ana_mastery_calculated_total`	Counter	—	掌握度计算次数
`data_ana_mastery_published_total`	Counter	status	mastery.updated 事件发布数
`data_ana_datascope_cache_hits_total`	Counter	—	DataScope 缓存命中

6.5 Tracer 初始化

位置：main.py init_tracer()（已具备）
endpoint：settings.otel_endpoint + /v1/traces
改进：gRPC server 注册 grpc.aio.ServerInterceptor 透传 W3C trace context

6.6 /healthz 检查逻辑

liveness：仅进程存活（已具备）

6.7 /readyz 检查逻辑

async def readyz() -> dict:
    return {
        "status": "ok" if all_ready else "degraded",
        "ready": all_ready,
        "degraded": not all_ready,
        "clickhouse": "ok" | "unreachable" | "not_configured",
        "cdc_consumer": "running" | "disabled" | "failed",
        "kafka_brokers": settings.kafka_brokers or None,
        "iam_grpc": "ok" | "unreachable",  # 新增：iam gRPC 连通性
        "timestamp": datetime.now(UTC).isoformat(),
    }

6.8 优雅关闭顺序

HTTP server stop accepting new requests（uvicorn graceful shutdown）
gRPC server graceful stop（等待在途 RPC 完成，30s 超时）
CDC consumer stop（等待在途消息处理完成，commit offset）
Kafka producer flush + close（确保 mastery.updated 事件已投递）
ClickHouse client close
iam gRPC channel close

信号处理：注册 signal.SIGTERM handler，触发上述顺序。

7. 与其他模块的交互点（契约清单）

方向	对方服务	协议	接口/事件	用途
被调用	api-gateway	HTTP	`/analytics/*`	Gateway 代理
被调用	teacher-bff / student-bff	gRPC	`AnalyticsService.*`	BFF 聚合查询
被调用	ai	gRPC	`AnalyticsService.GetStudentWeakness / GetLearningTrend`	AI 个性化出题上下文
调用	iam	gRPC	`IamService.GetEffectiveDataScope`（待 proto 新增）	DataScope 解析
消费	core-edu（CDC）	Kafka	`edu-cdc.next_edu_cloud.core_edu_grades/exams/homework_submissions`	学情数据投递
消费	core-edu（CDC）	Kafka	`edu-cdc.next_edu_cloud.classes`	班级维度同步
消费	iam（CDC）	Kafka	`edu-cdc.next_edu_cloud.iam_users`	用户 dataScope 同步
消费	ai	Kafka	`edu.insight.ai.usage`（待 coord 新增）	AI 用量落库
发布	—	Kafka	`edu.insight.mastery.updated`	掌握度更新通知 core-edu / msg

8. 风险与假设

8.1 假设

假设 1：iam 提供 GetEffectiveDataScope(userId) → DataScope gRPC API。若 iam 未提供，fallback 为：从 x-user-roles 头推导（admin=ALL, teacher=CLASS_TAUGHT, student=SELF），但无法支持细粒度年级/学校范围
假设 2：core-edu 的 core_edu_homework_submissions 表存在 binlog。若不存在，作业相关学情无法通过 CDC 获取，需 core-edu 补表或走 Outbox 事件
假设 3：ClickHouse ReplacingMergeTree 在查询时需 FINAL 关键字确保去重生效。当前查询未加 FINAL，可能读到重复版本。修复：所有查询加 FINAL 或使用 argMax 聚合
假设 4：coord 同意新增 edu.insight.ai.usage topic。若不同意，ai 服务的用量计费需自建记录（破坏 ai 无状态原则）

8.2 技术风险

风险	影响	缓解措施
ClickHouse 查询延迟超 5s	违反 P4 退出标准	宽表索引优化 + 物化视图预聚合 + 查询超时 3s 降级
CDC 消费者 lag 过大	学情数据延迟 > 5s	监控 lag + 告警 + 水平扩展消费者（分区数提升）
ExamCache 内存泄漏	长期运行 OOM	LRU 淘汰策略（max 10000 条）+ 定期清理过期 exam
mastery.updated 事件丢失	下游 core-edu/msg 收不到通知	Kafka producer `acks=all` + 本地失败表重试
iam gRPC 不可达	DataScope 无法解析 → 查询降级为 SELF	Redis 缓存 5min + fallback SELF 范围（最保守）
ClickHouse `FINAL` 查询性能	查询变慢	使用 `argMax` 替代 `FINAL`，或在写入时去重

8.3 未决设计决策（需 coord 仲裁）

mastery.updated 发布是否需要 Outbox：Python 无 Outbox 模式，建议直接 producer；但 004 §12.2 明确"事件发布：Outbox 模式 / 禁止直接 Kafka producer"。冲突：data-ana 是 Python 服务无 MySQL 写事务，Outbox 不适用。建议 coord 裁定：派生数据事件（非业务事务）允许直接 producer
iam GetEffectiveDataScope proto 新增：当前 iam.proto 仅有 GetUserInfo，无 DataScope 解析 API。需 coord 在 shared-proto 新增 GetEffectiveDataScope RPC
edu.insight.ai.usage topic 新增：004 §7.2 未列出，需 coord 确认是否新增

模块架构设计文档 — ai

1. 模块内部分层图

flowchart TB
    subgraph Entry["入口层"]
        HTTP[FastAPI HTTP Router<br/>/ai/* + /healthz + /readyz]
        GRPC[grpc.aio Server<br/>AiService 含 StreamChat]
    end

    subgraph Middleware["中间件层"]
        AUTH[AuthDepends<br/>校验 x-user-id / x-user-roles]
        RATE[RateLimitDepends<br/>Redis 令牌桶限流]
        TRACE[OTel + grpc interceptor]
    end

    subgraph Service["应用服务层"]
        S1[ChatService<br/>聊天编排 + Prompt 模板渲染]
        S2[QuestionGenerationService<br/>出题工作流编排]
        S3[ExpressionOptimizationService<br/>表达优化]
        S4[LessonPreparationWorkflow<br/>备课工作流 4 步编排]
    end

    subgraph Provider["LLM Provider 适配层"]
        P0[LLMProvider 抽象接口<br/>chat / stream_chat]
        P1[OpenAIProvider<br/>httpx 异步]
        P2[AnthropicProvider<br/>httpx 异步]
        P3[BaichuanProvider<br/>httpx 异步]
        P4[LocalOllamaProvider<br/>httpx 异步]
    end

    subgraph Template["Prompt 模板管理"]
        T1[PromptTemplateRegistry<br/>模板注册 + 渲染]
        T2[模板存储<br/>YAML 文件 / DB]
    end

    subgraph Client["下游 gRPC client"]
        C1[ContentClient<br/>查询知识点 / 题库]
        C2[DataAnaClient<br/>查询学情 / 薄弱点]
    end

    subgraph Usage["用量计费"]
        U1[UsageRecorder<br/>token 消耗统计]
        U2[KafkaProducer<br/>发布 edu.insight.ai.usage]
    end

    subgraph External["外部 / 存储"]
        LLM[LLM Provider API<br/>OpenAI/Anthropic/百川/Ollama]
        CONTENT[content:3005 gRPC]
        DATAANA[data-ana:3006 gRPC]
        KAFKA[(Kafka)]
        REDIS[(Redis<br/>限流 + 缓存)]
    end

    HTTP --> AUTH --> RATE --> S1
    HTTP --> S2
    HTTP --> S3
    GRPC --> S1
    GRPC --> S2
    S2 --> S4
    S4 --> C1
    S4 --> C2
    S1 --> T1
    S2 --> T1
    S1 --> P0
    S2 --> P0
    P0 --> P1
    P0 --> P2
    P0 --> P3
    P0 --> P4
    P1 --> LLM
    P2 --> LLM
    P3 --> LLM
    P4 --> LLM
    S1 --> U1
    S2 --> U1
    U1 --> U2
    U2 --> KAFKA
    RATE --> REDIS
    C1 --> CONTENT
    C2 --> DATAANA

分层规则：

入口层：HTTP（保留作 Gateway 直连）+ gRPC（主入口，含 StreamChat 流式 RPC）
中间件层：Auth + RateLimit（Redis 令牌桶，按 user_id 限流）
应用服务层：4 个 Service，每个对应一类 AI 能力
Provider 适配层：抽象 LLMProvider 接口，多适配器实现（策略模式）
Prompt 模板：模板注册 + 渲染，模板存储可配置（YAML 文件 or DB）
下游 client：gRPC 调 content / data-ana
用量计费：token 消耗统计 + Kafka 事件外发

2. 领域模型

ai 是无状态服务，不持有持久化聚合根。领域模型为请求/响应模型 + 工作流编排：

聚合根（请求型，无持久化）

聚合根	含义	生命周期
`ChatConversation`	单次聊天请求	单次请求
`QuestionGenerationTask`	出题任务	单次请求（备课工作流中多步）
`ExpressionOptimizationTask`	表达优化任务	单次请求
`LessonPreparationWorkflow`	备课工作流	跨多步（分析学情 → 推荐知识点 → 生成题目 → 教师审核 → 入库）

值对象

ChatMessage：role + content
Usage：prompt_tokens + completion_tokens + total_tokens
GeneratedQuestion：question + answer + explanation
PromptTemplate：name + system_prompt + user_template + variables

工作流编排（备课）

sequenceDiagram
    participant T as 教师
    participant BFF as teacher-bff
    participant AI as ai 服务
    participant Content as content
    participant DA as data-ana

    T->>BFF: 请求备课（class_id, subject_id）
    BFF->>AI: gRPC GenerateLessonPlan
    AI->>DA: gRPC GetStudentWeakness(class_id)
    DA-->>AI: 薄弱知识点列表
    AI->>Content: gRPC GetPrerequisites(knowledge_point_id)
    Content-->>AI: 前置依赖知识点
    AI->>AI: LLM 生成题目（基于学情 + 知识点）
    AI-->>BFF: 题目列表 + 推荐理由
    BFF-->>T: 题目供审核
    T->>BFF: 确认入库
    BFF->>Content: gRPC CreateQuestions
    Content-->>BFF: 入库成功

工作流实现：

简单场景（4 步内）：FastAPI BackgroundTasks + asyncio.gather 并行查询
复杂场景（含教师审核等待）：待 coord 仲裁是否引入 Temporal（004 §2.3 列出 Temporal 用于 AI 编排）

3. 数据模型

ai 服务无独占数据库，无 MySQL schema。所有数据通过 Kafka 事件外发（用量计费）或 gRPC 查询下游。

用量计费（Kafka 事件 → data-ana 落 ClickHouse）

{
  "event_id": "uuid",
  "user_id": "user-xxx",
  "request_id": "req-xxx",
  "provider": "openai",
  "model": "gpt-4o-mini",
  "prompt_tokens": 150,
  "completion_tokens": 80,
  "total_tokens": 230,
  "latency_ms": 1200,
  "success": true,
  "occurred_at": "2026-07-09T12:00:00Z"
}

Topic：edu.insight.ai.usage（待 coord 新增）

缓存策略

数据	存储	TTL	失效策略
Prompt 模板	Redis（模板变更事件驱动失效）	1 小时	文件/DB 变更时主动失效
LLM 响应（相同 prompt）	Redis（hash 缓存）	30 分钟	短 TTL，避免陈旧
DataScope（ai 不需要，仅 data-ana 用）	—	—	—
限流计数	Redis 令牌桶	滑动窗口	自动过期

4. API 设计

4.1 HTTP 端点（保留作 Gateway 直连降级）

method	path	权限	请求体	响应	说明
GET	`/healthz`	—	—	`{status, service}`	liveness
GET	`/readyz`	—	—	`{status, llm_configured, degraded}`	readiness
GET	`/metrics`	—	—	Prometheus	指标
POST	`/ai/chat`	`AI_CHAT`	`ChatRequest`	`ChatResponse`	LLM 聊天
POST	`/ai/chat/stream`	`AI_CHAT`	`ChatRequest`	SSE stream	流式聊天
POST	`/ai/generate/question`	`AI_QUESTION_GENERATE`	`GenerateQuestionRequest`	`GeneratedQuestionResponse`	生成题目
POST	`/ai/optimize/expression`	`AI_EXPRESSION_OPTIMIZE`	`OptimizeExpressionRequest`	`OptimizedExpressionResponse`	优化表达
POST	`/ai/lesson/preparation`	`AI_LESSON_PREPARE`	`LessonPreparationRequest`	`LessonPreparationResponse`	备课工作流（新增）

4.2 gRPC 契约（ai.proto，待实现 server）

RPC	请求	响应	权限	说明
`Chat`	`ChatRequest{messages, model, temperature}`	`ChatResponse{content, model, usage}`	`AI_CHAT`	非流式聊天
`StreamChat`	`ChatRequest`	`stream ChatChunk`	`AI_CHAT`	流式聊天（SSE over gRPC）
`GenerateQuestion`	`GenerateQuestionRequest{prompt, subject, difficulty}`	`GeneratedQuestion`	`AI_QUESTION_GENERATE`	生成题目
`OptimizeExpression`	`OptimizeExpressionRequest{text, context}`	`OptimizedExpression`	`AI_EXPRESSION_OPTIMIZE`	优化表达

4.3 Pydantic 请求/响应模型

class ChatRequest(BaseModel):
    messages: list[ChatMessage]
    model: str = "gpt-4o-mini"
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    stream: bool = False

class ChatMessage(BaseModel):
    role: Literal["system", "user", "assistant"]
    content: str

class ChatResponse(BaseModel):
    success: bool
    data: ChatData
    degraded: bool = False

class ChatData(BaseModel):
    content: str
    model: str
    usage: Usage

class Usage(BaseModel):
    prompt_tokens: int
    completion_tokens: int
    total_tokens: int

class GenerateQuestionRequest(BaseModel):
    prompt: str = Field(..., min_length=1, max_length=2000)
    subject: str
    difficulty: Literal["easy", "medium", "hard"]
    knowledge_point_ids: list[str] = []  # 可选：靶向知识点

class GeneratedQuestionResponse(BaseModel):
    success: bool
    data: GeneratedQuestionData
    degraded: bool = False

5. 事件设计

5.1 消费的事件

ai 服务不消费任何事件（无状态，纯请求-响应）。

5.2 发布的事件

事件	Topic	触发时机	消费者	Payload
`AIUsageRecorded`	`edu.insight.ai.usage`（待 coord 新增）	每次 LLM 调用完成	data-ana（落 ClickHouse ai_usage_log）	`{event_id, user_id, request_id, provider, model, prompt_tokens, completion_tokens, total_tokens, latency_ms, success, occurred_at}`

发布实现：

每次 LLM 调用后异步发布（不阻塞响应）
aiokafka.AIOKafkaProducer + acks=all
失败重试 3 次，仍失败记录日志（不影响主流程）

6. 横切关注点对齐清单

6.1 权限装饰器等价物

class Permissions:
    AI_CHAT = "ai:chat"
    AI_QUESTION_GENERATE = "ai:question:generate"
    AI_EXPRESSION_OPTIMIZE = "ai:expression:optimize"
    AI_LESSON_PREPARE = "ai:lesson:prepare"

async def require_permission(permission: str) -> UserContext:
    """从 x-user-id / x-user-roles 校验权限."""
    ...

6.2 错误码清单（前缀 `AI_*`）

错误码	触发条件	HTTP	gRPC status
`AI_UNAUTHORIZED`	缺失 x-user-id 或 token 无效	401	UNAUTHENTICATED
`AI_FORBIDDEN`	角色无对应权限	403	PERMISSION_DENIED
`AI_RATE_LIMITED`	触发限流	429	RESOURCE_EXHAUSTED
`AI_LLM_UNAVAILABLE`	LLM Provider 不可达（降级模式仍返回骨架）	200 + degraded:true	OK + degraded flag
`AI_LLM_TIMEOUT`	LLM 调用超时（30s）	504	DEADLINE_EXCEEDED
`AI_INVALID_MODEL`	model 名不支持	400	INVALID_ARGUMENT
`AI_INVALID_DIFFICULTY`	difficulty 不在 easy/medium/hard	400	INVALID_ARGUMENT
`AI_DOWNSTREAM_UNAVAILABLE`	content / data-ana gRPC 不可达	502	UNAVAILABLE
`AI_PROMPT_RENDER_FAILED`	Prompt 模板渲染失败（变量缺失）	500	INTERNAL
`AI_INTERNAL_ERROR`	未捕获异常	500	INTERNAL

6.3 Logger 初始化

位置：main.py（已具备）
改进：生产环境改用 JSONRenderer

6.4 Metrics 指标清单

指标名	类型	标签	描述
`ai_http_requests_total`	Counter	method, path, status	HTTP 请求总数
`ai_http_request_duration_seconds`	Histogram	method, path	HTTP 请求延迟
`ai_llm_calls_total`	Counter	provider, model, status	LLM 调用总数
`ai_llm_call_duration_seconds`	Histogram	provider, model	LLM 调用延迟
`ai_llm_tokens_total`	Counter	provider, model, type (prompt/completion)	token 消耗总数
`ai_llm_stream_chunks_total`	Counter	provider, model	流式 chunk 总数
`ai_grpc_calls_total`	Counter	downstream, method, status	下游 gRPC 调用总数
`ai_rate_limit_hits_total`	Counter	user_id	限流命中次数
`ai_usage_events_published_total`	Counter	status	用量事件发布数
`ai_prompt_template_renders_total`	Counter	template_name, status	模板渲染次数

6.5 Tracer 初始化

位置：main.py init_tracer()（已具备，dev_mode 跳过）
改进：gRPC server interceptor + 下游 gRPC client interceptor 透传 trace context

6.6 /healthz 检查逻辑

liveness：仅进程存活（已具备）

6.7 /readyz 检查逻辑

async def readyz() -> dict:
    return {
        "status": "ok",
        "service": "ai",
        "llm_configured": settings.llm_available,
        "degraded": not settings.llm_available,
        "providers": {
            "openai": bool(settings.openai_api_key),
            "anthropic": bool(settings.anthropic_api_key),
            "baichuan": bool(settings.baichuan_api_key),
            "local_ollama": bool(settings.ollama_base_url),
        },
        "downstream_grpc": {
            "content": "ok" | "unreachable",  # 新增
            "data_ana": "ok" | "unreachable",  # 新增
        },
        "redis": "ok" | "unreachable",  # 新增（限流依赖）
    }

6.8 优雅关闭顺序

HTTP server stop accepting new requests
gRPC server graceful stop（关键：等待在途 StreamChat 流式 RPC 完成，60s 超时，避免截断用户响应）
LLM 流式请求 drain（等待 httpx stream 完成）
Kafka producer flush + close（确保用量事件已投递）
下游 gRPC channels close（content / data-ana）
Redis connection close

7. 与其他模块的交互点（契约清单）

方向	对方服务	协议	接口/事件	用途
被调用	api-gateway	HTTP	`/ai/*`	Gateway 代理
被调用	teacher-bff	gRPC	`AiService.*`	BFF 聚合 AI 能力
调用	content	gRPC	`KnowledgeGraphService.GetPrerequisites / GetLearningPath`	出题上下文查询
调用	content	gRPC	`TextbookService.*`（若需教材上下文）	出题教材关联
调用	data-ana	gRPC	`AnalyticsService.GetStudentWeakness / GetLearningTrend`	个性化出题学情查询
调用	LLM Provider	HTTP	OpenAI 兼容 REST `/chat/completions`	LLM 推理
发布	—	Kafka	`edu.insight.ai.usage`（待 coord 新增）	用量计费外发

8. 风险与假设

8.1 假设

假设 1：content 服务实现了 KnowledgeGraphService gRPC server。当前 content.proto 已定义但未实现 gRPC server（ai05 阶段 2 设计中）。ai 调用前需确认 content gRPC 可用
假设 2：data-ana 实现了 AnalyticsService gRPC server（本设计文档已设计）。ai 调用前需确认 data-ana gRPC 可用
假设 3：coord 同意新增 edu.insight.ai.usage topic。若不同意，用量计费降级为 ai 本地日志（不落 ClickHouse，影响成本分析）
假设 4：Redis 可用（限流依赖）。若 Redis 不可用，限流降级为"无限制"（风险：LLM 成本失控），或降级为内存令牌桶（单实例有效，多实例不一致）

8.2 技术风险

风险	影响	缓解措施
LLM 调用延迟高（>30s）	用户体验差	超时 30s + 降级骨架响应 + 流式优先（用户感知首字延迟）
LLM 成本失控	财务风险	Redis 令牌桶限流（每用户每分钟 10 次）+ 用量计费监控 + 告警阈值
LLM Provider 单点故障	服务不可用	多 Provider 适配器 + 自动 failover（OpenAI 失败切 Anthropic）
流式 RPC 中断	用户响应截断	gRPC server graceful shutdown 60s drain + 客户端重连机制
Prompt 注入攻击	LLM 输出恶意内容	输入 sanitize + system prompt 加安全约束 + 输出过滤
下游 gRPC 不可达	备课工作流失败	降级：跳过学情查询，仅基于 prompt 生成题目 + `degraded: true`

8.3 未决设计决策（需 coord 仲裁）

备课工作流是否引入 Temporal：004 §2.3 列出 Temporal 用于 AI 编排，但简单 4 步工作流可用 FastAPI BackgroundTasks。建议：M14 用 BackgroundTasks，M15 评估是否迁移 Temporal
edu.insight.ai.usage topic 新增：需 coord 在 shared-proto events.proto 新增 AIUsageEvent message + 004 §7.2 新增 topic
LLM Provider 配置化路由：是否在 shared-py 建立通用 LLMProvider 抽象（供未来其他 Python 服务复用）。建议 P5 阶段在 ai 服务内部实现，P6 评估是否提取到 shared-py
Prompt 模板存储位置：YAML 文件（简单，无 DB）vs DB（动态更新）。建议 P5 用 YAML 文件（services/ai/src/ai/prompts/*.yaml），P6 评估迁移 DB

阶段 2 总结

ai06 已完成阶段 2 模块架构设计，产出 data-ana 与 ai 两份设计文档。核心设计决策：

data-ana 设计要点

分层：HTTP + gRPC 双入口共享 Application Service；CDC 消费者独立后台任务
数据模型：4 张 ClickHouse 宽表（student_dashboard_view / student_errors / mastery_snapshot / ai_usage_log），ReplacingMergeTree 引擎保证幂等
权限：FastAPI Depends 链（require_permission + inject_data_scope），DataScope WHERE 注入 ClickHouse 查询
事件：消费 6 个 CDC topic + 发布 edu.insight.mastery.updated（直接 producer，非 Outbox，因派生数据非事务写）
gRPC：实现 AnalyticsService server（analytics.proto 已定义）
降级：ClickHouse 不可达返回骨架 + degraded:true；iam gRPC 不可达降级为 SELF DataScope

ai 设计要点

分层：HTTP + gRPC 双入口；LLM Provider 适配层（策略模式，4 适配器）
无状态：无 DB，用量计费通过 Kafka 事件外发
权限：FastAPI Depends + Redis 令牌桶限流（每用户每分钟 10 次）
工作流：备课 4 步编排（学情查询 → 知识点推荐 → 题目生成 → 教师审核入库），简单场景用 BackgroundTasks
gRPC：实现 AiService server（含 StreamChat 流式 RPC）；gRPC client 调 content / data-ana
降级：LLM 不可达返回骨架 + degraded:true；下游 gRPC 不可达降级跳过

待 coord 交叉审查项（汇总）

#	议题	涉及文档
1	data-ana 发布 `edu.insight.mastery.updated` 用直接 producer（非 Outbox）是否合规	004 §12.2
2	新增 `edu.insight.ai.usage` topic + `AIUsageEvent` proto message	004 §7.2 + events.proto
3	iam 新增 `GetEffectiveDataScope` gRPC RPC	iam.proto
4	data-ana / ai 实现 gRPC server 决策	004 §4.1
5	ClickHouse DDL 管理位置（建议 `infra/clickhouse/ddl/`）	infra/
6	ai 备课工作流是否引入 Temporal	004 §2.3
7	端口冲突检查：data-ana 3006 / ai 3008（无冲突）	full-stack-runbook
8	错误码前缀检查：`DATA_ANA_` / `AI_`（与其他服务不重叠）	—
9	黄金模板对齐：Python 服务无 NestJS 装饰器，权限校验用 FastAPI Depends 等价物是否认可	—

下一步：等待 coord 交叉审查通过后，进入阶段 3（按图实施）。

AI Agent: ai06 (data-ana + ai) Branch: 单仓库并行模式（直接提交 main） Coordinator: coord-ai

48 KiB Raw Blame History Unescape Escape

ai06 阶段 2 交付物：模块架构设计文档

设计原则与全局约束

模块架构设计文档 — data-ana

1. 模块内部分层图

2. 领域模型

聚合根（视图型）

值对象

聚合间通信

3. 数据模型（ClickHouse DDL）

3.1 宽表 student_dashboard_view

3.2 错题本 student_errors

3.3 掌握度快照 mastery_snapshot（新增）

3.4 用量计费 ai_usage_log（新增，供 ai 服务写入）

3.5 读写分离策略

4. API 设计

4.1 HTTP 端点（保留作 Gateway 直连降级）

4.2 gRPC 契约（analytics.proto，待实现 server）

4.3 Pydantic 请求/响应模型

5. 事件设计

5.1 消费的事件

5.2 发布的事件

6. 横切关注点对齐清单

6.1 权限装饰器等价物（FastAPI Depends）

6.2 错误码清单（前缀 DATA_ANA_*）

6.3 Logger 初始化

6.4 Metrics 指标清单

6.5 Tracer 初始化

6.6 /healthz 检查逻辑

6.7 /readyz 检查逻辑

6.8 优雅关闭顺序

7. 与其他模块的交互点（契约清单）

8. 风险与假设

8.1 假设

8.2 技术风险

8.3 未决设计决策（需 coord 仲裁）

模块架构设计文档 — ai

1. 模块内部分层图

2. 领域模型

聚合根（请求型，无持久化）

值对象

工作流编排（备课）

3. 数据模型

用量计费（Kafka 事件 → data-ana 落 ClickHouse）

缓存策略

4. API 设计

4.1 HTTP 端点（保留作 Gateway 直连降级）

4.2 gRPC 契约（ai.proto，待实现 server）

4.3 Pydantic 请求/响应模型

5. 事件设计

5.1 消费的事件

5.2 发布的事件

6. 横切关注点对齐清单

6.1 权限装饰器等价物

6.2 错误码清单（前缀 AI_*）

6.3 Logger 初始化

6.4 Metrics 指标清单

6.5 Tracer 初始化

6.6 /healthz 检查逻辑

6.7 /readyz 检查逻辑

6.8 优雅关闭顺序

7. 与其他模块的交互点（契约清单）

8. 风险与假设

8.1 假设

8.2 技术风险

8.3 未决设计决策（需 coord 仲裁）

阶段 2 总结

data-ana 设计要点

ai 设计要点

待 coord 交叉审查项（汇总）

48 KiB

Raw Blame History

3.1 宽表 `student_dashboard_view`

3.2 错题本 `student_errors`

3.3 掌握度快照 `mastery_snapshot`（新增）

3.4 用量计费 `ai_usage_log`（新增，供 ai 服务写入）

6.2 错误码清单（前缀 `DATA_ANA_*`）

6.2 错误码清单（前缀 `AI_*`）