iven/hms
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(health+ai): P2 咨询联动 + AI 巡检消费 — 全链路打通

Browse Source 
业务链路打通 5/5 断点全部完成：
- 咨询→随访：医生端新增"创建随访"按钮，从咨询会话直接创建随访任务
- 咨询→AI：医生端新增"AI 分析"按钮，对咨询上下文触发 AI 分析
- 告警→咨询：小程序告警详情页新增"在线咨询"快捷入口
- AI 巡检消费：erp-ai 新增 patrol_consumer，订阅 ai.patrol.requested 事件
- 前端联动：Web ConsultationDetail + 小程序 alerts 页面联动实现

后端：2 新 API + 2 handler + 1 service + AI event consumer
前端：Web 2 API + 1 页面改造 + 小程序 2 页面改造
测试：Web consultations.test.ts 9/9 通过
This commit is contained in:
iven
2026-05-20 17:50:49 +08:00
parent 5f34e5715a
commit fa1dc764a3
15 changed files with 888 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
crates/erp-ai/src/event/mod.rs
View File

@@ -1 +1,2 @@
pub mod copilot_consumer;
pub mod patrol_consumer;

254
crates/erp-ai/src/event/patrol_consumer.rs Normal file
View File

@@ -0,0 +1,254 @@
//! AI 巡护事件消费者 — 订阅 health 模块发布的 `ai.patrol.requested` 事件，
//! 为未处理告警患者自动入队 AI 趋势分析。
//!
//! 事件来源：erp-health 每日定时扫描有未处理告警的患者，发布此事件。
//! 处理策略：幂等消费 + 入队分析（复用 AnalysisQueue），不直接调用 AI Provider。
use erp_core::events::{EventBus, SubscriptionHandle};
use tracing::{info, warn};
use uuid::Uuid;
/// 消费者标识，用于幂等检查（processed_events 表的 consumer_id）
const CONSUMER_ID: &str = "patrol_consumer";
/// 订阅事件类型前缀
const EVENT_PREFIX: &str = "ai.patrol.";
/// 巡护请求事件 payload 中提取的结构
#[derive(Debug)]
struct PatrolRequest {
patient_id: Uuid,
doctor_id: Option<Uuid>,
source: String,
reason: String,
}
/// 从事件 payload 中提取巡护请求字段
fn extract_patrol_request(payload: &serde_json::Value) -> Option<PatrolRequest> {
let patient_id = payload
.get("patient_id")
.or_else(|| payload.get("data").and_then(|d| d.get("patient_id")))
.and_then(|v| v.as_str())
.and_then(|s| Uuid::parse_str(s).ok())?;
let doctor_id = payload
.get("doctor_id")
.or_else(|| payload.get("data").and_then(|d| d.get("doctor_id")))
.and_then(|v| v.as_str())
.and_then(|s| Uuid::parse_str(s).ok());
let source = payload
.get("source")
.or_else(|| payload.get("data").and_then(|d| d.get("source")))
.and_then(|v| v.as_str())
.unwrap_or("daily_patrol")
.to_string();
let reason = payload
.get("reason")
.or_else(|| payload.get("data").and_then(|d| d.get("reason")))
.and_then(|v| v.as_str())
.unwrap_or("未指定原因")
.to_string();
Some(PatrolRequest {
patient_id,
doctor_id,
source,
reason,
})
}
/// 启动巡护事件消费者。
///
/// 订阅 `ai.patrol.*` 前缀事件，收到 `ai.patrol.requested` 时：
/// 1. 幂等检查（跳过已处理事件）
/// 2. 提取 patient_id / tenant_id / reason
/// 3. 入队 AI 趋势分析（priority=2，与 health_data.critical_alert 同级）
/// 4. 标记事件已处理
pub fn spawn(db: &sea_orm::DatabaseConnection, event_bus: &EventBus) -> SubscriptionHandle {
let (mut rx, handle) = event_bus.subscribe_filtered(EVENT_PREFIX.to_string());
let db = db.clone();
tokio::spawn(async move {
info!("AI 巡护事件消费者已启动，监听前缀: {EVENT_PREFIX}");
while let Some(event) = rx.recv().await {
if event.event_type != "ai.patrol.requested" {
continue;
}
// 幂等检查
match erp_core::events::is_event_processed(&db, event.id, CONSUMER_ID).await {
Ok(true) => {
info!(
event_id = %event.id,
"巡护事件已处理，跳过"
);
continue;
}
Ok(false) => {}
Err(e) => {
warn!(
event_id = %event.id,
error = %e,
"幂等检查失败，继续处理"
);
}
}
// 提取 payload
let request = match extract_patrol_request(&event.payload) {
Some(r) => r,
None => {
warn!(
event_id = %event.id,
"ai.patrol.requested 事件缺少 patient_id 字段，跳过"
);
// 仍然标记已处理，避免重复消费格式错误的事件
let _ =
erp_core::events::mark_event_processed(&db, event.id, CONSUMER_ID).await;
continue;
}
};
info!(
event_id = %event.id,
tenant_id = %event.tenant_id,
patient_id = %request.patient_id,
source = %request.source,
reason = %request.reason,
"收到巡护分析请求"
);
// 入队趋势分析 — 复用现有 AnalysisQueue
let queue = crate::service::analysis_queue::AnalysisQueue::new(db.clone());
let job = crate::service::analysis_queue::AnalysisJob {
tenant_id: event.tenant_id,
patient_id: request.patient_id,
analysis_type: "trend".to_string(),
priority: 2, // 与 critical_alert 同级，高于普通分析
source_event: Some(event.event_type.clone()),
source_ref: format!("patrol:{}", request.reason),
created_by: request.doctor_id,
};
match queue.enqueue(job).await {
Ok(queue_id) => {
info!(
queue_id = %queue_id,
patient_id = %request.patient_id,
tenant_id = %event.tenant_id,
"巡护分析已入队"
);
}
Err(e) => {
warn!(
patient_id = %request.patient_id,
tenant_id = %event.tenant_id,
error = %e,
"巡护分析入队失败"
);
}
}
// 标记事件已处理（无论入队成功与否，避免无限重试同一事件）
if let Err(e) = erp_core::events::mark_event_processed(&db, event.id, CONSUMER_ID).await
{
warn!(
event_id = %event.id,
error = %e,
"标记巡护事件已处理失败（非致命）"
);
}
}
info!("AI 巡护事件消费者已退出");
});
handle
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_extract_patrol_request_full_payload() {
let payload = serde_json::json!({
"patient_id": "01234567-89ab-7def-8000-000000000001",
"doctor_id": "01234567-89ab-7def-8000-000000000002",
"source": "daily_patrol",
"reason": "告警未处理: 血压异常"
});
let request = extract_patrol_request(&payload).unwrap();
assert_eq!(
request.patient_id.to_string(),
"01234567-89ab-7def-8000-000000000001"
);
assert!(request.doctor_id.is_some());
assert_eq!(request.source, "daily_patrol");
assert_eq!(request.reason, "告警未处理: 血压异常");
}
#[test]
fn test_extract_patrol_request_nested_in_data() {
let payload = serde_json::json!({
"schema_version": "v1",
"occurred_at": "2026-05-20T10:00:00Z",
"data": {
"patient_id": "01234567-89ab-7def-8000-000000000001",
"source": "daily_patrol",
"reason": "告警未处理: 血糖偏高"
}
});
let request = extract_patrol_request(&payload).unwrap();
assert_eq!(request.source, "daily_patrol");
assert!(request.doctor_id.is_none());
}
#[test]
fn test_extract_patrol_request_missing_patient_id() {
let payload = serde_json::json!({
"source": "daily_patrol",
"reason": "告警未处理"
});
assert!(extract_patrol_request(&payload).is_none());
}
#[test]
fn test_extract_patrol_request_invalid_uuid() {
let payload = serde_json::json!({
"patient_id": "not-a-uuid",
"source": "daily_patrol"
});
assert!(extract_patrol_request(&payload).is_none());
}
#[test]
fn test_extract_patrol_request_defaults() {
let payload = serde_json::json!({
"patient_id": "01234567-89ab-7def-8000-000000000001"
});
let request = extract_patrol_request(&payload).unwrap();
assert_eq!(request.source, "daily_patrol");
assert_eq!(request.reason, "未指定原因");
assert!(request.doctor_id.is_none());
}
#[test]
fn test_consumer_id_is_stable() {
// 确保消费者 ID 不变，否则幂等检查会失效
assert_eq!(CONSUMER_ID, "patrol_consumer");
}
#[test]
fn test_event_prefix_matches_expected() {
assert_eq!(EVENT_PREFIX, "ai.patrol.");
}
}

