基于 GenAI 的医疗器械说明书审核方案

prompt:
# 说明书差异对比提示词
# 使用位置: process/md_diff_compare.py -> compare_whole_document()
instruction_diff_prompt: |
您是一个说明书差异对比专家。我们会给您提供两个版本的说明书文档（Markdown 格式），您需要找出所有的**实质性内容变更**。
** 超级重要 - 必须严格遵守 **
**对比方法（必须按此步骤执行）：**
1. **第一步：格式清理**
- 将文档 A 的所有 Markdown 标记去掉，得到纯文本 A
- 将文档 B 的所有 Markdown 标记去掉，得到纯文本 B
- 例如：`# 标题` → `标题`，`## 标题` → `标题`，`**文字**` → `文字`
2. **第二步：纯文本对比**
- 只对比纯文本 A 和纯文本 B
- 逐字检查差异
3. **第三步：判断是否输出**
- 纯文本相同 → **不输出**（这是格式差异）
- 纯文本不同 → **输出**（这是内容差异）
**示例说明：**
- `# 使用说明` vs `## 使用说明` → 去掉#后都是`使用说明` → **不输出**
- `**警告**` vs `警告` → 去掉**后都是`警告` → **不输出**
- `使用说明` vs `使用方法` → 纯文本不同 → **输出**
**重要原则 - 区分格式差异和内容差异**：
**禁止输出的差异类型**（纯格式变化，不是实质性差异）：
1. **格式标记的增删**：
- 加粗标记：`**文字**` vs `文字`
- 斜体标记：`*文字*` vs `文字`
- 下划线、删除线等格式标记
- **空格调整**：
- 空格数量变化
- 空格位置变化
- 空格的增加或删除
- **完全忽略所有空格差异**
3. **页码引用变化**（如"见 32 页"改为"见 31 页"）
4. **锚点链接、HTML 标签的增删**（如 <span id="xxx"></span>）
5. **章节标题层级变化**（# vs ## vs ### vs ####，标题文字相同）
6. **换行、缩进的变化**（文字内容相同）
7. **列表格式变化**（有序 vs 无序，列表内容相同）
8. **目录的变化**（目录本身不对比）
9. **页眉、页脚、页码本身**
**必须输出的差异类型**（实质性内容变化）：
1. **文字内容的增删改**（非常重要，必须全部识别）：
- **单个字的变化**（如"使用"改为"使明"，"的"改为"地"）
- **词语替换**（如"使用"改为"应用"，"患者"改为"受试者"）
- **语序调整**（如"请勿使用"改为"使用请勿"）
- **标点符号的实质性变化**（如"。"改为"！"，"，"改为"；"）
- **助词、连词的增删**（如"的"、"了"、"和"、"或"的增删）
- **量词、单位的变化**（如"个"改为"只"，"次"改为"回"）
- **句子的增加或删除**
- **段落的增加或删除**
- **任何导致语义或语气变化的文字调整**
2. **数值、数据的变化**：
- 数字变化（如"100mg"改为"200mg"，"3 次"改为"2 次"）
- 百分比变化（如"50%"改为"60%"）
- 时间、日期的变化（如"2 小时"改为"3 小时"）
- 范围的变化（如"1-3 次"改为"2-4 次"）
3. **关键信息的变更**：
- 警告内容的改变
- 注意事项的改变
- 使用说明的改变
- 技术参数的改变
- 适应症、禁忌症的改变
**判断标准（严格模式）**：
- 去掉所有格式标记后，逐字对比纯文本内容
- **只要有任何一个字不同 → 就是差异，必须输出**
- **只有完全相同（包括标点符号）→ 才不是差异**
- 特别注意：
* "患者"和"受试者"虽然含义相近，但用词不同，**必须输出**
* "使用"和"应用"虽然含义相近，但用词不同，**必须输出**
* "的"和"地"虽然都是助词，但不同，**必须输出**
* 任何细微的表述差异都要识别

# =============================================================================
# 1. PDF 转换（PDF -> Markdown，保留图片）
#    来源：Phase1/process/pdf_to_md_with_images.py
#    核心：用 marker-pdf 模型把 PDF 转成 Markdown 并提取图片，带 MD5 缓存
# =============================================================================
# 全局模型缓存（避免每次都重新加载 10-30 秒的模型）
_MODEL_DICT = None
_CONVERTER = None
CACHE_DIR = Path(".cache/pdf_to_md")
def get_pdf_hash(pdf_file):
    """计算 PDF 的 MD5，用作缓存 key（同一份 PDF 不重复解析）。"""
    hash_md5 = hashlib.md5()
    with open(pdf_file, "rb") as f:
        for chunk in iter(lambda: f.read(4096), b""):
            hash_md5.update(chunk)
    return hash_md5.hexdigest()
