feat(rag): add shared retriever for audit pipeline

docs/合同模板搜索合同起草/合同模板上传与地区隔离改造方案.md

@@ -0,0 +1,470 @@
+# 合同模板上传与地区隔离改造方案
+
+## 1. 背景
+
+当前合同模板模块已具备以下只读能力：
+
+- `/api/v3/contract-templates/categories`
+- `/api/v3/contract-templates`
+- `/api/v3/contract-templates/search`
+- `/api/v3/contract-templates/{id}`
+
+当前实现可以支撑模板分类、列表、搜索、详情展示，但还不具备正式的模板管理上传能力。现阶段新增需求包括：
+
+- 在 `/contract-template/list` 页面支持上传合同模板
+- 新模板必须存入 `leaudit_platform` 主库，不再依赖旧项目库
+- 模板数据需要支持地区区分
+- 模板数据需要支持完整审计字段
+- 模板数据需要支持软删除
+- 前端与后端需统一按当前 LeAudit/FastAPI 风格实现
+
+本方案仅覆盖“合同模板管理与上传”，不扩展“合同起草”独立业务模块。
+
+## 2. 现状问题
+
+### 2.1 数据表不足
+
+当前 `contract_templates` 仅包含：
+
+- `template_code`
+- `title`
+- `category_id`
+- `description`
+- `file_path`
+- `file_format`
+- `pdf_file_path`
+- `is_featured`
+- `created_at`
+- `updated_at`
+
+存在以下问题：
+
+- 没有 `region`，无法做地区隔离
+- 没有 `created_by / updated_by`，无法追踪操作者
+- 没有 `deleted_at`，无法软删除
+- 没有 `original_file_name / mime_type / file_size`，无法完整表达上传文件元数据
+- 当前唯一约束仅为 `template_code` 全局唯一，不适合多地区模板管理
+
+### 2.2 读接口缺少统一数据范围控制
+
+当前合同模板读接口没有套用平台现有“按用户地区可见”的规则，无法满足：
+
+- 省级管理员查看全部或指定地区
+- 地市管理员仅查看本地区及公共模板
+- 非管理用户限制可见范围
+
+### 2.3 OSS 路径未带地区
+
+当前 `BuildContractTemplateKey()` 只按：
+
+- 分类
+- 模板编码
+- 文件角色
+
+生成路径，未带 `region`，多地区下会出现路径命名冲突与后期归档困难。
+
+### 2.4 权限不足
+
+当前仅具备读权限：
+
+- `contract_template:list:read`
+- `contract_template:search:read`
+- `contract_template:detail:read`
+
+尚未具备：
+
+- 上传创建权限
+- 编辑权限
+- 删除权限
+
+## 3. 设计目标
+
+本次改造目标如下：
+
+1. 为合同模板模块补齐上传能力
+2. 按地区隔离模板数据
+3. 支持审计字段与软删除
+4. 保持与现有 `govdoc` / `document` 模块相同的权限和地区控制风格
+5. 新上传模板统一走新 OSS 路径规范
+6. 不影响现有搜索、列表、详情页面的继续使用
+
+## 4. 数据模型设计
+
+### 4.1 `contract_categories`
+
+分类暂不做地区化，保留为全局字典表，但补齐审计与软删除字段。
+
+建议字段：
+
+- `id BIGSERIAL/SERIAL PRIMARY KEY`
+- `name VARCHAR(100) NOT NULL`
+- `icon VARCHAR(100) NULL`
+- `description TEXT NULL`
+- `sort_order INTEGER NOT NULL DEFAULT 0`
+- `created_by BIGINT NULL`
+- `created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()`
+- `updated_by BIGINT NULL`
+- `updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()`
+- `deleted_at TIMESTAMPTZ NULL`
+
+索引建议：
+
+- `UNIQUE INDEX uq_contract_categories_name_active ON contract_categories(name) WHERE deleted_at IS NULL`
+- `INDEX idx_contract_categories_sort_active ON contract_categories(sort_order) WHERE deleted_at IS NULL`
+
+### 4.2 `contract_templates`
+
+模板表补齐地区、上传元数据、审计字段、软删除字段。
+
+建议字段：
+
+- `id BIGSERIAL PRIMARY KEY`
+- `template_code VARCHAR(50) NOT NULL`
+- `title VARCHAR(200) NOT NULL`
+- `category_id INTEGER NOT NULL REFERENCES contract_categories(id)`
+- `region VARCHAR(50) NOT NULL DEFAULT '省级'`
+- `description TEXT NULL`
+- `file_path VARCHAR(500) NULL`
+- `pdf_file_path VARCHAR(500) NULL`
+- `file_format VARCHAR(10) NOT NULL`
+- `original_file_name VARCHAR(500) NOT NULL DEFAULT ''`
+- `mime_type VARCHAR(200) NULL`
+- `file_size BIGINT NOT NULL DEFAULT 0`
+- `pdf_file_size BIGINT NULL`
+- `is_featured BOOLEAN NOT NULL DEFAULT FALSE`
+- `created_by BIGINT NULL`
+- `created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()`
+- `updated_by BIGINT NULL`
+- `updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()`
+- `deleted_at TIMESTAMPTZ NULL`
+
+索引建议：
+
+- `UNIQUE INDEX uq_contract_templates_region_code_active ON contract_templates(region, template_code) WHERE deleted_at IS NULL`
+- `INDEX idx_contract_templates_region_active ON contract_templates(region) WHERE deleted_at IS NULL`
+- `INDEX idx_contract_templates_category_active ON contract_templates(category_id) WHERE deleted_at IS NULL`
+- `INDEX idx_contract_templates_updated_active ON contract_templates(updated_at DESC) WHERE deleted_at IS NULL`
+
+### 4.3 软删除策略
+
+所有列表、搜索、详情查询统一增加：
+
+- `deleted_at IS NULL`
+
+删除接口只做：
+
+- `deleted_at = NOW()`
+- `updated_at = NOW()`
+- `updated_by = 当前用户`
+
+本期不执行物理删除 OSS 对象，避免误删无法恢复。
+
+## 5. 地区隔离策略
+
+### 5.1 用户上下文来源
+
+复用现有用户上下文判断逻辑，基于：
+
+- `sso_users.area`
+- 用户角色 `super_admin / provincial_admin / admin / 普通用户`
+
+参考现有实现：
+
+- `documentServiceImpl._getCurrentUserContext`
+- `documentServiceImpl._buildDocumentScopeFilters`
+- `govdocServiceImpl._resolve_upload_region`
+
+### 5.2 模板可见性规则
+
+采用“省级公共模板 + 地区私有模板”的一层可见性模型：
+
+- `region = '省级'`：全局公共模板
+- `region = 用户地区`：当前地区私有模板
+
+读取规则：
+
+- `super_admin / provincial_admin`
+  - 默认可查看全部地区
+  - 支持显式传 `region` 做筛选
+- `admin`
+  - 默认可查看 `('省级', 自己地区)` 模板
+  - 若传入其他地区参数，返回空或直接拒绝
+- 普通用户
+  - 默认可查看 `('省级', 自己地区)` 模板
+  - 不允许跨地区筛选
+
+### 5.3 上传地区规则
+
+上传接口中，`region` 使用以下规则解析：
+
+- `super_admin / provincial_admin`
+  - 可选择上传到任意地区，含 `省级`
+- `admin`
+  - 只能上传到自己地区
+- 普通用户
+  - 默认不开放上传权限
+
+### 5.4 为什么本期不做“地区模板覆盖省级模板”
+
+本期不做同 `template_code` 的覆盖优先级逻辑，原因：
+
+- 列表/搜索会出现同编码多条模板
+- 详情页需要增加“实际命中版本”优先级
+- 后续还会影响起草、下载、预览引用
+
+本期先做“地区隔离 + 独立模板记录”，后续若业务明确需要，再做二期“本地覆盖省级模板”。
+
+## 6. OSS 存储设计
+
+### 6.1 新路径规范
+
+建议合同模板 OSS key 改为：
+
+`contract-templates/{region}/{category}/{template_code}/{role}__{filename}`
+
+例如：
+
+- `contract-templates/省级/房屋租赁/HT-RL-001/source__房屋租赁合同.docx`
+- `contract-templates/梅州/房屋租赁/HT-RL-001/preview__房屋租赁合同.pdf`
+
+### 6.2 文件角色
+
+- 主模板文件：`source`
+- 预览 PDF 文件：`preview`
+
+### 6.3 老数据策略
+
+老数据短期不强制迁移到新地区路径：
+
+- 已有历史数据继续可读
+- 新上传统一走新路径
+
+后续若需要统一整洁，再单独执行历史迁移脚本，将存量模板迁到 `contract-templates/省级/...`
+
+## 7. 接口设计
+
+### 7.1 新增接口
+
+#### 7.1.1 上传模板
+
+- `POST /api/v3/contract-templates`
+- 权限：`contract_template:create`
+- Content-Type：`multipart/form-data`
+
+表单字段：
+
+- `title: str`
+- `template_code: str`
+- `category_id: int`
+- `region: str`
+- `description: str | None`
+- `is_featured: bool`
+- `file: UploadFile`
+- `pdf_file: UploadFile | None`
+
+返回：
+
+- 模板基础信息
+- 文件路径
+- 地区信息
+- 审计时间
+
+#### 7.1.2 更新模板
+
+- `PUT /api/v3/contract-templates/{id}`
+- 权限：`contract_template:update`
+
+本期先可只支持元数据更新，文件替换可选一起补。
+
+#### 7.1.3 删除模板
+
+- `DELETE /api/v3/contract-templates/{id}`
+- 权限：`contract_template:delete`
+
+行为：
+
+- 软删除，不物理删 OSS
+
+### 7.2 既有接口增强
+
+#### 7.2.1 分类接口
+
+- 过滤 `deleted_at IS NULL`
+- 返回 `template_count` 时只统计当前用户可见地区且未删除模板
+
+#### 7.2.2 列表接口
+
+新增可选参数：
+
+- `region`
+
+逻辑：
+
+- 仅返回当前用户可见模板
+- 默认按 `updated_at DESC`
+
+#### 7.2.3 搜索接口
+
+新增可选参数：
+
+- `region`
+
+逻辑：
+
+- 仅搜索当前用户可见模板
+
+#### 7.2.4 详情接口
+
+逻辑：
+
+- 校验模板存在
+- 校验模板未删除
+- 校验当前用户对模板所在地区可见
+
+## 8. 权限设计
+
+新增权限：
+
+- `contract_template:create`
+- `contract_template:update`
+- `contract_template:delete`
+
+保留权限：
+
+- `contract_template:list:read`
+- `contract_template:search:read`
+- `contract_template:detail:read`
+
+角色建议：
+
+- `super_admin`
+  - 全部读写删
+- `provincial_admin`
+  - 全部读写删
+- `admin`
+  - 读
+  - 上传
+  - 编辑本地区模板
+  - 删除本地区模板
+- 普通用户
+  - 仅按需开放读权限
+
+## 9. 前端设计
+
+### 9.1 列表页入口
+
+在 `/contract-template/list` 页面右上角增加：
+
+- “上传模板”按钮
+
+仅有 `contract_template:create` 权限时展示。
+
+### 9.2 上传弹窗字段
+
+- 模板标题
+- 模板编码
+- 模板分类
+- 所属地区
+- 模板简介
+- 是否推荐
+- 模板主文件
+- 预览 PDF 文件（可选）
+
+### 9.3 前端交互规则
+
+- `admin` 用户地区默认锁定为自己地区
+- `provincial_admin` 可选择地区
+- 上传成功后：
+  - 提示成功
+  - 关闭弹窗
+  - 刷新当前列表
+- 上传失败时：
+  - 表单级错误走 toast
+  - 403 需给出明确无权限提示
+
+### 9.4 API 封装
+
+在 `lib/api/contract-template/index.ts` 增加：
+
+- `createContractTemplate`
+- `updateContractTemplate`
+- `deleteContractTemplate`
+
+上传使用 `FormData` 直接请求后端，不走多余代理语义。
+
+## 10. 数据迁移方案
+
+### 10.1 老表扩展
+
+通过增量 SQL：
+
+- 新增缺失列
+- 回填默认值
+- 修正索引
+- 增加 `updated_at` 触发器
+
+### 10.2 老数据回填
+
+建议回填：
+
+- `region = '省级'`
+- `original_file_name = ''` 或按旧路径推导
+- `file_size = 0`
+- `deleted_at = NULL`
+
+### 10.3 回滚策略
+
+若上传接口上线后发现问题：
+
+- 可先关闭前端上传入口
+- 已有新字段与索引保持兼容，不影响只读能力
+
+## 11. 开发步骤
+
+### 阶段 1：基础设施
+
+1. 扩展合同模板 SQL
+2. 扩展 RBAC seed
+3. 更新 OSS 路径工具
+
+### 阶段 2：后端接口
+
+1. 增加 DTO/VO
+2. 扩展 Service 接口
+3. 实现上传、更新、删除
+4. 为列表、搜索、详情补地区过滤
+
+### 阶段 3：前端页面
+
+1. 扩展 API 客户端
+2. 在列表页增加上传按钮和弹窗
+3. 接地区选择与权限控制
+
+### 阶段 4：验证
+
+1. 省级管理员上传省级模板
+2. 地市管理员上传本地区模板
+3. 非本地区参数校验
+4. 403 提示
+5. 列表、搜索、详情联调
+
+## 12. 本期明确不做
+
+- 合同起草模块重构
+- 模板占位符自动解析
+- docx 自动转 pdf
+- 地区模板覆盖省级模板优先级
+- 模板版本管理
+- OSS 历史模板统一搬迁
+
+## 13. 推荐本期交付范围
+
+建议本期落地以下完整闭环：
+
+1. 合同模板表完成地区化、审计化、软删除改造
+2. 合同模板后端新增上传接口
+3. 合同模板列表页新增上传弹窗
+4. 合同模板读接口支持地区可见性控制
+5. 权限种子补齐
+
+这套范围足以让合同模板模块从“只读展示”升级为“可管理上传的地区化模板库”。

fastapi_modules/fastapi_leaudit/controllers/contractTemplateController.py

@@ -79,7 +79,7 @@ class ContractTemplateController(BaseController):
             payload: dict = Depends(verify_access_token),
         ):
             if not await self._check_permission(int(payload["user_id"]), ["contract_template:create:write"]):
