AIclinicalresearch/docs/03-业务模块/ASL-AI智能文献/05-开发记录/2025-11-23_Day4_数据库设计与批处理服务开发.md

Day 4开发记录：数据库设计与批处理服务开发

日期：2025-11-23
开发者：ASL开发团队
阶段：全文复筛MVP - Day 4
状态：✅ 已完成

---

📋 开发目标

Day 4上午：完成数据库设计与迁移 Day 4下午：开发批处理服务（FulltextScreeningService）

---

✅ Day 4上午：数据库设计与迁移

1. Schema设计

1.1 修改 AslLiterature 表

新增13个全文复筛相关字段：

文献生命周期：

• stage - 阶段标记（imported/title_screened/fulltext_pending/fulltext_screened）

PDF管理：

• has_pdf - 是否有PDF
• pdf_storage_type - 存储类型（oss/dify/local）
• pdf_storage_ref - 存储引用
• pdf_status - 状态（pending/extracting/completed/failed）
• pdf_uploaded_at - 上传时间

全文管理（云原生）：

• full_text_storage_type - 存储类型（oss/dify）
• full_text_storage_ref - 存储引用
• full_text_url - 访问URL

全文元数据：

• full_text_format - 格式（markdown/plaintext）
• full_text_source - 提取方式（nougat/pymupdf）
• full_text_token_count - Token数量
• full_text_extracted_at - 提取时间

设计亮点：

• ✅ 云原生架构：全文存储在OSS/Dify，数据库只存引用
• ✅ 符合规范：遵循《云原生开发规范》，不在数据库存储大文本
• ✅ 可扩展性：支持多种存储方式的适配器模式

1.2 新建 AslFulltextScreeningTask 表

任务管理表，字段包括：

• 基础信息：id, project_id
• 模型配置：model_a, model_b, prompt_version
• 进度跟踪：total_count, processed_count, success_count, failed_count, degraded_count
• 成本统计：total_tokens, total_cost
• 状态管理：status, started_at, completed_at, estimated_end_at
• 错误记录：error_message, error_stack

设计亮点：

• ✅ 实时进度：支持前端轮询任务进度
• ✅ 成本跟踪：累计Token和费用
• ✅ 预估时间：动态计算剩余时间

1.3 新建 AslFulltextScreeningResult 表

结果存储表（12字段模板），字段包括：

• 双模型结果：Model A (DeepSeek-V3) 和 Model B (Qwen-Max) 的完整输出
• 验证结果：医学逻辑验证、证据链验证
• 冲突检测：字段级冲突对比、优先级排序
• 人工复核：最终决策、排除原因、复核笔记
• 可追溯性：原始输出、Prompt版本、处理时间

设计亮点：

• ✅ JSONB存储：12字段灵活存储，支持高效查询
• ✅ 双模型对比：完整保存两个模型的输出
• ✅ 冲突优先级：自动计算review_priority（0-100）
• ✅ 可审计：保留raw_output，可追溯LLM原始响应

2. 迁移策略

2.1 问题识别

在迁移过程中发现：

• ⚠️ 历史遗留问题：部分模块的表创建在 public schema
• ✅ ASL模块数据完全正确：所有表都在 asl_schema
• ⚠️ Prisma Migrate会尝试删除 public 中的重复表

2.2 解决方案：手动SQL迁移

策略：使用手动SQL脚本，只操作 asl_schema，不影响其他模块

-- 只操作asl_schema，不影响其他schema
ALTER TABLE asl_schema.literatures ADD COLUMN IF NOT EXISTS ...;
CREATE TABLE IF NOT EXISTS asl_schema.fulltext_screening_tasks (...);
CREATE TABLE IF NOT EXISTS asl_schema.fulltext_screening_results (...);

执行：

Get-Content manual_fulltext_screening.sql | docker exec -i ai-clinical-postgres psql ...

验证：

\dt asl_schema.*
-- 结果：6个表
-- ✅ literatures (已更新)
-- ✅ screening_projects
-- ✅ screening_tasks
-- ✅ screening_results
-- ✅ fulltext_screening_tasks (新建)
-- ✅ fulltext_screening_results (新建)

2.3 Schema隔离验证

检查结果：

