app.py

#!/usr/bin/env python3
# coding: utf-8
"""
医疗知识图谱智能问答系统 - Gradio 聊天界面（美化版）

基于 LangGraph 工作流的 Web 演示界面，支持：
- 聊天式问答
- 实时调试信息显示
- 三级降级策略可视化
- LangSmith 追踪集成

美化改进：
1. 现代主题配色（医疗蓝）
2. 响应式布局优化
3. 折叠调试信息面板
4. 交互细节增强（按钮状态、清空功能）
5. 页脚信息栏
"""
from __future__ import annotations

import sys
import os
import logging
from pathlib import Path
from typing import Tuple, List, Dict, Any, Optional

# 确保项目根目录在 sys.path 中
_PROJECT_ROOT = str(Path(__file__).resolve().parent)
if _PROJECT_ROOT not in sys.path:
    sys.path.insert(0, _PROJECT_ROOT)

# 加载 .env 环境变量
try:
    from dotenv import load_dotenv
    load_dotenv()
except ImportError:
    logging.warning("python-dotenv 未安装，环境变量需手动设置")

# 导入 basic_qa 的核心函数
from basic_qa import (
    create_app,
    DEFAULT_ANSWER,
    setup_langsmith,
    LANGSMITH_AVAILABLE,
)
from qa_engine.session import make_thread_config, DEFAULT_GRADIO_SESSION

# ---------------------------------------------------------------------------
# Gradio 界面配置
# ---------------------------------------------------------------------------

import gradio as gr
from gradio.themes import Soft

# 设置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(name)s] %(levelname)s: %(message)s",
)
log = logging.getLogger("gradio_app")

# ---------------------------------------------------------------------------
# 应用初始化
# ---------------------------------------------------------------------------

# 初始化 LangGraph 应用
log.info("正在初始化 LangGraph 应用...")
_app = None
_callback_manager = None


def get_app():
    """获取或初始化 LangGraph 应用（延迟加载）。"""
    global _app, _callback_manager
    if _app is None:
        log.info("首次调用，初始化 LangGraph 应用")
        _app = create_app()  # 含 MemorySaver，支持多轮会话
        setup_langsmith()
    return _app


def get_config() -> dict:
    """获取图执行配置（固定 gradio-session 线程）。"""
    return make_thread_config(DEFAULT_GRADIO_SESSION)


# ---------------------------------------------------------------------------
# 核心问答逻辑（保持不变）
# ---------------------------------------------------------------------------


def process_question(question: str) -> Tuple[str, Dict[str, Any]]:
    """
    处理用户问题，返回答案和调试信息。
    
    参数：
        question: 用户输入的问题
    
    返回：
        (answer, debug_info) - 答案和调试信息字典
    """
    if not question or not question.strip():
        return "请输入有效的问题。", {
            "analysis_level": "-",
            "analysis_level_desc": "",
            "route": "-",
            "route_desc": "",
            "intents": [],
            "entities": [],
            "cypher": [],
            "result_count": 0,
            "error": "",
            "no_results": False
        }
    
    question = question.strip()
    log.info(f"处理问题: {question}")
    
    try:
        # 调用 LangGraph 工作流
        app = get_app()
        config = get_config()
        
        result = app.invoke({"question": question}, config=config)
        
        # 提取答案
        answer = result.get("answer", DEFAULT_ANSWER)
        
        # 提取调试信息（增强版本，包含路由信息）
        debug_info = {
            "analysis_level": result.get("analysis_level", 0),
            "analysis_level_desc": {
                1: "Level 1 (全LLM)",
                2: "Level 2 (LLM实体+关键词)",
                3: "Level 3 (离线NER)"
            }.get(result.get("analysis_level", 0), "未知"),
            "route": result.get("route", "-"),
            "route_desc": {
                "template": "模板问答路径",
                "graphrag": "GraphRAG路径"
            }.get(result.get("route", "-"), "未知"),
            "intents": result.get("intent", []),
            "entities": _format_entities(result.get("entities", [])),
            "cypher": _format_cypher(result.get("cypher", [])),
            "result_count": len(result.get("raw_results", [])),
            "error": result.get("error", ""),
            "no_results": result.get("no_results", False)
        }
        
        log.info(f"问题处理完成，分析等级: {debug_info['analysis_level_desc']}, 路由: {debug_info['route_desc']}")
        
        return answer, debug_info
        
    except Exception as e:
        log.error(f"处理问题时发生异常: {e}", exc_info=True)
        return f"系统发生错误: {str(e)}", {
            "analysis_level": "-",
            "analysis_level_desc": "",
            "route": "-",
            "route_desc": "",
            "intents": [],
            "entities": [],
            "cypher": [],
            "result_count": 0,
            "error": str(e),
            "no_results": False
        }


