AIyaodianzhushou/diagnose.py

# -*- coding: utf-8 -*-
"""
Diagnostic script: Analyze why "penicillin allergy" query returns info about Qingkailing injection.

Traces the full RAG pipeline:
  User query -> Intent classification -> Vector retrieval -> Reranking -> LLM generation
"""
import sys
import os
import json
import re
import asyncio
import argparse
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent / "backend-python"))
sys.path.insert(0, str(Path(__file__).resolve().parent / "data-pipeline"))


def classify_intent(query: str) -> str:
    q = query.strip()
    negation_patterns = [
        r"不是", r"没有", r"并非", r"不算", r"不属于",
        r"这不是", r"我没有", r"不包含", r"不涉及",
    ]
    has_negation = any(re.search(pat, q) for pat in negation_patterns)
    if has_negation:
        return "drug_query"

    usage_keywords = [
        "怎么吃", "吃多少", "怎么用", "一天几次", "多长时间",
        "能一起吃", "孕妇能用", "儿童用量", "哺乳期",
        "饭前还是饭后", "空腹", "过量", "漏服", "停药",
        "副作用多大", "伤肝吗", "伤肾吗", "安全吗",
    ]
    safety_sections = [
        "副作用", "不良反应", "禁忌", "注意事项",
        "能不能", "可以吗", "会不会",
    ]
    regulation_keywords = [
        "凡例", "通则规定", "制剂通则",
        "一般规定", "通用技术要求", "检验方法通则",
    ]
    exam_keywords = [
        "执业药师考试", "考点", "历年真题", "考试大纲",
        "高频考点", "报名时间",
    ]
    symptom_keywords = [
        "吃了什么药", "吃什么药", "该吃", "推荐用药", "推荐下用药",
        "买什么药", "推荐什么药", "用什么药", "用药建议",
        "发烧", "咳嗽", "感冒", "腹泻", "头疼", "头痛",
        "嗓子疼", "流鼻涕", "鼻塞", "肚子疼", "胃疼",
        "过敏", "皮肤痒", "失眠", "便秘", "牙疼",
        "体温", "多少度", "退烧", "止痛", "止泻",
    ]

    if any(kw in q for kw in usage_keywords):
        return "usage_guide"
    if any(kw in q for kw in safety_sections):
        return "usage_guide"
    if any(kw in q for kw in symptom_keywords):
        return "symptom_advice"
    if any(kw in q for kw in regulation_keywords):
        return "regulation"
    if any(kw in q for kw in exam_keywords):
        return "exam_tutor"
    return "drug_query"


def tokenize_query(query: str) -> set:
    tokens = set()
    for n in range(2, 5):
        for i in range(len(query) - n + 1):
            seg = query[i:i + n]
            if all('一' <= c <= '鿿' for c in seg):
                tokens.add(seg)
    for word in re.findall(r'[a-zA-Z0-9]+', query):
        tokens.add(word.lower())
    return tokens


def keyword_overlap_score(query_terms: set, content: str) -> float:
    if not query_terms:
        return 0.0
    matched = sum(1 for t in query_terms if t in content)
    return matched / len(query_terms)