• ✅ ASL模块所有6个表都在 asl_schema
• ✅ 无数据泄漏到 public schema
• ✅ 外键约束全部指向 asl_schema 内部
• ✅ Prisma Model正确映射（@@schema("asl_schema")）

相关文档：

• 数据库迁移状态说明
• 数据库设计文档

3. 产出

• ✅ Prisma Schema更新（3个模型）
• ✅ 手动SQL迁移脚本（141行）
• ✅ 数据库迁移状态说明文档（435行）
• ✅ 数据库设计文档更新（v3.0）
• ✅ 模块状态文档更新（v1.2）

---

✅ Day 4下午：批处理服务开发

1. 核心服务：FulltextScreeningService

1.1 服务职责

职责	说明
任务调度	批量处理文献，并发控制
服务集成	调用LLM服务、验证器、冲突检测
进度跟踪	实时更新任务进度，计算预估时间
容错处理	重试机制、降级模式、错误记录
数据持久化	保存处理结果到数据库

1.2 核心方法

1. createAndProcessTask() - 任务创建入口

async createAndProcessTask(
  projectId: string,
  literatureIds: string[],
  config: FulltextScreeningConfig
): Promise<string>

功能：

• 验证项目和文献数据
• 创建任务记录
• 启动后台处理（不等待完成）
• 返回任务ID

2. processTaskInBackground() - 后台批处理逻辑

private async processTaskInBackground(
  taskId: string,
  literatures: any[],
  project: any,
  config: FulltextScreeningConfig
): Promise<void>

功能：

• 更新任务状态为"运行中"
• 构建PICOS上下文
• 使用 p-queue 实现并发控制（默认并发3）
• 调用 screenLiteratureWithRetry() 处理每篇文献
• 累计统计（success/failed/degraded/tokens/cost）
• 标记任务完成

3. screenLiteratureWithRetry() - 单篇处理（带重试）

private async screenLiteratureWithRetry(
  taskId: string,
  projectId: string,
  literature: any,
  picosContext: any,
  config: FulltextScreeningConfig
): Promise<SingleLiteratureResult>

功能：

• 最多重试2次（可配置）
• 指数退避策略（1s, 2s）
• 捕获并记录错误

4. screenLiterature() - 单篇处理核心逻辑

private async screenLiterature(
  taskId: string,
  projectId: string,
  literature: any,
  picosContext: any,
  config: FulltextScreeningConfig
): Promise<SingleLiteratureResult>

功能：

1. 获取全文内容（支持测试模式：跳过PDF提取）
2. 调用 LLM12FieldsService.processDualModels()（双模型并行）
3. 医学逻辑验证（MedicalLogicValidator）
4. 证据链验证（EvidenceChainValidator）
5. 冲突检测（ConflictDetectionService）
6. 保存结果到数据库（fulltext_screening_results表）
7. 返回处理结果（tokens、cost、isDegraded）

5. updateTaskProgress() - 进度更新

private async updateTaskProgress(
  taskId: string,
  progress: { ... }
): Promise<void>

功能：

• 计算平均处理时间
• 预估剩余时间（estimatedEndAt）
• 更新数据库（processed/success/failed/degraded/tokens/cost）

6. completeTask() - 任务完成

private async completeTask(
  taskId: string,
  summary: { ... }
): Promise<void>

功能：

• 标记任务状态（completed/failed）
• 更新最终统计
• 记录完成时间

1.3 查询接口

getTaskProgress() - 查询任务进度

async getTaskProgress(taskId: string): Promise<ScreeningProgress | null>

返回：

• 任务状态（pending/running/completed/failed）
• 进度统计（processed/success/failed/degraded）
• 成本统计（totalTokens/totalCost）
• 时间信息（started/completed/estimatedEnd）

getTaskResults() - 查询任务结果

async getTaskResults(
  taskId: string,
  filter?: { conflictOnly, page, pageSize }
): Promise<{ results, total }>

功能：

• 支持过滤（仅冲突项）
• 分页查询
• 按优先级排序（冲突优先、review_priority降序）

updateReviewDecision() - 更新人工复核决策

async updateReviewDecision(
  resultId: string,
  decision: { finalDecision, finalDecisionBy, ... }
): Promise<void>

功能：

• 更新最终决策（include/exclude）
• 记录复核人和时间
• 记录排除原因和笔记

2. 技术亮点

2.1 并发控制