def _format_entities(entities: List[Dict]) -> List[str]:
    """格式化实体列表为字符串。"""
    if not entities:
        return []
    
    formatted = []
    for entity in entities:
        name = entity.get("name", "")
        etype = entity.get("type", "")
        if name:
            formatted.append(f"{name} ({etype})")
    return formatted


def _format_cypher(cypher_list: List[Dict]) -> List[str]:
    """格式化 Cypher 查询列表为字符串。"""
    if not cypher_list:
        return []
    
    formatted = []
    for group in cypher_list:
        question_type = group.get("question_type", "")
        queries = group.get("queries", [])
        
        for query in queries:
            cypher = query.get("cypher", "")
            if cypher:
                # 简化 Cypher 显示
                simple_cypher = cypher.split("\n")[0]
                if len(simple_cypher) > 100:
                    simple_cypher = simple_cypher[:100] + "..."
                formatted.append(f"[{question_type}] {simple_cypher}")
    
    return formatted


# ---------------------------------------------------------------------------
# 调试信息格式化（美化版）
# ---------------------------------------------------------------------------


def create_debug_display(debug_state: Dict[str, Any]) -> str:
    """
    创建格式化的调试信息显示文本（美化版）。
    """
    lines = []
    
    # 分析等级（带颜色标记）
    level = debug_state.get("analysis_level", "-")
    level_desc = debug_state.get("analysis_level_desc", "")
    level_color = {
        1: "[L1]",
        2: "[L2]",
        3: "[L3]"
    }.get(debug_state.get("analysis_level", 0), "[-]")
    
    if level != "-":
        lines.append(f"## {level_color} 分析等级")
        lines.append(f"- 级别: {level}")
        lines.append(f"- 描述: {level_desc}")
    else:
        lines.append(f"## [-] 分析等级")
        lines.append(f"- 级别: {level}")
    
    # 路由路径
    route = debug_state.get("route", "-")
    route_desc = debug_state.get("route_desc", "")
    route_color = {
        "template": "[TP]",
        "graphrag": "[GR]"
    }.get(route, "[?]")
    
    lines.append(f"\n## {route_color} 路由路径")
    lines.append(f"- 路径: {route}")
    lines.append(f"- 描述: {route_desc}")
    
    # 意图
    intents = debug_state.get("intents", [])
    if intents:
        lines.append("\n## 意图列表")
        for intent in intents:
            lines.append(f"- {intent}")
    else:
        lines.append("\n## 意图列表")
        lines.append("- 无")
    
    # 实体
    entities = debug_state.get("entities", [])
    if entities:
        lines.append(f"\n## 识别实体 ({len(entities)} 个)")
        for entity in entities[:10]:
            lines.append(f"- {entity}")
        if len(entities) > 10:
            lines.append(f"- ... 还有 {len(entities) - 10} 个")
    else:
        lines.append("\n## 识别实体")
        lines.append("- 无")
    
    # Cypher 查询
    cypher = debug_state.get("cypher", [])
    if cypher:
        lines.append(f"\n## Cypher 查询 ({len(cypher)} 条)")
        for i, q in enumerate(cypher[:5], 1):
            lines.append(f"{i}. {q}")
        if len(cypher) > 5:
            lines.append(f"... 还有 {len(cypher) - 5} 条")
    else:
        lines.append("\n## Cypher 查询")
        lines.append("- 无")
    
    # 结果数量
    result_count = debug_state.get("result_count", 0)
    lines.append(f"\n## 查询结果")
    lines.append(f"- 返回记录数: {result_count}")
    
    # 错误信息
    error = debug_state.get("error", "")
    if error:
        lines.append("\n## 错误信息")
        lines.append(f"- {error}")
    
    # 无结果提示
    if debug_state.get("no_results", False) and not error:
        lines.append("\n## 提示")
        lines.append("- 知识库中暂无相关信息")
    
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# 构建美化后的 Gradio 界面
# ---------------------------------------------------------------------------


