273 lines
15 KiB
Python
273 lines
15 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, requested_facets
|
|
|
|
COMPLETED = (r'\b(?:(?:implementovan|natrenovan|vytvoren|extrahovan|vykonan|vyhodnoten|nameran|'
|
|
r'vyriesen|odstranen|otestovan|porovnan|dosiahnut|dokoncen)(?:y|a|e|u|ou|ych|ym|ymi|om)|'
|
|
r'urobene vyhodnotenie|implementoval\w*|natrenoval\w*|vyhodnotil\w*|funguje|funguju)\b')
|
|
PLANNED = r'\b(?:bude|budu|planuj\w*|planovan\w*|cielom|mal by|ma sa|treba|pripravte|navrhnite|otestujte|vyhodnotte)\b'
|
|
ADMIN = r'\b(?:konzultac\w*|struktura prace|pisomn\w*|pisat|odovzda\w*\s+(?:text\w*|pisomn\w*|prac\w*)|wip|termin\w*)\b'
|
|
NOT_COMPLETED = r'\b(?:nie je|nie su|nebol\w*|este nie|zatial nie|nebude|nebudu)\b'
|
|
|
|
|
|
def has_qualification(text: str) -> bool:
|
|
return bool(re.search(r'\b(?:iba|len|avsak|ale|zatial|podmien\w*|obmedzen\w*|nebol\w*|nie|nepotvrd\w*)\b',normalized(text)))
|
|
|
|
|
|
def administrative_only(text: str) -> bool:
|
|
n=normalized(text)
|
|
return bool(re.search(ADMIN,n) and not re.search(COMPLETED,n))
|
|
|
|
|
|
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 or has_qualification(part) 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 in {'other','progress'}:
|
|
from .presentation import GOAL_VERBS
|
|
first=n.split()[0] if n else ''
|
|
if (first in {normalized(word) for word in [*GOAL_VERBS,*GOAL_VERBS.values()]} or n.startswith('cielom je') or
|
|
re.search(PLANNED,n)):
|
|
return 'task'
|
|
if role == 'other' and re.search(COMPLETED, n) and not re.search(PLANNED, n):
|
|
return 'progress'
|
|
return role
|
|
|
|
|
|
def evidence_phase(text: str, role: str, source_phase: str = 'unknown') -> str:
|
|
if role=='assignment':return 'assignment'
|
|
if role=='task':return 'planned'
|
|
if administrative_only(text):return 'administrative'
|
|
if role=='progress':
|
|
if source_phase=='result' or (not re.search(NOT_COMPLETED,normalized(text)) and
|
|
re.search(COMPLETED,normalized(text))):
|
|
return 'result'
|
|
return 'progress'
|
|
return 'unknown'
|
|
|
|
|
|
def facets_for(text: str, role: str, question: str, phase: str | None = None) -> list[str]:
|
|
n = normalized(text); facets = []
|
|
if role == 'assignment': facets.append('assignment')
|
|
if role == 'task':facets.append('plans')
|
|
if role == 'progress' and not re.search(PLANNED,n) and not administrative_only(text):
|
|
facets.append('progress')
|
|
if (phase or evidence_phase(text,role))=='result':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):
|
|
actual_use=(re.search(r'\b(?:skutocne|pouziva\w*|pouzil\w*|vyuzival\w*|implementoval\w*)\b',normalized(question)) or
|
|
re.search(r'\b(?:pouzit\w*|aplikovan\w*|vyuzit\w*)\s+(?:metod\w*|technolog\w*|nastroj\w*)\b',normalized(question)))
|
|
if role not in {'assignment','task'} or not actual_use:
|
|
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*|vykonan\w*|otestoval\w*|uskutocnil\w*|uskutocnen\w*|realizovan\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 administrative_only(text): 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 prioritize_frames(frames: list[str], question: str) -> list[str]:
|
|
"""Order whole source windows before budgeting; do not rewrite their text."""
|
|
from .rag_evidence import content_sections, TITLE_LABELS
|
|
rankings=[]
|
|
for facet in requested_facets(question):
|
|
scores=[]
|
|
for index,frame in enumerate(frames):
|
|
relevant=[]
|
|
for section in content_sections(frame):
|
|
for text in split_units(section['text']):
|
|
role=effective_role(text,section['kind'])
|
|
phase=evidence_phase(text,role,section.get('phase','unknown'))
|
|
if len(text)<=2500 and facet in facets_for(text,role,question,phase):
|
|
relevant.append(unit_score(text,role,facet))
|
|
if relevant:scores.append((index,max(relevant)))
|
|
rankings.append([i for i,_ in sorted(scores,key=lambda item:-item[1])])
|
|
order=[]
|
|
if re.search(r'\bnaz(?:ov|v)\w*\b',normalized(question)) and frames:
|
|
order=[i for i,frame in enumerate(frames) if any(':' in line and
|
|
normalized(line.split(':',1)[0]) in TITLE_LABELS for line in frame.splitlines())] or [0]
|
|
seen=set(order)
|
|
for pos in range(max(map(len,rankings),default=0)):
|
|
for ranking in rankings:
|
|
if pos<len(ranking) and ranking[pos] not in seen:
|
|
seen.add(ranking[pos]);order.append(ranking[pos])
|
|
return [frames[i] for i in [*order,*(i for i in range(len(frames)) if i not in seen)]]
|
|
|
|
|
|
def evidence_units(context: dict, quotes: list[dict] | None = None) -> list[dict]:
|
|
from .rag_evidence import content_sections, quote_sections, quote_matches_passage
|
|
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_matches_passage(quote,passage)
|
|
for piece in quote_sections(quote, {'content_passages':[passage]})]
|
|
for piece in pieces:
|
|
for unit in split_units(piece['text']):
|
|
role = effective_role(unit, piece['kind'])
|
|
phase=evidence_phase(unit,role,piece.get('phase','unknown'))
|
|
rows.append({k: passage.get(k) for k in ['person','work_type','year','work_label','source_id','passage_id']} |
|
|
{'text':unit,'kind':role,'phase':phase,
|
|
'facets':facets_for(unit,role,context.get('question',''),phase)})
|
|
# 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']) + (normalized(row.get('work_label')),
|
|
row['phase'],' '.join(clean.split()))
|
|
old = unique.get(key)
|
|
if old is None or (old['kind']=='other' and row['kind']!='other'):
|
|
unique[key]=row
|
|
units=list(unique.values())
|
|
# Prefer a whole longer excerpt over its overlapping prefix, within one
|
|
# work and phase. Never merge people, years or plans with recorded results.
|
|
def plain(row):return ' '.join(re.sub(r'^\s*(?:\d+[.)]|[-+*])\s+','',row['text']).split())
|
|
return [row for row in units if not any(
|
|
all(row.get(k)==other.get(k) for k in ['person','work_type','year','phase']) and
|
|
normalized(row.get('work_label'))==normalized(other.get('work_label')) and
|
|
len(plain(row))<len(plain(other)) and ' '+plain(row)+' ' in ' '+plain(other)+' '
|
|
for other in units)]
|
|
|
|
|
|
def candidates(context: dict) -> list[dict]:
|
|
from .evidence_selection import identify
|
|
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']]
|
|
if facet=='progress' and 'results' in facets and any(u['phase']=='progress' for u in relevant):
|
|
relevant=[u for u in relevant if u['phase']=='progress']
|
|
if facet!='assignment':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.get('passage_id'),unit['source_id'],text,unit['phase'])
|
|
if key not in seen:
|
|
seen.add(key); selected.append(identify({**unit,'text':text},
|
|
required=unit['kind']=='assignment' and 'assignment' in facets))
|
|
accepted+=1
|
|
if accepted==2 and facet!='assignment': 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, evidence_subject, missing_aspects, comparison_table
|
|
req=context['requirements']; units=evidence_units(context,answer.get('evidence_quotes',[]))
|
|
facets=req.get('facets',[]);groups=[];labels={f:FACET_LABELS[f] for f in facets}
|
|
if 'experiments' in facets and planned_experiments(context.get('question','')):
|
|
labels['experiments']='Plánované experimenty a vyhodnotenie'
|
|
for person in req['people']:
|
|
group={'subject':evidence_subject(person,context),'items':{},'missing':[]};seen=set()
|
|
for facet in facets:
|
|
rows=[u for u in units if u['person']==person and facet in u['facets']]
|
|
if facet=='progress' and 'results' in facets and any(u['phase']=='progress' for u in rows):
|
|
rows=[u for u in rows if u['phase']=='progress']
|
|
if facet!='assignment':rows.sort(key=lambda u:-unit_score(u['text'],u['kind'],facet))
|
|
if not rows:group['missing'].append(facet)
|
|
rendered=[]
|
|
for row in rows:
|
|
key=(' '.join(re.sub(r'^\s*(?:\d+[.)]|[-+*])\s+','',row['text']).split()),row['phase'])
|
|
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','plans'} else ''
|
|
state='Zaznamenaný výsledok: ' if row['phase']=='result' and 'results' in facets and facet!='results' else ''
|
|
rendered.append(state+text+phase);seen.add(key)
|
|
if len(rendered)==2 and facet!='assignment':break
|
|
group['items'][facet]=rendered
|
|
groups.append(group)
|
|
if len(groups)>1 and re.search(r'\bporovnaj\w*\b',normalized(context.get('question',''))):
|
|
rows=[]
|
|
for facet in facets:
|
|
if not any(g['items'][facet] for g in groups):
|
|
if all(facet not in g['missing'] for g in groups):continue
|
|
values=['\n'.join(g['items'][facet]) or ('V podkladoch chýbajú údaje.' if facet in g['missing'] else 'Doložené vyššie.') for g in groups]
|
|
rows.append((labels[facet],values))
|
|
if any(any(g['items'][f] for f in facets) for g in groups):
|
|
table=comparison_table(['Oblasť',*(g['subject'] for g in groups)],rows)
|
|
if table is not None:return table
|
|
lines=[]
|
|
for group in groups:
|
|
title=group['subject']
|
|
if len(facets)==1 and group['items'][facets[0]]:title+=' — '+labels[facets[0]].lower()
|
|
lines.extend(['**'+title+'**',''])
|
|
for facet in facets:
|
|
items=group['items'][facet]
|
|
if not items:continue
|
|
if len(facets)>1:lines.append('**'+labels[facet]+':**')
|
|
lines.extend('- '+text for text in items);lines.append('')
|
|
if group['missing']:lines.extend([missing_aspects(group['missing'],context.get('question','')),''])
|
|
return lines
|