-                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有上传合同模板权限", "data": None})
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前仅允许地区管理员上传合同模板", "data": None})
             body = ContractTemplateCreateDTO(
                 title=title,
                 template_code=template_code,

fastapi_modules/fastapi_leaudit/controllers/crossReviewController.py

@@ -17,6 +17,7 @@ from fastapi_modules.fastapi_leaudit.domian.Dto.crossReviewDto import (
     CrossReviewTaskQueryDTO,
 )
 from fastapi_modules.fastapi_leaudit.domian.vo.crossReviewVo import (
+    CrossReviewTaskDocumentAppendVO,
     CrossReviewPendingVotesVO,
     CrossReviewPermissionVO,
     CrossReviewProposalCancelVO,
@@ -158,6 +159,30 @@ class CrossReviewController(BaseController):
             )
             return Result.success(data=Data, message="交叉评查任务文档上传成功")
 
+        @self.router.post("/tasks/{TaskId}/documents/{DocumentId}/attachments", response_model=Result[CrossReviewTaskDocumentAppendVO])
+        async def AppendTaskDocumentAttachments(
+            TaskId: int,
+            DocumentId: int,
+            files: list[UploadFile] = File(..., description="附件文件列表"),
+            remark: str | None = Form(None, description="本次追加附件备注"),
+            payload: dict[str, Any] = Depends(verify_access_token),
+        ):
+            """为交叉评查任务文档追加附件，并生成同版本链新版本。"""
+            if not await self._check_permission(int(payload["user_id"]), [self._PERMISSIONS["document_complete"]]):
+                return JSONResponse(status_code=403, content={"code": 403, "msg": "当前用户没有交叉评查任务追加附件权限", "data": None})
+            filePayloads: list[tuple[str, bytes, str | None]] = []
+            for file in files:
+                content = await file.read()
+                filePayloads.append((file.filename or "attachment.bin", content, file.content_type))
+            Data = await self.CrossReviewService.AppendTaskDocumentAttachments(
+                CurrentUserId=int(payload["user_id"]),
+                TaskId=TaskId,
+                DocumentId=DocumentId,
+                Files=filePayloads,
+                Remark=remark,
+            )
+            return Result.success(data=Data, message="附件追加成功")
+
         @self.router.post("/proposals", response_model=Result[CrossReviewProposalCreateVO])
         async def CreateProposal(
             Body: CrossReviewProposalCreateDTO,

fastapi_modules/fastapi_leaudit/controllers/documentController.py

@@ -221,6 +221,8 @@ class DocumentController(BaseController):
         async def AppendAttachments(
             DocumentId: int,
             files: list[UploadFile] = File(..., description="附件文件列表"),
+            mergeMode: str = Form("new", description="附件合并模式：overwrite/new"),
+            remark: str | None = Form(None, description="本次追加附件备注"),
             payload: dict[str, Any] = Depends(verify_access_token),
         ):
             """为现有文档追加附件（带数据隔离校验）。"""
@@ -232,6 +234,8 @@ class DocumentController(BaseController):
                 CurrentUserId=int(payload["user_id"]),
                 Id=DocumentId,
                 Files=filePayloads,
+                MergeMode=mergeMode,
+                Remark=remark,
             )
             return Result.success(data=Data, message="附件上传成功")
 

fastapi_modules/fastapi_leaudit/domian/vo/crossReviewVo.py

@@ -74,6 +74,39 @@ class CrossReviewTaskDocumentVO(BaseModel):
     fullScore: float = Field(0, description="满分")
     scoreSummary: str = Field("", description="得分摘要")
     scorePercent: float = Field(0, description="得分百分比")
+    historyVersions: list["CrossReviewTaskHistoryVersionVO"] = Field(default_factory=list, description="历史版本列表")
+
+
+class CrossReviewTaskHistoryVersionVO(BaseModel):
+    """任务文档历史版本项。"""
+
+    documentId: int = Field(..., description="文档ID")
+    name: str = Field("", description="文档名称")
+    documentNumber: str | None = Field(None, description="文号")
+    typeId: int | None = Field(None, description="文档类型ID")
+    typeName: str | None = Field(None, description="文档类型名称")
+    processingStatus: str | None = Field(None, description="处理状态")
+    versionNo: int = Field(1, description="版本号")
+    auditStatus: int = Field(0, description="任务内完成状态")
+    createdAt: datetime | None = Field(None, description="创建时间")
+    fileSize: int = Field(0, description="文件大小（字节）")
+    path: str | None = Field(None, description="文件存储路径")
+    uploadTime: datetime | None = Field(None, description="上传时间")
+    fileExt: str | None = Field(None, description="文件扩展名")
+    totalEvaluationPoints: int = Field(0, description="总评查点数")
+    passCount: int = Field(0, description="通过数")
+    warningCount: int = Field(0, description="警告数")
+    errorCount: int = Field(0, description="错误数")
+    manualCount: int = Field(0, description="人工审核数")
+    issueCount: int = Field(0, description="问题总数")
+    warningMessages: list[str] = Field(default_factory=list, description="警告消息")
+    errorMessages: list[str] = Field(default_factory=list, description="错误消息")
+    issueMessages: list[str] = Field(default_factory=list, description="问题消息")
+    manualMessages: list[str] = Field(default_factory=list, description="人工审核消息")
+    finalScore: float = Field(0, description="最终得分")
+    fullScore: float = Field(0, description="满分")
+    scoreSummary: str = Field("", description="得分摘要")
+    scorePercent: float = Field(0, description="得分百分比")
 
 
 class CrossReviewTaskDocumentPageVO(BaseModel):
@@ -194,3 +227,15 @@ class CrossReviewTaskDocumentUploadVO(BaseModel):
     documentId: int = Field(..., description="文档ID")
     auditStatus: int = Field(0, description="任务内评查状态")
     processingStatus: str | None = Field(None, description="文档处理状态")
+
+
+class CrossReviewTaskDocumentAppendVO(BaseModel):
+    """交叉评查任务文档追加附件结果。"""
+
+    taskId: int = Field(..., description="任务ID")
+    originalDocumentId: int = Field(..., description="原文档ID")
+    documentId: int = Field(..., description="新版本文档ID")
+    versionNo: int = Field(..., description="新版本号")
+    versionGroupKey: str = Field("", description="版本组Key")
+    auditStatus: int = Field(0, description="任务内评查状态")
+    processingStatus: str | None = Field(None, description="文档处理状态")

fastapi_modules/fastapi_leaudit/domian/vo/documentVo.py

@@ -59,6 +59,7 @@ class DocumentHistoryVersionVO(BaseModel):
     fileName: str | None = Field(None, description="文件名")
     fileExt: str | None = Field(None, description="文件扩展名")
     fileSize: int | None = Field(None, description="文件大小")
+    ossUrl: str | None = Field(None, description="OSS 路径")
     processingStatus: str | None = Field(None, description="处理状态")
     runStatus: str | None = Field(None, description="最新运行状态")
     resultStatus: str | None = Field(None, description="最新结果状态")

fastapi_modules/fastapi_leaudit/leaudit_bridge/pipeline.py

@@ -24,6 +24,7 @@ from leaudit.ocr.base import BaseOCRClient
 from leaudit.ocr.models import OcrResult
 
 from fastapi_modules.fastapi_leaudit.leaudit_bridge.storage_adapter import StorageAdapter
+from fastapi_modules.fastapi_leaudit.rag_engine.retriever import RagRetriever
 
 log = logging.getLogger(__name__)
 
@@ -89,10 +90,12 @@ class LauditPipeline:
         ocr_client: BaseOCRClient,
         llm_client: BaseLLMClient | None = None,
         storage_adapter: StorageAdapter | None = None,
+        rag_retriever: RagRetriever | None = None,
     ) -> None:
         self.ocr_client = ocr_client
         self.llm_client = llm_client
         self.storage = storage_adapter or StorageAdapter()
