368 lines
20 KiB
Python
368 lines
20 KiB
Python
"""Typed evidence from the RAG text actually delivered after its context budget.
|
|
|
|
No Neo4j query, source oracle, benchmark answer or unseen Markdown is used.
|
|
Taxonomy years are deliberately not treated as years of academic works.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from .context_policy import normalized, text_contains_value
|
|
|
|
WORK_TYPES = {
|
|
'bakalarska praca': 'bachelor_thesis', 'diplomova praca': 'diploma_thesis',
|
|
'dizertacna praca': 'doctoral_thesis', 'bakalarsky projekt': 'bachelor_project',
|
|
'diplomovy projekt': 'diploma_project', 'timovy projekt': 'team_project',
|
|
'vyskumny projekt': 'research_project', 'vedecky projekt': 'research_project',
|
|
}
|
|
WORK_LABELS = {
|
|
'bachelor_thesis': 'bakalárska práca', 'diploma_thesis': 'diplomová práca',
|
|
'doctoral_thesis': 'dizertačná práca', 'bachelor_project': 'bakalársky projekt',
|
|
'diploma_project': 'diplomový projekt', 'team_project': 'tímový projekt',
|
|
'research_project': 'výskumný projekt',
|
|
}
|
|
TITLE_LABELS = {'nazov', 'nazov prace', 'nazov bakalarskej prace', 'nazov diplomovej prace',
|
|
'predbezny nazov', 'tema', 'navrh na temu', 'navrh na nazov', 'navrh na nazov bp',
|
|
'navrh na nazov dp', 'pisomna praca'}
|
|
PROFILE = re.compile(r'^pages/(?:students/\d{4}/[^/]+|interns/[^/]+)/README\.md$')
|
|
TYPE_PATTERN = re.compile(r'\b(bakalarsk\w*|diplomov\w*|dizertacn\w*|timov\w*|vyskumn\w*|vedeck\w*)\s+(prac\w*|projekt\w*)\b')
|
|
|
|
|
|
def work_type(text: str) -> str | None:
|
|
value = normalized(text)
|
|
match = TYPE_PATTERN.search(value)
|
|
if not match:
|
|
return None
|
|
kind, noun = match.groups()
|
|
prefix = next((p for p in ['bakalarsk', 'diplomov', 'dizertacn', 'timov', 'vyskumn', 'vedeck'] if kind.startswith(p)), None)
|
|
lookup = {('bakalarsk', 'prac'): 'bachelor_thesis', ('bakalarsk', 'projekt'): 'bachelor_project',
|
|
('diplomov', 'prac'): 'diploma_thesis', ('diplomov', 'projekt'): 'diploma_project',
|
|
('dizertacn', 'prac'): 'doctoral_thesis', ('timov', 'projekt'): 'team_project',
|
|
('vyskumn', 'projekt'): 'research_project', ('vedeck', 'projekt'): 'research_project'}
|
|
return lookup.get((prefix, 'projekt' if noun.startswith('projekt') else 'prac'))
|
|
|
|
|
|
def academic_scope(path: str) -> tuple[str, int | None] | None:
|
|
for heading in reversed(path.split('>')):
|
|
if normalized(heading).startswith(('priprava na ', 'priprava k ')):
|
|
return None
|
|
kind = work_type(heading)
|
|
if kind:
|
|
years = [int(year) for year in re.findall(r'(?<!\d)([12]\d{3})(?!\d)', heading)]
|
|
short = re.search(r'([12]\d{3})\s*/\s*(\d{2})(?!\d)', heading)
|
|
if short:
|
|
years.append(int(short[1]) // 100 * 100 + int(short[2]))
|
|
return kind, years[-1] if years else None
|
|
return None
|
|
|
|
|
|
def clean_title(value: str) -> str | None:
|
|
value = re.sub(r'\[([^\]]+)\]\([^)]*\)', r'\1', value).strip(' \t*`"„“”')
|
|
value = value.rstrip('.')
|
|
if not 6 <= len(value) <= 300 or normalized(value) in {'resers', 'tutorial', 'projekt', 'dokument', 'pisomna praca'}:
|
|
return None
|
|
if re.match(r'https?://', value) or value.endswith(':'):
|
|
return None
|
|
return value
|
|
|
|
|
|
def source_blocks(payload: dict) -> list[dict]:
|
|
context = payload.get('context', '')
|
|
sources = payload.get('sources', [])
|
|
# Reject injected/ambiguous framing rather than attributing it to a different
|
|
# student. These markers are generated by the RAG renderer, not the model.
|
|
if re.findall(r'^ZAČIATOK ZDROJA (S\d+)\s*$', context, re.M) != [s['source_id'] for s in sources]:
|
|
raise ValueError('Invalid RAG evidence framing')
|
|
blocks = []
|
|
for source in sources:
|
|
sid = re.escape(source['source_id'])
|
|
matches = re.findall(rf'^ZAČIATOK ZDROJA {sid}\n([\s\S]*?)^KONIEC ZDROJA {sid}\s*$', context, re.M)
|
|
if len(matches) != 1:
|
|
raise ValueError('Invalid RAG source boundary')
|
|
full = matches[0]
|
|
_, marker, body = full.partition('OBSAH ZDROJA\n')
|
|
if not marker:
|
|
raise ValueError('Missing RAG source content')
|
|
blocks.append({**source, 'full_text': full, 'body': body})
|
|
return blocks
|
|
|
|
|
|
def requested_types(question: str, people: list[str]) -> dict[str, str]:
|
|
text = normalized(question)
|
|
types = [(match.start(), match.end(), work_type(match[0])) for match in TYPE_PATTERN.finditer(text)]
|
|
unique = {kind for _, _, kind in types if kind}
|
|
if len(unique) == 1:
|
|
return {person: next(iter(unique)) for person in people}
|
|
result = {}
|
|
# Explicit "pri osobe X ..." clauses outrank proximity: the next person's
|
|
# name can be closer to a type phrase than the name governing that clause.
|
|
for clause in re.split(r'\b(?:pri osobe|pre osobu|u osoby)\b', text)[1:]:
|
|
names = [person for person in people if text_contains_value(clause, person, person=True)]
|
|
kinds = {work_type(match[0]) for match in TYPE_PATTERN.finditer(clause)} - {None}
|
|
if len(names) == len(kinds) == 1:
|
|
result[names[0]] = next(iter(kinds))
|
|
if len(result) == len(people):
|
|
return result
|
|
suffix = r'(?:a|u|ovi|om|e|ou|y|ovej|oveho|ym|emu)?'
|
|
positions = []
|
|
for person in people:
|
|
pattern = r'\b' + r'\s+'.join(re.escape(token) + suffix for token in normalized(person).split()) + r'\b'
|
|
positions.extend((m.start(), m.end(), person) for m in re.finditer(pattern, text))
|
|
for start, end, kind in types:
|
|
distances = [(max(start - right, left - end, 0), person) for left, right, person in positions]
|
|
if not distances or not kind:
|
|
continue
|
|
best = min(distance for distance, _ in distances)
|
|
nearest = {person for distance, person in distances if distance == best}
|
|
if len(nearest) == 1:
|
|
person = nearest.pop()
|
|
if person not in result:
|
|
result[person] = kind
|
|
return result
|
|
|
|
|
|
def academic_windows(body: str) -> list[tuple[tuple[str, int | None] | None, str]]:
|
|
"""Keep literal passages bound to the academic section actually delivered."""
|
|
windows = []; scope = None; heading_stack = []; lines = []
|
|
def flush():
|
|
if lines:
|
|
windows.append((scope, '\n'.join(lines)))
|
|
lines.clear()
|
|
for line in body.splitlines():
|
|
heading = re.match(r'^(?:Sekcia(?: dôkazu)?:|#{1,6}\s)(.*)', line)
|
|
if line.startswith(('RELEVANTNÁ SEKCIA ', 'DÔKAZ Z ', 'ĎALŠÍ DÔKAZ ',
|
|
'ĎALŠÍ RELEVANTNÝ ', 'NAJRELEVANTNEJŠÍ ', 'ZAČIATOK RELEVANTNEJ ')):
|
|
flush(); scope = None; heading_stack.clear()
|
|
elif line.startswith('Dokument:'):
|
|
flush(); scope = None; heading_stack.clear()
|
|
elif heading:
|
|
flush()
|
|
if line.startswith('Sekcia'):
|
|
scope = academic_scope(heading[1]); heading_stack = [(0, scope)]
|
|
else:
|
|
level = len(line) - len(line.lstrip('#'))
|
|
while heading_stack and heading_stack[-1][0] >= level:
|
|
heading_stack.pop()
|
|
detected = academic_scope(heading[1])
|
|
scope = (None if normalized(heading[1]).startswith(('priprava na ', 'priprava k ')) else
|
|
detected or (heading_stack[-1][1] if heading_stack else None))
|
|
heading_stack.append((level, scope))
|
|
else:
|
|
lines.append(line)
|
|
flush()
|
|
return windows
|
|
|
|
|
|
def descriptive_text(text: str, titles=()) -> str:
|
|
"""Exclude labels/identity lines; a title or year alone is not a description."""
|
|
metadata = TITLE_LABELS | {'meno veduceho', 'veduci', 'veduci prace', 'autor', 'autor dokumentu',
|
|
'rok', 'rok prace', 'typ', 'typ prace', 'rok zaciatku studia', 'kategoria', 'kategorie',
|
|
'tagy', 'klucove slova', 'dokument', 'nazov dokumentu', 'url', 'zdroj', 'zdroje'}
|
|
titles = {normalized(title) for title in titles if title}
|
|
result = []
|
|
for raw in text.splitlines():
|
|
line = re.sub(r'^\s*(?:[-*+]\s+|\d+[.)]\s+)', '', raw).strip()
|
|
value = normalized(line)
|
|
if not value or value in titles or line.endswith(':') or line.startswith(('#', '[OBSAH ZDROJA')):
|
|
continue
|
|
if ':' in line and normalized(line.split(':', 1)[0]) in metadata:
|
|
continue
|
|
if value.startswith(('stretnutie ', 'virtualne stretnutie ', 'sekcia ', 'dokument ')):
|
|
continue
|
|
identity = normalized(re.sub(r'\b(?:ing|mgr|phd|rndr|doc|prof|bc)\b', '', value))
|
|
if identity in titles:
|
|
continue
|
|
if work_type(value) and re.fullmatch(TYPE_PATTERN.pattern + r'(?:\s+\d+)*', value):
|
|
continue
|
|
if len(re.findall(r'[^\W\d_]+', line, flags=re.UNICODE)) < 2:
|
|
continue
|
|
result.append(line)
|
|
return '\n'.join(result)
|
|
|
|
|
|
def academic_content_passages(blocks: list[dict], facts: list[dict], requirements: dict, question: str) -> list[dict]:
|
|
"""Project only retrieved text, respecting person, work type and year together."""
|
|
result = []; requested = requirements['people']
|
|
profiles = [row['document_path'].removesuffix('README.md') for row in blocks
|
|
if row.get('title') in requested and PROFILE.fullmatch(row.get('document_path', ''))]
|
|
query_types = {work_type(match[0]) for match in TYPE_PATTERN.finditer(normalized(question))}
|
|
for source in blocks:
|
|
path = source.get('document_path', '')
|
|
if requested and not any(path.startswith(profile) for profile in profiles):
|
|
continue
|
|
people = [person for person in requested if any(row.get('title') == person and
|
|
path.startswith(row['document_path'].removesuffix('README.md')) for row in blocks)]
|
|
types = {requirements['work_types'][person] for person in people if person in requirements['work_types']} or query_types
|
|
titles = [row.get(key) for row in facts if row.get('source_id') == source['source_id']
|
|
for key in ('title', 'person', 'document_author')]
|
|
for scope, text in academic_windows(source['body']):
|
|
if not scope or (types and scope[0] not in types):
|
|
continue
|
|
if requirements['unsupported_year'] or (requirements['years'] and scope[1] not in requirements['years']):
|
|
continue
|
|
if not descriptive_text(text, titles):
|
|
continue
|
|
owner = (source.get('title') if PROFILE.fullmatch(path) else people[0] if len(people) == 1 else None)
|
|
passage = {'source_id': source['source_id'], 'person': owner,
|
|
'work_type': scope[0], 'year': scope[1], 'text': text}
|
|
if passage not in result:
|
|
result.append(passage)
|
|
return result
|
|
|
|
|
|
def rag_evidence_context(question: str, payload: dict) -> dict:
|
|
blocks = source_blocks(payload) if payload.get('sources') else []
|
|
facts = []
|
|
people = []
|
|
for source in blocks:
|
|
if not PROFILE.fullmatch(source.get('document_path', '')) or not source.get('title'):
|
|
continue
|
|
person = source['title']
|
|
if person not in people:
|
|
people.append(person)
|
|
base = {'person': person, 'source_id': source['source_id'], 'source_url': source.get('source_url'),
|
|
'document_path': source['document_path']}
|
|
windows = academic_windows(source['body'])
|
|
unscoped = []
|
|
for section, text in windows:
|
|
titles = []
|
|
text_lines = text.splitlines()
|
|
for index, line in enumerate(text_lines):
|
|
if ':' in line:
|
|
label, value = line.split(':', 1)
|
|
if normalized(label) in TITLE_LABELS:
|
|
value_index = index
|
|
if not value.strip():
|
|
value_index = next((j for j in range(index+1, len(text_lines)) if text_lines[j].strip()), index)
|
|
value = text_lines[value_index] if value_index != index else ''
|
|
following = next((v.strip() for v in text_lines[value_index+1:] if v.strip()), '')
|
|
if following == '[OBSAH ZDROJA OREZANÝ]' or value.strip() == '[OBSAH ZDROJA OREZANÝ]':
|
|
continue
|
|
title = clean_title(value)
|
|
if title:
|
|
titles.append(title)
|
|
start = re.search(r'rok zaciatku studia\s+([12]\d{3})\b', normalized(line))
|
|
if start:
|
|
facts.append({**base, 'start_year': int(start[1])})
|
|
if section:
|
|
kind, year = section
|
|
if not titles:
|
|
# Wiki also puts an unlabelled title immediately after its
|
|
# academic heading (sometimes preceded by explicit study
|
|
# start). Never promote meeting notes, tasks or list items.
|
|
lead_lines = [v.strip() for v in text_lines if v.strip() and
|
|
not normalized(v).startswith('rok zaciatku studia')]
|
|
lead = lead_lines[0] if lead_lines else ''
|
|
next_line = lead_lines[1] if len(lead_lines) > 1 else ''
|
|
if (lead and re.match(r'^(?:Cie[lľe]+\s*:|Zadanie\s*:|Myšlienky\s*:|1[.)]\s)', next_line, re.I) and
|
|
not re.match(r'^(?:[-*+]\s|\d+[.)]\s|#{1,6}\s)', lead) and
|
|
not normalized(lead).startswith(('stretnutie','virtualne stretnutie','stav','ulohy',
|
|
'ciel','zadanie','zasobnik','rok ','zaciatok studia','projekt ','praca ','tema na '))):
|
|
title = clean_title(lead)
|
|
if title and len(normalized(title).split()) >= 3:
|
|
titles.append(title)
|
|
record = {**base, 'work_type': kind}
|
|
if year is not None:
|
|
record['year'] = year
|
|
for title in titles or [None]:
|
|
facts.append({**record, **({'title': title} if title else {})})
|
|
else:
|
|
unscoped.extend(titles)
|
|
# A root proposal is attached only when its exact wording also occurs in
|
|
# the academic section. A taxonomy year alone never licenses this join.
|
|
for title in unscoped:
|
|
for section, text in windows:
|
|
if section and normalized(title) in normalized(text):
|
|
facts.append({**base, 'title': title, 'work_type': section[0],
|
|
**({'year': section[1]} if section[1] is not None else {})})
|
|
# A root label explicitly naming BP/DP licenses only that type;
|
|
# attach it only if the delivered evidence has exactly one such
|
|
# academic scope. Taxonomy categories still cannot supply a year.
|
|
for label, kind in [('navrh na nazov dp', 'diploma_thesis'),
|
|
('navrh na nazov bp', 'bachelor_thesis')]:
|
|
explicit = any(re.search(r'\b'+re.escape(label)+r'\s+'+re.escape(normalized(title))+r'\b',
|
|
normalized(text)) for section, text in windows if section is None)
|
|
scopes = {s for s, _ in windows if s and s[0] == kind}
|
|
if explicit and len(scopes) == 1:
|
|
only = next(iter(scopes))
|
|
facts.append({**base, 'title': title, 'work_type': kind,
|
|
**({'year': only[1]} if only[1] is not None else {})})
|
|
for field, values in [('keyword', source.get('tags', [])), ('category', source.get('categories', [])),
|
|
('document_author', [source['author']] if source.get('author') else [])]:
|
|
facts.extend({**base, field: value} for value in values)
|
|
facts = list({tuple(sorted(row.items())): row for row in facts}.values())
|
|
requested = [person for person in people if text_contains_value(question, person, person=True)]
|
|
expected_types = requested_types(question, requested)
|
|
query = normalized(question)
|
|
intent_query = normalized(re.sub(r'[„“"«][^„“”"»]+[“”"»]', ' ', question))
|
|
academic = bool(TYPE_PATTERN.search(query) or re.search(r'\b(?:rok\w* prac\w*|nazov\w* prac\w*)\b', query))
|
|
content = bool(academic or (requested and re.search(r'\b(?:prac\w*|projekt\w*)\b', query))) and bool(re.search(
|
|
r'\b(?:comu(?:\s+sa)?\s+ven(?:uj|ov)\w*|o\s+com|co\s+ries\w*|obsah\w*|opis\w*|popis\w*|'
|
|
r'zhrn\w*|zhrnut\w*|vysvetl\w*|zamer\w*|zadani\w*|ciel\w*|metod\w*|technolog\w*|'
|
|
r'pouziv\w*|implement\w*|vysledk\w*|dosiahn\w*)\b', intent_query))
|
|
identity_first = re.search(r'^(?:aky|aka|ake|ktory|ktora|ktore|v ktorom|uved)'
|
|
r'(?:\s+(?:je|su|bol|bola|boli|presny|presne)){0,3}\s+(?:nazov\w*|rok\w*)\b', intent_query)
|
|
explicit_description = re.search(r'\b(?:comu|o com|opis\w*|popis\w*|zhrn\w*|obsah\w*|zadani\w*|ciel\w*)\b', intent_query)
|
|
if identity_first and not explicit_description:
|
|
content = False
|
|
academic = academic or content
|
|
start_year = bool(re.search(r'\b(?:zaciat\w* stud\w*|zacal\w* stud\w*)\b', query))
|
|
requirements = {'people': requested, 'work_types': expected_types,
|
|
'academic': academic, 'content': content,
|
|
'title': bool(re.search(r'\bnaz(?:ov|v)\w*\b', query)) or ('tema' in query and not content),
|
|
'year': academic and ('rok' in query or ('porovnaj' in query and not content)),
|
|
'start_year': start_year,
|
|
'response_mode': 'claims' if academic or start_year else 'quotes'}
|
|
if content:
|
|
requirements['response_mode'] = 'mixed' if any(requirements[key] for key in ['title', 'year', 'start_year']) else 'quotes'
|
|
stated_years = re.findall(r'\b(?:v roku|z roku|pre rok|na rok|roku|rok)\s+(\d+)\b', query) if academic else []
|
|
requirements['years'] = ([int(value) for value in stated_years if len(value) == 4]
|
|
if not requirements['start_year'] else [])
|
|
requirements['unsupported_year'] = any(len(value) != 4 for value in stated_years)
|
|
if requested:
|
|
facts = [row for row in facts if row['person'] in requested]
|
|
context = {'question': question, **{key: payload.get(key) for key in ['status', 'reason', 'complete', 'truncated',
|
|
'truncation_reasons', 'index_version']}, 'sources': payload.get('sources', []),
|
|
'facts': facts, 'source_blocks': blocks, 'requirements': requirements,
|
|
'aggregation': None, 'answer_candidates': []}
|
|
if content:
|
|
context['content_passages'] = academic_content_passages(blocks, facts, requirements, question)
|
|
return context
|
|
|
|
|
|
def rag_prose_issues(question: str, answer: str, payload: dict) -> list[str]:
|
|
"""Conservative guard for the legacy free-prose client; typed delivery is stricter."""
|
|
context = rag_evidence_context(question, payload)
|
|
if not context['requirements']['academic']:
|
|
return []
|
|
text = answer
|
|
from .context_policy import text_urls
|
|
for url in sorted(text_urls(text), key=len, reverse=True):
|
|
text = text.replace(url, '')
|
|
normal = normalized(text)
|
|
if re.search(r'nepodarilo.*najst|nenasli|nie je uveden|nie su uveden', normal):
|
|
return []
|
|
suffix = r'(?:a|u|ovi|om|e|ou|y|ovej|oveho|ym|emu)?'
|
|
positions = []
|
|
for person in context['requirements']['people']:
|
|
pattern = r'\b' + r'\s+'.join(re.escape(token) + suffix for token in normalized(person).split()) + r'\b'
|
|
positions.extend((m.start(), m.end(), person) for m in re.finditer(pattern, normal))
|
|
positions.sort()
|
|
issues = []
|
|
for index, (start, end, person) in enumerate(positions):
|
|
clause = normal[end:positions[index + 1][0] if index + 1 < len(positions) else len(normal)]
|
|
rows = [row for row in context['facts'] if row['person'] == person]
|
|
requested_type = context['requirements']['work_types'].get(person)
|
|
academic = [row for row in rows if row.get('work_type') and
|
|
(not requested_type or row['work_type'] == requested_type)]
|
|
named_titles = [row for row in academic if row.get('title') and normalized(row['title']) in clause]
|
|
relevant = named_titles or academic
|
|
years = {str(row['year']) for row in relevant if row.get('year') is not None}
|
|
if context['requirements']['start_year']:
|
|
years |= {str(row['start_year']) for row in rows if row.get('start_year') is not None}
|
|
if set(re.findall(r'\b[12]\d{3}\b', clause)) - years:
|
|
issues.append('unsupported_person_work_year_binding')
|
|
declared_type = work_type(clause)
|
|
if requested_type and declared_type and declared_type != requested_type:
|
|
issues.append('wrong_requested_work_type')
|
|
return list(dict.fromkeys(issues))
|