MedicalGraphRAGSystem/basic_qa.py at main · codewriterYang/MedicalGraphRAGSystem · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
#!/usr/bin/env python3
# coding: utf-8
"""
[兼容入口] 实际逻辑已迁移至 qa_engine/
此文件保留用于兼容旧导入路径和 CLI 入口

医药知识图谱智能问答系统 - 兼容入口文件。

此文件作为原有 basic_qa.py 的兼容层，将所有功能委托给 qa_engine 模块。
保持与原有接口完全兼容，确保 server/app.py 和 app.py 中的导入不会出错。
"""
from qa_engine import build_graph, build_workflow, create_app, stream_qa, QAState
from qa_engine.collect import collect_stream_result
from qa_engine.cli import (
    main,
    setup_langsmith,
    LANGSMITH_AVAILABLE,
    generate_workflow_html,
    render_graph_diagram,
)
from qa_engine.nodes.route import route_question
from qa_engine.nodes.error_handler import handle_error, should_handle_error
from qa_engine.nodes.normalize import normalize_entities
from qa_engine.nodes.template_path import generate_cypher, execute_query, format_answer
from qa_engine.nodes.graphrag_path import (
    extract_rag_entities,
    normalize_rag_entities,
    retrieve_subgraph,
    build_context,
    generate_rag_answer,
)
from settings import DEFAULT_ANSWER


def create_trace_metadata(result: dict, question: str) -> dict:
    """创建用于 LangSmith 追踪的元数据。

    Args:
        result: 工作流执行结果
        question: 用户问题

    Returns:
        元数据字典，包含路由信息、子图统计等
    """
    metadata = {
        "question": question,
        "route": result.get("route", "unknown"),
        "analysis_level": result.get("analysis_level"),
    }

    # 添加 GraphRAG 特有字段
    subgraph = result.get("subgraph", {})
    if isinstance(subgraph, dict):
        stats = subgraph.get("stats", {})
        if isinstance(stats, dict):
            metadata["subgraph_nodes"] = stats.get("total_nodes")
            metadata["subgraph_edges"] = stats.get("total_edges")

    context = result.get("context", {})
    if isinstance(context, dict):
        metadata["context_chars"] = context.get("char_count", context.get("length", 0))

    return metadata


# 保持向后兼容的导出
__all__ = [
    "build_graph",
    "build_workflow",
    "create_app",
    "stream_qa",
    "QAState",
    "DEFAULT_ANSWER",
    "setup_langsmith",
    "LANGSMITH_AVAILABLE",
    "route_question",
    "handle_error",
    "should_handle_error",
    "normalize_entities",
    "generate_cypher",
    "execute_query",
    "format_answer",
    "extract_rag_entities",
    "normalize_rag_entities",
    "retrieve_subgraph",
    "build_context",
    "generate_rag_answer",
    "create_trace_metadata",
    "generate_workflow_html",
    "render_graph_diagram",
    "collect_stream_result",
]

if __name__ == "__main__":
    main()