def build_interface():
    """构建美化后的 Gradio 界面。"""
    
    global theme, custom_css
    
    # 使用现代主题（Soft 主题 + 医疗蓝主色调）
    theme = Soft(
        primary_hue="blue",
        secondary_hue="cyan",
        neutral_hue="slate",
        radius_size="md",
        font=[
            gr.themes.GoogleFont("Inter"),
            "ui-sans-serif",
            "system-ui",
            "sans-serif"
        ]
    )
    
    # CSS 样式（增强版）
    custom_css = """
    /* 自定义样式增强 */
    .gradio-container {
        max-width: 1400px !important;
        margin: 0 auto !important;
        padding: 20px;
    }
    
    /* 聊天气泡：用户右对齐蓝渐变，助手左对齐浅灰卡片 */
    .message.user, .message.user * {
        margin-left: auto !important;
    }
    .message.bot, .message.assistant, .message.bot * {
        margin-right: auto !important;
    }
    div[data-testid="user"] {
        background: linear-gradient(135deg, #2563eb 0%, #3b82f6 50%, #60a5fa 100%) !important;
        color: #fff !important;
        border-radius: 18px 18px 4px 18px !important;
        box-shadow: 0 4px 14px rgba(37, 99, 235, 0.35) !important;
        padding: 4px 2px !important;
    }
    div[data-testid="bot"], div[data-testid="assistant"] {
        background: linear-gradient(180deg, #ffffff 0%, #f1f5f9 100%) !important;
        color: #1e293b !important;
        border: 1px solid #e2e8f0 !important;
        border-radius: 18px 18px 18px 4px !important;
        box-shadow: 0 4px 12px rgba(15, 23, 42, 0.08) !important;
    }
    .gr-chatbot {
        border-radius: 16px !important;
        box-shadow: 0 8px 24px rgba(15, 23, 42, 0.06) !important;
    }
    
    /* 消息内容区域 */
    .chat-message .message-content {
        padding: 12px 16px !important;
        font-size: 15px;
        line-height: 1.6;
    }
    
    /* 页脚样式 */
    .footer-text {
        font-size: 12px;
        color: #64748b;
        text-align: center;
        padding: 16px;
        border-top: 1px solid #e2e8f0;
        margin-top: 16px;
        background: #f8fafc;
        border-radius: 0 0 12px 12px;
    }
    
    /* 标题区域 */
    .title-section {
        text-align: center;
        padding: 24px 0;
        background: linear-gradient(135deg, #3b82f6 0%, #06b6d4 100%);
        border-radius: 12px 12px 0 0;
        color: white;
        margin-bottom: 0;
    }
    
    .title-section h1 {
        color: white !important;
        font-size: 28px !important;
        margin-bottom: 8px;
    }
    
    .title-section h3 {
        color: rgba(255, 255, 255, 0.9) !important;
        font-size: 14px !important;
        font-weight: normal;
        margin-bottom: 16px;
    }
    
    .title-section ul {
        text-align: left;
        max-width: 600px;
        margin: 0 auto;
        padding: 0;
        list-style: none;
    }
    
    .title-section li {
        display: inline-block;
        margin: 4px 8px;
        padding: 6px 14px;
        background: rgba(255, 255, 255, 0.2);
        border-radius: 20px;
        font-size: 13px;
    }
    
    /* 按钮样式增强 */
    .gr-button-primary {
        background: linear-gradient(135deg, #3b82f6 0%, #2563eb 100%) !important;
        border: none !important;
        box-shadow: 0 4px 12px rgba(59, 130, 246, 0.3);
        transition: all 0.2s ease;
    }
    
    .gr-button-primary:hover {
        transform: translateY(-1px);
        box-shadow: 0 6px 16px rgba(59, 130, 246, 0.4);
    }
    
    .gr-button-secondary {
        background: #f1f5f9 !important;
        color: #64748b !important;
        border: 1px solid #e2e8f0 !important;
    }
    
    .gr-button-secondary:hover {
        background: #e2e8f0 !important;
    }
    
    /* 输入框样式 */
    .gr-textbox {
        border-radius: 12px !important;
        border: 2px solid #e2e8f0 !important;
        transition: all 0.2s ease;
    }
    
    .gr-textbox:focus {
        border-color: #3b82f6 !important;
        box-shadow: 0 0 0 3px rgba(59, 130, 246, 0.1);
    }
    
    /* 聊天容器 */
    .gr-chatbot {
        background: #fafafa !important;
        border-radius: 12px;
        border: 1px solid #e2e8f0;
    }
    
    /* Accordion 样式 */
    .gr-accordion {
        border-radius: 12px !important;
        border: 1px solid #e2e8f0 !important;
        margin-bottom: 12px;
    }
    
    .gr-accordion-header {
        background: #f8fafc !important;
        border-bottom: 1px solid #e2e8f0 !important;
    }
    
    .gr-accordion-content {
        background: white !important;
    }
    
    /* 示例按钮 */
    .gr-example {
        background: #f8fafc !important;
        border: 1px solid #e2e8f0 !important;
        border-radius: 8px !important;
        padding: 8px 12px !important;
        font-size: 13px !important;
        transition: all 0.2s ease;
    }
    
    .gr-example:hover {
        background: #e0f2fe !important;
        border-color: #3b82f6 !important;
    }
    
    /* 响应式调整 */
    @media (max-width: 768px) {
        .gradio-container {
            padding: 10px;
        }
        
        .title-section h1 {
            font-size: 22px !important;
        }
        
        .chat-message .message-content {
            font-size: 14px;
            padding: 10px 12px !important;
        }
    }
    """
    
    with gr.Blocks(title="医药知识图谱智能问答系统") as demo:
        
        # 标题区域（居中显示，带渐变背景）
        with gr.Row(elem_classes=["title-section"]):
            with gr.Column():
                gr.Markdown(
                    """
                    # 医疗知识图谱智能问答系统
                    
                    基于 **LangChain 1.0** + **LangGraph** + **Neo4j** 构建的智能问答系统
                    
                    **核心能力：**
                    - 三级降级策略（全LLM → LLM实体 → 离线NER）
                    - 条件路由（模板问答 / GraphRAG）
                    - 实时调试信息展示
                    - LangSmith 深度追踪
                    """
                )
        
        # 状态管理
        chat_history = gr.State([])
        debug_state = gr.State({
            "analysis_level": "-",
            "analysis_level_desc": "",
            "route": "-",
            "route_desc": "",
            "intents": [],
            "entities": [],
            "cypher": [],
            "result_count": 0,
            "error": "",
            "no_results": False
        })
        
        # 主内容区域
        with gr.Row():
            # 聊天区域（主要空间）
            with gr.Column(scale=3):
                # Gradio 6.x：无 bubble_full_width / show_copy_button，改用 layout + buttons
                chatbot = gr.Chatbot(
                    label="对话历史",
                    height=520,
                    show_label=False,
                    avatar_images=(
                        None,
                        "https://api.iconify.design/noto/hospital.svg",
                    ),
                    layout="bubble",
                    buttons=["copy", "copy_all"],
                )
                last_answer_box = gr.Textbox(
                    label="最后一条回答（可复制）",
                    lines=4,
                    interactive=True,
                    visible=True,
                )
                
                # 输入区域（底部）
                with gr.Row(variant="compact"):
                    user_input = gr.Textbox(
                        label="",
                        placeholder="请输入您的医疗问题，例如：糖尿病有什么症状？",
                        lines=2,
                        scale=5,
                        container=False
                    )
                    submit_btn = gr.Button(
                        "发送",
                        variant="primary",
                        scale=1,
                        icon="https://api.iconify.design/material-symbols/send.svg?color=white",
                        size="lg"
                    )
                
                # 操作按钮
                with gr.Row(variant="compact"):
                    clear_btn = gr.Button("清空对话", variant="secondary", size="sm")
                    copy_hint_btn = gr.Button("刷新可复制回答", variant="secondary", size="sm")
                    stop_btn = gr.Button("停止生成", variant="stop", size="sm")
                    gr.Examples(
                        examples=[
                            ["感冒有什么症状？"],
                            ["高血压吃什么药？"],
                            ["糖尿病和高血压有什么共同的并发症？"],
                            ["布洛芬有哪些副作用？"],
                            ["糖尿病和高血压在用药上有什么关联？"],
                        ],
                        inputs=user_input,
                        label="示例（模板 / GraphRAG / LLM兜底）",
                        examples_per_page=3,
                    )
            
            # 调试信息区域（可折叠）
            with gr.Column(scale=2):
                with gr.Accordion("调试信息", open=False):
                    debug_display = gr.Markdown(
                        "# 等待输入...\n\n请输入问题查看调试信息",
                        label="",
                        elem_id="debug-display"
                    )
                
                # 系统信息
                with gr.Accordion("系统架构", open=False):
                    gr.Markdown(
                        """
                        ## 工作流节点
                        
                        __start__
                            ↓
                        1. 分析问题（三级降级）
                            ├─ 成功 → 2. 路由判断
                            └─ 失败 → X. 错误处理
                            ↓
                        2. 路由判断
                            ├─ template → 模板路径
                            ├─ graphrag → GraphRAG路径
                            └─ error → X. 错误处理
                            ↓
                    模板路径    GraphRAG路径
                        ↓           ↓
                    实体归一化   实体提取
                        ↓           ↓
                    生成Cypher   子图检索
                        ↓           ↓
                    执行查询     上下文构建
                        ↓           ↓
                    格式化答案   LLM生成
                        ↓           ↓
                        └─────┬─────┘
                              ↓
                            最终答案
                        
                        ## 三级降级策略
                        
                        | 级别 | 描述 | 依赖 |
                        |------|------|------|
                        | Level 1 | 全LLM分析 | LLM |
                        | Level 2 | LLM实体+关键词意图 | LLM |
                        | Level 3 | 词典NER+关键词意图 | 离线 |
                        
                        ## 数据来源
                        - **LLM**: 硅基流动 (deepseek-ai/DeepSeek-V4-Pro)
                        - **数据库**: Neo4j 图数据库
                        - **实体词典**: 7类别，44,111词条
                        """
                    )
        
        # 页脚信息栏
        gr.Markdown(
            """
            <div class="footer-text">
                基于 <strong>LangChain 1.0</strong> + <strong>LangGraph</strong> + <strong>Neo4j</strong> 构建 |
                支持三级降级 · 条件路由 · 多轮会话 · 仅供参考，不构成医疗建议 |
                v2.0.0
            </div>
            """,
            elem_classes=["footer-text"]
        )
        
        # -------------------------------------------------------------------
        # 事件绑定（美化版）
        # -------------------------------------------------------------------
        
        def on_submit(
            question: str,
            history: List[Dict],
            debug: Dict,
        ) -> Tuple[List[Dict], str, Dict, List[Dict], str]:
            """处理用户输入（同步；多轮由 MemorySaver + gradio-session 保持）"""
            if not question or not question.strip():
                return history, "", debug, history, ""

            answer, new_debug = process_question(question)

            new_history = history.copy()
            new_history.append({"role": "user", "content": question})
            new_history.append({"role": "assistant", "content": answer})

            return new_history, "", new_debug, new_history, answer

        def on_copy_refresh(history: List[Dict]) -> str:
            """提取最后一条助手回答供复制。"""
            for msg in reversed(history or []):
                if msg.get("role") == "assistant":
                    return str(msg.get("content", ""))
            return ""
        
        def on_clear(
            history: List[Dict],
            debug: Dict
        ) -> Tuple[List[Dict], Dict]:
            """清空对话历史和调试信息。"""
            return [], {
                "analysis_level": "-",
                "analysis_level_desc": "",
                "route": "-",
                "route_desc": "",
                "intents": [],
                "entities": [],
                "cypher": [],
                "result_count": 0,
                "error": "",
                "no_results": False
            }
        
        # 绑定提交事件（停止按钮可取消排队中的生成任务）
        submit_event = submit_btn.click(
            fn=on_submit,
            inputs=[user_input, chat_history, debug_state],
            outputs=[chat_history, user_input, debug_state, chatbot, last_answer_box],
            queue=True,
        )

        user_input.submit(
            fn=on_submit,
            inputs=[user_input, chat_history, debug_state],
            outputs=[chat_history, user_input, debug_state, chatbot, last_answer_box],
            queue=True,
        )

        stop_btn.click(fn=None, inputs=None, outputs=None, cancels=[submit_event])

        copy_hint_btn.click(
            fn=on_copy_refresh,
            inputs=[chatbot],
            outputs=[last_answer_box],
        )
        
        # 清空按钮
        clear_btn.click(
            fn=on_clear,
            inputs=[chat_history, debug_state],
            outputs=[chatbot, debug_state]
        )
        
        # 调试信息同步更新
        debug_state.change(
            fn=create_debug_display,
            inputs=[debug_state],
            outputs=[debug_display]
        )
        
        # 初始化调试显示
        demo.load(
            fn=create_debug_display,
            inputs=[debug_state],
            outputs=[debug_display]
        )
    
    return demo, theme, custom_css