+        self.rag_retriever = rag_retriever or RagRetriever()
 
     async def run(
         self,
@@ -219,6 +222,7 @@ class LauditPipeline:
             visual_manifest=visual_manifest,
             phase=detected_phase,
             external_mocks=external_mocks,
+            retriever=self.rag_retriever,
         )
         timing["evaluation"] = round(time.time() - t0, 2)
         log.info(

fastapi_modules/fastapi_leaudit/leaudit_bridge/tasks.py

@@ -532,8 +532,30 @@ async def _load_run_context(run_id: int) -> dict[str, Any]:
         if not document_file:
             raise ValueError(f"未找到 document_file_id={run.documentFileId} 对应的文件记录")
 
+        mergedPdfResult = await session.execute(
+            select(LeauditDocumentFile)
+            .where(
+                LeauditDocumentFile.documentId == document.Id,
+                LeauditDocumentFile.isActive.is_(True),
+                LeauditDocumentFile.fileRole == "merged_pdf",
+            )
+            .order_by(LeauditDocumentFile.Id.desc())
+            .limit(1)
+        )
+        mergedPdfFile = mergedPdfResult.scalar_one_or_none()
+        effective_document_file = mergedPdfFile or document_file
+
         resolver = FileSourceResolver()
-        payload = await resolver.ResolvePayload(document_file)
+        payload = await resolver.ResolvePayload(effective_document_file)
+        attachmentFiles: list[LeauditDocumentFile] = []
+        attachmentPayloads = []
+        if mergedPdfFile is not None:
+            attachmentFiles = []
+        elif str(getattr(document_file, "fileRole", "") or "").lower() != "primary":
+            attachmentFiles = []
+        elif str(getattr(document_file, "fileExt", "") or "").lower() not in {"pdf", "docx"}:
+            attachmentFiles = []
+        else:
             attachmentResult = await session.execute(
                 select(LeauditDocumentFile)
                 .where(

fastapi_modules/fastapi_leaudit/rag_engine/retriever.py

@@ -0,0 +1,470 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from typing import Any, Awaitable, Callable
+
+import httpx
+from sqlalchemy import text
+
+from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
+from fastapi_common.fastapi_common_web.domain.responses import StatusCodeEnum
+from fastapi_common.fastapi_common_web.exception.LeauditException import LeauditException
+from fastapi_modules.fastapi_leaudit.rag_engine.chroma_client import get_chroma
+from fastapi_modules.fastapi_leaudit.rag_engine.config import RAG_CONFIG, build_openai_embeddings_url
+
+
+EmbedTexts = Callable[[list[str], str], Awaitable[list[list[float]]] | list[list[float]]]
+
+
+@dataclass
+class RagRetrieveResult:
+  rag_context: str = ""
+  rag_resources: list[dict[str, Any]] = field(default_factory=list)
+  chunks: list[dict[str, Any]] = field(default_factory=list)
+  dataset_name: str = ""
+
+  def as_dict(self) -> dict[str, Any]:
+    return {
+      "rag_context": self.rag_context,
+      "rag_resources": self.rag_resources,
+    }
+
+
+class RagRetriever:
+  def __init__(
+    self,
+    *,
+    chroma_client: Any | None = None,
+    embed_texts: EmbedTexts | None = None,
+    hydrate_documents: bool = True,
+  ) -> None:
+    self._chroma_client = chroma_client
+    self._embed_texts_override = embed_texts
+    self._hydrate_documents_enabled = hydrate_documents
+
+  async def retrieve(
+    self,
+    query: str,
+    collection_name: str | None = None,
+    dataset_id: int | None = None,
+    top_k: int = 5,
+    source_names: list[str] | None = None,
+  ) -> RagRetrieveResult:
+    query_text = (query or "").strip()
+    if not query_text:
+      return RagRetrieveResult()
+
+    top_k = max(1, int(top_k or 5))
+    dataset = await self._load_dataset(dataset_id) if dataset_id else None
+    resolved_collection = (collection_name or (dataset or {}).get("collection_name") or "").strip()
+    if not resolved_collection:
+      return RagRetrieveResult(dataset_name=str((dataset or {}).get("name") or ""))
+
+    retrieval_model = (dataset or {}).get("retrieval_model") or {}
+    if dataset and not collection_name:
+      top_k = max(1, int(retrieval_model.get("top_k") or top_k))
+    score_threshold = self._resolve_score_threshold(retrieval_model)
+    dataset_name = str((dataset or {}).get("name") or "")
+    embedding_model = str((dataset or {}).get("embedding_model") or "")
+
+    chunks: list[dict[str, Any]] = []
+    try:
+      chunks = await self._vector_retrieve(
+        query=query_text,
+        collection_name=resolved_collection,
+        dataset_name=dataset_name,
+        embedding_model=embedding_model,
+        top_k=top_k,
+        score_threshold=score_threshold,
+        source_names=source_names,
+      )
+    except Exception:
+      chunks = []
+
+    if not chunks:
+      try:
+        chunks = await self._keyword_retrieve_context(
+          dataset_id=dataset_id,
+          collection_name=resolved_collection,
+          dataset_name=dataset_name,
+          query=query_text,
+          top_k=top_k,
+          score_threshold=score_threshold,
+          source_names=source_names,
+        )
+      except Exception:
+        chunks = []
+
+    if dataset_id and self._hydrate_documents_enabled:
+      chunks = await self._hydrate_document_hits(dataset_id, chunks)
+
+    chunks = chunks[:top_k]
+    return RagRetrieveResult(
+      rag_context=self._build_context(chunks),
+      rag_resources=self.build_sources(chunks, dataset_name),
+      chunks=chunks,
+      dataset_name=dataset_name,
+    )
+
+  async def _load_dataset(self, dataset_id: int | None) -> dict[str, Any] | None:
+    if not dataset_id:
+      return None
+    async with GetAsyncSession() as session:
+      row = (
+        await session.execute(
+          text(
+            """
+            SELECT id, name, collection_name, retrieval_model, embedding_model
+            FROM rag_dataset
+            WHERE id = :dataset_id AND deleted_at IS NULL
+            LIMIT 1
+            """
+          ),
+          {"dataset_id": dataset_id},
+        )
+      ).mappings().first()
+    return dict(row) if row else None
+
+  def _resolve_score_threshold(self, retrieval_model: dict[str, Any]) -> float | None:
+    if not retrieval_model.get("score_threshold_enabled"):
+      return None
+    try:
+      return float(retrieval_model.get("score_threshold"))
+    except (TypeError, ValueError):
+      return None
+
+  async def _vector_retrieve(
+    self,
+    *,
+    query: str,
+    collection_name: str,
+    dataset_name: str,
+    embedding_model: str,
+    top_k: int,
+    score_threshold: float | None,
+    source_names: list[str] | None,
+  ) -> list[dict[str, Any]]:
+    query_embedding = await self._embed_texts([query], embedding_model)
+    collection = self._get_chroma().get_or_create_collection(collection_name)
+    result = collection.query(
+      query_embeddings=query_embedding,
+      n_results=max(top_k, 1),
+      include=["documents", "metadatas", "distances"],
+    )
+    ids = (result.get("ids") or [[]])[0] if result.get("ids") else []
+    docs = (result.get("documents") or [[]])[0]
+    metas = (result.get("metadatas") or [[]])[0]
+    distances = (result.get("distances") or [[]])[0]
+    allowed_sources = self._normalize_source_filter(source_names)
+    chunks: list[dict[str, Any]] = []
+    for idx, doc in enumerate(docs):
+      meta = metas[idx] if idx < len(metas) and isinstance(metas[idx], dict) else {}
+      document_name = str(meta.get("document_name") or meta.get("source") or "")
+      if allowed_sources and document_name not in allowed_sources:
+        continue
+      dist = float(distances[idx]) if idx < len(distances) and distances[idx] is not None else 1.0
+      score = 1.0 / (1.0 + max(dist, 0.0))
+      if score_threshold is not None and score < score_threshold:
+        continue
+      chunks.append(
+        {
+          "id": str(ids[idx] if idx < len(ids) else meta.get("id") or idx),
+          "text": doc or "",
+          "source": meta.get("source") or meta.get("document_name") or dataset_name,
+          "score": score,
+          "chunk_index": int(meta.get("chunk_index") or idx),
+          "document_name": document_name,
+          "document_id": meta.get("document_id"),
+          "page": meta.get("page"),
+        }
+      )
+    return chunks
+
+  async def _embed_texts(self, texts: list[str], model_name: str = "") -> list[list[float]]:
+    if self._embed_texts_override is not None:
+      result = self._embed_texts_override(texts, model_name)
+      if hasattr(result, "__await__"):
+        return await result  # type: ignore[misc]
+      return result
+
+    embed_url = (RAG_CONFIG.get("EMBED_URL") or "").strip() or build_openai_embeddings_url(RAG_CONFIG["LLM_BASE_URL"])
+    embed_key = (RAG_CONFIG.get("EMBED_KEY") or "").strip() or RAG_CONFIG["LLM_API_KEY"]
+    embed_model = model_name or (RAG_CONFIG.get("EMBED_MODEL") or "").strip() or "text-embedding-v4"
+    batch_size = max(1, int(RAG_CONFIG.get("EMBED_BATCH_SIZE") or 10))
+    if not embed_url or not embed_key:
+      raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "未配置可用的向量化服务")
+
+    embeddings: list[list[float]] = []
+    async with httpx.AsyncClient(timeout=120.0) as client:
+      for start in range(0, len(texts), batch_size):
+        batch_texts = texts[start:start + batch_size]
+        try:
+          response = await client.post(
+            embed_url,
+            headers={
+              "Content-Type": "application/json",
+              "Authorization": f"Bearer {embed_key}",
+            },
+            json={"model": embed_model, "input": batch_texts},
+          )
+          response.raise_for_status()
+        except httpx.HTTPStatusError as exc:
+          error_message = exc.response.text.strip() or f"{exc.response.status_code} {exc.response.reason_phrase}"
+          raise LeauditException(
+            StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR,
+            f"向量化服务调用失败: {error_message[:300]}",
+          ) from exc
+
+        payload = response.json()
+        rows = payload.get("data") or []
+        batch_embeddings = [row.get("embedding") for row in rows if isinstance(row, dict) and row.get("embedding")]
+        if len(batch_embeddings) != len(batch_texts):
+          raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "向量化结果数量异常")
+        embeddings.extend(batch_embeddings)
+
+    if len(embeddings) != len(texts):
+      raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "向量化结果数量异常")
+    return embeddings
+
+  async def _keyword_retrieve_context(
+    self,
+    *,
+    dataset_id: int | None,
+    collection_name: str,
+    dataset_name: str,
+    query: str,
+    top_k: int,
+    score_threshold: float | None,
+    source_names: list[str] | None,
+  ) -> list[dict[str, Any]]:
+    collection = self._get_chroma().get_or_create_collection(collection_name)
+    raw = collection.get(include=["documents", "metadatas"])
+    ids = raw.get("ids") or []
+    docs = raw.get("documents") or []
+    metas = raw.get("metadatas") or []
+    allowed_sources = self._normalize_source_filter(source_names)
+
+    terms = self._build_keyword_terms(query)
+    if not terms:
+      return []
+
+    scored_chunks: list[dict[str, Any]] = []
+    for idx, chunk_id in enumerate(ids):
+      doc = docs[idx] if idx < len(docs) else ""
+      meta = metas[idx] if idx < len(metas) and isinstance(metas[idx], dict) else {}
+      document_name = str(meta.get("document_name") or meta.get("source") or "")
+      if allowed_sources and document_name not in allowed_sources:
+        continue
+      score = self._score_keyword_chunk(
+        query=query,
+        terms=terms,
+        content=doc or "",
+        document_name=document_name,
+      )
+      if score <= 0:
+        continue
+      if score_threshold is not None and score < score_threshold:
+        continue
+      scored_chunks.append(
+        {
+          "id": str(chunk_id),
+          "text": doc or "",
+          "source": meta.get("source") or meta.get("document_name") or dataset_name,
+          "score": score,
+          "chunk_index": int(meta.get("chunk_index") or idx),
+          "document_name": document_name,
+          "document_id": meta.get("document_id"),
+          "page": meta.get("page"),
+        }
+      )
+
+    scored_chunks.sort(key=lambda item: (-float(item.get("score") or 0.0), int(item.get("chunk_index") or 0)))
+    return scored_chunks[: max(top_k * 3, top_k)]
+
+  def _build_keyword_terms(self, query: str) -> list[str]:
+    normalized = self._normalize_keyword_query(query)
+    spans = [item.strip() for item in re.findall(r"[\u4e00-\u9fffA-Za-z0-9]+", normalized) if item.strip()]
+    if not spans:
+      return []
+
+    stop_terms = {
+      "什么",
+      "请问",
+      "一下",
+      "有关",
+      "关于",
+      "如何",
+      "哪些",
+      "怎么",
+      "是否",
+      "规定",
+      "办法",
+      "条例",
+      "法律",
+    }
+    terms: list[str] = []
+    for span in spans:
+      if span in stop_terms:
+        continue
+      terms.append(span)
+      if re.fullmatch(r"[\u4e00-\u9fff]+", span):
+        for size in (2, 3, 4):
+          if len(span) > size:
+            for start in range(0, len(span) - size + 1):
+              token = span[start:start + size]
+              if token not in stop_terms:
+                terms.append(token)
+
+    unique_terms: list[str] = []
+    seen: set[str] = set()
+    for term in sorted(terms, key=len, reverse=True):
+      if term and term not in seen:
+        unique_terms.append(term)
+        seen.add(term)
+    return unique_terms[:20]
+
+  def _normalize_keyword_query(self, query: str) -> str:
+    normalized = (query or "").strip().lower()
+    patterns = [
+      "是什么",
+      "什么是",
+      "有哪些",
+      "有什么",
+      "是什么？",
+      "是什么?",
+      "请问",
+      "介绍一下",
+      "解释一下",
+      "帮我分析",
+      "帮我看看",
+    ]
+    for pattern in patterns:
+      normalized = normalized.replace(pattern, " ")
+    return re.sub(r"\s+", " ", normalized).strip()
+
+  def _score_keyword_chunk(self, *, query: str, terms: list[str], content: str, document_name: str) -> float:
+    haystack = f"{document_name}\n{content}".lower()
+    if not haystack:
+      return 0.0
+
+    exact_query = self._normalize_keyword_query(query)
+    if exact_query and exact_query in haystack:
+      return 0.98
+
+    matched_weight = 0.0
+    total_weight = 0.0
+    name_bonus = 0.0
+    for term in terms:
+      weight = float(max(len(term), 1) ** 2)
+      total_weight += weight
+      if term.lower() in haystack:
+        matched_weight += weight
+        if term.lower() in document_name.lower():
+          name_bonus += min(0.15, 0.03 * len(term))
+
+    if total_weight <= 0:
+      return 0.0
+    score = (matched_weight / total_weight) + name_bonus
+    return round(min(score, 0.99), 6)
+
+  def _build_context(self, chunks: list[dict[str, Any]]) -> str:
+    lines: list[str] = []
+    for index, chunk in enumerate(chunks, start=1):
+      document_name = chunk.get("document_name") or chunk.get("source") or "未知来源"
+      text_value = str(chunk.get("text") or "").strip()
+      if not text_value:
+        continue
+      lines.append(f"[{index}] 来源：{document_name}\n{text_value}")
+    return "\n\n".join(lines)
+
+  def build_sources(self, context_chunks: list[dict[str, Any]], dataset_name: str = "") -> list[dict[str, Any]]:
+    return [
+      {
+        "position": index + 1,
+        "dataset_id": str(chunk.get("dataset_id") or ""),
+        "dataset_name": dataset_name,
+        "document_id": str(chunk.get("document_id") or ""),
+        "document_name": chunk.get("document_name") or chunk.get("source", ""),
+        "data_source_type": "upload_file",
+        "segment_id": chunk.get("id", ""),
+        "retriever_from": "rag",
+        "score": round(float(chunk.get("score") or 0.0), 4),
+        "hit_count": chunk.get("hit_count", 0),
+        "word_count": len(chunk.get("text", "")),
+        "segment_position": index + 1,
+        "index_node_hash": "",
+        "content": chunk.get("text", "")[:500],
+        "page": None,
+      }
+      for index, chunk in enumerate(context_chunks)
+    ]
+
+  async def _hydrate_document_hits(self, dataset_id: int, chunks: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    source_names = sorted(
+      {
+        str(chunk.get("document_name") or chunk.get("source") or "").strip()
+        for chunk in chunks
+        if str(chunk.get("document_name") or chunk.get("source") or "").strip()
+      }
+    )
+    if not source_names:
+      return chunks
+
+    async with GetAsyncSession() as session:
+      rows = (
+        await session.execute(
+          text(
+            """
+            SELECT id, original_name, enabled, hit_count
+            FROM rag_document
+            WHERE dataset_id = :dataset_id
+              AND deleted_at IS NULL
+              AND original_name = ANY(:source_names)
+            """
+          ),
+          {
+            "dataset_id": dataset_id,
+            "source_names": source_names,
+          },
+        )
+      ).mappings().all()
+
+    document_map = {str(row["original_name"]): row for row in rows}
+    visible_chunks: list[dict[str, Any]] = []
+    hit_document_ids: list[int] = []
+    for chunk in chunks:
+      source_name = str(chunk.get("document_name") or chunk.get("source") or "").strip()
+      document = document_map.get(source_name)
+      if document and not bool(document.get("enabled")):
+        continue
+      if document:
+        chunk["document_id"] = document["id"]
+        chunk["dataset_id"] = dataset_id
+        chunk["document_name"] = document["original_name"]
+        chunk["hit_count"] = document.get("hit_count") or 0
+        hit_document_ids.append(int(document["id"]))
+      visible_chunks.append(chunk)
+
+    if hit_document_ids:
+      async with GetAsyncSession() as session:
+        async with session.begin():
+          await session.execute(
+            text(
+              """
+              UPDATE rag_document
+              SET hit_count = hit_count + 1,
+                  updated_at = NOW()
+              WHERE id = ANY(:document_ids)
+              """
+            ),
+            {"document_ids": sorted(set(hit_document_ids))},
+          )
+
+    return visible_chunks
+
+  def _normalize_source_filter(self, source_names: list[str] | None) -> set[str]:
+    return {str(name).strip() for name in (source_names or []) if str(name).strip()}
+
+  def _get_chroma(self) -> Any:
+    return self._chroma_client or get_chroma()

fastapi_modules/fastapi_leaudit/services/crossReviewService.py

@@ -10,6 +10,7 @@ from fastapi_modules.fastapi_leaudit.domian.Dto.crossReviewDto import (
     CrossReviewTaskQueryDTO,
 )
 from fastapi_modules.fastapi_leaudit.domian.vo.crossReviewVo import (
+    CrossReviewTaskDocumentAppendVO,
     CrossReviewPendingVotesVO,
     CrossReviewPermissionVO,
     CrossReviewProposalCancelVO,
@@ -117,3 +118,15 @@ class ICrossReviewService(ABC):
     ) -> CrossReviewTaskDocumentUploadVO:
         """向交叉评查任务补传文档。"""
         ...
+
+    @abstractmethod
+    async def AppendTaskDocumentAttachments(
+        self,
+        CurrentUserId: int,
+        TaskId: int,
+        DocumentId: int,
+        Files: list[tuple[str, bytes, str | None]],
+        Remark: str | None = None,
+    ) -> CrossReviewTaskDocumentAppendVO:
+        """为交叉评查任务文档追加附件，并生成同版本链新版本。"""
+        ...

fastapi_modules/fastapi_leaudit/services/documentService.py

@@ -115,6 +115,8 @@ class IDocumentService(ABC):
         CurrentUserId: int,
         Id: int,
         Files: list[tuple[str, bytes, str | None]],
+        MergeMode: str = "new",
+        Remark: str | None = None,
     ) -> DocumentDetailVO:
         """为现有文档追加附件，并执行数据隔离校验。"""
         ...

fastapi_modules/fastapi_leaudit/services/impl/contractTemplateServiceImpl.py

@@ -681,18 +681,20 @@ class ContractTemplateServiceImpl(IContractTemplateService):
                 "area": str(row["area"] or ""),
                 "is_global": bool(row["is_global"]),
                 "can_manage": bool(row["can_manage"]),
+                "is_area_admin": bool(row["can_manage"]) and not bool(row["is_global"]),
             }
         finally:
             if own_session:
                 await session_cm.__aexit__(None, None, None)
 
     def _resolve_upload_region(self, currentUser: dict[str, Any], requestedRegion: str | None) -> str:
+        _ = requestedRegion
         area = str(currentUser["area"] or "").strip()
-        if currentUser["can_manage"]:
+        if not currentUser.get("is_area_admin"):
+            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前仅支持地区管理员上传合同模板")
         if not area:
-                raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前管理员账号未配置地区，无法上传合同模板")
+            raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前地区管理员账号未配置所属地区，无法上传合同模板")
         return area