使用 p-queue 实现优雅的并发控制：

const queue = new PQueue({ concurrency: 3 });

const tasks = literatures.map((literature, index) =>
  queue.add(async () => {
    // 处理单篇文献
  })
);

await Promise.all(tasks);

优势：

• ✅ 自动排队，避免同时发起过多LLM请求
• ✅ 控制API调用频率，防止触发限流
• ✅ 充分利用并发，提速3倍（串行→3并发）

2.2 容错机制

3层容错：

1. Retry层：单篇文献失败自动重试（最多2次）
2. Degraded层：LLM12FieldsService支持降级模式（单模型成功即可）
3. Continue层：单篇失败不影响整体，继续处理其他文献

效果：

• ✅ 降低失败率
• ✅ 提高任务完成率
• ✅ 完整记录失败原因

2.3 测试模式

支持 skipExtraction: true 测试模式：

if (config.skipExtraction) {
  // 使用标题+摘要作为全文
  fullText = `# ${literature.title}\n\n## Abstract\n${literature.abstract}`;
  fullTextFormat = 'markdown';
  fullTextSource = 'test';
}

优势：

• ✅ 快速验证服务逻辑
• ✅ 无需真实PDF文件
• ✅ 节省测试成本

2.4 实时进度跟踪

动态计算预估剩余时间：

const avgTimePerItem = elapsed / processedCount;
const remainingItems = totalCount - processedCount;
const estimatedRemainingTime = avgTimePerItem * remainingItems;

用户体验：

• ✅ 前端可轮询显示进度
• ✅ 显示预估完成时间
• ✅ 实时显示成本统计

3. 集成测试

创建了完整的集成测试脚本：

测试场景：

1. ✅ 准备测试数据（查找项目和文献）
2. ✅ 创建并处理任务（测试模式，3篇文献，2并发）
3. ✅ 轮询任务进度（每5秒）
4. ✅ 查询任务结果（分页，排序）
5. ✅ 更新人工复核决策

测试文件：

• service-integration-test.ts (约200行)

运行方式：

cd backend
npx ts-node src/modules/asl/fulltext-screening/services/__tests__/service-integration-test.ts

4. 产出

代码：

• ✅ FulltextScreeningService.ts (约700行)
• ✅ 集成测试脚本 (约200行)
• ✅ TypeScript类型定义完整
• ✅ 代码注释详细

依赖：

• ✅ 安装 p-queue 库

质量：

• ✅ 无Linter错误
• ✅ 完整的错误处理
• ✅ 详细的日志记录

---

📊 Day 4 总体统计

时间分配

阶段	任务	耗时	状态
上午	数据库设计	1h	✅
	Schema设计（3个模型）	30min	✅
	手动SQL迁移	20min	✅
	Schema隔离验证	10min	✅
	文档编写（迁移状态说明）	30min	✅
	文档更新（设计文档、状态文档）	20min	✅
下午	批处理服务开发	2h	✅
	服务核心逻辑	1h	✅
	集成测试脚本	30min	✅
	代码审查与优化	30min	✅
合计		3h	✅

代码产出

类别	文件	行数	说明
核心服务	FulltextScreeningService.ts	~700	批处理服务
测试	service-integration-test.ts	~200	集成测试
数据库	manual_fulltext_screening.sql	141	迁移脚本
文档	数据库迁移状态说明	435	详细记录
文档	Day 4开发记录	~800	本文档
合计		~2,276

功能完成度

功能模块	完成度	说明
数据库设计	100% ✅	3个表，13个新字段
数据库迁移	100% ✅	手动SQL，安全执行
任务创建与调度	100% ✅	支持并发控制
单篇文献处理	100% ✅	集成所有验证器
进度跟踪	100% ✅	实时更新，预估时间
容错处理	100% ✅	重试、降级、继续
查询接口	100% ✅	进度、结果、决策
集成测试	100% ✅	端到端测试脚本

---

🎯 关键决策

1. 云原生存储方案 ✅

决策：全文内容存储在OSS/Dify，数据库只存引用

理由：

• 符合《云原生开发规范》
• 避免数据库膨胀
• 支持大规模扩展

实现：

• full_text_storage_type - 存储类型（oss/dify）
• full_text_storage_ref - 存储引用（key或ID）
• full_text_url - 访问URL

