529 lines
30 KiB
Python
529 lines
30 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
|
|
import json
|
|
import sqlite3
|
|
import hashlib
|
|
from contextlib import closing
|
|
from .context_policy import normalized, text_contains_value, person_pattern
|
|
from .question_intent import requested_facets, content_question, facet_mode
|
|
from tools.retrieval.question_constraints import requested_years
|
|
|
|
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
|
|
if not TYPE_PATTERN.match(normalized(heading)):
|
|
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
|
|
positions = []
|
|
for person in people:
|
|
pattern = person_pattern(person)
|
|
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 project_number(text: str) -> int | None:
|
|
match=re.search(r'\bprojekt\w*\s+([1-9]\d?)(?!\d)\b',normalized(text))
|
|
return int(match[1]) if match else None
|
|
|
|
|
|
def academic_windows_metadata(body: str) -> list[tuple]:
|
|
"""Keep literal passages bound to the academic section actually delivered."""
|
|
windows = []; scope = None; label = None; heading_stack = []; lines = []
|
|
def flush():
|
|
if lines:
|
|
windows.append((scope, label, '\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; label = None; heading_stack.clear()
|
|
elif line.startswith('Dokument:'):
|
|
flush(); scope = None; label = None; heading_stack.clear()
|
|
elif heading:
|
|
flush()
|
|
if line.startswith('Sekcia'):
|
|
scope = academic_scope(heading[1]); label = None
|
|
parts=[p.strip() for p in heading[1].split('>')]
|
|
academic=[i for i,p in enumerate(parts) if TYPE_PATTERN.match(normalized(p))]
|
|
if scope and academic:
|
|
pos=academic[-1];label=parts[pos]
|
|
# Chunk ancestry can carry the only explicit role heading.
|
|
lines.extend('# '+part for part in parts[pos+1:] if part)
|
|
heading_stack = [(0, scope, label)]
|
|
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))
|
|
label = heading[1].strip() if detected else (heading_stack[-1][2] if heading_stack else None)
|
|
heading_stack.append((level, scope, label))
|
|
if scope is not None and detected is None:
|
|
# Keep semantic child headings (Stav, Úlohy, Metódy) as
|
|
# literal source text instead of losing their meaning.
|
|
lines.append(line)
|
|
else:
|
|
lines.append(line)
|
|
flush()
|
|
return windows
|
|
|
|
|
|
def academic_windows(body: str) -> list[tuple[tuple[str, int | None] | None, str]]:
|
|
return [(scope,text) for scope,_label,text in academic_windows_metadata(body)]
|
|
|
|
|
|
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, label, text in academic_windows_metadata(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 requirements.get('project_number') is not None and project_number(label or '')!=requirements['project_number']:
|
|
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,
|
|
'document_path':path, 'work_type': scope[0], 'year': scope[1], 'work_label':label, 'text': text}
|
|
passage['passage_id']='P'+hashlib.sha256(json.dumps(passage,ensure_ascii=False,sort_keys=True).encode()).hexdigest()[:20]
|
|
if passage not in result:
|
|
result.append(passage)
|
|
return result
|
|
|
|
|
|
def content_sections(text: str) -> list[dict]:
|
|
"""Separate explicit source goals from recorded progress, without inference."""
|
|
sections = []; lines = []; kind = 'other'; phase = 'unknown'
|
|
def flush():
|
|
if lines and '\n'.join(lines).strip():
|
|
sections.append({'kind': kind, 'phase':phase, 'text': '\n'.join(lines).strip()})
|
|
lines.clear()
|
|
for line in text.splitlines():
|
|
value = normalized(line)
|
|
label = line.strip().strip('*_# ').rstrip(':').strip()
|
|
heading = value == normalized(label) and len(value.split()) <= 8
|
|
role_value=re.sub(r'\s+(?:diplomov\w*|bakalarsk\w*|zaverecn\w*)\s+(?:prac\w*|projekt\w*)(?:\s+\d{4})?$','',value)
|
|
next_kind = None
|
|
if heading and role_value in {'ciel', 'ciele', 'ciele na semester', 'ciele prace', 'zadanie',
|
|
'zadanie diplomovej prace', 'zadanie bakalarskej prace',
|
|
'zadanie diplomoveho projektu', 'zadanie bakalarskeho projektu',
|
|
'navrh na zadanie', 'navrh zadania'}:
|
|
next_kind = 'assignment'
|
|
elif heading and role_value in {'stav', 'priebezny stav', 'vysledky', 'dosiahnute vysledky', 'dosiahnuty stav'}:
|
|
next_kind = 'progress'
|
|
elif heading and role_value in {'ulohy', 'zasobnik uloh', 'dalsie ulohy', 'plan', 'planovane ulohy'}:
|
|
next_kind = 'task'
|
|
elif value.startswith(('stretnutie ', 'virtualne stretnutie ', 'sekcia ')):
|
|
next_kind = 'other'
|
|
if next_kind is not None:
|
|
flush(); kind = next_kind
|
|
phase=({'assignment':'assignment','task':'planned','other':'unknown'}.get(kind) or
|
|
('result' if role_value in {'vysledky','dosiahnute vysledky'} else 'progress'))
|
|
else:
|
|
lines.append(line)
|
|
flush()
|
|
return sections
|
|
|
|
|
|
def quote_matches_passage(quote: dict, passage: dict) -> bool:
|
|
return (quote['source_id']==passage['source_id'] and
|
|
(not quote.get('passage_id') or quote['passage_id']==passage.get('passage_id')) and
|
|
' '.join(quote['quote'].split()) in ' '.join(passage['text'].split()))
|
|
|
|
|
|
def quote_sections(quote: dict, context: dict) -> list[dict]:
|
|
"""Classify only literal excerpts bound to the delivered academic passage."""
|
|
full = [section for passage in context.get('content_passages', [])
|
|
if quote_matches_passage(quote,passage)
|
|
for section in content_sections(passage['text'])]
|
|
pieces = content_sections(quote['quote'])
|
|
for piece in pieces:
|
|
if piece['kind'] == 'other':
|
|
roles = {(section['kind'],section['phase']) for section in full
|
|
if ' '.join(piece['text'].split()) in ' '.join(section['text'].split())}
|
|
if len(roles) == 1:
|
|
piece['kind'],piece['phase'] = roles.pop()
|
|
return pieces
|
|
|
|
|
|
def expand_comparison_sources(db_file, question: str, payload: dict) -> dict:
|
|
"""Read requested academic sections from the same SQLite snapshot.
|
|
|
|
Multi-person comparisons need each person's requested work, not the single
|
|
global section intent used by ordinary retrieval expansion. The caller
|
|
holds index_snapshot; the normal shared LLM budget is applied afterwards.
|
|
"""
|
|
context = rag_evidence_context(question, payload)
|
|
req = context['requirements']
|
|
if not req.get('content') or len(req['people']) < 2:
|
|
return payload
|
|
from tools.rag.rag_utils import build_context_text
|
|
selected = []; changed = False
|
|
families = [source['document_path'].removesuffix('README.md') for source in payload.get('sources', [])
|
|
if source.get('title') in req['people'] and PROFILE.fullmatch(source.get('document_path', ''))]
|
|
with closing(sqlite3.connect(db_file.resolve().as_uri() + '?mode=ro', uri=True)) as conn:
|
|
conn.row_factory = sqlite3.Row
|
|
columns = {row[1] for row in conn.execute('PRAGMA table_info(chunks)')}
|
|
if not {'text', 'heading_paths_json', 'document_path', 'published', 'chunk_index'} <= columns:
|
|
return payload
|
|
for source in payload.get('sources', []):
|
|
person = source.get('title')
|
|
if person not in req['people'] or not PROFILE.fullmatch(source.get('document_path', '')):
|
|
if any(source.get('document_path', '').startswith(prefix) for prefix in families):
|
|
selected.append(source)
|
|
continue
|
|
frames = []; seen = set()
|
|
rows = conn.execute('''SELECT text, heading_paths_json FROM chunks
|
|
WHERE document_path=? AND COALESCE(published,1)=1 ORDER BY chunk_index''',
|
|
(source['document_path'],))
|
|
for row in rows:
|
|
paths = json.loads(row['heading_paths_json'] or '[]')
|
|
headings = [' > '.join(path) if isinstance(path, list) else str(path) for path in paths]
|
|
first = headings[0] if headings else ''
|
|
for scope, label, text in academic_windows_metadata('Sekcia: ' + first + '\n' + row['text']):
|
|
if not scope or (req['work_types'].get(person) and scope[0] != req['work_types'][person]):
|
|
continue
|
|
if req['unsupported_year'] or (req['years'] and scope[1] not in req['years']):
|
|
continue
|
|
if not descriptive_text(text) or (scope, text) in seen:
|
|
continue
|
|
seen.add((scope, text))
|
|
frames.append('Sekcia: ' + (label or WORK_LABELS[scope[0]]) + '\n' + text)
|
|
selected.append({**source, **({'text': '\n\n'.join(frames)} if frames else {})})
|
|
changed |= bool(frames)
|
|
if not changed:
|
|
return payload
|
|
selected = [{**source, 'source_id': f'S{index}'} for index, source in enumerate(selected, 1)]
|
|
return {**payload, 'sources': selected, 'source_count': len(selected), 'context': build_context_text(selected)}
|
|
|
|
|
|
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,text) for scope,label,text in academic_windows_metadata(source['body'])
|
|
if project_number(question) is None or
|
|
(scope is not None and project_number(label or '')==project_number(question))]
|
|
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)]
|
|
requested=list(dict.fromkeys([*payload.get('named_subjects',[]),*requested]))
|
|
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 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*|problem\w*|tazkost\w*|prekazk\w*|experiment\w*|merani\w*)\b', intent_query)
|
|
or content_question(question))
|
|
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:
|
|
from .answer_coverage import response_scope
|
|
requirements['content_style'] = response_scope(question, content=True)
|
|
requirements['response_mode'] = 'mixed' if any(requirements[key] for key in ['title', 'year', 'start_year']) else 'quotes'
|
|
requirements['distinguish_progress'] = bool(re.search(r'\b(?:zadani\w*|ciel\w*)\b', query)
|
|
and re.search(r'\b(?:vysledk\w*|dosiahn\w*|stav\w*)\b', query))
|
|
if facet_mode(question):
|
|
requirements['facets'] = requested_facets(question)
|
|
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'] = requested_years(question) if academic and not requirements['start_year'] else []
|
|
requirements['unsupported_year'] = any(len(value) != 4 for value in stated_years)
|
|
requirements['project_number'] = project_number(question)
|
|
if requested:
|
|
facts = [row for row in facts if row['person'] in requested]
|
|
requirements['available_fields'] = {person:[key for key in ['title','year','start_year']
|
|
if any(row.get('person')==person and row.get(key) is not None and
|
|
(key=='start_year' or ((not expected_types.get(person) or row.get('work_type')==expected_types[person])
|
|
and (not requirements['years'] or row.get('year') in requirements['years']))) for row in facts)]
|
|
for person in requested}
|
|
context = {'question': question, 'clarification':payload.get('clarification'), **{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)
|
|
if requirements.get('facets'):
|
|
from .content_selection import candidates
|
|
context['content_candidates'] = candidates(context)
|
|
if requirements['distinguish_progress']:
|
|
requirements['content_roles'] = {person: sorted({section['kind']
|
|
for p in context['content_passages'] if p['person'] == person
|
|
for section in content_sections(p['text'])
|
|
if section['kind'] in {'assignment', 'progress'} and descriptive_text(section['text'])})
|
|
for person in requested}
|
|
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))
|