-        raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前用户没有上传合同模板权限")
 
     async def _ensureContractTemplateSchema(self, session) -> None:
         statements = [

fastapi_modules/fastapi_leaudit/services/impl/crossReviewServiceImpl.py

@@ -26,6 +26,8 @@ from fastapi_modules.fastapi_leaudit.domian.Dto.crossReviewDto import (
     CrossReviewTaskQueryDTO,
 )
 from fastapi_modules.fastapi_leaudit.domian.vo.crossReviewVo import (
+    CrossReviewTaskDocumentAppendVO,
+    CrossReviewTaskHistoryVersionVO,
     CrossReviewPendingProposalVO,
     CrossReviewPendingVotesVO,
     CrossReviewPermissionVO,
@@ -463,14 +465,16 @@ class CrossReviewServiceImpl(ICrossReviewService):
                 "limit": Body.pageSize,
                 "offset": (Body.page - 1) * Body.pageSize,
             }
-            whereClauses = [
+            baseWhereClauses = [
                 "td.task_id = :task_id",
                 "td.delete_time IS NULL",
+                "d.deleted_at IS NULL",
             ]
             if Body.keyword:
-                whereClauses.append("(d.normalized_name ILIKE :keyword OR CAST(d.biz_document_id AS TEXT) ILIKE :keyword)")
+                baseWhereClauses.append("(d.normalized_name ILIKE :keyword OR CAST(d.biz_document_id AS TEXT) ILIKE :keyword)")
                 params["keyword"] = f"%{Body.keyword.strip()}%"
-            whereSql = " AND ".join(whereClauses)
+            baseWhereSql = " AND ".join(baseWhereClauses)
+            latestWhereSql = f"{baseWhereSql} AND COALESCE(d.is_latest_version, false) = true"
 
             total = int(
                 (
@@ -481,7 +485,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
                             FROM leaudit_cross_review_task_documents td
                             JOIN leaudit_documents d
                               ON d.id = td.document_id
-                            WHERE {whereSql}
+                            WHERE {latestWhereSql}
                             """
                         ),
                         params,
@@ -510,7 +514,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
                             ) AS processing_status,
                             d.version_no,
                             d.is_latest_version,
-                            COALESCE(d.version_group_key, '') AS version_group_key,
+                            COALESCE(NULLIF(d.version_group_key, ''), CONCAT('root:', COALESCE(d.root_version_id, d.id)::text)) AS version_group_key,
                             COALESCE(vc.total_versions, 1)::int AS total_versions,
                             d.created_at,
                             td.audit_status,
@@ -549,7 +553,9 @@ class CrossReviewServiceImpl(ICrossReviewService):
                             LIMIT 1
                         ) df ON TRUE
                         LEFT JOIN (
-                            SELECT d2.version_group_key, COUNT(*) AS total_versions
+                            SELECT
+                                COALESCE(NULLIF(d2.version_group_key, ''), CONCAT('root:', COALESCE(d2.root_version_id, d2.id)::text)) AS version_group_key,
+                                COUNT(*) AS total_versions
                             FROM leaudit_documents d2
                             JOIN leaudit_cross_review_task_documents td2
                               ON td2.document_id = d2.id
@@ -557,7 +563,8 @@ class CrossReviewServiceImpl(ICrossReviewService):
                              AND td2.task_id = :task_id
                             WHERE d2.deleted_at IS NULL
                             GROUP BY d2.version_group_key
-                        ) vc ON vc.version_group_key = d.version_group_key
+                                 , COALESCE(d2.root_version_id, d2.id)
+                        ) vc ON vc.version_group_key = COALESCE(NULLIF(d.version_group_key, ''), CONCAT('root:', COALESCE(d.root_version_id, d.id)::text))
                         LEFT JOIN LATERAL (
                             SELECT
                                 COUNT(*)::int AS total_evaluation_points,
@@ -610,7 +617,7 @@ class CrossReviewServiceImpl(ICrossReviewService):
                               AND p.status = 'approved'
                               AND p.delete_time IS NULL
                         ) pd ON TRUE
-                        WHERE {whereSql}
+                        WHERE {latestWhereSql}
                         ORDER BY d.created_at DESC, d.id DESC
                         LIMIT :limit OFFSET :offset
                         """
@@ -619,7 +626,174 @@ class CrossReviewServiceImpl(ICrossReviewService):
                 )
             ).mappings().all()
 
-            items = [
+            latestDocumentIds = [int(row["document_id"]) for row in rows]
+            historyByGroup: dict[str, list[CrossReviewTaskHistoryVersionVO]] = {}
+            if latestDocumentIds:
+                historyRows = (
+                    await session.execute(
+                        text(
+                            """
+                            WITH target_groups AS (
+                                SELECT DISTINCT COALESCE(NULLIF(d.version_group_key, ''), CONCAT('root:', COALESCE(d.root_version_id, d.id)::text)) AS version_group_key
+                                FROM leaudit_cross_review_task_documents td
+                                JOIN leaudit_documents d
+                                  ON d.id = td.document_id
+                                WHERE td.task_id = :task_id
+                                  AND td.delete_time IS NULL
+                                  AND d.id = ANY(:document_ids)
+                            )
+                            SELECT
+                                d.id AS document_id,
+                                COALESCE(d.normalized_name, '') AS name,
+                                CAST(d.biz_document_id AS TEXT) AS document_number,
+                                d.type_id,
+                                COALESCE(
+                                    CASE
+                                        WHEN LOWER(COALESCE(ar.status, '')) IN ('pending', 'queued', 'running', 'retrying')
+                                            THEN ar.status
+                                        ELSE d.processing_status
+                                    END,
+                                    d.processing_status,
+                                    'waiting'
+                                ) AS processing_status,
+                                d.version_no,
+                                d.is_latest_version,
+                                COALESCE(NULLIF(d.version_group_key, ''), CONCAT('root:', COALESCE(d.root_version_id, d.id)::text)) AS version_group_key,
+                                td.audit_status,
+                                d.created_at,
+                                COALESCE(dt.name, '') AS type_name,
+                                COALESCE(df.file_size, 0) AS file_size,
+                                COALESCE(df.file_path, '') AS path,
+                                df.file_upload_time AS upload_time,
+                                COALESCE(df.file_ext, '') AS file_ext,
+                                COALESCE(es.total_evaluation_points, 0) AS total_evaluation_points,
+                                COALESCE(es.pass_count, 0) AS pass_count,
+                                COALESCE(es.warning_count, 0) AS warning_count,
+                                COALESCE(es.error_count, 0) AS error_count,
+                                COALESCE(es.manual_count, 0) AS manual_count,
+                                COALESCE(es.issue_count, 0) AS issue_count,
+                                COALESCE(es.warning_messages, ARRAY[]::text[]) AS warning_messages,
+                                COALESCE(es.error_messages, ARRAY[]::text[]) AS error_messages,
+                                COALESCE(es.issue_messages, ARRAY[]::text[]) AS issue_messages,
+                                COALESCE(es.manual_messages, ARRAY[]::text[]) AS manual_messages,
+                                COALESCE(es.final_score, 0) AS final_score,
+                                COALESCE(es.full_score, 0) AS full_score,
+                                COALESCE(pd.approved_delta, 0) AS approved_delta
+                            FROM leaudit_cross_review_task_documents td
+                            JOIN leaudit_documents d
+                              ON d.id = td.document_id
+                            JOIN target_groups tg
+                              ON tg.version_group_key = COALESCE(NULLIF(d.version_group_key, ''), CONCAT('root:', COALESCE(d.root_version_id, d.id)::text))
+                            LEFT JOIN leaudit_audit_runs ar
+                              ON ar.id = d.current_run_id
+                            LEFT JOIN leaudit_document_types dt
+                              ON dt.id = d.type_id
+                            LEFT JOIN LATERAL (
+                                SELECT file_size, local_path AS file_path, created_at AS file_upload_time,
+                                       COALESCE(file_ext, '') AS file_ext
+                                FROM leaudit_document_files
+                                WHERE document_id = d.id
+                                ORDER BY id ASC
+                                LIMIT 1
+                            ) df ON TRUE
+                            LEFT JOIN LATERAL (
+                                SELECT
+                                    COUNT(*)::int AS total_evaluation_points,
+                                    COUNT(*) FILTER (WHERE rr.passed IS TRUE)::int AS pass_count,
+                                    COUNT(*) FILTER (WHERE rr.passed IS FALSE AND rr.risk = 'high')::int AS error_count,
+                                    COUNT(*) FILTER (WHERE rr.passed IS FALSE AND rr.risk IN ('low', 'medium'))::int AS warning_count,
+                                    0::int AS manual_count,
+                                    COUNT(*) FILTER (WHERE rr.passed IS FALSE)::int AS issue_count,
+                                    ARRAY_AGG(rr.fail_message ORDER BY rr.id) FILTER (
+                                        WHERE rr.passed IS FALSE AND rr.risk = 'high' AND rr.fail_message IS NOT NULL AND rr.fail_message != ''
+                                    ) AS error_messages,
+                                    ARRAY_AGG(rr.fail_message ORDER BY rr.id) FILTER (
+                                        WHERE rr.passed IS FALSE AND rr.risk IN ('low', 'medium') AND rr.fail_message IS NOT NULL AND rr.fail_message != ''
+                                    ) AS warning_messages,
+                                    ARRAY_AGG(rr.fail_message ORDER BY rr.id) FILTER (
+                                        WHERE rr.passed IS FALSE AND rr.fail_message IS NOT NULL AND rr.fail_message != ''
+                                    ) AS issue_messages,
+                                    ARRAY[]::text[] AS manual_messages,
+                                    COALESCE(SUM(rr.score) FILTER (WHERE rr.passed IS TRUE), 0) AS final_score,
+                                    COALESCE(SUM(rr.score), 0) AS full_score
+                                FROM leaudit_rule_results rr
+                                WHERE rr.document_id = d.id
+                                  AND rr.run_id = d.current_run_id
+                            ) es ON TRUE
+                            LEFT JOIN LATERAL (
+                                SELECT COALESCE(SUM(proposed_score_delta), 0) AS approved_delta
+                                FROM leaudit_cross_review_proposals p
+                                WHERE p.document_id = d.id
+                                  AND p.status = 'approved'
+                                  AND p.delete_time IS NULL
+                            ) pd ON TRUE
+                            WHERE td.task_id = :task_id
+                              AND td.delete_time IS NULL
+                              AND d.deleted_at IS NULL
+                            ORDER BY COALESCE(NULLIF(d.version_group_key, ''), CONCAT('root:', COALESCE(d.root_version_id, d.id)::text)), d.version_no DESC, d.id DESC
+                            """
+                        ).bindparams(bindparam("document_ids", expanding=False)),
+                        {"task_id": TaskId, "document_ids": latestDocumentIds},
+                    )
+                ).mappings().all()
+
+                groupedRows: dict[str, list[dict]] = {}
+                for row in historyRows:
+                    groupedRows.setdefault(str(row.get("version_group_key") or ""), []).append(dict(row))
+
+                for groupKey, groupRows in groupedRows.items():
+                    orderedRows = sorted(
+                        groupRows,
+                        key=lambda item: (int(item.get("version_no") or 1), int(item.get("document_id") or 0)),
+                    )
+                    localVersionMap = {
+                        int(item["document_id"]): index + 1 for index, item in enumerate(orderedRows)
+                    }
+                    historyItems: list[CrossReviewTaskHistoryVersionVO] = []
+                    for item in reversed(orderedRows[:-1]):
+                        finalScore = float(item.get("final_score") or 0) + float(item.get("approved_delta") or 0)
+                        fullScore = float(item.get("full_score") or 0)
+                        historyItems.append(
+                            CrossReviewTaskHistoryVersionVO(
+                                documentId=int(item["document_id"]),
+                                name=str(item.get("name") or ""),
+                                documentNumber=item.get("document_number"),
+                                typeId=self._to_int(item.get("type_id")),
+                                typeName=item.get("type_name"),
+                                processingStatus=item.get("processing_status"),
+                                versionNo=int(localVersionMap.get(int(item["document_id"]), 1)),
+                                auditStatus=int(item.get("audit_status") or 0),
+                                createdAt=item.get("created_at"),
+                                fileSize=int(item.get("file_size") or 0),
+                                path=str(item.get("path") or ""),
+                                uploadTime=item.get("upload_time"),
+                                fileExt=str(item.get("file_ext") or "") or None,
+                                totalEvaluationPoints=int(item.get("total_evaluation_points") or 0),
+                                passCount=int(item.get("pass_count") or 0),
+                                warningCount=int(item.get("warning_count") or 0),
+                                errorCount=int(item.get("error_count") or 0),
+                                manualCount=int(item.get("manual_count") or 0),
+                                issueCount=int(item.get("issue_count") or 0),
+                                warningMessages=self._parse_text_array(item.get("warning_messages")),
+                                errorMessages=self._parse_text_array(item.get("error_messages")),
+                                issueMessages=self._parse_text_array(item.get("issue_messages")),
+                                manualMessages=self._parse_text_array(item.get("manual_messages")),
+                                finalScore=finalScore,
+                                fullScore=fullScore,
+                                scoreSummary=self._build_score_summary(finalScore, fullScore),
+                                scorePercent=self._build_score_percent(finalScore, fullScore),
+                            )
+                        )
+                    historyByGroup[groupKey] = historyItems
+
+            items: list[CrossReviewTaskDocumentVO] = []
+            for row in rows:
+                groupKey = str(row.get("version_group_key") or "")
+                historyVersions = historyByGroup.get(groupKey, [])
+                finalScore = float(row.get("final_score") or 0) + float(row.get("approved_delta") or 0)
+                fullScore = float(row.get("full_score") or 0)
+                versionNo = int(row.get("total_versions") or 1)
+                items.append(
                     CrossReviewTaskDocumentVO(
                         documentId=int(row["document_id"]),
                         name=str(row["name"] or ""),
@@ -627,9 +801,9 @@ class CrossReviewServiceImpl(ICrossReviewService):
                         typeId=self._to_int(row.get("type_id")),
                         typeName=row.get("type_name"),
                         processingStatus=row.get("processing_status"),
-                    versionNo=int(row.get("version_no") or 1),
-                    isLatestVersion=bool(row.get("is_latest_version")),
-                    versionGroupKey=str(row.get("version_group_key") or ""),
+                        versionNo=versionNo,
+                        isLatestVersion=True,
+                        versionGroupKey=groupKey,
                         totalVersions=int(row.get("total_versions") or 1),
                         auditStatus=int(row.get("audit_status") or 0),
                         createdAt=row.get("created_at"),
@@ -647,19 +821,13 @@ class CrossReviewServiceImpl(ICrossReviewService):
                         errorMessages=self._parse_text_array(row.get("error_messages")),
                         issueMessages=self._parse_text_array(row.get("issue_messages")),
                         manualMessages=self._parse_text_array(row.get("manual_messages")),
-                    finalScore=float(row.get("final_score") or 0) + float(row.get("approved_delta") or 0),
-                    fullScore=float(row.get("full_score") or 0),
-                    scoreSummary=self._build_score_summary(
-                        float(row.get("final_score") or 0) + float(row.get("approved_delta") or 0),
-                        float(row.get("full_score") or 0),
-                    ),
-                    scorePercent=self._build_score_percent(
-                        float(row.get("final_score") or 0) + float(row.get("approved_delta") or 0),
-                        float(row.get("full_score") or 0),
-                    ),
+                        finalScore=finalScore,
+                        fullScore=fullScore,
+                        scoreSummary=self._build_score_summary(finalScore, fullScore),
+                        scorePercent=self._build_score_percent(finalScore, fullScore),
+                        historyVersions=historyVersions,
+                    )
                 )
-                for row in rows
-            ]
             return CrossReviewTaskDocumentPageVO(
                 taskId=TaskId,
                 total=total,
@@ -1242,6 +1410,84 @@ class CrossReviewServiceImpl(ICrossReviewService):
             processingStatus=uploadResult.processingStatus,
         )
 
+    async def AppendTaskDocumentAttachments(
+        self,
+        CurrentUserId: int,
+        TaskId: int,
+        DocumentId: int,
+        Files: list[tuple[str, bytes, str | None]],
+        Remark: str | None = None,
+    ) -> CrossReviewTaskDocumentAppendVO:
+        """为交叉评查任务文档追加附件，并生成同版本链新版本。"""
+        async with GetAsyncSession() as session:
+            await self._ensure_tables_ready(session)
+            permission = await self.CanConfirmTaskDocument(CurrentUserId, TaskId)
+            if not permission.canConfirm:
+                raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, permission.reason)
+            await self._ensure_task_document(session, TaskId, DocumentId)
+
+        appendResult = await self.DocumentService.AppendAttachments(
+            CurrentUserId=CurrentUserId,
+            Id=DocumentId,
+            Files=Files,
+            MergeMode="new",
+            Remark=Remark,
+        )
+
+        async with GetAsyncSession() as session:
+            await self._ensure_tables_ready(session)
+            await self._reset_transaction_for_write(session)
+            async with session.begin():
+                exists = bool(
+                    await session.scalar(
+                        text(
+                            """
+                            SELECT 1
+                            FROM leaudit_cross_review_task_documents
+                            WHERE task_id = :task_id
+                              AND document_id = :document_id
+                              AND delete_time IS NULL
+                            LIMIT 1
+                            """
+                        ),
+                        {"task_id": TaskId, "document_id": appendResult.documentId},
+                    )
+                )
+                if not exists:
+                    await session.execute(
+                        text(
+                            """
+                            INSERT INTO leaudit_cross_review_task_documents
+                                (task_id, document_id, audit_status)
+                            VALUES
+                                (:task_id, :document_id, 0)
+                            """
+                        ),
+                        {"task_id": TaskId, "document_id": appendResult.documentId},
+                    )
+                await session.execute(
+                    text(
+                        """
+                        UPDATE leaudit_cross_review_tasks
+                        SET status = 'in_progress',
+                            update_time = NOW()
+                        WHERE id = :task_id
+                          AND delete_time IS NULL
+                        """
+                    ),
+                    {"task_id": TaskId},
+                )
+
+        return CrossReviewTaskDocumentAppendVO(
+            taskId=TaskId,
+            originalDocumentId=DocumentId,
+            documentId=appendResult.documentId,
+            versionNo=appendResult.versionNo,
+            versionGroupKey=appendResult.versionGroupKey,
+            auditStatus=0,
+            processingStatus=appendResult.processingStatus,
+        )
+
     async def _build_document_proposals_page(
         self,
         session,

fastapi_modules/fastapi_leaudit/services/impl/documentServiceImpl.py

@@ -11,12 +11,17 @@ from datetime import date as date_type, datetime
 import hashlib
 import mimetypes
 import re
+import tempfile
 import time
 import unicodedata
 import uuid
+from copy import deepcopy
 from pathlib import Path
 
-from sqlalchemy import text
+import fitz
+from leaudit.converters import doc2pdf
+from sqlalchemy import bindparam, text
+from docx import Document as DocxDocument
 
 from fastapi_common.fastapi_common_sqlalchemy.database import GetAsyncSession
 from fastapi_common.fastapi_common_web.domain.responses import StatusCodeEnum
@@ -49,6 +54,7 @@ from fastapi_modules.fastapi_leaudit.domian.vo.reviewPointVo import (
     ReviewPointsAggregateVO,
 )
 from fastapi_modules.fastapi_leaudit.models import LeauditDocument, LeauditDocumentFile
+from fastapi_modules.fastapi_leaudit.leaudit_bridge.fileSourceResolver import FileSourceResolver
 from fastapi_modules.fastapi_leaudit.services import IAuditService, IDocumentService, IOssService
 from fastapi_modules.fastapi_leaudit.services.impl.auditServiceImpl import AuditServiceImpl
 from fastapi_modules.fastapi_leaudit.services.impl.ossServiceImpl import OssServiceImpl
@@ -313,7 +319,12 @@ class DocumentServiceImpl(IDocumentService):
             documentColumns = await self._loadDocumentColumns(Session)
 
         currentUser = await self._getCurrentUserContext(CurrentUserId)
-        filters = ["d.is_latest_version = true", "d.deleted_at IS NULL", "f.is_active = true", "f.file_role = 'primary'"]
+        filters = [
+            "d.is_latest_version = true",
+            "d.deleted_at IS NULL",
+            "f.is_active = true",
+            "f.file_role = 'primary'",
+        ]
         params: dict[str, object] = {"limit": page_size, "offset": offset}
         filters.extend(
             self._buildDocumentScopeFilters(
@@ -516,6 +527,7 @@ class DocumentServiceImpl(IDocumentService):
                 f.file_name,
                 f.file_ext,
                 f.file_size,
+                f.oss_url,
                 ar.status AS run_status,
                 ar.result_status,
                 ar.total_score,
@@ -555,6 +567,7 @@ class DocumentServiceImpl(IDocumentService):
                             fileName=row["file_name"],
                             fileExt=row["file_ext"],
                             fileSize=int(row["file_size"]) if row["file_size"] is not None else None,
+                            ossUrl=row["oss_url"],
                             processingStatus=row["processing_status"],
                             runStatus=row["run_status"],
                             resultStatus=row["result_status"],
@@ -1049,11 +1062,687 @@ class DocumentServiceImpl(IDocumentService):
 
         await Session.commit()
 
+    async def _load_active_primary_file_row(
+        self,
+        Session,
+        *,
+        DocumentId: int,
+    ) -> dict[str, Any]:
+        row = (
+            await Session.execute(
+                text(
+                    """
+                    SELECT
+                        id,
+                        file_name,
+                        file_ext,
+                        mime_type,
+                        file_role,
+                        oss_url,
+                        local_path
+                    FROM leaudit_document_files
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role = 'primary'
+                    ORDER BY id DESC
+                    LIMIT 1
+                    """
+                ),
+                {"document_id": DocumentId},
+            )
+        ).mappings().first()
+        if not row:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前文档缺少主文件")
+        return dict(row)
+
+    def _normalize_document_source_kind(self, file_name: str, file_ext: str | None, mime_type: str | None) -> str:
+        suffix = (f".{str(file_ext).lstrip('.')}" if file_ext else Path(file_name).suffix).lower()
+        normalizedMime = (mime_type or "").lower()
+        if suffix == ".pdf" or normalizedMime == "application/pdf":
+            return "pdf"
+        if suffix == ".docx" or normalizedMime == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
+            return "docx"
+        raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前仅支持 PDF 或 DOCX 主文档追加附件")
+
+    def _normalize_attachment_source_kind(self, file_name: str, file_ext: str | None, mime_type: str | None) -> str:
+        suffix = (f".{str(file_ext).lstrip('.')}" if file_ext else Path(file_name).suffix).lower()
+        normalizedMime = (mime_type or "").lower()
+        if suffix == ".pdf" or normalizedMime == "application/pdf":
+            return "pdf"
+        if suffix == ".docx" or normalizedMime == "application/vnd.openxmlformats-officedocument.wordprocessingml.document":
+            return "docx"
+        raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, f"附件 {file_name} 仅支持 PDF 或 DOCX 格式")
+
+    def _validate_attachment_matrix(
+        self,
+        *,
+        MainSourceKind: str,
+        Files: list[tuple[str, bytes, str | None]],
+    ) -> None:
+        for fileName, _, contentType in Files:
+            attachmentKind = self._normalize_attachment_source_kind(fileName, Path(fileName).suffix.lstrip(".").lower() or None, contentType)
+            if MainSourceKind == "docx" and attachmentKind != "docx":
+                raise LeauditException(
+                    StatusCodeEnum.HTTP_400_BAD_REQUEST,
+                    "源文档为 DOCX 时，仅允许追加 DOCX 附件，且合并结果仍为 DOCX",
+                )
+
+    def _merge_docx_paths_for_merge(
+        self,
+        *,
+        DocxPaths: list[str],
+        TempPaths: list[str],
+    ) -> str:
+        if not DocxPaths:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "缺少可合并的 DOCX 文件")
+
+        mergedTemp = tempfile.NamedTemporaryFile(suffix=".docx", delete=False)
+        mergedTemp.close()
+        TempPaths.append(mergedTemp.name)
+
+        baseDoc = DocxDocument(DocxPaths[0])
+        for attachmentPath in DocxPaths[1:]:
+            attachmentDoc = DocxDocument(attachmentPath)
+            if baseDoc.paragraphs:
+                baseDoc.add_page_break()
+            for element in attachmentDoc.element.body:
+                baseDoc.element.body.append(deepcopy(element))
+        baseDoc.save(mergedTemp.name)
+        return mergedTemp.name
+
+    async def _cloneActiveFilesToNewDocument(
+        self,
+        Session,
+        *,
+        SourceDocumentId: int,
+        TargetDocumentId: int,
+        CreatedBy: int | None,
+        IncludeAttachments: bool = True,
+    ) -> None:
+        """复制当前文档的主文件与现有附件到新版本文档。"""
+        fileRoles = ["primary", "attachment"] if IncludeAttachments else ["primary"]
+        sourceFiles = (
+            await Session.execute(
+                text(
+                    """
+                    SELECT
+                        id,
+                        file_role,
+                        file_name,
+                        file_ext,
+                        mime_type,
+                        file_size,
+                        sha256,
+                        local_path,
+                        oss_url,
+                        storage_provider
+                    FROM leaudit_document_files
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role = ANY(:file_roles)
+                    ORDER BY
+                      CASE WHEN file_role = 'primary' THEN 0 ELSE 1 END,
+                      id ASC
+                    """
+                ).bindparams(bindparam("file_roles", expanding=False)),
+                {"document_id": SourceDocumentId, "file_roles": fileRoles},
+            )
+        ).mappings().all()
+        if not sourceFiles:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "源文档缺少可复制的有效文件")
+
+        for row in sourceFiles:
+            clonedFile = LeauditDocumentFile(
+                documentId=TargetDocumentId,
+                fileRole=str(row["file_role"] or "attachment"),
+                fileName=str(row["file_name"] or ""),
+                fileExt=row["file_ext"],
+                mimeType=row["mime_type"],
+                fileSize=int(row["file_size"]) if row["file_size"] is not None else None,
+                sha256=row["sha256"],
+                localPath=row["local_path"],
+                ossUrl=row["oss_url"],
+                storageProvider=row["storage_provider"],
+                isActive=True,
+                createdBy=CreatedBy,
+            )
+            Session.add(clonedFile)
+
+        await Session.flush()
+
+    async def _buildMergedPdfBytesForDocument(
+        self,
+        Session,
+        *,
+        DocumentId: int,
+    ) -> tuple[bytes, str]:
+        """读取当前主文件与附件，生成新的主 PDF 内容。"""
+        primaryFile = (
+            await Session.execute(
+                text(
+                    """
+                    SELECT id
+                    FROM leaudit_document_files
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role = 'primary'
+                    ORDER BY id DESC
+                    LIMIT 1
+                    """
+                ),
+                {"document_id": DocumentId},
+            )
+        ).scalar_one_or_none()
+        if primaryFile is None:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前文档缺少主文件，无法生成合并文件")
+
+        attachmentRows = (
+            await Session.execute(
+                text(
+                    """
+                    SELECT id
+                    FROM leaudit_document_files
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role = 'attachment'
+                    ORDER BY id ASC
+                    """
+                ),
+                {"document_id": DocumentId},
+            )
+        ).scalars().all()
+        if not attachmentRows:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前文档没有附件，无法生成合并文件")
+
+        resolver = FileSourceResolver()
+        primaryModel = await Session.get(LeauditDocumentFile, int(primaryFile))
+        if primaryModel is None:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "主文件记录不存在，无法生成合并文件")
+        primaryPayload = await resolver.ResolvePayload(primaryModel)
+
+        attachmentModels: list[LeauditDocumentFile] = []
+        for attachmentId in attachmentRows:
+            attachmentModel = await Session.get(LeauditDocumentFile, int(attachmentId))
+            if attachmentModel is not None:
+                attachmentModels.append(attachmentModel)
+        attachmentPayloads = await resolver.ResolvePayloads(attachmentModels)
+
+        tempPaths: list[str] = []
+        try:
+            mainLocalPath = self._write_temp_file_for_merge(
+                FileName=primaryPayload.fileName,
+                Content=primaryPayload.fileContent,
+                TempPaths=tempPaths,
+            )
+            mainPdfPath = self._convert_source_to_pdf(
+                SourcePath=mainLocalPath,
+                TempPaths=tempPaths,
+            )
+            pdfPaths = [mainPdfPath]
+
+            for attachmentPayload in attachmentPayloads:
+                attachmentLocalPath = self._write_temp_file_for_merge(
+                    FileName=attachmentPayload.fileName,
+                    Content=attachmentPayload.fileContent,
+                    TempPaths=tempPaths,
+                )
+                attachmentPdfPath = self._convert_source_to_pdf(
+                    SourcePath=attachmentLocalPath,
+                    TempPaths=tempPaths,
+                )
+                pdfPaths.append(attachmentPdfPath)
+
+            mergedPdfPath = self._merge_pdf_paths_for_merge(
+                PdfPaths=pdfPaths,
+                TempPaths=tempPaths,
+            )
+            mergedBytes = Path(mergedPdfPath).read_bytes()
+            mergedName = f"{Path(primaryPayload.fileName).stem}.pdf"
+            return mergedBytes, mergedName
+        except LeauditException:
+            raise
+        except Exception as error:
+            raise LeauditException(
+                StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR,
+                f"生成合并PDF失败: {error}",
+            ) from error
+        finally:
+            for tempPath in reversed(tempPaths):
+                try:
+                    pathObj = Path(tempPath)
+                    if pathObj.is_file():
+                        pathObj.unlink(missing_ok=True)
+                except Exception:
+                    pass
+
+    async def _buildMergedDocxBytesForDocument(
+        self,
+        Session,
+        *,
+        DocumentId: int,
+    ) -> tuple[bytes, str]:
+        """读取当前主文件与附件，生成新的主 DOCX 内容。"""
+        primaryFile = await self._load_active_primary_file_row(Session, DocumentId=DocumentId)
+        if self._normalize_document_source_kind(
+            str(primaryFile["file_name"] or ""),
+            primaryFile.get("file_ext"),
+            primaryFile.get("mime_type"),
+        ) != "docx":
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "仅 DOCX 主文档支持生成 DOCX 合并结果")
+
+        attachmentRows = (
+            await Session.execute(
+                text(
+                    """
+                    SELECT id
+                    FROM leaudit_document_files
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role = 'attachment'
+                    ORDER BY id ASC
+                    """
+                ),
+                {"document_id": DocumentId},
+            )
+        ).scalars().all()
+        if not attachmentRows:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前文档没有附件，无法生成合并文件")
+
+        resolver = FileSourceResolver()
+        primaryModel = await Session.get(LeauditDocumentFile, int(primaryFile["id"]))
+        if primaryModel is None:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "主文件记录不存在，无法生成合并文件")
+        primaryPayload = await resolver.ResolvePayload(primaryModel)
+
+        attachmentModels: list[LeauditDocumentFile] = []
+        for attachmentId in attachmentRows:
+            attachmentModel = await Session.get(LeauditDocumentFile, int(attachmentId))
+            if attachmentModel is not None:
+                attachmentModels.append(attachmentModel)
+        attachmentPayloads = await resolver.ResolvePayloads(attachmentModels)
+
+        tempPaths: list[str] = []
+        try:
+            docxPaths = [
+                self._write_temp_file_for_merge(
+                    FileName=primaryPayload.fileName,
+                    Content=primaryPayload.fileContent,
+                    TempPaths=tempPaths,
+                )
+            ]
+            for attachmentPayload in attachmentPayloads:
+                attachmentKind = self._normalize_attachment_source_kind(
+                    attachmentPayload.fileName,
+                    Path(attachmentPayload.fileName).suffix.lstrip(".").lower() or None,
+                    None,
+                )
+                if attachmentKind != "docx":
+                    raise LeauditException(
+                        StatusCodeEnum.HTTP_400_BAD_REQUEST,
+                        f"DOCX 主文档仅允许追加 DOCX 附件，当前附件 {attachmentPayload.fileName} 不符合要求",
+                    )
+                docxPaths.append(
+                    self._write_temp_file_for_merge(
+                        FileName=attachmentPayload.fileName,
+                        Content=attachmentPayload.fileContent,
+                        TempPaths=tempPaths,
+                    )
+                )
+
+            mergedDocxPath = self._merge_docx_paths_for_merge(
+                DocxPaths=docxPaths,
+                TempPaths=tempPaths,
+            )
+            mergedBytes = Path(mergedDocxPath).read_bytes()
+            mergedName = f"{Path(primaryPayload.fileName).stem}.docx"
+            return mergedBytes, mergedName
+        finally:
+            for tempPath in reversed(tempPaths):
+                try:
+                    pathObj = Path(tempPath)
+                    if pathObj.is_file():
+                        pathObj.unlink(missing_ok=True)
+                except Exception:
+                    pass
+
+    def _write_temp_file_for_merge(
+        self,
+        *,
+        FileName: str,
+        Content: bytes,
+        TempPaths: list[str],
+    ) -> str:
+        """为持久化合并流程写入临时文件。"""
+        suffix = Path(FileName).suffix or ".bin"
+        with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as tempFile:
+            tempFile.write(Content)
+            tempPath = tempFile.name
+        TempPaths.append(tempPath)
+        return tempPath
+
+    def _convert_source_to_pdf(
+        self,
+        *,
+        SourcePath: str,
+        TempPaths: list[str],
+    ) -> str:
+        """将源文件转换为 PDF。"""
+        source = Path(SourcePath)
+        if source.suffix.lower() == ".pdf":
+            return str(source)
+        pdfTemp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False)
+        pdfTemp.close()
+        TempPaths.append(pdfTemp.name)
+        doc2pdf.convert(source, pdfTemp.name, soffice="auto", pdfa=False, force=True, verify=False)
+        return pdfTemp.name
+
+    def _merge_pdf_paths_for_merge(
+        self,
+        *,
+        PdfPaths: list[str],
+        TempPaths: list[str],
+    ) -> str:
+        """合并多个 PDF 为一个临时 PDF。"""
+        mergedTemp = tempfile.NamedTemporaryFile(suffix=".pdf", delete=False)
+        mergedTemp.close()
+        TempPaths.append(mergedTemp.name)
+        output = fitz.open()
+        try:
+            for pdfPath in PdfPaths:
+                source = fitz.open(pdfPath)
+                try:
+                    output.insert_pdf(source)
+                finally:
+                    source.close()
+            output.save(mergedTemp.name)
+        finally:
+            output.close()
+        return mergedTemp.name
+
+    async def _persistMergedPdfForDocument(
+        self,
+        Session,
+        *,
+        DocumentId: int,
+        TypeCode: str,
+        Region: str,
+        VersionNo: int,
+        CreatedBy: int | None,
+        FileRole: str = "primary",
+    ) -> None:
+        """为当前文档生成并替换主 PDF 文件。"""
+        mergedBytes, mergedName = await self._buildMergedPdfBytesForDocument(
+            Session,
+            DocumentId=DocumentId,
+        )
+        mergedSha256 = hashlib.sha256(mergedBytes).hexdigest()
+        mergedSize = len(mergedBytes)
+        uploadedAt = datetime.now()
+        versionLabel = f"v{VersionNo}"
+        objectKey = OssPathUtils.BuildBusinessDocKey(
+            Region=Region,
+            TypeCode=TypeCode,
+            DocumentId=DocumentId,
+            Version=versionLabel,
+            FileRole=FileRole,
+            FileName=mergedName,
+            Year=uploadedAt.year,
+            Month=uploadedAt.month,
+        )
+        ossUrl = await self.OssService.UploadBytes(
+            ObjectKey=objectKey,
+            Content=mergedBytes,
+            ContentType="application/pdf",
+        )
+
+        if FileRole == "primary":
+            await Session.execute(
+                text(
+                    """
+                    UPDATE leaudit_document_files
+                    SET is_active = false
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role IN ('primary', 'merged_pdf', 'merged_docx')
+                    """
+                ),
+                {"document_id": DocumentId},
+            )
+        else:
+            await Session.execute(
+                text(
+                    """
+                    UPDATE leaudit_document_files
+                    SET is_active = false
+                    WHERE document_id = :document_id
+                      AND deleted_at IS NULL
+                      AND is_active = true
+                      AND file_role = :file_role
+                    """
+                ),
+                {"document_id": DocumentId, "file_role": FileRole},
+            )
+
+        Session.add(
+            LeauditDocumentFile(
+                documentId=DocumentId,
+                fileRole=FileRole,
+                fileName=mergedName,
+                fileExt="pdf",
+                mimeType="application/pdf",
+                fileSize=mergedSize,
+                sha256=mergedSha256,
+                localPath=None,
+                ossUrl=ossUrl,
+                storageProvider="minio",
+                isActive=True,
+                createdBy=CreatedBy,
+            )
+        )
+        await Session.flush()
+
+    async def _persistMergedDocxForDocument(
+        self,
+        Session,
+        *,
+        DocumentId: int,
+        TypeCode: str,
+        Region: str,
+        VersionNo: int,
+        CreatedBy: int | None,
+    ) -> None:
+        """为当前文档生成并替换主 DOCX 文件。"""
+        mergedBytes, mergedName = await self._buildMergedDocxBytesForDocument(
+            Session,
+            DocumentId=DocumentId,
+        )
+        mergedSha256 = hashlib.sha256(mergedBytes).hexdigest()
+        mergedSize = len(mergedBytes)
+        uploadedAt = datetime.now()
+        versionLabel = f"v{VersionNo}"
+        objectKey = OssPathUtils.BuildBusinessDocKey(
+            Region=Region,
+            TypeCode=TypeCode,
+            DocumentId=DocumentId,
+            Version=versionLabel,
+            FileRole="primary",
+            FileName=mergedName,
+            Year=uploadedAt.year,
+            Month=uploadedAt.month,
+        )
+        ossUrl = await self.OssService.UploadBytes(
+            ObjectKey=objectKey,
+            Content=mergedBytes,
+            ContentType="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+        )
+
+        await Session.execute(
+            text(
+                """
+                UPDATE leaudit_document_files
+                SET is_active = false
+                WHERE document_id = :document_id
+                  AND deleted_at IS NULL
+                  AND is_active = true
+                  AND file_role IN ('primary', 'merged_pdf', 'merged_docx')
+                """
+            ),
+            {"document_id": DocumentId},
+        )
+
+        Session.add(
+            LeauditDocumentFile(
+                documentId=DocumentId,
+                fileRole="primary",
+                fileName=mergedName,
+                fileExt="docx",
+                mimeType="application/vnd.openxmlformats-officedocument.wordprocessingml.document",
+                fileSize=mergedSize,
+                sha256=mergedSha256,
+                localPath=None,
+                ossUrl=ossUrl,
+                storageProvider="minio",
+                isActive=True,
+                createdBy=CreatedBy,
+            )
+        )
+        await Session.flush()
+
+    async def _createNextVersionFromExistingDocument(
+        self,
+        Session,
+        *,
+        SourceDocumentId: int,
+        CreatedBy: int,
+        Remark: str | None = None,
+    ) -> tuple[LeauditDocument, str, str]:
+        """基于现有文档创建一个新版本，并复制当前主文件/附件。"""
+        documentColumns = await self._loadDocumentColumns(Session)
+        optionalSelects = [
+            "d.document_number AS document_number" if "document_number" in documentColumns else "NULL::text AS document_number",
+            "d.remark AS remark" if "remark" in documentColumns else "NULL::text AS remark",
+            "d.is_test_document AS is_test_document" if "is_test_document" in documentColumns else "FALSE AS is_test_document",
+            "d.audit_status AS audit_status" if "audit_status" in documentColumns else "NULL::integer AS audit_status",
+        ]
+        sourceRow = (
+            await Session.execute(
+                text(
+                    """
+                    SELECT
+                        d.id,
+                        d.biz_document_id,
+                        d.type_id,
+                        d.group_id,
+                        d.region,
+                        d.processing_status,
+                        d.version_group_key,
+                        d.version_no,
+                        d.previous_version_id,
+                        d.root_version_id,
+                        d.is_latest_version,
+                        d.normalized_name,
+                        d.review_scope,
+                        """
+                    + ",\n                        ".join(optionalSelects)
+                    + """
+                        ,
+                        dt.code AS type_code
+                    FROM leaudit_documents d
+                    LEFT JOIN leaudit_document_types dt
+                      ON dt.id = d.type_id
+                    WHERE d.id = :document_id
+                      AND d.deleted_at IS NULL
+                    LIMIT 1
+                    """
+                ),
+                {"document_id": SourceDocumentId},
+            )
+        ).mappings().first()
+        if not sourceRow:
+            raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "文档不存在或无权访问")
+
+        resolvedTypeCode = str(sourceRow["type_code"] or "").strip()
+        if not resolvedTypeCode:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前文档缺少文档类型编码，无法创建新版本")
+
+        previousDocument = await Session.get(LeauditDocument, SourceDocumentId)
+        if previousDocument is None:
+            raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "文档不存在或无权访问")
+        previousDocument.isLatestVersion = False
+
+        documentFields: dict[str, object] = {
+            "bizDocumentId": time.time_ns(),
+            "typeId": int(sourceRow["type_id"]) if sourceRow["type_id"] is not None else None,
+            "groupId": int(sourceRow["group_id"]) if sourceRow["group_id"] is not None else None,
+            "region": str(sourceRow["region"] or "default").strip() or "default",
+            "processingStatus": "waiting",
+            "versionGroupKey": str(sourceRow["version_group_key"] or uuid.uuid4().hex),
+            "versionNo": int(sourceRow["version_no"] or 1) + 1,
+            "previousVersionId": SourceDocumentId,
+            "rootVersionId": int(sourceRow["root_version_id"] or SourceDocumentId),
+            "isLatestVersion": True,
+            "normalizedName": sourceRow["normalized_name"],
+            "reviewScope": str(sourceRow["review_scope"] or "standard"),
+        }
+
+        newDocument = await LeauditDocument.create_new(Session, **documentFields)
+        if newDocument.rootVersionId is None:
+            newDocument.rootVersionId = newDocument.Id
+
+        assignments: list[str] = []
+        params: dict[str, object] = {"id": int(newDocument.Id)}
+        if "document_number" in documentColumns:
+            assignments.append("document_number = :document_number")
+            params["document_number"] = sourceRow["document_number"]
+        if "remark" in documentColumns:
+            assignments.append("remark = :remark")
+            params["remark"] = Remark.strip() if Remark and Remark.strip() else sourceRow["remark"]
+        if "is_test_document" in documentColumns:
+            assignments.append("is_test_document = :is_test_document")
+            params["is_test_document"] = bool(sourceRow["is_test_document"])
+        if "audit_status" in documentColumns:
+            assignments.append("audit_status = :audit_status")
+            params["audit_status"] = int(sourceRow["audit_status"]) if sourceRow["audit_status"] is not None else None
+        if assignments:
+            assignments.append("updated_at = NOW()")
+            await Session.execute(
+                text(f"UPDATE leaudit_documents SET {', '.join(assignments)} WHERE id = :id"),
+                params,
+            )
+
+        await self._cloneActiveFilesToNewDocument(
+            Session,
+            SourceDocumentId=SourceDocumentId,
+            TargetDocumentId=newDocument.Id,
+            CreatedBy=CreatedBy,
+            IncludeAttachments=False,
+        )
+        await Session.flush()
+        return newDocument, resolvedTypeCode, str(sourceRow["region"] or "default").strip() or "default"
+
+    def _normalizeAttachmentMergeMode(self, MergeMode: str | None) -> str:
+        """标准化附件合并模式。"""
+        normalizedMode = (MergeMode or "new").strip().lower()
+        if normalizedMode not in {"overwrite", "new"}:
+            raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "mergeMode 仅支持 overwrite 或 new")
+        return normalizedMode
+
     async def DeleteDocument(self, CurrentUserId: int, Id: int) -> None:
         """软删除文档，并执行数据隔离校验。"""
         async with GetAsyncSession() as Session:
             currentUser = await self._getCurrentUserContext(CurrentUserId)
