iven/hms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
d6420f4e274ca2e45418d73c30245cd8d78261cc
hms/docs/superpowers/plans/2026-04-25-erp-ai-phase1-mvp.md
iven 2963e7ce63
Some checks failed
CI / rust-check (push) Has been cancelled
Details
CI / rust-test (push) Has been cancelled
Details
CI / frontend-build (push) Has been cancelled
Details
CI / security-audit (push) Has been cancelled
Details
docs(ai): 实施计划 Chunk 6 (SSE Handler + 端到端验证 + 后续任务)
2026-04-25 13:41:23 +08:00

2218 lines
64 KiB
Markdown
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# erp-ai Phase 1 MVP 实施计划
> **For agentic workers:** REQUIRED: Use superpowers:subagent-driven-development (if subagents available) or superpowers:executing-plans to implement this plan. Steps use checkbox (`- [ ]`) syntax for tracking.
**Goal:** 为 HMS 新建 erp-ai crate，实现 AI 智能分析流 SSE 端点，支持化验单解读/趋势分析/个性化方案/报告摘要
**Architecture:** 新建独立 erp-ai crate，通过 HealthDataProvider trait 从 erp-health 获取脱敏数据，AiProvider trait 抽象 AI 提供商（Phase 1 实现 Claude SSE），请求驱动管道 + SSE 流式返回
**Tech Stack:** Rust/Axum/SeaORM/PostgreSQL + futures/tokio-stream/async-stream (SSE) + serde_json/uuid/chrono/thiserror/utoipa
**设计规格:** `docs/superpowers/specs/2026-04-25-erp-ai-module-design.md`
---
## Chunk 1: Crate 骨架 + 错误类型 + erp-core 扩展
### Task 1: 创建 erp-ai crate 骨架
**Files:**
- Create: `crates/erp-ai/Cargo.toml`
- Create: `crates/erp-ai/src/lib.rs`
- Create: `crates/erp-ai/src/error.rs`
- Modify: `Cargo.toml` (workspace root) — 添加 erp-ai 到 workspace
- [ ] **Step 1: 创建 crate 目录**
```bash
mkdir -p crates/erp-ai/src
```
- [ ] **Step 2: 创建 Cargo.toml**
```toml
# crates/erp-ai/Cargo.toml
[package]
name = "erp-ai"
version.workspace = true
edition.workspace = true
[dependencies]
erp-core.workspace = true
tokio = { workspace = true, features = ["full"] }
tokio-stream.workspace = true
futures.workspace = true
async-stream.workspace = true
serde.workspace = true
serde_json.workspace = true
uuid.workspace = true
chrono.workspace = true
axum.workspace = true
sea-orm.workspace = true
tracing.workspace = true
thiserror.workspace = true
utoipa.workspace = true
async-trait.workspace = true
reqwest = { version = "0.12", features = ["stream", "json"] }
handlebars = "6"
sha2 = "0.10"
hex = "0.4"
```
> 注意: `futures`, `tokio-stream`, `async-stream`, `reqwest`, `handlebars`, `sha2` 需要加入 workspace 依赖或在此声明版本。参照 `crates/erp-health/Cargo.toml` 模式。
- [ ] **Step 3: 创建 error.rs**
```rust
// crates/erp-ai/src/error.rs
use erp_core::AppError;
#[derive(Debug, thiserror::Error)]
pub enum AiError {
#[error("验证失败: {0}")]
Validation(String),
#[error("分析未找到: {0}")]
AnalysisNotFound(String),
#[error("Prompt 模板未找到: {0}")]
PromptNotFound(String),
#[error("AI 提供商不可用: {0}")]
ProviderUnavailable(String),
#[error("AI 提供商错误: {0}")]
ProviderError(String),
#[error("数据脱敏失败: {0}")]
SanitizationError(String),
#[error("模板渲染失败: {0}")]
TemplateError(String),
#[error("速率超限")]
RateLimitExceeded,
#[error("版本不匹配")]
VersionMismatch,
#[error("数据库错误: {0}")]
DbError(String),
}
impl From<AiError> for AppError {
fn from(e: AiError) -> Self {
match e {
AiError::Validation(msg) => AppError::Validation(msg),
AiError::AnalysisNotFound(id) => AppError::NotFound(format!("分析结果: {id}")),
AiError::PromptNotFound(name) => AppError::NotFound(format!("Prompt 模板: {name}")),
AiError::ProviderUnavailable(p) => AppError::ServiceUnavailable(format!("AI 提供商 {p} 不可用")),
AiError::RateLimitExceeded => AppError::TooManyRequests,
AiError::VersionMismatch => AppError::VersionMismatch,
AiError::DbError(msg) => AppError::Internal(msg),
other => AppError::Internal(other.to_string()),
}
}
}
impl From<sea_orm::DbErr> for AiError {
fn from(e: sea_orm::DbErr) -> Self {
AiError::DbError(e.to_string())
}
}
pub type AiResult<T> = Result<T, AiError>;
```
> 注意: 检查 `AppError` 是否有 `ServiceUnavailable` 变体。如果没有，使用 `AppError::Internal` 替代。
- [ ] **Step 4: 创建 lib.rs (最小骨架)**
```rust
// crates/erp-ai/src/lib.rs
pub mod error;
pub use error::{AiError, AiResult};
```
- [ ] **Step 5: 注册到 workspace**
在根 `Cargo.toml` 的 `[workspace] members` 数组中添加 `"crates/erp-ai"`，在 `[workspace.dependencies]` 中添加:
```toml
erp-ai = { path = "crates/erp-ai" }
```
同时确认以下依赖在 workspace dependencies 中存在（如不存在则添加）:
```toml
futures = "0.3"
tokio-stream = "0.1"
async-stream = "0.3"
reqwest = { version = "0.12", features = ["stream", "json"] }
handlebars = "6"
sha2 = "0.10"
hex = "0.4"
```
- [ ] **Step 6: 验证编译**
```bash
cargo check -p erp-ai
```
Expected: 编译通过，无错误
- [ ] **Step 7: 提交**
```bash
git add crates/erp-ai/ Cargo.toml
git commit -m "feat(ai): 创建 erp-ai crate 骨架 + 错误类型"
```
---
### Task 2: erp-core 扩展 — HealthDataProvider trait + AI 权限码 + 事件类型
**Files:**
- Create: `crates/erp-core/src/health_provider.rs` — trait + DTO 定义
- Modify: `crates/erp-core/src/lib.rs` — 添加 pub mod
- Modify: `crates/erp-health/src/health_provider_impl.rs` — trait 实现 (stub)
- Modify: `crates/erp-health/src/lib.rs` — 添加 pub mod
- Modify: `crates/erp-health/src/module.rs` — permissions() 中声明 AI 权限
> 注意: AI 权限码放在 erp-ai 模块的 permissions() 中，不在 erp-health。此处仅做 erp-core 的 trait 扩展。
- [ ] **Step 1: 创建 HealthDataProvider trait + DTO**
```rust
// crates/erp-core/src/health_provider.rs
use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use uuid::Uuid;
use crate::AppResult;
/// 健康数据提供者 trait，由 erp-health 实现
/// 返回的 DTO 已脱去 PII（姓名、身份证号等），只包含年龄/性别/医疗数据
#[async_trait]
pub trait HealthDataProvider: Send + Sync {
/// 获取化验报告（指标列表）
async fn get_lab_report(
&self,
tenant_id: Uuid,
report_id: Uuid,
) -> AppResult<LabReportDto>;
/// 获取生命体征趋势数据
async fn get_vital_signs(
&self,
tenant_id: Uuid,
patient_id: Uuid,
metrics: &[String],
range: &TimeRange,
) -> AppResult<Vec<VitalSignDto>>;
/// 获取患者摘要（用于个性化方案）
async fn get_patient_summary(
&self,
tenant_id: Uuid,
patient_id: Uuid,
) -> AppResult<PatientSummaryDto>;
/// 获取完整健康报告（用于摘要生成）
async fn get_full_report(
&self,
tenant_id: Uuid,
report_id: Uuid,
) -> AppResult<HealthReportDto>;
}
// === DTO 定义 ===
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TimeRange {
pub start: chrono::DateTime<chrono::Utc>,
pub end: chrono::DateTime<chrono::Utc>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LabReportDto {
pub age_group: String,
pub sex: String,
pub department: String,
pub report_date: String,
pub items: Vec<LabItemDto>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LabItemDto {
pub name: String,
pub value: f64,
pub unit: String,
pub reference_range: String,
pub is_abnormal: bool,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct VitalSignDto {
pub metric: String,
pub values: Vec<(String, f64)>,
pub unit: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PatientSummaryDto {
pub age_group: String,
pub sex: String,
pub chronic_conditions: Vec<String>,
pub medications: Vec<String>,
pub family_history: Vec<String>,
pub last_checkup_date: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HealthReportDto {
pub age_group: String,
pub sex: String,
pub department: String,
pub report_date: String,
pub sections: Vec<ReportSectionDto>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ReportSectionDto {
pub title: String,
pub findings: Vec<String>,
pub abnormal_items: Vec<String>,
}
```
- [ ] **Step 2: 在 erp-core/src/lib.rs 中添加 pub mod**
```rust
pub mod health_provider;
```
并添加 re-export:
```rust
pub use health_provider::{
HealthDataProvider, LabItemDto, LabReportDto, PatientSummaryDto,
HealthReportDto, ReportSectionDto, TimeRange, VitalSignDto,
};
```
- [ ] **Step 3: 验证 erp-core 编译**
```bash
cargo check -p erp-core
```
- [ ] **Step 4: 提交 erp-core 扩展**
```bash
git add crates/erp-core/src/health_provider.rs crates/erp-core/src/lib.rs
git commit -m "feat(core): 新增 HealthDataProvider trait + DTO 定义"
```
---
### Task 3: erp-health 实现 HealthDataProvider (stub)
**Files:**
- Create: `crates/erp-health/src/health_provider_impl.rs`
- Modify: `crates/erp-health/src/lib.rs`
> 注意: Phase 1 先创建 stub 实现（返回 todo! 或空数据），确保编译通过。实际数据查询在 Chunk 5 集成时完善。
- [ ] **Step 1: 创建 stub 实现**
```rust
// crates/erp-health/src/health_provider_impl.rs
use async_trait::async_trait;
use erp_core::{
AppResult, HealthDataProvider, LabReportDto, VitalSignDto,
PatientSummaryDto, HealthReportDto, TimeRange,
};
use uuid::Uuid;
pub struct HealthDataProviderImpl {
pub db: sea_orm::DatabaseConnection,
}
#[async_trait]
impl HealthDataProvider for HealthDataProviderImpl {
async fn get_lab_report(
&self,
_tenant_id: Uuid,
_report_id: Uuid,
) -> AppResult<LabReportDto> {
todo!("Chunk 5: 实现化验报告数据查询")
}
async fn get_vital_signs(
&self,
_tenant_id: Uuid,
_patient_id: Uuid,
_metrics: &[String],
_range: &TimeRange,
) -> AppResult<Vec<VitalSignDto>> {
todo!("Chunk 5: 实现生命体征趋势查询")
}
async fn get_patient_summary(
&self,
_tenant_id: Uuid,
_patient_id: Uuid,
) -> AppResult<PatientSummaryDto> {
todo!("Chunk 5: 实现患者摘要查询")
}
async fn get_full_report(
&self,
_tenant_id: Uuid,
_report_id: Uuid,
) -> AppResult<HealthReportDto> {
todo!("Chunk 5: 实现完整报告查询")
}
}
```
- [ ] **Step 2: 在 erp-health/src/lib.rs 添加 pub mod**
```rust
pub mod health_provider_impl;
pub use health_provider_impl::HealthDataProviderImpl;
```
- [ ] **Step 3: 验证全 workspace 编译**
```bash
cargo check --workspace
```
Expected: 编译通过（stub 的 todo! 不影响编译）
- [ ] **Step 4: 提交**
```bash
git add crates/erp-health/src/health_provider_impl.rs crates/erp-health/src/lib.rs
git commit -m "feat(health): 添加 HealthDataProvider stub 实现"
```
---
## Chunk 2: 数据库迁移 + SeaORM Entity
### Task 4: 创建 ai_prompts / ai_analysis_results / ai_usage_logs 迁移
**Files:**
- Create: `crates/erp-server/migration/src/m20260425_000050_create_ai_tables.rs`
- Modify: `crates/erp-server/migration/src/lib.rs` — 注册到 Migrator
> 先查看 `migration/src/lib.rs` 最后一个 migration 编号，确保 `000050` 不冲突。
- [ ] **Step 1: 创建迁移文件**
```rust
// crates/erp-server/migration/src/m20260425_000050_create_ai_tables.rs
use sea_orm_migration::prelude::*;
#[derive(DeriveMigrationName)]
pub struct Migration;
#[async_trait::async_trait]
impl MigrationTrait for Migration {
async fn up(&self, manager: &SchemaManager) -> Result<(), DbErr> {
// 1. ai_prompts — Prompt 模板存储
manager.create_table(
Table::create()
.table(AiPrompt::Table)
.if_not_exists()
.col(ColumnDef::new(AiPrompt::Id).uuid().not_null().primary_key())
.col(ColumnDef::new(AiPrompt::TenantId).uuid().not_null())
.col(ColumnDef::new(AiPrompt::Name).string_len(100).not_null())
.col(ColumnDef::new(AiPrompt::Description).text().not_null().default(""))
.col(ColumnDef::new(AiPrompt::SystemPrompt).text().not_null())
.col(ColumnDef::new(AiPrompt::UserPromptTemplate).text().not_null())
.col(ColumnDef::new(AiPrompt::VariablesSchema).json().null())
.col(ColumnDef::new(AiPrompt::ModelConfig).json().not_null())
.col(ColumnDef::new(AiPrompt::Version).integer().not_null().default(1))
.col(ColumnDef::new(AiPrompt::IsActive).boolean().not_null().default(true))
.col(ColumnDef::new(AiPrompt::Category).string_len(50).not_null().default("analysis"))
.col(ColumnDef::new(AiPrompt::Tags).json().null())
.col(ColumnDef::new(AiPrompt::CreatedAt).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
.col(ColumnDef::new(AiPrompt::UpdatedAt).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
.col(ColumnDef::new(AiPrompt::CreatedBy).uuid().null())
.col(ColumnDef::new(AiPrompt::UpdatedBy).uuid().null())
.col(ColumnDef::new(AiPrompt::DeletedAt).timestamp_with_time_zone().null())
.col(ColumnDef::new(AiPrompt::VersionLock).integer().not_null().default(1))
.to_owned(),
).await?;
// ai_prompts 索引
manager.create_index(
Index::create()
.if_not_exists()
.name("idx_ai_prompts_tenant_name")
.table(AiPrompt::Table)
.col(AiPrompt::TenantId)
.col(AiPrompt::Name)
.col(AiPrompt::IsActive)
.to_owned(),
).await?;
// 2. ai_analysis_results — 分析结果存储
manager.create_table(
Table::create()
.table(AiAnalysis::Table)
.if_not_exists()
.col(ColumnDef::new(AiAnalysis::Id).uuid().not_null().primary_key())
.col(ColumnDef::new(AiAnalysis::TenantId).uuid().not_null())
.col(ColumnDef::new(AiAnalysis::PatientId).uuid().not_null())
.col(ColumnDef::new(AiAnalysis::AnalysisType).string_len(50).not_null())
.col(ColumnDef::new(AiAnalysis::SourceRef).string_len(200).not_null())
.col(ColumnDef::new(AiAnalysis::PromptId).uuid().not_null())
.col(ColumnDef::new(AiAnalysis::PromptVersion).integer().not_null())
.col(ColumnDef::new(AiAnalysis::ModelUsed).string_len(100).not_null())
.col(ColumnDef::new(AiAnalysis::InputDataHash).string_len(64).not_null())
.col(ColumnDef::new(AiAnalysis::SanitizedInput).json().null())
.col(ColumnDef::new(AiAnalysis::ResultContent).text().null())
.col(ColumnDef::new(AiAnalysis::ResultMetadata).json().null())
.col(ColumnDef::new(AiAnalysis::Status).string_len(20).not_null().default("pending"))
.col(ColumnDef::new(AiAnalysis::ErrorMessage).text().null())
.col(ColumnDef::new(AiAnalysis::CreatedAt).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
.col(ColumnDef::new(AiAnalysis::UpdatedAt).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
.col(ColumnDef::new(AiAnalysis::CreatedBy).uuid().null())
.col(ColumnDef::new(AiAnalysis::UpdatedBy).uuid().null())
.col(ColumnDef::new(AiAnalysis::DeletedAt).timestamp_with_time_zone().null())
.col(ColumnDef::new(AiAnalysis::VersionLock).integer().not_null().default(1))
.to_owned(),
).await?;
// ai_analysis 索引
manager.create_index(
Index::create()
.if_not_exists()
.name("idx_ai_analysis_tenant_patient")
.table(AiAnalysis::Table)
.col(AiAnalysis::TenantId)
.col(AiAnalysis::PatientId)
.to_owned(),
).await?;
manager.create_index(
Index::create()
.if_not_exists()
.name("idx_ai_analysis_cache_hash")
.table(AiAnalysis::Table)
.col(AiAnalysis::InputDataHash)
.to_owned(),
).await?;
// 3. ai_usage_logs — AI 调用计量
manager.create_table(
Table::create()
.table(AiUsage::Table)
.if_not_exists()
.col(ColumnDef::new(AiUsage::Id).uuid().not_null().primary_key())
.col(ColumnDef::new(AiUsage::TenantId).uuid().not_null())
.col(ColumnDef::new(AiUsage::Provider).string_len(50).not_null())
.col(ColumnDef::new(AiUsage::Model).string_len(100).not_null())
.col(ColumnDef::new(AiUsage::AnalysisType).string_len(50).not_null())
.col(ColumnDef::new(AiUsage::InputTokens).integer().not_null().default(0))
.col(ColumnDef::new(AiUsage::OutputTokens).integer().not_null().default(0))
.col(ColumnDef::new(AiUsage::DurationMs).integer().not_null().default(0))
.col(ColumnDef::new(AiUsage::CostCents).integer().not_null().default(0))
.col(ColumnDef::new(AiUsage::IsCacheHit).boolean().not_null().default(false))
.col(ColumnDef::new(AiUsage::CreatedAt).timestamp_with_time_zone().not_null().default(Expr::current_timestamp()))
.to_owned(),
).await?;
// ai_usage 索引
manager.create_index(
Index::create()
.if_not_exists()
.name("idx_ai_usage_tenant_created")
.table(AiUsage::Table)
.col(AiUsage::TenantId)
.col(AiUsage::CreatedAt)
.to_owned(),
).await?;
Ok(())
}
async fn down(&self, manager: &SchemaManager) -> Result<(), DbErr> {
manager.drop_table(Table::drop().table(AiUsage::Table).to_owned()).await?;
manager.drop_table(Table::drop().table(AiAnalysis::Table).to_owned()).await?;
manager.drop_table(Table::drop().table(AiPrompt::Table).to_owned()).await?;
Ok(())
}
}
#[derive(DeriveIden)]
enum AiPrompt {
Table,
Id,
TenantId,
Name,
Description,
SystemPrompt,
UserPromptTemplate,
VariablesSchema,
ModelConfig,
Version,
IsActive,
Category,
Tags,
CreatedAt,
UpdatedAt,
CreatedBy,
UpdatedBy,
DeletedAt,
VersionLock,
}
#[derive(DeriveIden)]
enum AiAnalysis {
Table,
Id,
TenantId,
PatientId,
AnalysisType,
SourceRef,
PromptId,
PromptVersion,
ModelUsed,
InputDataHash,
SanitizedInput,
ResultContent,
ResultMetadata,
Status,
ErrorMessage,
CreatedAt,
UpdatedAt,
CreatedBy,
UpdatedBy,
DeletedAt,
VersionLock,
}
#[derive(DeriveIden)]
enum AiUsage {
Table,
Id,
TenantId,
Provider,
Model,
AnalysisType,
InputTokens,
OutputTokens,
DurationMs,
CostCents,
IsCacheHit,
CreatedAt,
}
```
- [ ] **Step 2: 在 migration/src/lib.rs 注册**
在 `Migrator::migrations()` 的 `vec![]` 末尾添加:
```rust
Box::new(m20260425_000050_create_ai_tables::Migration),
```
- [ ] **Step 3: 验证迁移编译**
```bash
cargo check -p erp-server
```
- [ ] **Step 4: 提交**
```bash
git add crates/erp-server/migration/src/
git commit -m "feat(db): 添加 ai_prompts / ai_analysis_results / ai_usage_logs 迁移"
```
---
### Task 5: 创建 SeaORM Entity
**Files:**
- Create: `crates/erp-ai/src/entity/mod.rs`
- Create: `crates/erp-ai/src/entity/ai_prompt.rs`
- Create: `crates/erp-ai/src/entity/ai_analysis.rs`
- Create: `crates/erp-ai/src/entity/ai_usage.rs`
- [ ] **Step 1: 创建 entity/mod.rs**
```rust
// crates/erp-ai/src/entity/mod.rs
pub mod ai_analysis;
pub mod ai_prompt;
pub mod ai_usage;
```
- [ ] **Step 2: 创建 ai_prompt.rs**
```rust
// crates/erp-ai/src/entity/ai_prompt.rs
use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
#[sea_orm(table_name = "ai_prompts")]
pub struct Model {
#[sea_orm(primary_key, auto_increment = false)]
pub id: Uuid,
pub tenant_id: Uuid,
pub name: String,
pub description: String,
pub system_prompt: String,
pub user_prompt_template: String,
pub variables_schema: Option<serde_json::Value>,
pub model_config: serde_json::Value,
pub version: i32,
pub is_active: bool,
pub category: String,
pub tags: Option<serde_json::Value>,
pub created_at: DateTimeUtc,
pub updated_at: DateTimeUtc,
pub created_by: Option<Uuid>,
pub updated_by: Option<Uuid>,
pub deleted_at: Option<DateTimeUtc>,
pub version_lock: i32,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {}
impl ActiveModelBehavior for ActiveModel {}
```
- [ ] **Step 3: 创建 ai_analysis.rs**
```rust
// crates/erp-ai/src/entity/ai_analysis.rs
use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
#[sea_orm(table_name = "ai_analysis_results")]
pub struct Model {
#[sea_orm(primary_key, auto_increment = false)]
pub id: Uuid,
pub tenant_id: Uuid,
pub patient_id: Uuid,
pub analysis_type: String,
pub source_ref: String,
pub prompt_id: Uuid,
pub prompt_version: i32,
pub model_used: String,
pub input_data_hash: String,
pub sanitized_input: Option<serde_json::Value>,
pub result_content: Option<String>,
pub result_metadata: Option<serde_json::Value>,
pub status: String,
pub error_message: Option<String>,
pub created_at: DateTimeUtc,
pub updated_at: DateTimeUtc,
pub created_by: Option<Uuid>,
pub updated_by: Option<Uuid>,
pub deleted_at: Option<DateTimeUtc>,
pub version_lock: i32,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {}
impl ActiveModelBehavior for ActiveModel {}
```
- [ ] **Step 4: 创建 ai_usage.rs**
```rust
// crates/erp-ai/src/entity/ai_usage.rs
use sea_orm::entity::prelude::*;
use serde::{Deserialize, Serialize};
#[derive(Clone, Debug, PartialEq, DeriveEntityModel, Serialize, Deserialize)]
#[sea_orm(table_name = "ai_usage_logs")]
pub struct Model {
#[sea_orm(primary_key, auto_increment = false)]
pub id: Uuid,
pub tenant_id: Uuid,
pub provider: String,
pub model: String,
pub analysis_type: String,
pub input_tokens: i32,
pub output_tokens: i32,
pub duration_ms: i32,
pub cost_cents: i32,
pub is_cache_hit: bool,
pub created_at: DateTimeUtc,
}
#[derive(Copy, Clone, Debug, EnumIter, DeriveRelation)]
pub enum Relation {}
impl ActiveModelBehavior for ActiveModel {}
```
- [ ] **Step 5: 更新 lib.rs 添加 entity mod**
在 `crates/erp-ai/src/lib.rs` 中添加:
```rust
pub mod entity;
```
- [ ] **Step 6: 验证编译**
```bash
cargo check -p erp-ai
```
- [ ] **Step 7: 提交**
```bash
git add crates/erp-ai/src/entity/ crates/erp-ai/src/lib.rs
git commit -m "feat(ai): 添加 SeaORM Entity (ai_prompt/ai_analysis/ai_usage)"
```
---
## Chunk 3: AI Provider 抽象 + Claude SSE + 数据脱敏
### Task 6: AiProvider trait + Claude 实现
**Files:**
- Create: `crates/erp-ai/src/provider/mod.rs`
- Create: `crates/erp-ai/src/provider/claude.rs`
- Create: `crates/erp-ai/src/dto.rs`
- [ ] **Step 1: 创建 DTO 定义**
```rust
// crates/erp-ai/src/dto.rs
use serde::{Deserialize, Serialize};
// === 分析请求 ===
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AnalyzeRequest {
pub analysis_type: AnalysisType,
pub source_ref: String, // report_id 或 patient_id + metrics
pub options: AnalyzeOptions,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
#[serde(rename_all = "snake_case")]
pub enum AnalysisType {
LabReport,
Trends,
CheckupPlan,
ReportSummary,
}
impl AnalysisType {
pub fn as_str(&self) -> &str {
match self {
Self::LabReport => "lab_report",
Self::Trends => "trend",
Self::CheckupPlan => "checkup_plan",
Self::ReportSummary => "report_summary",
}
}
pub fn prompt_name(&self) -> &str {
match self {
Self::LabReport => "lab_report_interpretation",
Self::Trends => "health_trend_analysis",
Self::CheckupPlan => "personalized_checkup_plan",
Self::ReportSummary => "report_summary_generation",
}
}
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct AnalyzeOptions {
pub detail_level: Option<String>, // patient_friendly | professional
pub language: Option<String>, // zh-CN
}
impl Default for AnalyzeOptions {
fn default() -> Self {
Self {
detail_level: Some("patient_friendly".into()),
language: Some("zh-CN".into()),
}
}
}
// === AI Provider 请求/响应 ===
#[derive(Debug, Clone)]
pub struct GenerateRequest {
pub system_prompt: String,
pub user_prompt: String,
pub model: String,
pub temperature: f32,
pub max_tokens: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct GenerateResponse {
pub content: String,
pub model: String,
pub input_tokens: u32,
pub output_tokens: u32,
pub duration_ms: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamChunk {
pub content: String,
pub index: u32,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamDone {
pub analysis_id: uuid::Uuid,
pub status: String,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamMetadata {
pub model: String,
pub tokens: TokenUsage,
pub duration_ms: u64,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct TokenUsage {
pub input: u32,
pub output: u32,
}
// === SSE 事件 ===
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum AnalysisSseEvent {
#[serde(rename = "chunk")]
Chunk { content: String, index: u32 },
#[serde(rename = "metadata")]
Metadata { model: String, tokens: TokenUsage, duration_ms: u64 },
#[serde(rename = "done")]
Done { analysis_id: uuid::Uuid, status: String },
#[serde(rename = "error")]
Error { message: String },
}
```
- [ ] **Step 2: 创建 provider/mod.rs**
```rust
// crates/erp-ai/src/provider/mod.rs
pub mod claude;
use async_trait::async_trait;
use futures::Stream;
use pin_project_lite::pin_project;
use std::pin::Pin;
use crate::dto::GenerateRequest;
use crate::error::AiResult;
/// AI 提供商 trait
#[async_trait]
pub trait AiProvider: Send + Sync {
/// 流式生成 — 返回 Pin<Box<dyn Stream>> 避免 async_trait + impl Trait 不兼容
async fn stream_generate(
&self,
req: GenerateRequest,
) -> AiResult<Pin<Box<dyn Stream<Item = AiResult<String>> + Send>>>;
/// 非流式生成
async fn generate(&self, req: GenerateRequest) -> AiResult<crate::dto::GenerateResponse>;
/// 提供商名称
fn name(&self) -> &str;
/// 健康检查
async fn health_check(&self) -> AiResult<bool>;
}
```
> 注意: 需要在 Cargo.toml 中添加 `pin-project-lite` 或使用 `futures` 的 `pin_mut!` 宏。检查 `futures` 是否已导出 `Stream` trait。
- [ ] **Step 3: 创建 provider/claude.rs (Claude API SSE 实现)**
```rust
// crates/erp-ai/src/provider/claude.rs
use async_trait::async_trait;
use async_stream::stream;
use futures::{Stream, StreamExt};
use reqwest::Client;
use serde::{Deserialize, Serialize};
use std::pin::Pin;
use super::AiProvider;
use crate::dto::GenerateRequest;
use crate::error::{AiError, AiResult};
#[derive(Debug, Clone)]
pub struct ClaudeProvider {
client: Client,
api_key: String,
base_url: String,
}
impl ClaudeProvider {
pub fn new(api_key: String) -> Self {
Self {
client: Client::new(),
api_key,
base_url: "https://api.anthropic.com".into(),
}
}
pub fn with_base_url(mut self, url: String) -> Self {
self.base_url = url;
self
}
}
// Claude API 请求/响应结构
#[derive(Serialize)]
struct ClaudeRequest {
model: String,
max_tokens: u32,
temperature: f32,
system: String,
messages: Vec<ClaudeMessage>,
stream: bool,
}
#[derive(Serialize)]
struct ClaudeMessage {
role: String,
content: String,
}
#[derive(Deserialize)]
struct ClaudeStreamEvent {
#[serde(rename = "type")]
event_type: String,
delta: Option<ClaudeDelta>,
message: Option<ClaudeMessageResp>,
}
#[derive(Deserialize)]
struct ClaudeDelta {
text: Option<String>,
}
#[derive(Deserialize)]
struct ClaudeMessageResp {
usage: Option<ClaudeUsage>,
}
#[derive(Deserialize)]
struct ClaudeUsage {
input_tokens: u32,
output_tokens: u32,
}
#[async_trait]
impl AiProvider for ClaudeProvider {
async fn stream_generate(
&self,
req: GenerateRequest,
) -> AiResult<Pin<Box<dyn Stream<Item = AiResult<String>> + Send>>> {
let claude_req = ClaudeRequest {
model: req.model,
max_tokens: req.max_tokens,
temperature: req.temperature,
system: req.system_prompt,
messages: vec![ClaudeMessage {
role: "user".into(),
content: req.user_prompt,
}],
stream: true,
};
let response = self
.client
.post(format!("{}/v1/messages", self.base_url))
.header("x-api-key", &self.api_key)
.header("anthropic-version", "2023-06-01")
.header("content-type", "application/json")
.json(&claude_req)
.send()
.await
.map_err(|e| AiError::ProviderError(format!("Claude API 请求失败: {e}")))?;
if !response.status().is_success() {
let status = response.status();
let body = response.text().await.unwrap_or_default();
return Err(AiError::ProviderError(format!("Claude API 错误 {status}: {body}")));
}
let stream = Box::pin(stream! {
let mut stream = response.bytes_stream();
while let Some(chunk_result) = stream.next().await {
let bytes = match chunk_result {
Ok(b) => b,
Err(e) => {
yield Err(AiError::ProviderError(format!("流读取错误: {e}")));
break;
}
};
let text = String::from_utf8_lossy(&bytes);
for line in text.lines() {
if let Some(data) = line.strip_prefix("data: ") {
if data == "[DONE]" {
return;
}
if let Ok(event) = serde_json::from_str::<ClaudeStreamEvent>(data) {
if event.event_type == "content_block_delta" {
if let Some(delta) = event.delta {
if let Some(text) = delta.text {
yield Ok(text);
}
}
}
}
}
}
}
});
Ok(stream)
}
async fn generate(&self, req: GenerateRequest) -> AiResult<crate::dto::GenerateResponse> {
let start = std::time::Instant::now();
let claude_req = ClaudeRequest {
model: req.model.clone(),
max_tokens: req.max_tokens,
temperature: req.temperature,
system: req.system_prompt,
messages: vec![ClaudeMessage {
role: "user".into(),
content: req.user_prompt,
}],
stream: false,
};
let resp = self
.client
.post(format!("{}/v1/messages", self.base_url))
.header("x-api-key", &self.api_key)
.header("anthropic-version", "2023-06-01")
.header("content-type", "application/json")
.json(&claude_req)
.send()
.await
.map_err(|e| AiError::ProviderError(e.to_string()))?;
let status = resp.status();
let body = resp.text().await.map_err(|e| AiError::ProviderError(e.to_string()))?;
if !status.is_success() {
return Err(AiError::ProviderError(format!("Claude {status}: {body}")));
}
let start_time = start;
// 解析非流式响应
let parsed: serde_json::Value = serde_json::from_str(&body)
.map_err(|e| AiError::ProviderError(format!("解析响应失败: {e}")))?;
let content = parsed["content"][0]["text"]
.as_str()
.unwrap_or("")
.to_string();
let input_tokens = parsed["usage"]["input_tokens"].as_u64().unwrap_or(0) as u32;
let output_tokens = parsed["usage"]["output_tokens"].as_u64().unwrap_or(0) as u32;
Ok(crate::dto::GenerateResponse {
content,
model: req.model,
input_tokens,
output_tokens,
duration_ms: start_time.elapsed().as_millis() as u64,
})
}
fn name(&self) -> &str {
"claude"
}
async fn health_check(&self) -> AiResult<bool> {
// 简单检查: 发一个最小请求验证 API key 有效
let resp = self
.client
.post(format!("{}/v1/messages", self.base_url))
.header("x-api-key", &self.api_key)
.header("anthropic-version", "2023-06-01")
.header("content-type", "application/json")
.json(&serde_json::json!({
"model": "claude-sonnet-4-6",
"max_tokens": 1,
"messages": [{"role": "user", "content": "hi"}]
}))
.send()
.await;
match resp {
Ok(r) => Ok(r.status().is_success() || r.status().as_u16() == 400),
Err(_) => Ok(false),
}
}
}
```
- [ ] **Step 4: 更新 lib.rs 添加新模块**
```rust
// crates/erp-ai/src/lib.rs
pub mod dto;
pub mod entity;
pub mod error;
pub mod provider;
pub use error::{AiError, AiResult};
```
- [ ] **Step 5: 验证编译**
```bash
cargo check -p erp-ai
```
> 如果 `pin-project-lite` 缺失，在 Cargo.toml 中添加，或直接使用 `std::pin::Pin<Box<...>>` 无需 pin-project。
- [ ] **Step 6: 提交**
```bash
git add crates/erp-ai/src/
git commit -m "feat(ai): AiProvider trait + Claude SSE 流式实现 + DTO 定义"
```
---
### Task 7: 数据脱敏服务 + Prompt 模板引擎
**Files:**
- Create: `crates/erp-ai/src/sanitization/mod.rs`
- Create: `crates/erp-ai/src/prompt/mod.rs`
- [ ] **Step 1: 创建 sanitization/mod.rs**
```rust
// crates/erp-ai/src/sanitization/mod.rs
use erp_core::{
HealthReportDto, LabReportDto, PatientSummaryDto, VitalSignDto,
};
use serde_json::Value;
use crate::error::{AiError, AiResult};
/// 数据脱敏服务 — 确保发送给 AI 的数据不含 PII
/// HealthDataProvider 返回的 DTO 已经是脱敏的（只有年龄/性别/医疗数据）
/// 此服务做二次检查和安全约束注入
pub struct SanitizationService;
impl SanitizationService {
pub fn new() -> Self {
Self
}
/// 验证 DTO 中不包含意外泄漏的 PII 字段
/// 并生成安全的 JSON 数据供 Prompt 模板使用
pub fn sanitize_lab_report(&self, report: &LabReportDto) -> AiResult<Value> {
// LabReportDto 已由 HealthDataProvider 脱敏
// 此处做二次验证 + 转换为模板友好的 JSON
let sanitized = serde_json::to_value(report)
.map_err(|e| AiError::SanitizationError(format!("序列化失败: {e}")))?;
self.verify_no_pii(&sanitized)?;
Ok(sanitized)
}
pub fn sanitize_vital_signs(&self, signs: &[VitalSignDto]) -> AiResult<Value> {
let sanitized = serde_json::to_value(signs)
.map_err(|e| AiError::SanitizationError(format!("序列化失败: {e}")))?;
self.verify_no_pii(&sanitized)?;
Ok(sanitized)
}
pub fn sanitize_patient_summary(&self, summary: &PatientSummaryDto) -> AiResult<Value> {
let sanitized = serde_json::to_value(summary)
.map_err(|e| AiError::SanitizationError(format!("序列化失败: {e}")))?;
self.verify_no_pii(&sanitized)?;
Ok(sanitized)
}
pub fn sanitize_health_report(&self, report: &HealthReportDto) -> AiResult<Value> {
let sanitized = serde_json::to_value(report)
.map_err(|e| AiError::SanitizationError(format!("序列化失败: {e}")))?;
self.verify_no_pii(&sanitized)?;
Ok(sanitized)
}
/// 二次验证: 确保没有意外泄漏的 PII
fn verify_no_pii(&self, value: &Value) -> AiResult<()> {
let pii_keys = ["name", "phone", "id_number", "address", "birth_date", "email"];
if let Value::Object(map) = value {
for key in pii_keys {
if map.contains_key(key) {
return Err(AiError::SanitizationError(
format!("检测到疑似 PII 字段: {key}"),
));
}
}
}
Ok(())
}
}
```
- [ ] **Step 2: 创建 prompt/mod.rs**
```rust
// crates/erp-ai/src/prompt/mod.rs
use handlebars::Handlebars;
use serde_json::Value;
use crate::error::{AiError, AiResult};
/// Prompt 模板渲染引擎
pub struct PromptRenderer {
registry: Handlebars<'static>,
}
impl PromptRenderer {
pub fn new() -> Self {
let mut registry = Handlebars::new();
registry.set_strict_mode(true);
Self { registry }
}
/// 渲染 Prompt 模板 — 使用 Handlebars {{variable}} 语法
/// JSON 序列化注入，不做字符串拼接，防止 Prompt 注入
pub fn render(&self, template: &str, data: &Value) -> AiResult<String> {
self.registry
.render_template(template, data)
.map_err(|e| AiError::TemplateError(format!("模板渲染失败: {e}")))
}
}
```
- [ ] **Step 3: 更新 lib.rs**
```rust
pub mod prompt;
pub mod sanitization;
```
- [ ] **Step 4: 验证编译**
```bash
cargo check -p erp-ai
```
- [ ] **Step 5: 提交**
```bash
git add crates/erp-ai/src/
git commit -m "feat(ai): 数据脱敏服务 + Prompt 模板渲染引擎"
```
---
## Chunk 4: Service 层 — AnalysisService 核心编排
### Task 8: AnalysisService — 分析管道核心逻辑
**Files:**
- Create: `crates/erp-ai/src/service/mod.rs`
- Create: `crates/erp-ai/src/service/analysis.rs`
- Create: `crates/erp-ai/src/service/prompt.rs`
- Create: `crates/erp-ai/src/service/usage.rs`
- [ ] **Step 1: 创建 service/mod.rs**
```rust
// crates/erp-ai/src/service/mod.rs
pub mod analysis;
pub mod prompt;
pub mod usage;
```
- [ ] **Step 2: 创建 service/prompt.rs — Prompt CRUD**
```rust
// crates/erp-ai/src/service/prompt.rs
use sea_orm::{ActiveModelTrait, ColumnTrait, EntityTrait, QueryFilter, Set};
use uuid::Uuid;
use crate::entity::ai_prompt;
use crate::error::{AiError, AiResult};
pub struct PromptService {
pub db: sea_orm::DatabaseConnection,
}
impl PromptService {
pub fn new(db: sea_orm::DatabaseConnection) -> Self {
Self { db }
}
/// 获取当前激活的 Prompt 模板
pub async fn get_active_prompt(
&self,
tenant_id: Uuid,
name: &str,
) -> AiResult<ai_prompt::Model> {
ai_prompt::Entity::find()
.filter(ai_prompt::Column::TenantId.eq(tenant_id))
.filter(ai_prompt::Column::Name.eq(name))
.filter(ai_prompt::Column::IsActive.eq(true))
.filter(ai_prompt::Column::DeletedAt.is_null())
.one(&self.db)
.await?
.ok_or_else(|| AiError::PromptNotFound(name.into()))
}
/// 新建 Prompt 版本
pub async fn create_prompt(
&self,
tenant_id: Uuid,
user_id: Uuid,
name: String,
system_prompt: String,
user_prompt_template: String,
model_config: serde_json::Value,
category: String,
) -> AiResult<ai_prompt::Model> {
let id = Uuid::now_v7();
let now = chrono::Utc::now();
let active = ai_prompt::ActiveModel {
id: Set(id),
tenant_id: Set(tenant_id),
name: Set(name),
description: Set(String::new()),
system_prompt: Set(system_prompt),
user_prompt_template: Set(user_prompt_template),
variables_schema: Set(None),
model_config: Set(model_config),
version: Set(1),
is_active: Set(true),
category: Set(category),
tags: Set(None),
created_at: Set(now),
updated_at: Set(now),
created_by: Set(Some(user_id)),
updated_by: Set(Some(user_id)),
deleted_at: Set(None),
version_lock: Set(1),
};
Ok(active.insert(&self.db).await?)
}
}
```
- [ ] **Step 3: 创建 service/usage.rs — 用量记录**
```rust
// crates/erp-ai/src/service/usage.rs
use sea_orm::ActiveModelTrait;
use uuid::Uuid;
use crate::entity::ai_usage;
use crate::error::AiResult;
pub struct UsageService {
pub db: sea_orm::DatabaseConnection,
}
impl UsageService {
pub fn new(db: sea_orm::DatabaseConnection) -> Self {
Self { db }
}
pub async fn log_usage(
&self,
tenant_id: Uuid,
provider: &str,
model: &str,
analysis_type: &str,
input_tokens: u32,
output_tokens: u32,
duration_ms: u64,
cost_cents: i32,
is_cache_hit: bool,
) -> AiResult<ai_usage::Model> {
let id = Uuid::now_v7();
let active = ai_usage::ActiveModel {
id: Set(id),
tenant_id: Set(tenant_id),
provider: Set(provider.into()),
model: Set(model.into()),
analysis_type: Set(analysis_type.into()),
input_tokens: Set(input_tokens as i32),
output_tokens: Set(output_tokens as i32),
duration_ms: Set(duration_ms as i32),
cost_cents: Set(cost_cents),
is_cache_hit: Set(is_cache_hit),
created_at: Set(chrono::Utc::now()),
};
Ok(active.insert(&self.db).await?)
}
}
```
> `Set` 需要 `use sea_orm::Set;`
- [ ] **Step 4: 创建 service/analysis.rs — 核心编排**
```rust
// crates/erp-ai/src/service/analysis.rs
use async_stream::stream;
use futures::{Stream, StreamExt};
use sha2::{Digest, Sha256};
use std::pin::Pin;
use uuid::Uuid;
use crate::dto::{AnalysisType, GenerateRequest, StreamChunk};
use crate::entity::ai_analysis;
use crate::error::{AiError, AiResult};
use crate::prompt::PromptRenderer;
use crate::provider::AiProvider;
use crate::sanitization::SanitizationService;
pub struct AnalysisService {
pub provider: Box<dyn AiProvider>,
pub sanitizer: SanitizationService,
pub renderer: PromptRenderer,
pub db: sea_orm::DatabaseConnection,
}
impl AnalysisService {
pub fn new(
provider: Box<dyn AiProvider>,
db: sea_orm::DatabaseConnection,
) -> Self {
Self {
provider,
sanitizer: SanitizationService::new(),
renderer: PromptRenderer::new(),
db,
}
}
/// 执行流式分析 — 返回 SSE 事件流
pub async fn stream_analyze(
&self,
tenant_id: Uuid,
user_id: Uuid,
patient_id: Uuid,
analysis_type: AnalysisType,
source_ref: String,
system_prompt: String,
user_template: String,
sanitized_data: serde_json::Value,
model: String,
temperature: f32,
max_tokens: u32,
) -> AiResult<(
Pin<Box<dyn Stream<Item = AiResult<String>> + Send>>,
uuid::Uuid,
String,
)> {
let analysis_id = Uuid::now_v7();
let input_hash = self.compute_hash(&sanitized_data);
let provider_name = self.provider.name().to_string();
// 1. 渲染 Prompt
let user_prompt = self.renderer.render(&user_template, &sanitized_data)?;
// 2. 创建分析记录
self.create_analysis_record(
analysis_id, tenant_id, patient_id,
analysis_type.as_str(), &source_ref,
&input_hash, &provider_name, &model,
).await?;
// 3. 调用 AI 流式生成
let req = GenerateRequest {
system_prompt,
user_prompt,
model,
temperature,
max_tokens,
};
let stream = self.provider.stream_generate(req).await?;
Ok((stream, analysis_id, provider_name))
}
/// 更新分析记录为完成
pub async fn complete_analysis(
&self,
analysis_id: Uuid,
content: String,
metadata: serde_json::Value,
) -> AiResult<()> {
let entity: Option<ai_analysis::Model> = ai_analysis::Entity::find_by_id(analysis_id)
.one(&self.db)
.await?
.ok_or_else(|| AiError::AnalysisNotFound(analysis_id.to_string()))?;
let mut active: ai_analysis::ActiveModel = entity.into();
active.status = sea_orm::Set("completed".into());
active.result_content = sea_orm::Set(Some(content));
active.result_metadata = sea_orm::Set(Some(metadata));
active.updated_at = sea_orm::Set(chrono::Utc::now());
active.update(&self.db).await?;
Ok(())
}
/// 标记分析失败
pub async fn fail_analysis(&self, analysis_id: Uuid, error: String) -> AiResult<()> {
let entity: Option<ai_analysis::Model> = ai_analysis::Entity::find_by_id(analysis_id)
.one(&self.db)
.await?
.ok_or_else(|| AiError::AnalysisNotFound(analysis_id.to_string()))?;
let mut active: ai_analysis::ActiveModel = entity.into();
active.status = sea_orm::Set("failed".into());
active.error_message = sea_orm::Set(Some(error));
active.updated_at = sea_orm::Set(chrono::Utc::now());
active.update(&self.db).await?;
Ok(())
}
/// 查找缓存
pub async fn find_cached(
&self,
tenant_id: Uuid,
input_hash: &str,
prompt_version: i32,
) -> AiResult<Option<ai_analysis::Model>> {
let result = ai_analysis::Entity::find()
.filter(ai_analysis::Column::TenantId.eq(tenant_id))
.filter(ai_analysis::Column::InputDataHash.eq(input_hash))
.filter(ai_analysis::Column::PromptVersion.eq(prompt_version))
.filter(ai_analysis::Column::Status.eq("completed"))
.filter(ai_analysis::Column::DeletedAt.is_null())
.one(&self.db)
.await?;
Ok(result)
}
fn compute_hash(&self, data: &serde_json::Value) -> String {
let canonical = serde_json::to_string(data).unwrap_or_default();
let mut hasher = Sha256::new();
hasher.update(canonical.as_bytes());
hex::encode(hasher.finalize())
}
async fn create_analysis_record(
&self,
id: Uuid,
tenant_id: Uuid,
patient_id: Uuid,
analysis_type: &str,
source_ref: &str,
input_hash: &str,
provider: &str,
model: &str,
) -> AiResult<()> {
let now = chrono::Utc::now();
let active = ai_analysis::ActiveModel {
id: sea_orm::Set(id),
tenant_id: sea_orm::Set(tenant_id),
patient_id: sea_orm::Set(patient_id),
analysis_type: sea_orm::Set(analysis_type.into()),
source_ref: sea_orm::Set(source_ref.into()),
prompt_id: sea_orm::Set(Uuid::nil()), // Phase 1 填充
prompt_version: sea_orm::Set(1),
model_used: sea_orm::Set(model.into()),
input_data_hash: sea_orm::Set(input_hash.into()),
sanitized_input: sea_orm::Set(None),
result_content: sea_orm::Set(None),
result_metadata: sea_orm::Set(None),
status: sea_orm::Set("streaming".into()),
error_message: sea_orm::Set(None),
created_at: sea_orm::Set(now),
updated_at: sea_orm::Set(now),
created_by: sea_orm::Set(None),
updated_by: sea_orm::Set(None),
deleted_at: sea_orm::Set(None),
version_lock: sea_orm::Set(1),
};
active.insert(&self.db).await?;
Ok(())
}
}
```
> 注意: 需要添加 `use sea_orm::{ActiveModelTrait, EntityTrait, QueryFilter, Set, ColumnTrait};`
- [ ] **Step 5: 更新 lib.rs**
```rust
pub mod service;
```
- [ ] **Step 6: 验证编译**
```bash
cargo check -p erp-ai
```
- [ ] **Step 7: 提交**
```bash
git add crates/erp-ai/src/
git commit -m "feat(ai): AnalysisService 核心编排 + PromptService + UsageService"
```
---
## Chunk 5: Handler + State + Module + erp-server 集成
### Task 9: State + Module 定义
**Files:**
- Create: `crates/erp-ai/src/state.rs`
- Create: `crates/erp-ai/src/module.rs`
- [ ] **Step 1: 创建 state.rs**
```rust
// crates/erp-ai/src/state.rs
use std::sync::Arc;
use erp_core::EventBus;
use sea_orm::DatabaseConnection;
use crate::provider::AiProvider;
use crate::service::analysis::AnalysisService;
use crate::service::prompt::PromptService;
use crate::service::usage::UsageService;
#[derive(Clone)]
pub struct AiState {
pub db: DatabaseConnection,
pub event_bus: EventBus,
pub analysis: Arc<AnalysisService>,
pub prompt: Arc<PromptService>,
pub usage: Arc<UsageService>,
}
```
- [ ] **Step 2: 创建 module.rs — ErpModule + 路由注册**
```rust
// crates/erp-ai/src/module.rs
use async_trait::async_trait;
use axum::Router;
use erp_core::module::{ErpModule, ModuleContext, ModuleType, PermissionDescriptor};
use erp_core::AppResult;
use std::any::Any;
use uuid::Uuid;
use crate::handler;
pub struct AiModule;
#[async_trait]
impl ErpModule for AiModule {
fn name(&self) -> &str { "ai" }
fn module_type(&self) -> ModuleType { ModuleType::Builtin }
fn dependencies(&self) -> Vec<&str> { vec!["health"] }
async fn on_startup(&self, _ctx: &ModuleContext) -> AppResult<()> {
Ok(())
}
fn permissions(&self) -> Vec<PermissionDescriptor> {
vec![
PermissionDescriptor {
code: "ai.analysis.list".into(),
name: "查看分析历史".into(),
description: "查看 AI 分析结果历史记录".into(),
module: "ai".into(),
},
PermissionDescriptor {
code: "ai.analysis.manage".into(),
name: "请求分析".into(),
description: "发起 AI 分析请求".into(),
module: "ai".into(),
},
PermissionDescriptor {
code: "ai.prompt.list".into(),
name: "查看 Prompt".into(),
description: "查看 AI Prompt 模板列表".into(),
module: "ai".into(),
},
PermissionDescriptor {
code: "ai.prompt.manage".into(),
name: "管理 Prompt".into(),
description: "创建/编辑/激活/回滚 Prompt 模板".into(),
module: "ai".into(),
},
PermissionDescriptor {
code: "ai.usage.list".into(),
name: "查看用量".into(),
description: "查看 AI 用量统计".into(),
module: "ai".into(),
},
PermissionDescriptor {
code: "ai.provider.manage".into(),
name: "管理提供商".into(),
description: "管理 AI 提供商配置".into(),
module: "ai".into(),
},
]
}
fn as_any(&self) -> &dyn Any { self }
}
impl AiModule {
pub fn public_routes<S>() -> Router<S>
where
crate::state::AiState: axum::extract::FromRef<S>,
S: Clone + Send + Sync + 'static,
{
Router::new()
}
pub fn protected_routes<S>() -> Router<S>
where
crate::state::AiState: axum::extract::FromRef<S>,
S: Clone + Send + Sync + 'static,
{
Router::new()
.route("/ai/analyze/lab-report", axum::routing::post(handler::stream_lab_report::<S>))
.route("/ai/analyze/trends", axum::routing::post(handler::stream_trends::<S>))
.route("/ai/analyze/checkup-plan", axum::routing::post(handler::stream_checkup_plan::<S>))
.route("/ai/analyze/report-summary", axum::routing::post(handler::stream_report_summary::<S>))
.route("/ai/analysis/history", axum::routing::get(handler::list_analysis::<S>))
.route("/ai/analysis/{id}", axum::routing::get(handler::get_analysis::<S>))
}
}
```
- [ ] **Step 3: 更新 lib.rs 添加 re-export**
```rust
pub use module::AiModule;
pub use state::AiState;
```
以及 `pub mod module; pub mod state;`
- [ ] **Step 4: 验证编译 (handler 还没写，先创建 stub)**
创建 `crates/erp-ai/src/handler/mod.rs`:
```rust
// crates/erp-ai/src/handler/mod.rs — stub，Chunk 6 完善
use axum::response::IntoResponse;
pub async fn stream_lab_report<S>() -> impl IntoResponse { "stub" }
pub async fn stream_trends<S>() -> impl IntoResponse { "stub" }
pub async fn stream_checkup_plan<S>() -> impl IntoResponse { "stub" }
pub async fn stream_report_summary<S>() -> impl IntoResponse { "stub" }
pub async fn list_analysis<S>() -> impl IntoResponse { "stub" }
pub async fn get_analysis<S>() -> impl IntoResponse { "stub" }
```
```bash
cargo check -p erp-ai
```
- [ ] **Step 5: 提交**
```bash
git add crates/erp-ai/src/
git commit -m "feat(ai): AiState + AiModule (ErpModule impl + 权限 + 路由骨架)"
```
---
### Task 10: erp-server 集成 — Config + State + 路由注册
**Files:**
- Modify: `crates/erp-server/src/config.rs` — 添加 AiConfig
- Modify: `crates/erp-server/src/state.rs` — 添加 FromRef<AiState>
- Modify: `crates/erp-server/src/main.rs` — 注册模块 + 合并路由
- Modify: `crates/erp-server/Cargo.toml` — 添加 erp-ai 依赖
- Modify: `crates/erp-server/config/default.toml` — 添加 [ai] 段
- [ ] **Step 1: erp-server/Cargo.toml 添加 erp-ai**
```toml
erp-ai.workspace = true
```
- [ ] **Step 2: config.rs 添加 AiConfig**
```rust
// 在 AppConfig 结构体中添加:
pub ai: AiConfig,
// 新增结构体:
#[derive(Debug, Clone, Deserialize)]
pub struct AiConfig {
pub default_provider: String,
pub api_key: String,
pub base_url: Option<String>,
pub model: String,
pub max_tokens: u32,
pub temperature: f32,
pub cache_ttl_seconds: u64,
pub rate_limit_patient_daily: u32,
}
```
- [ ] **Step 3: config/default.toml 添加 [ai] 段**
```toml
[ai]
default_provider = "claude"
api_key = ""
base_url = "https://api.anthropic.com"
model = "claude-sonnet-4-6"
max_tokens = 2048
temperature = 0.3
cache_ttl_seconds = 604800
rate_limit_patient_daily = 10
```
- [ ] **Step 4: state.rs 添加 FromRef**
```rust
impl FromRef<AppState> for erp_ai::AiState {
fn from_ref(state: &AppState) -> Self {
// 从 config 构建 ClaudeProvider
let provider = erp_ai::provider::claude::ClaudeProvider::new(
state.config.ai.api_key.clone(),
);
let db = state.db.clone();
let event_bus = state.event_bus.clone();
let analysis = std::sync::Arc::new(
erp_ai::service::analysis::AnalysisService::new(
Box::new(provider), db.clone(),
)
);
let prompt = std::sync::Arc::new(
erp_ai::service::prompt::PromptService::new(db.clone())
);
let usage = std::sync::Arc::new(
erp_ai::service::usage::UsageService::new(db.clone())
);
Self { db, event_bus, analysis, prompt, usage }
}
}
```
- [ ] **Step 5: main.rs 注册模块 + 路由**
```rust
// 1. 创建 AiModule
let ai_module = erp_ai::AiModule;
// 2. 注册到 registry
let registry = ModuleRegistry::new()
.register(auth_module)
// ...existing...
.register(ai_module);
// 3. 合并路由 (protected_routes 中)
.merge(erp_ai::AiModule::protected_routes())
```
- [ ] **Step 6: 验证全 workspace 编译**
```bash
cargo check --workspace
```
- [ ] **Step 7: 提交**
```bash
git add crates/erp-server/
git commit -m "feat(server): erp-ai 模块集成 — Config/State/路由注册"
```
---
## Chunk 6: 完善 SSE Handler + 端到端验证
### Task 11: 实现 SSE 分析 Handler
**Files:**
- Rewrite: `crates/erp-ai/src/handler/mod.rs`
- [ ] **Step 1: 完善 handler/mod.rs — SSE 流式分析**
```rust
// crates/erp-ai/src/handler/mod.rs
use axum::{
extract::{Extension, Path, Query, State},
response::sse::{Event, KeepAlive, Sse},
Json,
};
use axum::extract::FromRef;
use erp_core::rbac::require_permission;
use erp_core::tenant::TenantContext;
use futures::StreamExt;
use serde::Deserialize;
use std::convert::Infallible;
use crate::dto::{AnalyzeOptions, AnalysisSseEvent, AnalysisType, TokenUsage};
use crate::state::AiState;
// === 分析请求 Query/Body ===
#[derive(Debug, Deserialize)]
pub struct AnalyzeBody {
pub report_id: Option<uuid::Uuid>,
pub patient_id: Option<uuid::Uuid>,
pub metrics: Option<Vec<String>>,
pub options: Option<AnalyzeOptions>,
}
// === SSE 分析端点 ===
pub async fn stream_lab_report<S>(
State(state): State<AiState>,
Extension(ctx): Extension<TenantContext>,
Json(body): Json<AnalyzeBody>,
) -> Result<Sse<impl futures::Stream<Item = Result<Event, Infallible>>>, erp_core::AppError>
where
AiState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "ai.analysis.manage")?;
let report_id = body.report_id.ok_or_else(|| erp_core::AppError::Validation("report_id 必填".into()))?;
let prompt = state.prompt.get_active_prompt(ctx.tenant_id(), "lab_report_interpretation").await?;
let model_config: serde_json::Value = serde_json::from_str(&serde_json::to_string(&prompt.model_config).unwrap_or_default()).unwrap_or_default();
let model = model_config["model"].as_str().unwrap_or("claude-sonnet-4-6").to_string();
let temperature = model_config["temperature"].as_f32().unwrap_or(0.3);
let max_tokens = model_config["max_tokens"].as_u64().unwrap_or(2048) as u32;
let source_ref = report_id.to_string();
let (stream, analysis_id, provider_name) = state.analysis.stream_analyze(
ctx.tenant_id(), ctx.user_id(), uuid::Uuid::nil(),
AnalysisType::LabReport, source_ref,
prompt.system_prompt, prompt.user_prompt_template,
serde_json::json!({"placeholder": true}),
model, temperature, max_tokens,
).await?;
let analysis_id_clone = analysis_id;
let provider_name_clone = provider_name;
let model_clone = model.clone();
let state_clone = state.clone();
let sse_stream = async_stream::stream! {
let mut full_content = String::new();
let mut index: u32 = 0;
let mut stream = std::pin::pin!(stream);
while let Some(result) = stream.next().await {
match result {
Ok(chunk) => {
full_content.push_str(&chunk);
index += 1;
let event = AnalysisSseEvent::Chunk { content: chunk, index };
let data = serde_json::to_string(&event).unwrap_or_default();
yield Ok(Event::default().data(data));
}
Err(e) => {
let event = AnalysisSseEvent::Error { message: e.to_string() };
let data = serde_json::to_string(&event).unwrap_or_default();
yield Ok(Event::default().data(data));
let _ = state_clone.analysis.fail_analysis(analysis_id_clone, e.to_string()).await;
return;
}
}
}
// 完成: 存储结果
let metadata = serde_json::json!({
"model": model_clone,
"provider": provider_name_clone,
});
let _ = state_clone.analysis.complete_analysis(analysis_id_clone, full_content, metadata).await;
let done_event = AnalysisSseEvent::Done {
analysis_id: analysis_id_clone,
status: "completed".into(),
};
let data = serde_json::to_string(&done_event).unwrap_or_default();
yield Ok(Event::default().data(data));
};
Ok(Sse::new(sse_stream).keep_alive(KeepAlive::default()))
}
// === 趋势/方案/摘要端点 — 结构类似，切换 analysis_type 和 prompt name ===
macro_rules! analyze_endpoint {
($fn_name:ident, $analysis_type:expr, $prompt_name:literal) => {
pub async fn $fn_name<S>(
State(state): State<AiState>,
Extension(ctx): Extension<TenantContext>,
Json(body): Json<AnalyzeBody>,
) -> Result<Sse<impl futures::Stream<Item = Result<Event, Infallible>>>, erp_core::AppError>
where
AiState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "ai.analysis.manage")?;
// 与 stream_lab_report 相同的流程，区别:
// - analysis_type: $analysis_type
// - prompt name: $prompt_name
// - source_ref 来自 body.patient_id 或 body.report_id
// 实现时从 stream_lab_report 复制并修改对应参数
todo!("参照 stream_lab_report 实现")
}
};
}
analyze_endpoint!(stream_trends, AnalysisType::Trends, "health_trend_analysis");
analyze_endpoint!(stream_checkup_plan, AnalysisType::CheckupPlan, "personalized_checkup_plan");
analyze_endpoint!(stream_report_summary, AnalysisType::ReportSummary, "report_summary_generation");
// === 分析历史 ===
#[derive(Debug, Deserialize)]
pub struct ListAnalysisQuery {
pub patient_id: Option<uuid::Uuid>,
pub analysis_type: Option<String>,
pub page: Option<u64>,
pub page_size: Option<u64>,
}
pub async fn list_analysis<S>(
State(_state): State<AiState>,
Extension(ctx): Extension<TenantContext>,
Query(_params): Query<ListAnalysisQuery>,
) -> Result<Json<erp_core::ApiResponse<()>>, erp_core::AppError>
where
AiState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "ai.analysis.list")?;
// TODO: 查询 ai_analysis_results 表
Ok(Json(erp_core::ApiResponse::success(())))
}
pub async fn get_analysis<S>(
State(_state): State<AiState>,
Extension(ctx): Extension<TenantContext>,
Path(_id): Path<uuid::Uuid>,
) -> Result<Json<erp_core::ApiResponse<()>>, erp_core::AppError>
where
AiState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "ai.analysis.list")?;
// TODO: 查询单条分析结果
Ok(Json(erp_core::ApiResponse::success(())))
}
```
> 注意: `erp_core::ApiResponse::success` 和 `erp_core::rbac::require_permission` 的实际签名需参照现有代码。宏 `analyze_endpoint!` 展开的函数暂时用 `todo!()`，实际实现时从 `stream_lab_report` 复制并修改参数。
- [ ] **Step 2: 验证编译**
```bash
cargo check --workspace
```
- [ ] **Step 3: 提交**
```bash
git add crates/erp-ai/src/handler/
git commit -m "feat(ai): SSE 流式分析 Handler 实现"
```
---
### Task 12: 端到端验证
- [ ] **Step 1: 启动 PostgreSQL + 后端服务**
```bash
cd crates/erp-server && cargo run
```
验证:
- 迁移自动执行，`ai_prompts`/`ai_analysis_results`/`ai_usage_logs` 三张表创建成功
- 服务启动无错误日志
- [ ] **Step 2: 检查 API 文档**
访问 `http://localhost:3000/api/docs/openapi.json`，确认 `/api/v1/ai/analyze/*` 端点已注册
- [ ] **Step 3: 数据库验证**
```sql
SELECT table_name FROM information_schema.tables WHERE table_name LIKE 'ai_%';
```
Expected: `ai_prompts`, `ai_analysis_results`, `ai_usage_logs`
- [ ] **Step 4: 提交最终状态 + 推送**
```bash
git push
```
---
## 验证清单
- [ ] `cargo check --workspace` — 全 workspace 编译通过
- [ ] `cargo test --workspace` — 所有测试通过
- [ ] PostgreSQL 三张 AI 表存在
- [ ] `/api/v1/ai/analyze/lab-report` 端点在 Swagger UI 可见
- [ ] AI 权限码已同步到数据库 `permissions` 表
- [ ] `AiModule` 出现在启动日志的已注册模块列表中
---
## 后续计划 (Phase 1 MVP 剩余)
以下任务在实际实现阶段细化:
- **Task 13:** HealthDataProvider 实际数据查询实现 (替换 erp-health stub)
- **Task 14:** 种子 Prompt 模板数据 (4 个默认模板通过迁移插入)
- **Task 15:** Redis 缓存层 (缓存命中时跳过 AI 调用)
- **Task 16:** 降级规则引擎 (50 条常见指标本地规则)
- **Task 17:** 速率限制中间件
- **Task 18:** 小程序 AI 解读页面 (报告详情页 + SSE 渲染)
Reference in New Issue View Git Blame Copy Permalink