def analyze_data_files():
    data_dir = Path(__file__).resolve().parent / "data-pipeline" / "data"
    results = {
        "penicillin_drugs": [],
        "qingkailing_drugs": [],
        "penicillin_allergy_mentions": [],
        "allergy_related_drugs": [],
        "all_drugs_with_allergy_sections": [],
    }

    for json_file in sorted(data_dir.glob("*.json")):
        try:
            with open(json_file, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            continue
        if not isinstance(data, list):
            continue

        for item in data:
            if not isinstance(item, dict):
                continue
            name = item.get("name", "")
            sections = item.get("sections", {})

            if "青霉素" in name:
                results["penicillin_drugs"].append({
                    "file": json_file.name, "name": name,
                    "sections": list(sections.keys()),
                    "allergy_info": any("过敏" in str(v) for v in sections.values()),
                })
            if "清开灵" in name:
                results["qingkailing_drugs"].append({
                    "file": json_file.name, "name": name,
                    "sections": list(sections.keys()),
                    "allergy_info": any("过敏" in str(v) for v in sections.values()),
                })
            for sk, sv in sections.items():
                if "青霉素" in str(sv) and "过敏" in str(sv):
                    results["penicillin_allergy_mentions"].append({
                        "file": json_file.name, "name": name, "section": sk,
                    })
            for sk, sv in sections.items():
                if sk in ("禁忌", "不良反应", "注意事项") and "过敏" in str(sv):
                    if name not in [x["name"] for x in results["all_drugs_with_allergy_sections"]]:
                        results["all_drugs_with_allergy_sections"].append({
                            "file": json_file.name, "name": name,
                        })
    return results


def analyze_chunk_examples():
    """Simulate what chunks look like in the database for key drugs."""
    data_dir = Path(__file__).resolve().parent / "data-pipeline" / "data"
    examples = []

    for json_file in sorted(data_dir.glob("*.json")):
        try:
            with open(json_file, "r", encoding="utf-8") as f:
                data = json.load(f)
        except Exception:
            continue
        if not isinstance(data, list):
            continue

        for item in data:
            if not isinstance(item, dict):
                continue
            name = item.get("name", "")
            if "青霉素" in name or "清开灵" in name:
                sections = item.get("sections", {})
                source = item.get("source", {})
                source_str = f"{source.get('version', '')} {source.get('volume', '')}"

                for sk, sv in sections.items():
                    if not sv or len(sv.strip()) < 10:
                        continue
                    chunk = f"[{name} - {sk}]\n{sv}\n\nSource: {source_str}"
                    examples.append({
                        "drug": name, "section": sk,
                        "chunk_preview": chunk[:300] + "..." if len(chunk) > 300 else chunk,
                    })
    return examples


async def get_embedding(query: str) -> list:
    """Call DashScope text-embedding-v3 API to get embedding vector."""
    import httpx
    api_key = os.environ.get("QWEN_API_KEY", "")
    if not api_key:
        env_file = Path(__file__).resolve().parent / ".env"
        if env_file.exists():
            for line in open(env_file, encoding="utf-8"):
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    key, _, val = line.partition("=")
                    os.environ.setdefault(key.strip(), val.strip())
        api_key = os.environ.get("QWEN_API_KEY", "")

    if not api_key:
        return []

    EMBEDDING_URL = "https://dashscope.aliyuncs.com/api/v1/services/embeddings/text-embedding/text-embedding"
    async with httpx.AsyncClient(timeout=30) as client:
        resp = await client.post(
            EMBEDDING_URL,
            headers={
                "Content-Type": "application/json",
                "Authorization": f"Bearer {api_key}",
            },
            json={
                "model": "text-embedding-v3",
                "input": {"texts": [query]},
                "parameters": {"text_type": "query"},
            },
        )
        data = resp.json()
    if data.get("code") and data.get("code") != "":
        raise RuntimeError(f"Embedding error: {data.get('message')}")
    return data["output"]["embeddings"][0]["embedding"]


async def analyze_embedding_space():
    queries = [
        "青霉素过敏处理",
        "青霉素过敏",
        "过敏反应处理",
        "清开灵注射液不良反应",
        "清开灵注射液禁忌",
        "清开灵注射液过敏",
        "阿莫西林过敏",
        "头孢克肟过敏",
    ]
    print("\n" + "=" * 70)
    print("[Embedding Space Analysis]")
    print("=" * 70)

    api_key = os.environ.get("QWEN_API_KEY", "")
    if not api_key:
        env_file = Path(__file__).resolve().parent / ".env"
        if env_file.exists():
            for line in open(env_file, encoding="utf-8"):
                line = line.strip()
                if line and not line.startswith("#") and "=" in line:
                    key, _, val = line.partition("=")
                    os.environ.setdefault(key.strip(), val.strip())
        api_key = os.environ.get("QWEN_API_KEY", "")

    if not api_key:
        print("  SKIP: QWEN_API_KEY not set")
        return {}

    import math
    embeddings = {}
    for q in queries:
        try:
            vec = await get_embedding(q)
            embeddings[q] = vec
            print(f"  OK: {q} (dim={len(vec)})")
        except Exception as e:
            print(f"  FAIL: {q}: {e}")

    if not embeddings:
        return {}

    def cosine_sim(a, b):
        dot = sum(x * y for x, y in zip(a, b))
        na = math.sqrt(sum(x * x for x in a))
        nb = math.sqrt(sum(y * y for y in b))
        return dot / (na * nb) if na and nb else 0

    print("\nCosine Similarity Matrix:")
    header = [q[:10] for q in queries if q in embeddings]
    print(f"{'':22s}", end="")
    for h in header:
        print(f"{h:>10s}", end="")
    print()
    for q1 in queries:
        if q1 not in embeddings:
            continue
        print(f"{q1[:22]:22s}", end="")
        for q2 in queries:
            if q2 not in embeddings:
                continue
            sim = cosine_sim(embeddings[q1], embeddings[q2])
            print(f"{sim:10.4f}", end="")
        print()

    target = "青霉素过敏处理"
    if target in embeddings:
        print(f"\nSimilarity from '{target}' to others:")
        others = sorted(
            [(q, cosine_sim(embeddings[target], embeddings[q]))
             for q in embeddings if q != target],
            key=lambda x: -x[1]
        )
        for q, sim in others:
            bar = "=" * int(sim * 30)
            print(f"  {q[:30]:30s}: {sim:.4f} {bar}")

    return embeddings


def main():
    print("=" * 70)
    print("DIAGNOSIS: Why 'penicillin allergy' returns Qingkailing injection info")
    print("=" * 70)

    # === Step 1: Intent Classification ===
    query = "青霉素过敏处理"
    intent = classify_intent(query)
    print(f"\n[Step 1] Intent Classification:")
    print(f"  Query: '{query}'")
    print(f"  Intent: '{intent}'")
    print(f"  Matched keyword '过敏' in symptom_keywords")
    print(f"  >> PROBLEM: Query about managing a known allergy is routed to")
    print(f"     'symptom_advice' (symptom-based drug recommendation).")
    print(f"     This tells LLM to recommend drugs, not manage an allergy.")

    # Also test alternative queries
    alt_queries = ["青霉素过敏了怎么办", "青霉素过敏急救", "对青霉素过敏", "青霉素过敏替代药"]
    print(f"\n  Alternative query intents:")
    for q in alt_queries:
        print(f"    '{q}' -> '{classify_intent(q)}'")

    # === Step 2: Query Tokenization (for reranker) ===
    tokens = tokenize_query(query)
    print(f"\n[Step 2] Reranker Tokenization:")
    print(f"  Tokens ({len(tokens)}): {sorted(tokens)}")
    print(f"  >> PROBLEM: '过敏' is one of the most common terms in drug")
    print(f"     adverse reaction sections. Any chunk containing '过敏'")
    print(f"     gets a keyword score boost in reranking.")
    print(f"     '青霉素' (2-gram) only appears in penicillin-related chunks.")

    # === Step 3: Data file analysis ===
    print(f"\n[Step 3] Data File Analysis:")
    data_results = analyze_data_files()

    for category, items in data_results.items():
        if items:
            print(f"\n  [{category}] ({len(items)} entries):")
            for item in items[:15]:
                if isinstance(item, dict):
                    name = item.get("name", "?")
                    fname = item.get("file", "?")
                    print(f"    - {name} (in {fname})", end="")
                    if "section" in item:
                        print(f" [section: {item['section']}]")
                    elif "sections" in item:
                        print(f" [sections: {item.get('sections', [])}]")
                    else:
                        print()

    if not data_results.get("qingkailing_drugs"):
        print(f"\n  >> KEY FINDING: No Qingkailing injection content in local data files!")
        print(f"     Qingkailing only appears in drug_index.json (catalog index).")
        print(f"     If the production API returns Qingkailing info, it means:")
        print(f"     1) Full pharmacopoeia DOCX data (2280+ entries) was ingested")
        print(f"     2) Qingkailing chunks were matched by vector similarity")

    # === Step 4: Chunk examples ===
    chunks = analyze_chunk_examples()
    if chunks:
        print(f"\n[Step 4] Example Chunks from Data Files ({len(chunks)} found):")
        for c in chunks[:10]:
            print(f"\n  --- {c['drug']} / {c['section']} ---")
            print(f"  {c['chunk_preview']}")

    # === Step 5: Reranker simulation ===
    print(f"\n[Step 5] Reranker Simulation (query='{query}'):")
    # Simulate keyword overlap for sample chunks
    sample_contents = [
        ("阿莫西林-禁忌", "对青霉素类药物过敏者或青霉素皮肤试验阳性患者禁用。"),
        ("青霉素钠-不良反应", "过敏反应较常见，包括荨麻疹等各类皮疹、白细胞减少、间质性肾炎、哮喘发作等；偶见过敏性休克。"),
        ("清开灵注射液-不良反应", "偶有过敏反应，可见皮疹、瘙痒等。本品可能引起严重过敏反应，用药期间应注意观察。"),
        ("感冒用药指引-常用药物", "1. 对乙酰氨基酚：成人一次0.3-0.6g..."),
        ("过敏用药指引-用药原则", "首选第二代抗组胺药...过敏性休克需立即注射肾上腺素并送急救。"),
    ]
    for name, content in sample_contents:
        score = keyword_overlap_score(tokens, content)
        print(f"  {name}: keyword_score={score:.2f}  (content: {content[:60]}...)")

    # === Step 6: Embedding analysis ===
    print(f"\n[Step 6] Embedding Space Analysis:")
    print(f"  Run with --embedding flag to call DashScope API for vector comparison.")

    # === DIAGNOSIS SUMMARY ===
    print("\n" + "=" * 70)
    print("ROOT CAUSE ANALYSIS")
    print("=" * 70)
    print("""
ROOT CAUSE 1 [MOST CRITICAL]: Intent Classification Error
  - "青霉素过敏处理" is classified as "symptom_advice" because
    "过敏" appears in symptom_keywords.
  - symptom_advice uses a prompt template that tells LLM to
    recommend drugs for symptoms.
  - The correct classification should be "usage_guide" (safety
    consultation) which focuses on contraindications and precautions.
  - "过敏" was intentionally removed from safety_sections in the
    code, but the symptom_keywords trigger takes priority anyway.

ROOT CAUSE 2: Vector Embedding Semantic Confusion
  - text-embedding-v3 is a general-purpose model, not fine-tuned
    for pharmaceutical domain.
  - Query "penicillin allergy management" and chunks like
    "Qingkailing injection - adverse reactions: may cause allergies"
    are semantically similar in the embedding space because both
    involve "drug + allergy/adverse reaction" domain.

ROOT CAUSE 3: Data Imbalance + Chunking Strategy
  - TCM (一部) has 2280+ drugs, chemical drugs (二部) far fewer.
  - Each drug's sections (adverse reactions, contraindications,
    precautions) are vectorized independently.
  - Sections mentioning "allergy" cluster in embedding space,
    so any allergy query matches many unrelated drugs.

ROOT CAUSE 4: Weak Reranker
  - Current reranker only does keyword overlap + dedup.
  - Cross-Encoder reranking (BGE-Reranker-v2-m3) is NOT implemented
    (the _load_model method is empty).
  - As a result, top-20 vector results cannot be semantically
    re-ranked for true relevance.

ROOT CAUSE 5: LLM Web Search Amplification
  - enable_web_search=True may introduce external info about
    drugs not directly relevant to the user's query.
""")

    print("=" * 70)
    print("RECOMMENDED FIXES (in priority order)")
    print("=" * 70)
    print("""
1. [INTENT] Add negation and known-allergy detection:
   - "XX过敏" with no symptom description -> usage_guide or drug_query
   - Add pattern: r'.+过敏(处理|怎么办|急救|替代)' -> usage_guide
   - Distinguish "我过敏了该吃什么药" vs "我对青霉素过敏怎么办"

2. [RETRIEVAL] Add query expansion:
   - "青霉素过敏处理" -> add "青霉素 禁忌 替代药物 交叉过敏"
   - Or add pre-filtering: if query mentions a specific drug name,
     boost chunks where drug_name matches.

3. [RERANKER] Implement Cross-Encoder:
   - Load BGE-Reranker-v2-m3 model in Reranker._load_model()
   - Use it for semantic re-ranking of top-20 results to top-5

4. [PROMPT] Fix SYMPTOM_ADVICE template:
   - Add constraint: "If user asks about a KNOWN drug allergy,
     provide allergy management and alternative drug info,
     NOT recommend new drugs for treating the allergy symptom."

5. [DATA] Add drug_name weight:
   - In vector search, add a metadata filter or hybrid scoring
     that boosts chunks whose drug_name contains query terms.

6. [EVAL] Add test cases:
   - "青霉素过敏处理" should return content about penicillin
     alternatives (cephalosporins, macrolides, etc.), NOT
     random TCM drugs that mention "allergy" in their sections.
""")


if __name__ == "__main__":
    main()