-            filters = ["d.id = :id", "d.deleted_at IS NULL", "f.is_active = true", "f.file_role = 'primary'"]
+            filters = [
+                "d.id = :id",
+                "d.deleted_at IS NULL",
+                "f.is_active = true",
+                "f.file_role = 'primary'",
+            ]
             params: dict[str, object] = {"id": Id}
             filters.extend(
                 self._buildDocumentScopeFilters(
@@ -1141,10 +1830,13 @@ class DocumentServiceImpl(IDocumentService):
         CurrentUserId: int,
         Id: int,
         Files: list[tuple[str, bytes, str | None]],
+        MergeMode: str = "overwrite",
+        Remark: str | None = None,
     ) -> DocumentDetailVO:
         """为现有文档追加附件，并执行数据隔离校验。"""
         if not Files:
             raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "至少上传一个附件文件")
+        normalizedMergeMode = self._normalizeAttachmentMergeMode(MergeMode)
 
         async with GetAsyncSession() as Session:
             await self._ensureDocumentGroupColumn(Session)
@@ -1182,18 +1874,72 @@ class DocumentServiceImpl(IDocumentService):
             if not resolvedTypeCode:
                 raise LeauditException(StatusCodeEnum.HTTP_400_BAD_REQUEST, "当前文档缺少文档类型编码，无法追加附件")
 
