business-assistant/src/step3_question_answer/synthesis_input.py

"""Lossless source-local reuse of repeated synthesis text and list entries."""

from collections import Counter
from copy import deepcopy
import json
import re


TEXT_REUSE_RULE = '''
输入压缩规则：每条来源的text_pool只存本来源的原文字串。字段为{"text_parts":[...]}时，字符串直接读取，{"ref":"t1"}取同条text_pool对应全文，按顺序无缝拼接即原字段。所有部分必须完整读取，不遗漏末尾的条件、否定、例外和版本，不跨来源取字串。字段、原路径和状态各自有效，文字相同不表示资格满足。此编码不改变原文；输出仍用原source_id和JSON协议，不展示内部ref/text_pool。'''
TEXT_REUSE_MIN_BYTES = 96
TEXT_REUSE_MAX_FRAGMENTS = 256
TEXT_REUSE_MAX_PATTERN_CHARS = 32768
TEXT_REUSE_MIN_NET_BYTES = 1024
TEXT_REUSE_MIN_NET_PERCENT = 5


def text_reuse_minimum(payload_bytes):
    """Avoid encoding complexity for byte-level wins that cannot justify it."""
    return max(TEXT_REUSE_MIN_NET_BYTES, (payload_bytes * TEXT_REUSE_MIN_NET_PERCENT + 99) // 100)


def _bytes(value):
    return len(json.dumps(value, ensure_ascii=False).encode('utf-8'))


def _text_locations(entry):
    """Only prose fields are encoded; IDs, paths, names and markers stay literal."""
    locations = []
    def visit(value, path, root):
        if isinstance(value, dict):
            for key, child in value.items():
                visit(child, (*path, key), root)
        elif isinstance(value, list):
            for index, child in enumerate(value):
                visit(child, (*path, index), root)
        elif isinstance(value, str) and len(value.encode('utf-8')) >= TEXT_REUSE_MIN_BYTES:
            if root == 'citations' and path[-1] != 'quote':
                return
            if root == 'card' and len(path) > 1 and path[1] == 'name':
                return
            if root == 'conditions' and path[-1] not in ('quote', 'requirement', 'user_evidence', 'question'):
                return
            if root == 'match_basis' and path[-1] not in ('fact_quote', 'explanation'):
                return
            locations.append((path, value))
    for root in ('answer', 'citations', 'card', 'conditions', 'match_basis', 'pending_questions'):
        if root in entry:
            visit(entry[root], (root,), root)
    return locations


def _assign(entry, path, value):
    target = entry
    for key in path[:-1]:
        target = target[key]
    target[path[-1]] = value


def expand_text_entry(entry):
    """Restore the exact source-local representation for validation and tests."""
    pool = entry.get('text_pool')
    if not isinstance(pool, dict):
        return deepcopy(entry)
    def expand(value):
        if isinstance(value, dict):
            if set(value) == {'text_parts'}:
                parts = value['text_parts']
                if not isinstance(parts, list):
                    raise ValueError('invalid_synthesis_text_parts')
                result = []
                for part in parts:
                    if isinstance(part, str):
                        result.append(part)
                    elif (isinstance(part, dict) and set(part) == {'ref'} and isinstance(part['ref'], str)
                          and isinstance(pool.get(part['ref']), str)):
                        result.append(pool[part['ref']])
                    else:
                        raise ValueError('invalid_synthesis_text_reference')
                return ''.join(result)
            return {k: expand(v) for k, v in value.items()}
        if isinstance(value, list):
            return [expand(v) for v in value]
        return value
    return expand({k: v for k, v in entry.items() if k != 'text_pool'})


def _reuse_entry(entry):
    if not isinstance(entry, dict) or not isinstance(entry.get('source_id'), str) or 'text_pool' in entry:
        return entry
    locations = _text_locations(entry)
    candidates = Counter()
    for _, text in locations:
        sentences = re.split(r'(?<=[。！？；;!?\n])', text)
        candidates.update(s for s in sentences if len(s.encode('utf-8')) >= TEXT_REUSE_MIN_BYTES)
        if text not in sentences:
            candidates[text] += 1
    # Bound indexing work; unselected or unique prose remains inline in full.
    ordered = sorted((s for s, count in candidates.items() if count >= 2),
                     key=lambda s: (len(s.encode('utf-8')) * (candidates[s] - 1), len(s)), reverse=True)
    selected, remaining = [], TEXT_REUSE_MAX_PATTERN_CHARS
    for text in ordered:
        if len(text) <= remaining:
            selected.append(text)
            remaining -= len(text)
        if len(selected) >= TEXT_REUSE_MAX_FRAGMENTS:
            break
    if not selected:
        return entry
    for _ in range(3):
        pattern = re.compile('|'.join(re.escape(s) for s in sorted(selected, key=len, reverse=True)))
        uses = Counter(m.group() for _, text in locations for m in pattern.finditer(text))
        kept = [s for s in selected if uses[s] >= 2]
        if kept == selected:
            break
        selected = kept
        if not selected:
            return entry
    pattern = re.compile('|'.join(re.escape(s) for s in sorted(selected, key=len, reverse=True)))
    ids = {text: f't{index}' for index, text in enumerate(selected)}
    result = deepcopy(entry)
    count = 0
    for path, text in locations:
        parts, start = [], 0
        for match in pattern.finditer(text):
            if match.group() not in ids:
                continue
            if match.start() > start:
                parts.append(text[start:match.start()])
            parts.append({'ref': ids[match.group()]})
            start = match.end()
            count += 1
        if parts:
            if start < len(text):
                parts.append(text[start:])
            _assign(result, path, {'text_parts': parts})
    if not count:
        return entry
    result['text_pool'] = {identity: text for text, identity in ids.items()}
    # No semantic selection or truncation: every field, path and condition must
    # reconstruct exactly. Skip any representation whose JSON overhead grows.
    try:
        valid = expand_text_entry(result) == entry
    except (ValueError, TypeError, KeyError):
        valid = False
    return result if valid and _bytes(result) < _bytes(entry) else entry


def text_reuse_prompt(prompt, payload):
    entries = payload.get('evidence', payload.get('cards', []))
    return prompt + TEXT_REUSE_RULE if any(isinstance(e, dict) and 'text_pool' in e for e in entries) else prompt


def final_model_budget(model, prompt, payload, diagnostics):
    """Choose a call-local output limit; never mutate shared models or evidence."""
    maximum = getattr(model, 'MAX_OUTPUT_TOKENS', None)
    minimum = getattr(model, 'MIN_OUTPUT_TOKENS', None)
    config = getattr(model, 'config', None)
    context = getattr(config, 'context_tokens', None)
    margin = getattr(config, 'margin', None)
    if not all(type(v) is int for v in (maximum, minimum, context, margin)):
        return model
    margin = max(904, margin)
    size = len((prompt + json.dumps(payload, ensure_ascii=False)).encode('utf-8'))
    available = context - margin - size
    # Below the floor, keep the normal budget and let hierarchical reduction run.
    reserve = min(maximum, available) if available >= minimum else maximum
    diagnostics.update(configured_output_token_reserve=maximum, min_output_token_reserve=minimum,
                       output_token_reserve=reserve, output_reserve_reduction=maximum - reserve,
                       final_call_input_bytes=size, context_limit=context, margin_reserve=margin,
                       reserved_total=size + reserve + margin)
    return model.with_output_tokens(reserve) if reserve != maximum else model


def compact_payload(payload, diagnostics=None):
    result = deepcopy(payload)
    removed = 0

    def unique(rows):
        nonlocal removed
        seen, kept = set(), []
        for row in rows:
            key = json.dumps(row, ensure_ascii=False, sort_keys=True)
            if key in seen:
                removed += 1
            else:
                seen.add(key)
                kept.append(row)
        return kept

    # Do not merge sources, rewrite prose, or normalize quotes and qualifiers.
    for entry in result.get('evidence', result.get('cards', [])):
        for field in ('citations', 'conditions', 'match_basis'):
            if isinstance(entry.get(field), list):
                entry[field] = unique(entry[field])
    if isinstance(result.get('known_subject_facts'), list):
        result['known_subject_facts'] = unique(result['known_subject_facts'])
    key = 'evidence' if 'evidence' in result else 'cards' if 'cards' in result else None
    if key and isinstance(result[key], list):
        packed = {**result, key: [_reuse_entry(entry) for entry in result[key]]}
        savings = _bytes(result) - _bytes(packed)
        already_packed = any(isinstance(e, dict) and 'text_pool' in e for e in result[key])
        net = savings - (0 if already_packed else len(TEXT_REUSE_RULE.encode('utf-8')))
        minimum = text_reuse_minimum(_bytes(result))
        if savings > 0 and diagnostics is not None:
            diagnostics['text_reuse_candidate_net_bytes'] = max(0, net)
            diagnostics['text_reuse_minimum_net_bytes'] = minimum
            diagnostics['text_reuse_applied'] = int(net >= minimum)
        if net >= minimum:
            if diagnostics is not None:
                diagnostics['reused_text_bytes_saved'] = savings
                diagnostics['pooled_texts'] = sum(len(e.get('text_pool', {})) for e in packed[key] if isinstance(e, dict))
            result = packed
    return result, removed