242 lines
12 KiB
Python
242 lines
12 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 ')):
|
|
continue
|
|
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 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 = []
|
|
scope = None
|
|
lines = []
|
|
def flush():
|
|
if lines:
|
|
windows.append((scope, '\n'.join(lines)))
|
|
lines.clear()
|
|
for line in source['body'].splitlines():
|
|
heading = re.match(r'^(?:Sekcia(?: dôkazu)?:|#{1,6}\s)(.*)', line)
|
|
if line.startswith('Dokument:'):
|
|
flush(); scope = None
|
|
elif heading:
|
|
flush()
|
|
detected = academic_scope(heading[1])
|
|
if line.startswith('Sekcia') or detected:
|
|
scope = detected
|
|
else:
|
|
lines.append(line)
|
|
flush()
|
|
unscoped = []
|
|
for section, text in windows:
|
|
titles = []
|
|
for line in text.splitlines():
|
|
if ':' in line:
|
|
label, value = line.split(':', 1)
|
|
if normalized(label) in TITLE_LABELS:
|
|
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
|
|
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 {})})
|
|
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)
|
|
academic = bool(TYPE_PATTERN.search(query) or re.search(r'\brok\w* prac\w*\b', query))
|
|
requirements = {'people': requested, 'work_types': expected_types,
|
|
'academic': academic, 'title': 'nazov' in query or 'tema' in query,
|
|
'year': academic and ('rok' in query or 'porovnaj' in query),
|
|
'start_year': 'zaciat' in query and 'studi' in query}
|
|
requirements['years'] = ([int(value) for value in re.findall(r'\b(?:v roku|z roku|roku)\s+([12]\d{3})\b', query)]
|
|
if academic and not requirements['start_year'] else [])
|
|
requirements['unsupported_year'] = academic and bool(re.search(r'\b(?:v roku|z roku|roku)\s+\d{5,}\b', query))
|
|
if requested:
|
|
facts = [row for row in facts if row['person'] in requested]
|
|
return {**{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': []}
|
|
|
|
|
|
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))
|