+            primaryFileRow = await self._load_active_primary_file_row(Session, DocumentId=Id)
+            mainSourceKind = self._normalize_document_source_kind(
+                str(primaryFileRow["file_name"] or ""),
+                primaryFileRow.get("file_ext"),
+                primaryFileRow.get("mime_type"),
+            )
+            self._validate_attachment_matrix(
+                MainSourceKind=mainSourceKind,
+                Files=Files,
+            )
+
+            targetDocumentId = int(documentMeta["document_id"])
+            targetVersionNo = int(documentMeta["version_no"] or 1)
             normalizedRegion = str(documentMeta["region"] or "default").strip() or "default"
+
+            if normalizedMergeMode == "new":
+                newDocument, resolvedTypeCode, normalizedRegion = await self._createNextVersionFromExistingDocument(
+                    Session,
+                    SourceDocumentId=Id,
+                    CreatedBy=CurrentUserId,
+                    Remark=Remark,
+                )
+                targetDocumentId = int(newDocument.Id)
+                targetVersionNo = int(newDocument.versionNo or (int(documentMeta["version_no"] or 1) + 1))
+
             await self._appendAttachmentFiles(
                 Session=Session,
-                DocumentId=int(documentMeta["document_id"]),
+                DocumentId=targetDocumentId,
                 TypeCode=resolvedTypeCode,
                 Region=normalizedRegion,
-                VersionNo=int(documentMeta["version_no"] or 1),
+                VersionNo=targetVersionNo,
                 Files=Files,
                 CreatedBy=CurrentUserId,
             )
 
-            refreshed = await self._getDocumentDetail(Session, Id, CurrentUserId, currentUser, documentColumns)
+            if mainSourceKind == "docx":
+                await self._persistMergedDocxForDocument(
+                    Session,
+                    DocumentId=targetDocumentId,
+                    TypeCode=resolvedTypeCode,
+                    Region=normalizedRegion,
+                    VersionNo=targetVersionNo,
+                    CreatedBy=CurrentUserId,
+                )
+                await self._persistMergedPdfForDocument(
+                    Session,
+                    DocumentId=targetDocumentId,
+                    TypeCode=resolvedTypeCode,
+                    Region=normalizedRegion,
+                    VersionNo=targetVersionNo,
+                    CreatedBy=CurrentUserId,
+                    FileRole="merged_pdf",
+                )
+            else:
+                await self._persistMergedPdfForDocument(
+                    Session,
+                    DocumentId=targetDocumentId,
+                    TypeCode=resolvedTypeCode,
+                    Region=normalizedRegion,
+                    VersionNo=targetVersionNo,
+                    CreatedBy=CurrentUserId,
+                )
+
+            await Session.commit()
+
+            refreshed = await self._getDocumentDetail(Session, targetDocumentId, CurrentUserId, currentUser, documentColumns)
             if not refreshed:
                 raise LeauditException(StatusCodeEnum.HTTP_404_NOT_FOUND, "文档不存在或无权访问")
             return refreshed
@@ -1949,7 +2695,12 @@ class DocumentServiceImpl(IDocumentService):
     ) -> DocumentDetailVO | None:
         """查询单文档详情，并附带历史版本。"""
         params: dict[str, object] = {"id": DocumentId}