def convert_pdf_to_md_with_images(pdf_file, output_dir=None, use_cache=True):
    """
    将 PDF 转换为 Markdown，保留图片位置信息。
    Returns:
        dict: {'markdown', 'images', 'image_paths', 'metadata', 'output_dir', ...}
    """
    global _MODEL_DICT, _CONVERTER
    pdf_path = Path(pdf_file)
    if not pdf_path.exists():
        raise FileNotFoundError(f"PDF 文件不存在: {pdf_file}")
    # ——缓存命中则直接返回（此处省略缓存图片重存的细节，见原文件）——
    if use_cache:
        cache_path = CACHE_DIR / f"{pdf_path.stem}_{get_pdf_hash(pdf_file)}.pkl"
        if cache_path.exists():
            with open(cache_path, "rb") as f:
                result = pickle.load(f)
            result["from_cache"] = True
            logger.info("✓ 使用缓存结果，跳过 PDF 转换")
            return result
    if output_dir is None:
        output_dir = pdf_path.parent / f"{pdf_path.stem}_images"
    output_dir = Path(output_dir)
    output_dir.mkdir(parents=True, exist_ok=True)
    # ——关键：调用 marker-pdf 做转换（首次加载模型较慢，之后复用）——
    from marker.converters.pdf import PdfConverter
    from marker.models import create_model_dict
    if _MODEL_DICT is None:
        logger.info("首次加载 marker-pdf 模型（这可能需要 10-30 秒）...")
        _MODEL_DICT = create_model_dict()
    if _CONVERTER is None:
        _CONVERTER = PdfConverter(artifact_dict=_MODEL_DICT)
    rendered = _CONVERTER(str(pdf_file))          # <-- 实际转换发生处
    markdown_text = rendered.markdown
    images = rendered.images
    # 把提取出的图片落盘，并记录 id -> 路径 的映射
    image_paths = {}
    for img_id, img_data in images.items():
        img_filename = img_id if "." in img_id else f"{img_id}.png"
        img_path = output_dir / img_filename
        img_data.save(img_path)
        image_paths[img_id] = str(img_path)
    metadata = {
        "pages": len(rendered.pages) if hasattr(rendered, "pages") else 0,
        "has_images": len(images) > 0,
        "image_count": len(images),
    }
    result = {
        "markdown": markdown_text,
        "images": images,
        "image_paths": image_paths,
        "metadata": metadata,
        "output_dir": str(output_dir),
        "from_cache": False,
    }
    if use_cache:
        CACHE_DIR.mkdir(parents=True, exist_ok=True)
        cache_path = CACHE_DIR / f"{pdf_path.stem}_{get_pdf_hash(pdf_file)}.pkl"
        with open(cache_path, "wb") as f:
            pickle.dump(result, f)
    return result

# =============================================================================
# 2. Bedrock 调用
#    来源：Phase1/llms/aws_claude.py
#    核心：初始化 bedrock-runtime 客户端 + invoke_model_bedrock() 发起推理
#          同时兼容 Claude 与 Amazon Nova 两种请求/响应格式，带重试
# =============================================================================
# —— 配置从 config.yaml 读取（此处给出等价默认值，实际以配置为准）——
# USE_BEDROCK        = config['model']['use_bedrock']
# BEDROCK_AWS_REGION = config['model']['bedrock_region']       (默认 us-east-1)
# BEDROCK_MODEL_ID   = config['model']['bedrock_model']        
# BEDROCK_PROFILE    = config['model']['bedrock_profile']
BEDROCK_AWS_REGION = "us-east-1"
BEDROCK_MODEL_ID = "anthropic.claude-3-5-sonnet-20240620-v1:0"
BEDROCK_PROFILE = None
# 初始化 Bedrock 客户端（配置更长的超时与重试）
_boto_cfg = boto3.session.Config(
    read_timeout=600,      # 10 分钟读取超时
    connect_timeout=120,   # 2 分钟连接超时
    retries={"max_attempts": 5},
)
if BEDROCK_PROFILE:
    _session = boto3.Session(profile_name=BEDROCK_PROFILE)
    bedrock_client = _session.client(
        "bedrock-runtime", region_name=BEDROCK_AWS_REGION, config=_boto_cfg)
else:
    bedrock_client = boto3.client(
        "bedrock-runtime", region_name=BEDROCK_AWS_REGION, config=_boto_cfg)
