确认信息的错漏情况

diagnose_online.py

@@ -0,0 +1,664 @@
+# -*- coding: utf-8 -*-
+"""
+Production Diagnostic Script
+Deploy to server and run to trace the full RAG pipeline for any query.
+
+Usage:
+  python diagnose_online.py "青霉素过敏处理"
+  python diagnose_online.py "青霉素过敏处理" --top-k 30
+  python diagnose_online.py "青霉素过敏处理" --call-api  # also call the API
+"""
+import sys
+import os
+import json
+import re
+import asyncio
+import argparse
+import math
+from pathlib import Path
+
+# ============================================================
+# Environment Setup
+# ============================================================
+def load_env():
+    """Load .env file from project root."""
+    candidates = [
+        Path(__file__).resolve().parent / ".env",
+        Path.cwd() / ".env",
+        Path("/opt/pharmacopoeia-ai/.env"),
+    ]
+    for env_file in candidates:
+        if env_file.exists():
+            with open(env_file, encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if line and not line.startswith("#") and "=" in line:
+                        key, _, val = line.partition("=")
+                        os.environ.setdefault(key.strip(), val.strip())
+            print(f"[ENV] Loaded from {env_file}")
+            return True
+    print("[ENV] WARNING: No .env file found, using env vars directly")
+    return False
+
+
+load_env()
+
+# ============================================================
+# Intent Classification (copy from retriever.py)
+# ============================================================
+def classify_intent(query: str) -> str:
+    q = query.strip()
+    negation_patterns = [
+        r"不是", r"没有", r"并非", r"不算", r"不属于",
+        r"这不是", r"我没有", r"不包含", r"不涉及",
+    ]
+    if any(re.search(pat, q) for pat in negation_patterns):
+        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"
+
+
+# ============================================================
+# Tokenization (copy from reranker.py)
+# ============================================================
+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)
+
+
+# ============================================================
+# Database Connection & Vector Search
+# ============================================================
+def get_db_url() -> str:
+    host = os.environ.get("POSTGRES_HOST", "localhost")
+    port = os.environ.get("POSTGRES_PORT", "5432")
+    db = os.environ.get("POSTGRES_DB", "pharmacopoeia")
+    user = os.environ.get("POSTGRES_USER", "postgres")
+    password = os.environ.get("POSTGRES_PASSWORD", "postgres")
+    return f"postgresql+asyncpg://{user}:{password}@{host}:{port}/{db}"
+
+
+async def get_query_embedding(query: str) -> list:
+    """Call DashScope embedding API."""
+    import httpx
+    api_key = os.environ.get("QWEN_API_KEY", "")
+    if not api_key:
+        raise RuntimeError("QWEN_API_KEY not set")
+
+    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(
+            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 API error: {data.get('message')}")
+    return data["output"]["embeddings"][0]["embedding"]
+
+
+async def vector_search(query_vec: list, top_k: int = 20) -> list:
+    """Run pgvector cosine similarity search."""
+    from sqlalchemy.ext.asyncio import create_async_engine
+    from sqlalchemy import text
+
+    db_url = get_db_url()
+    vec_str = "[" + ",".join(str(v) for v in query_vec) + "]"
+
+    engine = create_async_engine(db_url, pool_size=5, max_overflow=10, pool_pre_ping=True)
+    try:
+        async with engine.connect() as conn:
+            result = await conn.execute(
+                text("""
+                    SELECT content, source, drug_id, section,
+                           1 - (vec <=> CAST(:qv AS vector)) AS similarity
+                    FROM drug_chunks
+                    WHERE vec IS NOT NULL
+                    ORDER BY vec <=> CAST(:qv AS vector)
+                    LIMIT :k
+                """),
+                {"qv": vec_str, "k": top_k},
+            )
+            rows = result.fetchall()
+
+        docs = []
+        for row in rows:
+            docs.append({
+                "content": row[0],
+                "source": row[1],
+                "drug_id": row[2],
+                "section": row[3],
+                "score": float(row[4]),
+            })
+        return docs
+    finally:
+        await engine.dispose()
+
+
+async def get_db_stats() -> dict:
+    """Get statistics about the drug_chunks table."""
+    from sqlalchemy.ext.asyncio import create_async_engine
+    from sqlalchemy import text
+
+    db_url = get_db_url()
+    engine = create_async_engine(db_url)
+    try:
+        async with engine.connect() as conn:
+            # Total chunks
+            r = await conn.execute(text("SELECT COUNT(*) FROM drug_chunks WHERE vec IS NOT NULL"))
+            total_chunks = r.fetchone()[0]
+
+            # By source
+            r = await conn.execute(text("""
+                SELECT source, COUNT(*) as cnt
+                FROM drug_chunks
+                WHERE vec IS NOT NULL
+                GROUP BY source
+                ORDER BY cnt DESC
+                LIMIT 20
+            """))
+            by_source = [(row[0], row[1]) for row in r.fetchall()]
+
+            # By section
+            r = await conn.execute(text("""
+                SELECT section, COUNT(*) as cnt
+                FROM drug_chunks
+                WHERE vec IS NOT NULL
+                GROUP BY section
+                ORDER BY cnt DESC
+                LIMIT 20
+            """))
+            by_section = [(row[0], row[1]) for row in r.fetchall()]
+
+            # Drugs with penicillin in name
+            r = await conn.execute(text("""
+                SELECT DISTINCT d.name, d.category, d.source_volume
+                FROM drugs d
+                WHERE d.name LIKE '%青霉素%'
+            """))
+            penicillin_drugs = [(row[0], row[1], row[2]) for row in r.fetchall()]
+
+            # Drugs with 清开灵 in name
+            r = await conn.execute(text("""
+                SELECT DISTINCT d.name, d.category, d.source_volume
+                FROM drugs d
+                WHERE d.name LIKE '%清开灵%'
+            """))
+            qkl_drugs = [(row[0], row[1], row[2]) for row in r.fetchall()]
+
+            # Count drugs that have allergy in their sections
+            r = await conn.execute(text("""
+                SELECT COUNT(DISTINCT drug_id)
+                FROM drug_chunks
+                WHERE content LIKE '%过敏%' AND vec IS NOT NULL
+            """))
+            allergy_chunk_count = r.fetchone()[0]
+
+            # Count total unique drugs
+            r = await conn.execute(text("SELECT COUNT(*) FROM drugs WHERE is_active = TRUE"))
+            total_drugs = r.fetchone()[0]
+
+            return {
+                "total_chunks": total_chunks,
+                "total_drugs": total_drugs,
+                "allergy_chunk_count": allergy_chunk_count,
+                "by_source": by_source,
+                "by_section": by_section,
+                "penicillin_drugs": penicillin_drugs,
+                "qkl_drugs": qkl_drugs,
+            }
+    finally:
+        await engine.dispose()
+
+
+# ============================================================
+# Reranker Simulation
+# ============================================================
+def simulate_reranker(query: str, documents: list, top_k: int = 5) -> list:
+    """Simulate the reranker behavior on retrieval results."""
+    query_terms = tokenize_query(query)
+    MIN_THRESHOLD = 0.3
+
+    # Step 1: Filter low similarity
+    filtered = [d for d in documents if d.get("score", 0) >= MIN_THRESHOLD]
+
+    # Step 2: Keyword bonus
+    for doc in filtered:
+        content = doc.get("content", "")
+        bonus = keyword_overlap_score(query_terms, content)
+        doc["keyword_score"] = bonus
+        doc["fused_score"] = doc.get("score", 0) + bonus * 0.3
+
+    # Step 3: Sort by fused score
+    sorted_docs = sorted(filtered, key=lambda d: d.get("fused_score", 0), reverse=True)
+
+    # Step 4: Dedup
+    deduped = []
+    seen_texts = []
+    for doc in sorted_docs:
+        content = doc.get("content", "")
+        # Simple dedup: check first 200 chars
+        sample = set(content[:200])
+        is_dup = False
+        for seen in seen_texts[-5:]:
+            seen_sample = set(seen[:200])
+            intersection = len(sample & seen_sample)
+            union = len(sample | seen_sample)
+            if union > 0 and intersection / union > 0.8:
+                is_dup = True
+                break
+        if not is_dup:
+            deduped.append(doc)
+            seen_texts.append(content)
+
+    return deduped[:top_k]
+
+
+# ============================================================
+# Extract drug name from chunk content
+# ============================================================
+def extract_drug_name(content: str) -> str:
+    m = re.match(r'【(.+?) - ', content)
+    return m.group(1) if m else "unknown"
+
+
+# ============================================================
+# API Caller
+# ============================================================
+async def call_production_api(query: str, api_base: str = "https://pharmacopoeia.kailin.com.cn"):
+    """Call the production chat API and capture full SSE response."""
+    import httpx
+
+    # Step 1: Get guest token
+    print(f"\n[API] Getting guest token from {api_base}/api/v1/auth/guest ...")
+    async with httpx.AsyncClient(timeout=30) as client:
+        resp = await client.post(f"{api_base}/api/v1/auth/guest")
+        if resp.status_code != 200:
+            print(f"  FAILED: {resp.status_code} {resp.text}")
+            return None
+        token_data = resp.json()
+        token = token_data.get("access_token", "")
+        print(f"  OK: token={token[:20]}...")
+
+        # Step 2: Call chat stream
+        print(f"\n[API] Calling {api_base}/api/v1/chat/stream ...")
+        print(f"  Query: {query}")
+
+        events = {"intent": None, "status": [], "content": [], "meta": None}
+
+        async with client.stream(
+            "POST",
+            f"{api_base}/api/v1/chat/stream",
+            headers={
+                "Content-Type": "application/json",
+                "Authorization": f"Bearer {token}",
+            },
+            json={"message": query, "conversation_id": f"diag-{os.urandom(4).hex()}"},
+        ) as stream:
+            current_event = None
+            async for line in stream.aiter_lines():
+                if line.startswith("event: "):
+                    current_event = line[7:].strip()
+                elif line.startswith("data: "):
+                    data = line[6:]
+                    if current_event == "intent":
+                        events["intent"] = data
+                        print(f"  [SSE] intent: {data}")
+                    elif current_event == "status":
+                        events["status"].append(data)
+                        print(f"  [SSE] status: {data}")
+                    elif current_event == "content":
+                        if data != "[DONE]":
+                            events["content"].append(data)
+                    elif current_event == "meta":
+                        try:
+                            events["meta"] = json.loads(data)
+                            print(f"  [SSE] meta: sources={len(events['meta'].get('sources', []))} items")
+                        except json.JSONDecodeError:
+                            pass
+
+        answer = "".join(events["content"])
+        print(f"\n  Answer length: {len(answer)} chars")
+
+        return {
+            "intent": events["intent"],
+            "sources": events["meta"].get("sources", []) if events["meta"] else [],
+            "answer_preview": answer[:500] + ("..." if len(answer) > 500 else ""),
+            "status": events["status"],
+        }
+
+
+# ============================================================
+# Main Diagnostic
+# ============================================================
+async def run_diagnosis(query: str, top_k: int = 20, call_api: bool = False,
+                        api_base: str = "https://pharmacopoeia.kailin.com.cn"):
+    print("=" * 80)
+    print(f"  RAG Pipeline Diagnosis: '{query}'")
+    print("=" * 80)
+
+    # ===== Phase 1: Intent Classification =====
+    print(f"\n{'='*80}")
+    print(f"  PHASE 1: Intent Classification")
+    print(f"{'='*80}")
+    intent = classify_intent(query)
+    print(f"  Query: '{query}'")
+    print(f"  Intent: '{intent}'")
+
+    # Show why
+    symptom_keywords = [
+        "发烧", "咳嗽", "感冒", "腹泻", "头疼", "头痛",
+        "嗓子疼", "流鼻涕", "鼻塞", "肚子疼", "胃疼",
+        "过敏", "皮肤痒", "失眠", "便秘", "牙疼",
+    ]
+    matched = [kw for kw in symptom_keywords if kw in query]
+    if matched:
+        print(f"  Matched symptom keywords: {matched}")
+        print(f"  >> This triggers 'symptom_advice' template -> LLM is told to recommend drugs")
+
+    print(f"\n  PROMPT TEMPLATE (intent={intent}):")
+    # Inline prompt templates
+    PROMPT_MAP = {
+        "symptom_advice": "SYMPTOM_ADVICE: 病情评估 -> 用药方案(推荐药品) -> 注意事项 -> 就医指征",
+        "drug_query": "DRUG_QUERY: 结论 -> 详细说明 -> 注意事项 -> 来源明细",
+        "usage_guide": "USAGE_GUIDE: 结论 -> 用法用量 -> 禁忌 -> 不良反应 -> 注意事项",
+        "regulation": "REGULATION: 摘要 -> 原文引用 -> 条款出处 -> 关联条款",
+        "exam_tutor": "EXAM_TUTOR: 考点定位 -> 知识要点 -> 记忆技巧 -> 考试频率",
+        "no_docs": "NO_DOCS: 结论 -> 详细说明(通用知识) -> 来源说明",
+    }
+    template = PROMPT_MAP.get(intent, "UNKNOWN")
+    print(f"    {template}")
+
+    # ===== Phase 2: Database Statistics =====
+    print(f"\n{'='*80}")
+    print(f"  PHASE 2: Database Overview")
+    print(f"{'='*80}")
+    try:
+        stats = await get_db_stats()
+        print(f"  Total active drugs: {stats['total_drugs']}")
+        print(f"  Total chunks (with vec): {stats['total_chunks']}")
+        print(f"  Chunks containing '过敏': {stats['allergy_chunk_count']} "
+              f"({stats['allergy_chunk_count']/max(stats['total_chunks'],1)*100:.1f}%)")
+
+        print(f"\n  Penicillin-related drugs in DB:")
+        for name, cat, vol in stats['penicillin_drugs']:
+            print(f"    - {name} ({cat}, {vol})")
+
+        print(f"\n  Qingkailing-related drugs in DB:")
+        for name, cat, vol in stats['qkl_drugs']:
+            print(f"    - {name} ({cat}, {vol})")
+
+        print(f"\n  Chunks by source (top 10):")
+        for src, cnt in stats['by_source'][:10]:
+            print(f"    {src}: {cnt} chunks")
+
+        print(f"\n  Chunks by section (top 15):")
+        for sec, cnt in stats['by_section'][:15]:
+            print(f"    {sec}: {cnt} chunks")
+    except Exception as e:
+        print(f"  DB STATS FAILED: {e}")
+        stats = None
+
+    # ===== Phase 3: Vector Search =====
+    print(f"\n{'='*80}")
+    print(f"  PHASE 3: Vector Search (pgvector cosine similarity, top-{top_k})")
+    print(f"{'='*80}")
+
+    try:
+        print(f"  Getting embedding for query...")
+        query_vec = await get_query_embedding(query)
+        print(f"  Embedding dim: {len(query_vec)}")
+
+        print(f"  Searching drug_chunks...")
+        docs = await vector_search(query_vec, top_k=top_k)
+        print(f"  Retrieved: {len(docs)} documents")
+
+        if docs:
+            print(f"\n  Similarity score range: {docs[0]['score']:.4f} ~ {docs[-1]['score']:.4f}")
+            above_threshold = sum(1 for d in docs if d['score'] >= 0.3)
+            print(f"  Above reranker threshold (0.3): {above_threshold}/{len(docs)}")
+
+            # Show all results
+            print(f"\n  --- VECTOR SEARCH RESULTS ---")
+            for i, doc in enumerate(docs):
+                drug_name = extract_drug_name(doc['content'])
+                content_preview = doc['content'][:120].replace('\n', ' ')
+                flag = ""
+                if drug_name and "青霉素" in drug_name:
+                    flag = " <<< PENICILLIN MATCH"
+                elif drug_name and "清开灵" in drug_name:
+                    flag = " <<< QINGKAILING MATCH"
+                print(f"\n  [{i+1}] score={doc['score']:.4f} | drug='{drug_name}' | section='{doc['section']}'{flag}")
+                print(f"      {content_preview}...")
+    except Exception as e:
+        print(f"  VECTOR SEARCH FAILED: {e}")
+        import traceback
+        traceback.print_exc()
+        docs = []
+
+    # ===== Phase 4: Reranker Simulation =====
+    print(f"\n{'='*80}")
+    print(f"  PHASE 4: Reranker Simulation (keyword overlap + dedup)")
+    print(f"{'='*80}")
+
+    query_terms = tokenize_query(query)
+    print(f"  Query tokens ({len(query_terms)}): {sorted(query_terms)}")
+
+    if docs:
+        reranked = simulate_reranker(query, docs, top_k=5)
+        print(f"\n  After filtering (score>=0.3), keyword bonus, dedup: {len(reranked)} docs")
+
+        print(f"\n  --- RERANKED TOP-5 ---")
+        for i, doc in enumerate(reranked):
+            drug_name = extract_drug_name(doc['content'])
+            content_preview = doc['content'][:150].replace('\n', ' ')
+            print(f"\n  [{i+1}] fused_score={doc['fused_score']:.4f} "
+                  f"(vec={doc['score']:.4f} + kw_bonus={doc['keyword_score']:.2f}*0.3)")
+            print(f"      drug='{drug_name}' | section='{doc['section']}' | source='{doc.get('source','')}'")
+            print(f"      {content_preview}")
+
+        # Find Qingkailing results
+        qkl_in_results = [d for d in docs if "清开灵" in d.get("content", "")]
+        qkl_in_reranked = [d for d in reranked if "清开灵" in d.get("content", "")]
+        print(f"\n  Qingkailing in top-{top_k} vector results: {len(qkl_in_results)}")
+        print(f"  Qingkailing in top-5 reranked: {len(qkl_in_reranked)}")
+
+        if qkl_in_reranked:
+            print(f"\n  *** QINGKAILING MADE IT TO TOP-5! ***")
+            for d in qkl_in_reranked:
+                print(f"  drug='{extract_drug_name(d['content'])}' "
+                      f"score={d['score']:.4f} kw_bonus={d['keyword_score']:.2f}")
+
+        # Show all Qingkailing in vector results
+        if qkl_in_results:
+            print(f"\n  All Qingkailing entries in vector results:")
+            for d in qkl_in_results:
+                drug_name = extract_drug_name(d['content'])
+                preview = d['content'][:100].replace('\n', ' ')
+                print(f"    [{drug_name}/{d['section']}] score={d['score']:.4f}: {preview}...")
+
+    # ===== Phase 5: API Call (optional) =====
+    if call_api:
+        print(f"\n{'='*80}")
+        print(f"  PHASE 5: Production API Call")
+        print(f"{'='*80}")
+        try:
+            api_result = await call_production_api(query, api_base)
+            if api_result:
+                print(f"\n  API Intent: {api_result['intent']}")
+                print(f"  API Sources ({len(api_result['sources'])}):")
+                for s in api_result['sources']:
+                    print(f"    - {s.get('name', '?')} | section={s.get('section', '?')} "
+                          f"| score={s.get('score', 0):.4f} | source={s.get('source', '?')}")
+                print(f"\n  Answer preview:")
+                print(f"  {api_result['answer_preview']}")
+        except Exception as e:
+            print(f"  API CALL FAILED: {e}")
+
+    # ===== Phase 6: Summary =====
+    print(f"\n{'='*80}")
+    print(f"  DIAGNOSIS SUMMARY")
+    print(f"{'='*80}")
+
+    issues = []
+
+    # Issue 1: Intent
+    if intent == "symptom_advice" and "过敏" in query:
+        issues.append({
+            "severity": "CRITICAL",
+            "category": "Intent Classification",
+            "detail": f"Query '{query}' classified as 'symptom_advice'. "
+                      f"The keyword '过敏' triggers symptom-based drug recommendation, "
+                      f"but this query is about managing a known drug allergy, not seeking "
+                      f"drugs for allergy symptoms.",
+            "fix": "Add known-allergy detection: r'.+过敏(处理|怎么办|急救|替代)' -> usage_guide"
+        })
+
+    # Issue 2: Vector search data imbalance
+    if stats:
+        allergy_pct = stats['allergy_chunk_count'] / max(stats['total_chunks'], 1) * 100
+        if allergy_pct > 30:
+            issues.append({
+                "severity": "HIGH",
+                "category": "Data Imbalance",
+                "detail": f"{allergy_pct:.0f}% of chunks contain '过敏'. "
+                          f"Any allergy-related query will match a huge number of "
+                          f"unrelated drug entries.",
+                "fix": "Consider section-specific indexing or adding drug-name boost "
+                       "in vector search scoring."
+            })
+
+        if stats['qkl_drugs'] and not stats['penicillin_drugs']:
+            issues.append({
+                "severity": "HIGH",
+                "category": "Missing Data",
+                "detail": f"Qingkailing entries found ({len(stats['qkl_drugs'])}), "
+                          f"but no penicillin entries in database. "
+                          f"System cannot return relevant results for penicillin queries.",
+                "fix": "Ensure chemical drug data (volume 2) is imported."
+            })
+
+    # Issue 3: Reranker weakness
+    if docs:
+        top_scores = [d['score'] for d in docs[:5]]
+        score_spread = top_scores[0] - top_scores[-1] if len(top_scores) > 1 else 0
+        if score_spread < 0.1:
+            issues.append({
+                "severity": "MEDIUM",
+                "category": "Weak Reranker",
+                "detail": f"Top-5 vector scores too close (spread={score_spread:.4f}). "
+                          f"Keyword-based reranker cannot effectively distinguish relevance.",
+                "fix": "Implement Cross-Encoder reranker (BGE-Reranker-v2-m3)."
+            })
+
+    # Issue 4: Always present
+    issues.append({
+        "severity": "MEDIUM",
+        "category": "Prompt Template",
+        "detail": "SYMPTOM_ADVICE template forces LLM to recommend drugs "
+                  "('用药方案: 方案一, 方案二...'). When query is about "
+                  "allergy management, this produces incorrect responses.",
+        "fix": "Add constraint in SYMPTOM_ADVICE: if user asks about known drug "
+               "allergy, provide management advice, not drug recommendations."
+    })
+
+    for issue in issues:
+        print(f"\n  [{issue['severity']}] {issue['category']}")
+        print(f"  {issue['detail']}")
+        print(f"  Fix: {issue['fix']}")
+
+    print(f"\n{'='*80}")
+    print(f"  Diagnosis complete. {len(issues)} issues found.")
+    print(f"{'='*80}")
+
+
+# ============================================================
+# CLI Entry Point
+# ============================================================
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(
+        description="Production RAG pipeline diagnostics"
+    )
+    parser.add_argument(
+        "query", nargs="?", default="青霉素过敏处理",
+        help="Query to diagnose (default: 青霉素过敏处理)"
+    )
+    parser.add_argument(
+        "--top-k", type=int, default=20,
+        help="Number of vector search results to retrieve (default: 20)"
+    )
+    parser.add_argument(
+        "--call-api", action="store_true",
+        help="Also call the production API endpoint"
+    )
+    parser.add_argument(
+        "--api-base", default="https://pharmacopoeia.kailin.com.cn",
+        help="API base URL (default: https://pharmacopoeia.kailin.com.cn)"
+    )
+    args = parser.parse_args()
+
+    asyncio.run(run_diagnosis(
+        query=args.query,
+        top_k=args.top_k,
+        call_api=args.call_api,
+        api_base=args.api_base,
+    ))