6
crates/erp-ai/src/module.rs
View File

@@ -363,6 +363,10 @@ impl ErpModule for AiModule {
let copilot_handles = crate::event::copilot_consumer::spawn(&ctx.db, &ctx.event_bus);
std::mem::forget(copilot_handles);
// 巡护事件消费者 — 订阅 ai.patrol.requested，为未处理告警患者入队趋势分析
let patrol_handle = crate::event::patrol_consumer::spawn(&ctx.db, &ctx.event_bus);
std::mem::forget(patrol_handle);
// 每日凌晨 2:00 批量刷新所有在管患者风险快照
let refresh_db = ctx.db.clone();
let refresh_event_bus = ctx.event_bus.clone();
@@ -406,7 +410,7 @@ impl ErpModule for AiModule {
tracing::info!(
module = "ai",
"AI 模块事件处理器已注册（监听 ai.* 事件 + Copilot 事件）"
"AI 模块事件处理器已注册（监听 ai.* 事件 + Copilot 事件 + 巡护事件）"
);
Ok(())
}

33
crates/erp-health/src/dto/consultation_dto.rs
View File

@@ -61,3 +61,36 @@ pub struct SessionQuery {
pub page: Option<u64>,
pub page_size: Option<u64>,
}
/// 从咨询会话创建随访任务请求体 — 仅需填写随访类型和计划日期，
/// patient_id / source_type / source_id 由服务端自动填充。
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct CreateFollowUpFromConsultationReq {
pub follow_up_type: String,
pub planned_date: chrono::NaiveDate,
pub assigned_to: Option<Uuid>,
pub content_template: Option<String>,
}
/// 从咨询会话触发 AI 分析请求体 — 可选指定分析类型。
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct TriggerAiAnalysisReq {
/// 分析类型，默认 "consultation_summary"
pub analysis_type: Option<String>,
}
/// 从咨询会话创建随访任务响应体。
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct FollowUpFromConsultationResp {
pub task_id: Uuid,
pub session_id: Uuid,
pub patient_id: Uuid,
}
/// 从咨询会话触发 AI 分析响应体。
#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
pub struct AiAnalysisTriggeredResp {
pub session_id: Uuid,
pub patient_id: Uuid,
pub analysis_type: String,
}

61
crates/erp-health/src/event/alert.rs
View File

@@ -1,3 +1,5 @@
use sea_orm::{ColumnTrait, EntityTrait, QueryFilter};
/// alert.triggered → 告警消息通知 + 告警聚合
pub fn spawn(state: &crate::state::HealthState) -> Vec<erp_core::events::SubscriptionHandle> {
let mut handles = Vec::new();
@@ -28,7 +30,54 @@ pub fn spawn(state: &crate::state::HealthState) -> Vec<erp_core::events::Subscri
.get("rule_name")
.and_then(|v| v.as_str())
.unwrap_or("健康告警");
if let Some(pid) = patient_id {
// 检查患者是否有活跃咨询会话（active / waiting）
let patient_uuid = uuid::Uuid::parse_str(pid).ok();
let active_session = if let Some(puid) = patient_uuid {
crate::entity::consultation_session::Entity::find()
.filter(
crate::entity::consultation_session::Column::PatientId.eq(puid),
)
.filter(
crate::entity::consultation_session::Column::TenantId
.eq(event.tenant_id),
)
.filter(
crate::entity::consultation_session::Column::DeletedAt
.is_null(),
)
.filter(
sea_orm::Condition::any()
.add(
crate::entity::consultation_session::Column::Status
.eq("active"),
)
.add(
crate::entity::consultation_session::Column::Status
.eq("waiting"),
),
)
.one(&alert_db)
.await
.ok()
.flatten()
} else {
None
};
let consultation_session_id =
active_session.as_ref().map(|s| s.id.to_string());
let mut params = serde_json::json!({
"rule_name": rule_name,
"severity": severity,
"suggested_action": "consult",
});
if let Some(ref sid) = consultation_session_id {
params["consultation_session_id"] = serde_json::json!(sid);
}
let notify_event = erp_core::events::DomainEvent::new(
"message.send",
event.tenant_id,
@@ -37,14 +86,16 @@ pub fn spawn(state: &crate::state::HealthState) -> Vec<erp_core::events::Subscri
"recipient_type": "patient",
"recipient_id": pid,
"template_key": if severity == "critical" { "CRITICAL_HEALTH_ALERT" } else { "HEALTH_DATA_ABNORMAL" },
"params": {
"rule_name": rule_name,
"severity": severity,
}
"params": params,
})),
);
alert_bus.publish(notify_event, &alert_db).await;
tracing::info!(patient_id = %pid, severity = %severity, "告警通知已发送");
tracing::info!(
patient_id = %pid,
severity = %severity,
consultation_session_id = ?consultation_session_id,
"告警通知已发送（含咨询联动建议）"
);
}
let _ = erp_core::events::mark_event_processed(
&alert_db,

67
crates/erp-health/src/handler/consultation_handler.rs
View File

@@ -305,3 +305,70 @@ where
.await?;
Ok(Json(ApiResponse::ok(result)))
}
/// 从咨询会话创建随访任务 — 自动从 session 中提取 patient_id，
/// source_type = "consultation", source_id = session_id。
#[utoipa::path(
post,
path = "/consultation-sessions/{id}/follow-up",
request_body = CreateFollowUpFromConsultationReq,
responses(
(status = 200, description = "随访任务已创建"),
(status = 404, description = "会话不存在"),
),
tag = "咨询联动",
)]
pub async fn create_follow_up_from_session<S>(
State(state): State<HealthState>,
Extension(ctx): Extension<TenantContext>,
Path(id): Path<Uuid>,
Json(req): Json<CreateFollowUpFromConsultationReq>,
) -> Result<Json<ApiResponse<FollowUpFromConsultationResp>>, AppError>
where
HealthState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "health.follow-up.manage")?;
let result = consultation_service::create_follow_up_from_session(
&state,
ctx.tenant_id,
Some(ctx.user_id),
id,
req,
)
.await?;
Ok(Json(ApiResponse::ok(result)))
}
/// 从咨询会话触发 AI 分析 — 加载最近消息作为上下文，发布事件。
#[utoipa::path(
post,
path = "/consultation-sessions/{id}/ai-analysis",
request_body = TriggerAiAnalysisReq,
responses(
(status = 200, description = "AI 分析已触发"),
(status = 404, description = "会话不存在"),
),
tag = "咨询联动",
)]
pub async fn trigger_ai_analysis_from_session<S>(
State(state): State<HealthState>,
Extension(ctx): Extension<TenantContext>,
Path(id): Path<Uuid>,
Json(req): Json<TriggerAiAnalysisReq>,
) -> Result<Json<ApiResponse<AiAnalysisTriggeredResp>>, AppError>
where
HealthState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
require_permission(&ctx, "health.consultation.manage")?;
let result = consultation_service::trigger_ai_analysis_from_session(
&state,
ctx.tenant_id,
Some(ctx.user_id),
id,
req,
)
.await?;
Ok(Json(ApiResponse::ok(result)))
}