-        filters = ["d.id = :id", "d.deleted_at IS NULL", "f.is_active = true", "f.file_role = 'primary'"]
+        filters = [
+            "d.id = :id",
+            "d.deleted_at IS NULL",
+            "f.is_active = true",
+            "f.file_role = 'primary'",
+        ]
         if not BypassScopeCheck:
             filters.extend(
                 self._buildDocumentScopeFilters(
@@ -2065,6 +2816,7 @@ class DocumentServiceImpl(IDocumentService):
                             f.id AS file_id,
                             f.file_name,
                             f.file_ext,
+                            f.oss_url,
                             ar.status AS run_status,
                             ar.result_status
                         FROM leaudit_documents d
@@ -2090,6 +2842,7 @@ class DocumentServiceImpl(IDocumentService):
                     versionNo=int(row["version_no"]),
                     fileName=row["file_name"],
                     fileExt=row["file_ext"],
+                    ossUrl=row["oss_url"],
                     processingStatus=row["processing_status"],
                     runStatus=row["run_status"],
                     resultStatus=row["result_status"],
@@ -2913,8 +3666,9 @@ class DocumentServiceImpl(IDocumentService):
                     ORDER BY
                       CASE file_role
                         WHEN 'converted_pdf' THEN 0
-                        WHEN 'merged_pdf' THEN 1
-                        WHEN 'primary' THEN 2
+                        WHEN 'primary' THEN 1
+                        WHEN 'merged_docx' THEN 2
+                        WHEN 'merged_pdf' THEN 3
                         ELSE 9
                       END,
                       id DESC

fastapi_modules/fastapi_leaudit/services/impl/ragChatServiceImpl.py

@@ -33,11 +33,10 @@ from fastapi_modules.fastapi_leaudit.domian.vo.ragChatVo import (
 from fastapi_modules.fastapi_leaudit.rag_engine.config import (
     RAG_CONFIG,
     build_openai_chat_completions_url,
-    build_openai_embeddings_url,
 )
-from fastapi_modules.fastapi_leaudit.rag_engine.chroma_client import get_chroma
 from fastapi_modules.fastapi_leaudit.rag_engine.generator import generate_stream
 from fastapi_modules.fastapi_leaudit.rag_engine.question_chains import generate_followups
+from fastapi_modules.fastapi_leaudit.rag_engine.retriever import RagRetriever
 from fastapi_modules.fastapi_leaudit.services.ragChatService import IRagChatService
 
 
@@ -54,6 +53,9 @@ class RagChatServiceImpl(IRagChatService):
     _task_locks: dict[str, asyncio.Lock] = {}
     _title_tasks: dict[str, asyncio.Task] = {}
 
+    def __init__(self, retriever: RagRetriever | None = None) -> None:
+        self.retriever = retriever or RagRetriever()
+
     async def GetApps(self, CurrentUserId: int, UserArea: str | None, UserRole: str | None) -> RagChatAppListVO:
         apps = await self._load_apps(UserArea, UserRole, only_default=False)
         return RagChatAppListVO(data=apps, total=len(apps))
@@ -592,121 +594,11 @@ class RagChatServiceImpl(IRagChatService):
             raise LeauditException(StatusCodeEnum.HTTP_403_FORBIDDEN, "当前用户无权访问该会话")
 
     async def _retrieve_context(self, dataset_id: int | None, query: str) -> tuple[list[dict], str]:
-        if not dataset_id:
-            return [], ""
-        async with GetAsyncSession() as session:
-            dataset = (
-                await session.execute(
-                    text(
-                        """
-                        SELECT id, name, collection_name, retrieval_model, embedding_model
-                        FROM rag_dataset
-                        WHERE id = :dataset_id AND deleted_at IS NULL
-                        LIMIT 1
-                        """
-                    ),
-                    {"dataset_id": dataset_id},
-                )
-            ).mappings().first()
-        if not dataset:
-            return [], ""
-        retrieval_model = dataset.get("retrieval_model") or {}
-        top_k = int(retrieval_model.get("top_k") or 5)
-        score_threshold = None
-        if retrieval_model.get("score_threshold_enabled"):
-            try:
-                score_threshold = float(retrieval_model.get("score_threshold"))
-            except (TypeError, ValueError):
-                score_threshold = None
-        try:
-            query_embedding = await self._embed_texts([query], dataset.get("embedding_model") or "")
-            collection = get_chroma().get_or_create_collection(dataset["collection_name"])
-            result = collection.query(
-                query_embeddings=query_embedding,
-                n_results=max(top_k, 1),
-                include=["documents", "metadatas", "distances"],
-            )
-            ids = (result.get("ids") or [[]])[0] if result.get("ids") else []
-            docs = (result.get("documents") or [[]])[0]
-            metas = (result.get("metadatas") or [[]])[0]
-            distances = (result.get("distances") or [[]])[0]
-            chunks: list[dict] = []
-            for idx, doc in enumerate(docs):
-                meta = metas[idx] if idx < len(metas) else {}
-                dist = float(distances[idx]) if idx < len(distances) and distances[idx] is not None else 1.0
-                score = 1.0 / (1.0 + max(dist, 0.0))
-                if score_threshold is not None and score < score_threshold:
-                    continue
-                chunks.append(
-                    {
-                        "id": str(ids[idx] if idx < len(ids) else meta.get("id") or idx),
-                        "text": doc,
-                        "source": meta.get("source") or meta.get("document_name") or dataset.get("name") or "",
-                        "score": score,
-                        "chunk_index": int(meta.get("chunk_index") or idx),
-                        "document_name": meta.get("document_name") or meta.get("source") or "",
-                        "document_id": meta.get("document_id"),
-                        "page": meta.get("page"),
-                    }
-                )
-            chunks = await self._hydrate_document_hits(dataset_id, chunks)
-            if chunks:
-                return chunks[:top_k], dataset.get("name") or ""
-        except Exception:
-            pass
-
-        try:
-            chunks = await self._keyword_retrieve_context(
-                dataset_id=dataset_id,
-                collection_name=str(dataset["collection_name"]),
-                dataset_name=str(dataset.get("name") or ""),
-                query=query,
-                top_k=top_k,
-                score_threshold=score_threshold,
-            )
-            return chunks[:top_k], dataset.get("name") or ""
-        except Exception:
-            return [], dataset.get("name") or ""
+        result = await self.retriever.retrieve(query=query, dataset_id=dataset_id)
+        return result.chunks, result.dataset_name
 
     async def _embed_texts(self, texts: list[str], model_name: str) -> list[list[float]]:
-        embed_url = (RAG_CONFIG.get("EMBED_URL") or "").strip() or build_openai_embeddings_url(RAG_CONFIG["LLM_BASE_URL"])
-        embed_key = (RAG_CONFIG.get("EMBED_KEY") or "").strip() or RAG_CONFIG["LLM_API_KEY"]
-        embed_model = model_name or (RAG_CONFIG.get("EMBED_MODEL") or "").strip() or "text-embedding-v4"
-        batch_size = max(1, int(RAG_CONFIG.get("EMBED_BATCH_SIZE") or 10))
-        if not embed_url or not embed_key:
-            raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "未配置可用的向量化服务")
-
-        embeddings: list[list[float]] = []
-        async with httpx.AsyncClient(timeout=120.0) as client:
-            for start in range(0, len(texts), batch_size):
-                batch_texts = texts[start:start + batch_size]
-                try:
-                    response = await client.post(
-                        embed_url,
-                        headers={
-                            "Content-Type": "application/json",
-                            "Authorization": f"Bearer {embed_key}",
-                        },
-                        json={"model": embed_model, "input": batch_texts},
-                    )
-                    response.raise_for_status()
-                except httpx.HTTPStatusError as exc:
-                    error_message = exc.response.text.strip() or f"{exc.response.status_code} {exc.response.reason_phrase}"
-                    raise LeauditException(
-                        StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR,
-                        f"向量化服务调用失败: {error_message[:300]}",
-                    ) from exc
-
-                payload = response.json()
-                rows = payload.get("data") or []
-                batch_embeddings = [row.get("embedding") for row in rows if isinstance(row, dict) and row.get("embedding")]
-                if len(batch_embeddings) != len(batch_texts):
-                    raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "向量化结果数量异常")
-                embeddings.extend(batch_embeddings)
-
-        if len(embeddings) != len(texts):
-            raise LeauditException(StatusCodeEnum.HTTP_500_INTERNAL_SERVER_ERROR, "向量化结果数量异常")
-        return embeddings
+        return await self.retriever._embed_texts(texts, model_name)
 
     async def _start_message_task(
         self,
@@ -1219,220 +1111,42 @@ class RagChatServiceImpl(IRagChatService):
         top_k: int,
         score_threshold: float | None,
     ) -> list[dict]:
-        collection = get_chroma().get_or_create_collection(collection_name)
-        raw = collection.get(include=["documents", "metadatas"])
-        ids = raw.get("ids") or []
-        docs = raw.get("documents") or []
-        metas = raw.get("metadatas") or []
-
-        terms = self._build_keyword_terms(query)
-        if not terms:
-            return []
-
-        scored_chunks: list[dict] = []
-        for idx, chunk_id in enumerate(ids):
-            doc = docs[idx] if idx < len(docs) else ""
-            meta = metas[idx] if idx < len(metas) and isinstance(metas[idx], dict) else {}
-            score = self._score_keyword_chunk(
+        chunks = await self.retriever._keyword_retrieve_context(
+            dataset_id=dataset_id,
+            collection_name=collection_name,
+            dataset_name=dataset_name,
             query=query,
-                terms=terms,
-                content=doc or "",
-                document_name=str(meta.get("document_name") or meta.get("source") or ""),
+            top_k=top_k,
+            score_threshold=score_threshold,
+            source_names=None,
         )
-            if score <= 0:
-                continue
-            if score_threshold is not None and score < score_threshold:
-                continue
-            scored_chunks.append(
-                {
-                    "id": str(chunk_id),
-                    "text": doc or "",
-                    "source": meta.get("source") or meta.get("document_name") or dataset_name,
-                    "score": score,
-                    "chunk_index": int(meta.get("chunk_index") or idx),
-                    "document_name": meta.get("document_name") or meta.get("source") or "",
-                    "document_id": meta.get("document_id"),
-                    "page": meta.get("page"),
-                }
-            )
-
-        scored_chunks.sort(key=lambda item: (-float(item.get("score") or 0.0), int(item.get("chunk_index") or 0)))
-        hydrated = await self._hydrate_document_hits(dataset_id, scored_chunks[: max(top_k * 3, top_k)])
-        return hydrated[:top_k]
+        return chunks[:top_k]
 
     def _build_keyword_terms(self, query: str) -> list[str]:
-        normalized = self._normalize_keyword_query(query)
-        spans = [item.strip() for item in re.findall(r"[\u4e00-\u9fffA-Za-z0-9]+", normalized) if item.strip()]
-        if not spans:
-            return []
-
-        stop_terms = {
-            "什么",
-            "请问",
-            "一下",
-            "有关",
-            "关于",
-            "如何",
-            "哪些",
-            "怎么",
-            "是否",
-            "规定",
-            "办法",
-            "条例",
-            "法律",
-        }
-        terms: list[str] = []
-        for span in spans:
-            if span in stop_terms:
-                continue
-            terms.append(span)
-            if re.fullmatch(r"[\u4e00-\u9fff]+", span):
-                for size in (2, 3, 4):
-                    if len(span) > size:
-                        for start in range(0, len(span) - size + 1):
-                            token = span[start:start + size]
-                            if token not in stop_terms:
-                                terms.append(token)
-
-        unique_terms: list[str] = []
-        seen: set[str] = set()
-        for term in sorted(terms, key=len, reverse=True):
-            if term and term not in seen:
-                unique_terms.append(term)
-                seen.add(term)
-        return unique_terms[:20]
+        return self.retriever._build_keyword_terms(query)
 
     def _normalize_keyword_query(self, query: str) -> str:
-        normalized = (query or "").strip().lower()
-        patterns = [
-            "是什么",
-            "什么是",
-            "有哪些",
-            "有什么",
-            "是什么？",
-            "是什么?",
-            "请问",
-            "介绍一下",
-            "解释一下",
-            "帮我分析",
-            "帮我看看",
-        ]
-        for pattern in patterns:
-            normalized = normalized.replace(pattern, " ")
-        return re.sub(r"\s+", " ", normalized).strip()
+        return self.retriever._normalize_keyword_query(query)
 
     def _score_keyword_chunk(self, *, query: str, terms: list[str], content: str, document_name: str) -> float:
-        haystack = f"{document_name}\n{content}".lower()
-        if not haystack:
-            return 0.0
-
-        exact_query = self._normalize_keyword_query(query)
-        if exact_query and exact_query in haystack:
-            return 0.98
-
-        matched_weight = 0.0
-        total_weight = 0.0
-        name_bonus = 0.0
-        for term in terms:
-            weight = float(max(len(term), 1) ** 2)
-            total_weight += weight
-            if term.lower() in haystack:
-                matched_weight += weight
-                if term.lower() in document_name.lower():
-                    name_bonus += min(0.15, 0.03 * len(term))
-
-        if total_weight <= 0:
-            return 0.0
-        score = (matched_weight / total_weight) + name_bonus
-        return round(min(score, 0.99), 6)
+        return self.retriever._score_keyword_chunk(
+            query=query,
+            terms=terms,
+            content=content,
+            document_name=document_name,
+        )
 
     def _format_sse(self, payload: dict) -> bytes:
         return f"data: {json.dumps(payload, ensure_ascii=False)}\n\n".encode("utf-8")
 
     def _build_sources(self, context_chunks: list[dict], dataset_name: str) -> list[dict]:
-        return [
-            {
-                "position": index + 1,
-                "dataset_id": str(chunk.get("dataset_id") or ""),
-                "dataset_name": dataset_name,
-                "document_id": str(chunk.get("document_id") or ""),
-                "document_name": chunk.get("document_name") or chunk.get("source", ""),
-                "data_source_type": "upload_file",
-                "segment_id": chunk.get("id", ""),
-                "retriever_from": "rag",
-                "score": round(chunk.get("score", 0.0), 4),
-                "hit_count": chunk.get("hit_count", 0),
-                "word_count": len(chunk.get("text", "")),
-                "segment_position": index + 1,
-                "index_node_hash": "",
-                "content": chunk.get("text", "")[:500],
-                "page": None,
-            }
-            for index, chunk in enumerate(context_chunks)
-        ]
+        build_sources = getattr(self.retriever, "build_sources", None)
+        if callable(build_sources):
+            return build_sources(context_chunks, dataset_name)
+        return RagRetriever(hydrate_documents=False).build_sources(context_chunks, dataset_name)
 
     async def _hydrate_document_hits(self, dataset_id: int, chunks: list[dict]) -> list[dict]:
-        source_names = sorted(
-            {
-                str(chunk.get("document_name") or chunk.get("source") or "").strip()
-                for chunk in chunks
-                if str(chunk.get("document_name") or chunk.get("source") or "").strip()
-            }
-        )
-        if not source_names:
-            return chunks
-
-        async with GetAsyncSession() as session:
-            rows = (
-                await session.execute(
-                    text(
-                        """
-                        SELECT id, original_name, enabled, hit_count
-                        FROM rag_document
-                        WHERE dataset_id = :dataset_id
-                          AND deleted_at IS NULL
-                          AND original_name = ANY(:source_names)
-                        """
-                    ),
-                    {
-                        "dataset_id": dataset_id,
-                        "source_names": source_names,
-                    },
-                )
-            ).mappings().all()
-
-        document_map = {str(row["original_name"]): row for row in rows}
-        visible_chunks: list[dict] = []
-        hit_document_ids: list[int] = []
-        for chunk in chunks:
-            source_name = str(chunk.get("document_name") or chunk.get("source") or "").strip()
-            document = document_map.get(source_name)
-            if document and not bool(document.get("enabled")):
-                continue
-            if document:
-                chunk["document_id"] = document["id"]
-                chunk["dataset_id"] = dataset_id
-                chunk["document_name"] = document["original_name"]
-                chunk["hit_count"] = document.get("hit_count") or 0
-                hit_document_ids.append(int(document["id"]))
-            visible_chunks.append(chunk)
-
-        if hit_document_ids:
-            async with GetAsyncSession() as session:
-                async with session.begin():
-                    await session.execute(
-                        text(
-                            """
-                            UPDATE rag_document
-                            SET hit_count = hit_count + 1,
-                                updated_at = NOW()
-                            WHERE id = ANY(:document_ids)
-                            """
-                        ),
-                        {"document_ids": sorted(set(hit_document_ids))},
-                    )
-
-        return visible_chunks
+        return await self.retriever._hydrate_document_hits(dataset_id, chunks)
 
     def _parse_sse_event(self, chunk: str) -> dict | None:
         data_lines: list[str] = []

fastapi_modules/fastapi_leaudit/services/impl/rbacAdminServiceImpl.py

@@ -119,6 +119,41 @@ class RbacAdminServiceImpl(IRbacAdminService):
             "is_cache": True,
             "meta": {"group": "cross-review"},
         },
+        {
+            "route_path": "/contract-template",
+            "route_name": "contract-template",
+            "component": "contract-template",
+            "route_title": "合同管理",
+            "icon": "ri-file-search-line",
+            "sort_order": 50,
+            "is_hidden": False,
+            "is_cache": True,
+            "meta": {"group": "contract"},
+        },
+        {
+            "route_path": "/contract-template/search",
+            "route_name": "contract-template-search",
+            "component": "contract-template.search",
+            "route_title": "模板搜索",
+            "icon": "ri-search-line",
+            "sort_order": 1,
+            "parent_path": "/contract-template",
+            "is_hidden": False,
+            "is_cache": True,
+            "meta": {"group": "contract"},
+        },
+        {
+            "route_path": "/contract-template/list",
+            "route_name": "contract-template-list",
+            "component": "contract-template.list",
+            "route_title": "模板列表",
+            "icon": "ri-folder-line",
+            "sort_order": 2,
+            "parent_path": "/contract-template",
+            "is_hidden": False,
+            "is_cache": True,
+            "meta": {"group": "contract"},
+        },
         {
             "route_path": "/cross-checking/upload",
             "route_name": "cross-checking-upload",

rules/行政处罚/rules.yaml

@@ -1392,6 +1392,67 @@ rules:
     references_laws:
     - 《中华人民共和国行政处罚法》第五十九条
     type: deterministic
+  - rule_id: JZ-JD-005
+    name: 案由及裁量标准适用准确性
+    desc: 结合处罚决定书认定依据、处罚依据、罚款项目和罚款金额，检索案由与裁量标准，判断处罚种类和罚款幅度是否适用准确。
+    risk: medium
+    score: 10
+    scope:
+    - 处罚决定书
+    rag:
+      collection: general_legal_kb
+      top_k: 5
+      source_names:
+      - 广东省烟草专卖行政处罚裁量执行标准-rag.md
+      - 案由_行政处罚与反走私管理治理办法.md
+      query_template: |
+        认定依据：{{处罚决定书.认定依据}}
+        处罚依据：{{处罚决定书.处罚依据}}
+        罚款项目：{{处罚决定书.罚款项目}}
+        罚款基数：{{处罚决定书.罚款基数}}
+        罚款比例：{{处罚决定书.罚款比例}}
+        罚款总额：{{处罚决定书.罚款总额}}
+        问题：检索对应案由、裁量档次、处罚种类和罚款幅度
+      inject_as: rag_context
+      resources_as: rag_resources
+    stages:
+    - id: '1'
+      check: required
+      fields:
+      - 处罚决定书.认定依据
+      - 处罚决定书.处罚依据
+      - 处罚决定书.罚款项目
+      - 处罚决定书.罚款基数
+      - 处罚决定书.罚款比例
+      - 处罚决定书.罚款总额
+    - id: '2'
+      check: ai
+      prompt: |
+        请结合检索到的法律知识和卷宗处罚决定书字段，判断案由、裁量档次、处罚种类和罚款幅度是否适用准确。
+
+        【检索依据】
+        {{rag_context}}
+
+        【处罚决定书字段】
+        认定依据：{{处罚决定书.认定依据}}
+        处罚依据：{{处罚决定书.处罚依据}}
+        罚款项目：{{处罚决定书.罚款项目}}
+        罚款基数：{{处罚决定书.罚款基数}}
+        罚款比例：{{处罚决定书.罚款比例}}
+        罚款总额：{{处罚决定书.罚款总额}}
+
+        【判断要求】
+        1. 判断违法事实对应案由是否准确；
+        2. 判断处罚依据是否能支撑对应处罚种类；
+        3. 判断罚款基数、比例、总额是否落在裁量标准允许幅度内；
+        4. 若检索依据不足以确认，应给出 warn，不要编造依据。
+    logic: 1 AND 2
+    messages:
+      pass: 案由、裁量档次、处罚种类和罚款幅度适用准确。
+      fail: 案由、裁量档次、处罚种类或罚款幅度可能适用不准确，请核对。
+    references_laws:
+    - 《中华人民共和国行政处罚法》第五十九条
+    type: ai_rule
 - group: JZG-SD
   rules:
   - rule_id: JZ-SD-001

scripts/创建sql/seed_contract_templates_rbac.sql

@@ -4,7 +4,7 @@ BEGIN;
 -- LeAudit Platform Contract Template RBAC Seed
 -- 目标：
 --   1. 补齐合同模板读写删权限
---   2. 给 super_admin / provincial_admin / admin 分配模板权限
+--   2. 给角色分配模板权限，其中上传/更新/删除仅开放给地区管理员 admin
 -- 说明：
 --   - 依赖 user_rbac_schema_patch.sql
 --   - 依赖合同模板前端路由已存在于 sys_routes
@@ -108,16 +108,10 @@ seed(role_key, permission_key, grant_type, data_scope) AS (
         ('super_admin', 'contract_template:list:read', 'GRANT', 'ALL'),
         ('super_admin', 'contract_template:search:read', 'GRANT', 'ALL'),
         ('super_admin', 'contract_template:detail:read', 'GRANT', 'ALL'),
-        ('super_admin', 'contract_template:create:write', 'GRANT', 'ALL'),
-        ('super_admin', 'contract_template:update:write', 'GRANT', 'ALL'),
-        ('super_admin', 'contract_template:delete:delete', 'GRANT', 'ALL'),
 
         ('provincial_admin', 'contract_template:list:read', 'GRANT', 'ALL'),
         ('provincial_admin', 'contract_template:search:read', 'GRANT', 'ALL'),
         ('provincial_admin', 'contract_template:detail:read', 'GRANT', 'ALL'),
-        ('provincial_admin', 'contract_template:create:write', 'GRANT', 'ALL'),
-        ('provincial_admin', 'contract_template:update:write', 'GRANT', 'ALL'),
-        ('provincial_admin', 'contract_template:delete:delete', 'GRANT', 'ALL'),
 
         ('admin', 'contract_template:list:read', 'GRANT', 'DEPT'),
         ('admin', 'contract_template:search:read', 'GRANT', 'DEPT'),
@@ -149,4 +143,15 @@ ON CONFLICT (role_id, permission_id) DO UPDATE SET
     data_scope = EXCLUDED.data_scope,
     updated_at = NOW();
 
+DELETE FROM role_permissions rp
+USING roles r, permissions p
+WHERE rp.role_id = r.id
+  AND rp.permission_id = p.id
+  AND r.role_key IN ('super_admin', 'provincial_admin')
+  AND p.permission_key IN (
+      'contract_template:create:write',
+      'contract_template:update:write',
+      'contract_template:delete:delete'
+  );
+
 COMMIT;

tests/test_leaudit_rag_bridge.py

@@ -0,0 +1,128 @@
+from __future__ import annotations
+
+import tempfile
+import sys
+import types
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+from leaudit.dsl.schema import Metadata, Rule, RuleAuthoringGroup, RulesFile, Stage
+from leaudit.engine.models import EvaluationResult
+from leaudit.extraction.bundle import bundle_from_single
+from leaudit.extraction.models import ExtractionResult, FieldValue
+from leaudit.ocr.models import OcrResult, Page
+
+from fastapi_modules.fastapi_leaudit.leaudit_bridge.pipeline import LauditPipeline
+
+
+class FakeOcrClient:
+  async def ocr(self, file_path):
+    return OcrResult(
+      pages=[Page(page_num=1, text="处罚决定书")],
+      full_text="处罚决定书",
+    )
+
+
+class FakeStorage:
+  async def update_document_status(self, document_id, status):
+    return None
+
+  async def save_ocr_result(self, document_id, ocr_result):
+    return None
+
+  async def save_extraction_result(self, document_id, extraction_bundle):
+    return None
+
+  async def save_evaluation_results(self, document_id, rules_file, evaluation_result, extraction_bundle):
+    return None
+
+
+class FakeRetriever:
+  pass
+
+
+def _rules_file() -> RulesFile:
+  return RulesFile(
+    metadata=Metadata(
+      type_id="test.rag_bridge",
+      name="RAG bridge test",
+      version="1.0",
+      last_updated="2026-05-21",
+    ),
+    rules=[
+      RuleAuthoringGroup(
+        group="测试",
+        rules=[
+          Rule(
+            rule_id="R-RAG",
+            name="RAG 规则",
+            risk="medium",
+            score=1,
+            stages=[Stage(id="1", check="required", field="处罚决定书.处罚依据")],
+            logic="1",
+            messages={"pass": "ok", "fail": "missing"},
+          )
+        ],
+      )
+    ],
+  )
+
+
+class LeauditRagBridgeTest(unittest.IsolatedAsyncioTestCase):
+  async def test_pipeline_passes_injected_rag_retriever_to_evaluation(self):
+    captured = {}
+    retriever = FakeRetriever()
+    field_value = FieldValue(value="依据", confidence=0.9)
+    object.__setattr__(field_value, "position", None)
+    extraction = bundle_from_single(
+      ExtractionResult(
+        fields={"处罚决定书.处罚依据": field_value},
+        source_text="处罚决定书",
+      )
+    )
+
+    async def fake_dispatch_extract(*args, **kwargs):
+      return extraction
+
+    async def fake_determine_phase(*args, **kwargs):
+      return "executed"
+
+    async def fake_evaluate_extraction(*args, **kwargs):
+      captured["retriever"] = kwargs.get("retriever")
+      return EvaluationResult()
+
+    class TestPipeline(LauditPipeline):
+      async def _extract_and_save_case_number(self, document_id, ocr_result):
+        return None
+
+    with tempfile.NamedTemporaryFile() as tmp:
+      pipeline = TestPipeline(
+        ocr_client=FakeOcrClient(),
+        storage_adapter=FakeStorage(),
+        rag_retriever=retriever,
+      )
+      with patch(
+        "fastapi_modules.fastapi_leaudit.leaudit_bridge.pipeline.dispatch_extract",
+        fake_dispatch_extract,
+      ), patch(
+        "fastapi_modules.fastapi_leaudit.leaudit_bridge.pipeline.determine_phase",
+        fake_determine_phase,
+      ), patch(
+        "fastapi_modules.fastapi_leaudit.leaudit_bridge.pipeline.evaluate_extraction",
+        fake_evaluate_extraction,
+      ), patch.dict(
+        sys.modules,
+        {
+          "leaudit.extraction.coordinate_resolver": types.SimpleNamespace(
+            resolve_bundle_positions=lambda *args, **kwargs: None
+          )
+        },
+      ):
+        await pipeline.run(1, Path(tmp.name), _rules_file())
+
+    self.assertIs(captured["retriever"], retriever)
+
+
+if __name__ == "__main__":
+  unittest.main()

tests/test_rag_retriever.py

@@ -0,0 +1,102 @@
+from __future__ import annotations
+
+import unittest
+
+from fastapi_modules.fastapi_leaudit.rag_engine.retriever import RagRetriever
+
+
+class FakeCollection:
+  def query(self, **kwargs):
+    return {
+      "ids": [["seg-1", "seg-2"]],
+      "documents": [["烟草处罚裁量标准内容", "其他来源内容"]],
+      "metadatas": [[
+        {
+          "document_name": "广东省烟草专卖行政处罚裁量执行标准-rag.md",
+          "chunk_index": 0,
+        },
+        {
+          "document_name": "其他.md",
+          "chunk_index": 1,
+        },
+      ]],
+      "distances": [[0.0, 0.2]],
+    }
+
+
+class FallbackCollection:
+  def query(self, **kwargs):
+    raise RuntimeError("vector unavailable")
+
+  def get(self, **kwargs):
+    return {
+      "ids": ["seg-fallback"],
+      "documents": ["未在当地烟草专卖批发企业进货，对应裁量档次内容"],
+      "metadatas": [
+        {
+          "document_name": "案由_行政处罚与反走私管理治理办法.md",
+          "chunk_index": 3,
+        }
+      ],
+    }
+
+
+class FakeChroma:
+  def __init__(self):
+    self.collection_name = None
+
+  def get_or_create_collection(self, name):
+    self.collection_name = name
+    return FakeCollection()
+
+
+class FallbackChroma:
+  def get_or_create_collection(self, name):
+    return FallbackCollection()
+
+
+class RagRetrieverTest(unittest.IsolatedAsyncioTestCase):
+  async def test_retrieve_from_collection_filters_sources_and_builds_resources(self):
+    chroma = FakeChroma()
+    retriever = RagRetriever(
+      chroma_client=chroma,
+      embed_texts=lambda texts, model_name="": [[0.1, 0.2] for _ in texts],
+      hydrate_documents=False,
+    )
+
+    result = await retriever.retrieve(
+      query="罚款幅度",
+      collection_name="general_legal_kb",
+      top_k=5,
+      source_names=["广东省烟草专卖行政处罚裁量执行标准-rag.md"],
+    )
+
+    self.assertEqual(chroma.collection_name, "general_legal_kb")
+    self.assertIn("烟草处罚裁量标准内容", result.rag_context)
+    self.assertNotIn("其他来源内容", result.rag_context)
+    self.assertEqual(len(result.rag_resources), 1)
+    self.assertEqual(
+      result.rag_resources[0]["document_name"],
+      "广东省烟草专卖行政处罚裁量执行标准-rag.md",
+    )
+
+  async def test_retrieve_uses_keyword_fallback_when_vector_search_fails(self):
+    retriever = RagRetriever(
+      chroma_client=FallbackChroma(),
+      embed_texts=lambda texts, model_name="": [[0.1, 0.2] for _ in texts],
+      hydrate_documents=False,
+    )
+
+    result = await retriever.retrieve(
+      query="未在当地烟草专卖批发企业进货",
+      collection_name="general_legal_kb",
+      source_names=["案由_行政处罚与反走私管理治理办法.md"],
+    )
+
+    self.assertIn("对应裁量档次内容", result.rag_context)
+    self.assertEqual(len(result.chunks), 1)
+    self.assertEqual(result.rag_resources[0]["segment_id"], "seg-fallback")
+
+
+if __name__ == "__main__":
+  unittest.main()