AI-assistant/coverage_checker.py at main · mianhua157/AI-assistant · 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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
from __future__ import annotations

import re
from dataclasses import asdict, dataclass
from typing import Any

from retrieval_planner import RetrievalPlan


QUESTION_STOPWORDS = {
    "what",
    "what's",
    "is",
    "are",
    "the",
    "a",
    "an",
    "of",
    "how",
    "does",
    "do",
    "did",
    "why",
    "explain",
    "define",
    "meaning",
    "summary",
    "summarize",
    "compare",
    "difference",
    "versus",
    "vs",
    "quiz",
    "plan",
    "study",
    "course",
    "material",
    "coverage",
    "什么是",
    "啥是",
    "定义",
    "解释一下",
    "解释",
    "介绍一下",
    "总结",
    "概括",
    "归纳",
    "区别",
    "不同",
    "对比",
    "比较",
    "帮我",
    "生成",
    "复习",
    "学习计划",
    "资料里",
    "课程里",
    "有没有",
    "没讲",
    "没提到",
    "覆盖",
}


@dataclass
class CoverageReport:
    status: str
    can_answer: bool
    reason: str
    matched_terms: list[str]
    total_terms: int
    strong_doc_count: int
    raw_doc_count: int
    wiki_doc_count: int

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


def _extract_question_terms(question: str) -> list[str]:
    tokens = re.findall(r"[a-zA-Z][a-zA-Z0-9_\-]+|[\u4e00-\u9fff]{2,}", question.lower())
    filtered = [token for token in tokens if len(token) > 1 and token not in QUESTION_STOPWORDS]
    seen: set[str] = set()
    deduped: list[str] = []
    for token in filtered:
        if token not in seen:
            seen.add(token)
            deduped.append(token)
    return deduped


def check_coverage(question: str, docs: list[Any], plan: RetrievalPlan) -> CoverageReport:
    if not docs:
        return CoverageReport(
            status="uncovered",
            can_answer=False,
            reason="No relevant documents were retrieved from the course material.",
            matched_terms=[],
            total_terms=0,
            strong_doc_count=0,
            raw_doc_count=0,
            wiki_doc_count=0,
        )

    terms = _extract_question_terms(question)
    joined_content = "\n".join(doc.page_content.lower()[:1500] for doc in docs)
    matched_terms = [term for term in terms if term in joined_content]

    strong_docs = [doc for doc in docs if float(doc.metadata.get("retrieval_score", 0.0)) >= plan.min_score]
    raw_doc_count = sum(1 for doc in docs if doc.metadata.get("chunk_type") == "raw")
    wiki_doc_count = sum(1 for doc in docs if doc.metadata.get("chunk_type") == "wiki")

    required_docs = 1 if plan.intent == "definition" else 2
    if plan.intent in {"summary", "quiz", "study_plan"}:
        required_docs = 3

    if not strong_docs:
        return CoverageReport(
            status="uncovered",
            can_answer=False,
            reason="Retrieved chunks are too weakly related to support a grounded answer.",
            matched_terms=matched_terms,
            total_terms=len(terms),
            strong_doc_count=0,
            raw_doc_count=raw_doc_count,
            wiki_doc_count=wiki_doc_count,
        )

    if len(strong_docs) < required_docs:
        return CoverageReport(
            status="partial",
            can_answer=False,
            reason="Some relevant material was found, but not enough independent evidence was retrieved.",
            matched_terms=matched_terms,
            total_terms=len(terms),
            strong_doc_count=len(strong_docs),
            raw_doc_count=raw_doc_count,
            wiki_doc_count=wiki_doc_count,
        )

    if terms and len(matched_terms) / len(terms) < 0.4:
        return CoverageReport(
            status="partial",
            can_answer=False,
            reason="The retrieved context overlaps with the question only loosely, so it may be a hard-negative match.",
            matched_terms=matched_terms,
            total_terms=len(terms),
            strong_doc_count=len(strong_docs),
            raw_doc_count=raw_doc_count,
            wiki_doc_count=wiki_doc_count,
        )

    if plan.intent != "definition" and raw_doc_count == 0:
        return CoverageReport(
            status="partial",
            can_answer=False,
            reason="Only wiki-style support was found; the answer lacks enough raw course evidence.",
            matched_terms=matched_terms,
            total_terms=len(terms),
            strong_doc_count=len(strong_docs),
            raw_doc_count=raw_doc_count,
            wiki_doc_count=wiki_doc_count,
        )

    return CoverageReport(
        status="covered",
        can_answer=True,
        reason="The retrieved set is sufficiently grounded for answer generation.",
        matched_terms=matched_terms,
        total_terms=len(terms),
        strong_doc_count=len(strong_docs),
        raw_doc_count=raw_doc_count,
        wiki_doc_count=wiki_doc_count,
    )