9
crates/erp-health/src/routes/consultation.rs
View File

@@ -38,6 +38,15 @@ where
"/health/consultation-sessions/{id}/read",
axum::routing::put(consultation_handler::mark_session_read),
)
// 咨询联动
.route(
"/health/consultation-sessions/{id}/follow-up",
axum::routing::post(consultation_handler::create_follow_up_from_session),
)
.route(
"/health/consultation-sessions/{id}/ai-analysis",
axum::routing::post(consultation_handler::trigger_ai_analysis_from_session),
)
.route(
"/health/consultation-messages",
axum::routing::post(consultation_handler::create_message),

162
crates/erp-health/src/service/consultation_service.rs
View File

@@ -857,3 +857,165 @@ pub async fn enrich_doctor_dashboard_health(
Ok(())
}
// ---------------------------------------------------------------------------
// 咨询联动 — 创建随访 / 触发 AI 分析
// ---------------------------------------------------------------------------
/// 从咨询会话创建随访任务。
///
/// 自动从 session 中提取 patient_id，设置 source_type = "consultation",
/// source_id = session_id。
pub async fn create_follow_up_from_session(
state: &HealthState,
tenant_id: Uuid,
operator_id: Option<Uuid>,
session_id: Uuid,
req: CreateFollowUpFromConsultationReq,
) -> HealthResult<FollowUpFromConsultationResp> {
tracing::info!(
action = "create_follow_up_from_session",
session_id = %session_id,
"Creating follow-up task from consultation session"
);
// 1. 查询会话，获取 patient_id
let session = consultation_session::Entity::find()
.filter(consultation_session::Column::Id.eq(session_id))
.filter(consultation_session::Column::TenantId.eq(tenant_id))
.filter(consultation_session::Column::DeletedAt.is_null())
.one(&state.db)
.await?
.ok_or(HealthError::ConsultationNotFound)?;
let patient_id = session.patient_id;
// 2. 构造 CreateFollowUpTaskReq 并委托给 follow_up_service
let follow_up_req = crate::dto::follow_up_dto::CreateFollowUpTaskReq {
patient_id,
assigned_to: req.assigned_to,
follow_up_type: req.follow_up_type,
planned_date: req.planned_date,
content_template: req.content_template,
related_appointment_id: None,
source_type: Some("consultation".to_string()),
source_id: Some(session_id),
};
let task = crate::service::follow_up_service::create_task(
state,
tenant_id,
operator_id,
follow_up_req,
)
.await?;
Ok(FollowUpFromConsultationResp {
task_id: task.id,
session_id,
patient_id,
})
}
/// 从咨询会话触发 AI 分析。
///
/// 加载会话最近消息作为上下文，发布 `ai.analysis.requested` 事件。
pub async fn trigger_ai_analysis_from_session(
state: &HealthState,
tenant_id: Uuid,
operator_id: Option<Uuid>,
session_id: Uuid,
req: TriggerAiAnalysisReq,
) -> HealthResult<AiAnalysisTriggeredResp> {
tracing::info!(
action = "trigger_ai_analysis_from_session",
session_id = %session_id,
"Triggering AI analysis from consultation session"
);
// 1. 查询会话，获取 patient_id
let session = consultation_session::Entity::find()
.filter(consultation_session::Column::Id.eq(session_id))
.filter(consultation_session::Column::TenantId.eq(tenant_id))
.filter(consultation_session::Column::DeletedAt.is_null())
.one(&state.db)
.await?
.ok_or(HealthError::ConsultationNotFound)?;
let patient_id = session.patient_id;
let doctor_id = session.doctor_id;
let analysis_type = req
.analysis_type
.unwrap_or_else(|| "consultation_summary".to_string());
// 2. 加载最近消息作为上下文（最多 50 条）
let recent_messages = consultation_message::Entity::find()
.filter(consultation_message::Column::SessionId.eq(session_id))
.filter(consultation_message::Column::TenantId.eq(tenant_id))
.filter(consultation_message::Column::DeletedAt.is_null())
.order_by_desc(consultation_message::Column::CreatedAt)
.limit(50)
.all(&state.db)
.await?;
// 解密消息内容，构造上下文摘要
let kek = state.crypto.kek();
let context_messages: Vec<serde_json::Value> = recent_messages
.into_iter()
.rev() // 恢复时间正序
.map(|m| {
let content = pii::decrypt(kek, &m.content).unwrap_or(m.content);
serde_json::json!({
"sender_role": m.sender_role,
"content_type": m.content_type,
"content": content,
"created_at": m.created_at.to_rfc3339(),
})
})
.collect();
// 3. 发布 ai.analysis.requested 事件
let event = DomainEvent::new(
"ai.analysis.requested",
tenant_id,
erp_core::events::build_event_payload(serde_json::json!({
"analysis_type": analysis_type,
"patient_id": patient_id.to_string(),
"doctor_id": doctor_id.map(|id| id.to_string()).unwrap_or_default(),
"source": "consultation",
"source_id": session_id.to_string(),
"triggered_by": operator_id.map(|id| id.to_string()).unwrap_or_default(),
"context": {
"session_id": session_id.to_string(),
"consultation_type": session.consultation_type,
"messages": context_messages,
}
})),
);
state.event_bus.publish(event, &state.db).await;
audit_service::record(
AuditLog::new(
tenant_id,
operator_id,
"consultation.ai_analysis_triggered",
"consultation",
)
.with_resource_id(session_id),
&state.db,
)
.await;
tracing::info!(
session_id = %session_id,
patient_id = %patient_id,
analysis_type = %analysis_type,
"AI 分析已从咨询会话触发"
);
Ok(AiAnalysisTriggeredResp {
session_id,
patient_id,
analysis_type,
})
}