"""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