2. 手动SQL迁移策略 ✅

决策：不使用 prisma migrate，而是手动编写SQL脚本

理由：

• Prisma Migrate会尝试删除 public schema中的重复表
• 可能影响其他模块（AIA、PKB、Platform）
• 手动SQL更安全、可控、可审计

原则：

• "管好自己"：只操作 asl_schema
• 不动 public schema，不影响其他模块

3. 测试模式设计 ✅

决策：支持 skipExtraction: true 测试模式

理由：

• 快速验证服务逻辑
• 无需准备真实PDF文件
• 节省测试成本和时间

实现：

if (config.skipExtraction) {
  fullText = `# ${title}\n\n## Abstract\n${abstract}`;
}

4. 并发控制策略 ✅

决策：使用 p-queue，默认并发3

理由：

• 提速3倍（相比串行处理）
• 避免触发API限流
• 自动排队，优雅控制

配置：

const queue = new PQueue({ concurrency: 3 });

---

🐛 遇到的问题与解决

问题1：数据库迁移冲突

问题：prisma db push 检测到会删除 public schema中的表

现象：

⚠️  There might be data loss when applying the changes:
  • You are about to drop the `users` table, which is not empty (2 rows).
  • You are about to drop the `projects` table, which is not empty (2 rows).

根因：

• 历史遗留问题：部分模块的表创建在 public schema
• Prisma Migrate会尝试同步所有schema

解决方案：

1. 不使用 prisma migrate 或 prisma db push
2. 编写手动SQL脚本，只操作 asl_schema
3. 执行：Get-Content xxx.sql | docker exec -i postgres psql ...
4. 验证：\dt asl_schema.*

预防措施：

• 未来继续使用手动SQL迁移
• 明确记录在文档中
• 提醒其他模块开发者

问题2：Prisma Client类型生成

问题：修改Schema后，Prisma Client类型未更新

解决：

npx prisma generate

预防措施：

• 每次修改Schema后立即执行
• 加入迁移流程文档

---

📚 相关文档

本次更新的文档：

1. 数据库迁移状态说明 ← 新建
2. 数据库设计文档 ← 更新v3.0
3. 模块当前状态与开发指南 ← 更新v1.2
4. 技术债务清单 ← 更新债务7状态
5. 全文复筛开发计划 ← 更新Day 4进度

参考的规范文档：

1. 云原生开发规范
2. 数据库架构说明
3. 系统当前状态与开发指南

---

🚀 下一步计划

Day 5：后端API开发（预计1天）

任务清单：

1. 创建 FulltextScreeningController.ts
   • createTask() - 创建任务
   • getTaskProgress() - 获取进度
   • getTaskResults() - 获取结果列表
   • getResultDetail() - 获取结果详情
   • updateDecision() - 人工审核决策
2. 创建 fulltext-screening.ts 路由
3. 集成到Fastify应用
4. API测试（Postman或集成测试）
5. 错误处理完善

预计产出：

• 5个API接口
• API文档
• 后端完成✅

---

🎉 总结

Day 4核心成果：

• ✅ 完成数据库设计（云原生架构）
• ✅ 完成数据库迁移（安全执行，无影响其他模块）
• ✅ 完成批处理服务开发（700行核心代码）
• ✅ 完成集成测试（端到端验证）
• ✅ 完成详细文档（5篇文档更新）

技术亮点：

• ✅ 云原生存储方案（全文存OSS/Dify）
• ✅ 手动SQL迁移策略（安全可控）
• ✅ 并发控制（p-queue，提速3倍）
• ✅ 容错机制（重试、降级、继续）
• ✅ 测试模式（快速验证）

质量保障：

• ✅ Schema隔离100%正确（所有表在asl_schema）
• ✅ 代码无Linter错误
• ✅ 完整的错误处理和日志
• ✅ 详细的文档记录

开发效率：

• ⏱️ 上午1h完成数据库设计与迁移
• ⏱️ 下午2h完成批处理服务开发
• ⏱️ 合计3h完成Day 4全部任务

MVP进度：

• Week 1：50% → 75% ✅
• Day 1-3：通用能力层完成 ✅
• Day 4：批处理服务完成 ✅
• Day 5：API开发（下一步）

---

开发人员：ASL开发团队
文档编写时间：2025-11-23
文档版本：v1.0