192 lines
9.8 KiB
Python
192 lines
9.8 KiB
Python
"""Role-aware, bounded excerpt selection and presentation from actual evidence."""
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from .context_policy import normalized
|
|
from .question_intent import FACET_LABELS, planned_experiments
|
|
|
|
COMPLETED = r'\b(?:implementovan\w*|natrenovan\w*|vytvoren\w*|extrahovan\w*|vykonan\w*|urobene vyhodnotenie|vyhodnoteny\w*|nameran\w*)\b'
|
|
PLANNED = r'\b(?:bude|budu|planuj\w*|cielom|mal by|ma sa|treba|pripravte|navrhnite|otestujte|vyhodnotte)\b'
|
|
ADMIN = r'\b(?:konzultac\w*|struktura prace|pisomn\w*|pisat|odovzda\w*|wip|termin\w*)\b'
|
|
|
|
|
|
def split_units(text: str) -> list[str]:
|
|
from .rag_evidence import descriptive_text
|
|
parts = re.split(r'\n\s*\n|(?=^[ \t]*(?:\d+[.)]|[-+*])\s+)', text, flags=re.M)
|
|
units = []
|
|
for part in parts:
|
|
part = part.strip()
|
|
if not descriptive_text(part):
|
|
continue
|
|
# A source paragraph is kept literal, including its negations and qualifications.
|
|
for unit in ([part] if len(part) <= 900 else re.split(r'(?<=[.!?])\s+(?=[A-ZÁČĎÉÍĽŇÓŠŤÚÝŽ])', part)):
|
|
if descriptive_text(unit):
|
|
units.append(unit.strip())
|
|
return units
|
|
|
|
|
|
def effective_role(text: str, role: str) -> str:
|
|
n = normalized(text)
|
|
if role=='other':
|
|
from .presentation import GOAL_VERBS
|
|
first=n.split()[0] if n else ''
|
|
if first in {normalized(word) for word in GOAL_VERBS} or n.startswith('cielom je'):
|
|
return 'task'
|
|
if role == 'other' and re.search(COMPLETED, n) and not re.search(PLANNED, n):
|
|
return 'progress'
|
|
return role
|
|
|
|
|
|
def facets_for(text: str, role: str, question: str) -> list[str]:
|
|
n = normalized(text); facets = []
|
|
if role == 'assignment': facets.append('assignment')
|
|
if role == 'progress' and (not re.search(PLANNED,n) or re.search(COMPLETED,n)):
|
|
facets.append('results')
|
|
if re.search(r'\b(?:problem\w*|tazkost\w*|prekazk\w*|chyb\w*|zlyha\w*|nepodar\w*|nevyries\w*)\b', n) and role not in {'assignment','task'}:
|
|
facets.append('problems')
|
|
if re.search(r'\b(?:metod\w*|technolog\w*|nastroj\w*|pouzi\w*|vyuziv\w*|pomocou|implementovan\w*|natrenovan\w* model\w*)\b', n):
|
|
if role not in {'assignment','task'} or not re.search(r'\b(?:skutocne|pouziva\w*|pouzil\w*|vyuzival\w*|implementoval\w*)\b',normalized(question)):
|
|
facets.append('methods')
|
|
if re.search(r'\b(?:experiment\w*|meran\w*|test\w*|otest\w*|vyhodnot\w*|benchmark\w*|f1|presnost\w*)\b', n):
|
|
wants_done=re.search(r'\b(?:vykonal\w*|otestoval\w*|uskutocnil\w*|vyhodnotil\w*)\b',normalized(question))
|
|
if wants_done and role in {'assignment','task'}:
|
|
pass
|
|
elif not planned_experiments(question) or role in {'assignment','task'} or re.search(PLANNED, n):
|
|
facets.append('experiments')
|
|
return facets
|
|
|
|
|
|
def unit_score(text: str, role: str, facet: str = '') -> int:
|
|
n = normalized(text)
|
|
score = 25 if role == 'progress' else 10
|
|
if re.search(COMPLETED, n): score += 45
|
|
if re.search(r'\b(?:model|system|algoritm\w*|vyhodnot\w*|experiment\w*|presnost\w*|pagerank|bert)\b', n): score += 15
|
|
if re.search(ADMIN, n): score -= 70
|
|
if facet == 'problems' and re.search(r'\b(?:problem\w*|chyb\w*|zlyha\w*)\b', n): score += 35
|
|
if role in {'assignment','task'} and facet == 'results': score -= 100
|
|
return score
|
|
|
|
|
|
def evidence_units(context: dict, quotes: list[dict] | None = None) -> list[dict]:
|
|
from .rag_evidence import content_sections, quote_sections
|
|
rows = []
|
|
for passage in context.get('content_passages', []):
|
|
pieces = content_sections(passage['text']) if quotes is None else [
|
|
piece for quote in quotes if quote['source_id'] == passage['source_id'] and
|
|
' '.join(quote['quote'].split()) in ' '.join(passage['text'].split())
|
|
for piece in quote_sections(quote, context)]
|
|
for piece in pieces:
|
|
for unit in split_units(piece['text']):
|
|
role = effective_role(unit, piece['kind'])
|
|
rows.append({k: passage.get(k) for k in ['person','work_type','year','source_id']} |
|
|
{'text':unit,'kind':role,'facets':facets_for(unit,role,context.get('question',''))})
|
|
# Overlapping chunks/quotes can repeat the same fact with different roles.
|
|
# An explicit role wins over an unlabelled duplicate, within the same work.
|
|
unique = {}
|
|
for row in rows:
|
|
clean=re.sub(r'^\s*(?:\d+[.)]|[-+*])\s+','',row['text'])
|
|
key = tuple(row.get(k) for k in ['person','work_type','year']) + (' '.join(clean.split()),)
|
|
old = unique.get(key)
|
|
if old is None or (old['kind']=='other' and row['kind']!='other'):
|
|
unique[key]=row
|
|
return list(unique.values())
|
|
|
|
|
|
def candidates(context: dict) -> list[dict]:
|
|
facets = context['requirements'].get('facets', [])
|
|
selected = []; seen = set(); units = evidence_units(context)
|
|
for person in context['requirements']['people']:
|
|
for facet in facets:
|
|
relevant = [u for u in units if u['person']==person and facet in u['facets']]
|
|
relevant.sort(key=lambda u: -unit_score(u['text'],u['kind'],facet))
|
|
accepted=0
|
|
for unit in relevant:
|
|
text = unit['text']
|
|
if len(text)>2500:
|
|
context['complete']=False; context['truncated']=True
|
|
context['truncation_reasons']=list(dict.fromkeys([*(context.get('truncation_reasons') or []),'evidence_unit_limit']))
|
|
continue
|
|
key=(unit['source_id'],text)
|
|
if key not in seen:
|
|
seen.add(key); selected.append({**unit,'text':text})
|
|
accepted+=1
|
|
if accepted==2: break
|
|
return selected
|
|
|
|
|
|
def missing_facets(answer: dict, context: dict) -> list[tuple[str,str]]:
|
|
found = evidence_units(context, answer.get('evidence_quotes', []))
|
|
return [(person,facet) for person in context['requirements']['people']
|
|
for facet in context['requirements'].get('facets', [])
|
|
if not any(row['person']==person and facet in row['facets'] for row in found)]
|
|
|
|
|
|
def fit_candidates(sent: dict, context: dict, limit: int, encoding: str) -> None:
|
|
"""Budget the actual candidate JSON, sharing space across named people."""
|
|
import json
|
|
from tools.rag.build_chunks import TokenCounter
|
|
counter=TokenCounter(encoding)
|
|
if counter.encoding is None:
|
|
raise RuntimeError('Missing context tokenizer')
|
|
original=list(context.get('content_candidates',[]))
|
|
groups=[[row for row in original if row.get('person')==person]
|
|
for person in context['requirements']['people']]
|
|
ordered=[group[i] for i in range(max(map(len,groups),default=0)) for group in groups if i<len(group)]
|
|
selected=[]
|
|
for row in ordered:
|
|
if counter.count(json.dumps({**sent,'content_candidates':[*selected,row]},ensure_ascii=False))<=limit:
|
|
selected.append(row)
|
|
if len(selected)!=len(original):
|
|
context['complete']=False;context['truncated']=True
|
|
context['truncation_reasons']=list(dict.fromkeys([*(context.get('truncation_reasons') or []),'answer_candidate_budget']))
|
|
sent.update({key:context[key] for key in ['complete','truncated','truncation_reasons']})
|
|
context['content_candidates']=selected;sent['content_candidates']=selected
|
|
# Account for the added limitation metadata as well.
|
|
while selected and counter.count(json.dumps(sent,ensure_ascii=False))>limit:
|
|
selected.pop()
|
|
if counter.count(json.dumps(sent,ensure_ascii=False))>limit:
|
|
raise ValueError('Mandatory answer metadata exceeds context budget')
|
|
|
|
|
|
def facet_lines(answer: dict, context: dict) -> list[str]:
|
|
from .presentation import goal_text, work_subject, owner_in_question
|
|
from .rag_evidence import WORK_LABELS
|
|
req=context['requirements']; units=evidence_units(context,answer.get('evidence_quotes',[]))
|
|
lines=[]
|
|
for person in req['people']:
|
|
scopes={(p['work_type'],p.get('year')) for p in context.get('content_passages',[]) if p.get('person')==person}
|
|
kind=req['work_types'].get(person) or (next(iter(scopes))[0] if len(scopes)==1 else None)
|
|
subject=work_subject(person,kind,context.get('question',''))
|
|
if not owner_in_question(person,context.get('question','')):
|
|
subject=person+(' — '+WORK_LABELS[kind] if kind in WORK_LABELS else '')
|
|
if len(scopes)==1 and next(iter(scopes))[1] is not None:
|
|
subject+=' ('+str(next(iter(scopes))[1])+')'
|
|
lines.extend(['**'+subject+'**', ''])
|
|
seen=set()
|
|
for facet in req.get('facets', []):
|
|
label=FACET_LABELS[facet]
|
|
if facet=='experiments' and planned_experiments(context.get('question','')):
|
|
label='Plánované experimenty a vyhodnotenie'
|
|
rows=[u for u in units if u['person']==person and facet in u['facets']]
|
|
rows.sort(key=lambda u:-unit_score(u['text'],u['kind'],facet))
|
|
lines.append('**'+label+':**')
|
|
shown=0
|
|
for row in rows:
|
|
key=' '.join(re.sub(r'^\s*(?:\d+[.)]|[-+*])\s+','',row['text']).split())
|
|
if key in seen:
|
|
continue
|
|
text=row['text']
|
|
if row['kind'] in {'assignment','task'}:
|
|
text,_=goal_text(text)
|
|
text=re.sub(r'^\s*(?:\d+[.)]|[-+*])\s+','',text)
|
|
phase=' (plán)' if row['kind'] in {'assignment','task'} and facet not in {'assignment'} else ''
|
|
lines.append('- '+text+phase); seen.add(key); shown+=1
|
|
if shown==2: break
|
|
if not shown:
|
|
if rows:
|
|
lines.append('Je doložené rovnakým výňatkom uvedeným vyššie.')
|
|
else:
|
|
lines.append('Táto časť odpovede nie je v poskytnutých podkladoch doložená.')
|
|
lines.append('')
|
|
return lines
|