# ---------------------------------------------------------------------------
# 主入口
# ---------------------------------------------------------------------------

if __name__ == "__main__":
    print("=" * 70)
    print("  医疗知识图谱智能问答系统 - Gradio 界面")
    print("=" * 70)
    print()
    print("  正在初始化系统，请稍候...")
    print("  -- LangGraph 工作流加载中...")
    print("  -- Neo4j 连接就绪")
    print("  -- LLM 模型准备完成")
    if LANGSMITH_AVAILABLE:
        tracing_enabled = os.environ.get("LANGCHAIN_TRACING_V2", "false").lower() == "true"
        if tracing_enabled:
            print("  -- LangSmith 追踪已启用")
        else:
            print("  -- LangSmith 追踪未启用")
    print()
    print("  请在浏览器中打开: http://127.0.0.1:7860")
    print("  按 Ctrl+C 可停止服务")
    print("=" * 70)
    print()
    
    # 构建并启动界面
    demo, theme, css = build_interface()
    
    # 启动 Gradio 服务（Gradio 6.x 将主题和CSS移到launch方法）
    demo.launch(
        server_name="127.0.0.1",
        server_port=7860,
        inbrowser=False,
        share=False,
        show_error=True,
        favicon_path=None,
        theme=theme,
        css=css
    )