@retry(
    stop=stop_after_attempt(8),
    wait=wait_exponential(multiplier=2, min=5, max=120),
    retry=retry_if_exception_type((Exception,)),
    before_sleep=before_sleep_log(logger, logging.WARNING),
)
def invoke_model_bedrock(system_prompt, messages, max_tokens=4096):
    """
    使用 AWS Bedrock 调用模型（支持 Claude 和 Amazon Nova）。
    这是整个项目里真正“调大模型”的地方。
    """
    if "anthropic.claude" in BEDROCK_MODEL_ID:
        # Claude 请求体格式
        body = {
            "anthropic_version": "bedrock-2023-05-31",
            "max_tokens": max_tokens,
            "temperature": 0,
            "system": system_prompt,
            "messages": messages,
        }
    elif "amazon.nova" in BEDROCK_MODEL_ID:
        # Amazon Nova 请求体格式：把 system 合并进 user 消息
        if system_prompt:
            nova_messages = [{
                "role": "user",
                "content": [{"text": f"System: {system_prompt}\n\nUser: {messages[0]['content']}"}],
            }]
        else:
            nova_messages = [{"role": "user", "content": [{"text": messages[0]["content"]}]}]
        body = {
            "messages": nova_messages,
            "inferenceConfig": {"maxTokens": max_tokens, "temperature": 0},
        }
    else:
        raise ValueError(f"Unsupported model: {BEDROCK_MODEL_ID}")
    response = bedrock_client.invoke_model(          # <-- 实际调用发生处
        modelId=BEDROCK_MODEL_ID, body=json.dumps(body))
    response_body = json.loads(response["body"].read())
    # 按模型类型解析响应文本
    if "anthropic.claude" in BEDROCK_MODEL_ID:
        return response_body["content"][0]["text"]
    elif "amazon.nova" in BEDROCK_MODEL_ID:
        return response_body["output"]["message"]["content"][0]["text"]
def user_message(content):
    """把字符串包装成 Bedrock messages 结构。"""
    return [{"role": "user", "content": content}]

# =============================================================================
# 3. 置信度分析（相似度打分）—— 分两层
#    3a. 文本块相似度：Phase1/process/md_diff_compare.py
#        用 Jaccard 相似度判断“变更前/后段落是否是同一块”，用于对齐
#    3b. 差异高亮阈值：Phase1/process/diff_result_export_with_images.py
#        用 difflib 字符级相似度，低于阈值(默认 0.3)则判为“差异过大不高亮”
# =============================================================================
def calculate_similarity(text1, text2):
    """
    计算两个文本的相似度（Jaccard：共同词汇占比）。
    返回 0~1，越大越可能是同一段内容。
    """
    if not text1 or not text2:
        return 0.0
    words1 = set(text1.lower().split())
    words2 = set(text2.lower().split())
    if not words1 or not words2:
        return 0.0
    intersection = len(words1 & words2)
    union = len(words1 | words2)
    return intersection / union if union > 0 else 0.0
def align_chunks_by_similarity(chunks_before, chunks_after, titles_before, titles_after):
    """
    基于相似度把变更前/后的块两两对齐（节选自 align_chunks 的“相似度匹配”一遍）。
    综合分 = 标题相似度*0.7 + 内容相似度*0.3；超过阈值 0.5 才认为匹配。
    """
    aligned_pairs = []
    used_before, used_after = set(), set()
    for i in range(len(chunks_before)):
        if i in used_before:
            continue
        best_match = -1
        best_similarity = 0.5  # 阈值：低于 0.5 不认为是同一块
        for j in range(len(chunks_after)):
            if j in used_after:
                continue
            title_sim = calculate_similarity(titles_before[i], titles_after[j])
            content_sim = calculate_similarity(chunks_before[i][:500], chunks_after[j][:500])
            similarity = title_sim * 0.7 + content_sim * 0.3   # <-- 置信度综合分
            if similarity > best_similarity:
                best_similarity = similarity
                best_match = j
        if best_match >= 0:
            aligned_pairs.append((i, best_match, "modified"))
            used_before.add(i)
            used_after.add(best_match)
            logger.info(f"≈ 相似匹配: 块{i+1} <-> 块{best_match+1} (相似度: {best_similarity:.2f})")
    return aligned_pairs
# ---- 3b. 字符级相似度阈值控制高亮（diff_result_export_with_images.py）-----
def load_diff_highlight_config():
    """从 config.yaml 读取相似度阈值（diff_highlight.similarity_threshold，默认 0.3）。"""
    try:
        import yaml
        config_path = Path(__file__).parent / "config" / "config.yaml"
        with open(config_path, "r", encoding="utf-8") as f:
            config = yaml.safe_load(f)
            return config.get("diff_highlight", {}).get("similarity_threshold", 0.3)
    except Exception as e:
        logger.warning(f"无法加载差异高亮配置，使用默认值 0.3: {e}")
        return 0.3
def should_highlight(text, other_text, similarity_threshold=0.3):
    """
    判断两段文本是否值得做“字符级差异高亮”。
    用 difflib.SequenceMatcher 算相似度比率，低于阈值说明差异太大，
    直接显示原文而不做逐字高亮（避免满屏红绿的误报）。
    Returns:
        (bool 是否高亮, float 相似度比率)
    """
    if not text or not other_text:
        return True, 0.0  # 只有一侧有内容 = 纯新增/删除，整体高亮
    matcher = difflib.SequenceMatcher(None, text, other_text)
    similarity_ratio = matcher.ratio()               # <-- 置信度：字符级相似度
    if similarity_ratio < similarity_threshold:
        logger.info(f"文本相似度 {similarity_ratio:.2%} 低于阈值 "
                    f"{similarity_threshold:.0%}，跳过高亮")
        return False, similarity_ratio
    # 高于阈值时，用 matcher.get_opcodes() 逐段标红(删除)/标绿(新增)
    # 具体渲染见原文件 highlight_text_diff()
    return True, similarity_ratio