On main: B4 role reframe + F29 expansion experiments — held during Option E (phase_z2_pipeline/classifier/placement_planner/content_extractor)

This commit is contained in:
2026-05-07 11:14:49 +09:00
4 changed files with 67 additions and 97 deletions
-4
View File
@@ -313,8 +313,6 @@ def _build_placement_diagnostic_for_zone(
"zone_position": zone_position, "zone_position": zone_position,
"mapper_frame_template_id": mapper_template_id, "mapper_frame_template_id": mapper_template_id,
"b4_selected_template_id": None, "b4_selected_template_id": None,
"frame_selection_matches_mapper": None,
"frame_selection_match_note": "no placement_trace recorded",
"region_count": 0, "region_count": 0,
"slot_assignment_count": 0, "slot_assignment_count": 0,
"rejection_count": 0, "rejection_count": 0,
@@ -325,8 +323,6 @@ def _build_placement_diagnostic_for_zone(
placement_trace.get("mapper_frame_template_id") or mapper_template_id placement_trace.get("mapper_frame_template_id") or mapper_template_id
), ),
"b4_selected_template_id": placement_trace.get("selected_template_id"), "b4_selected_template_id": placement_trace.get("selected_template_id"),
"frame_selection_matches_mapper": placement_trace.get("frame_selection_matches_mapper"),
"frame_selection_match_note": placement_trace.get("frame_selection_match_note"),
"region_count": len(placement_trace.get("internal_regions") or []), "region_count": len(placement_trace.get("internal_regions") or []),
"slot_assignment_count": len(placement_trace.get("slot_assignments") or []), "slot_assignment_count": len(placement_trace.get("slot_assignments") or []),
"rejection_count": len(placement_trace.get("rejection") or []), "rejection_count": len(placement_trace.get("rejection") or []),
+11 -7
View File
@@ -216,25 +216,29 @@ def extract_content_objects(section, source_shape: Optional[str] = None) -> list
content = section.raw_content content = section.raw_content
section_id = section.section_id section_id = section.section_id
if source_shape == "top_bullets": if source_shape in ("top_bullets", "h3_subsections"):
from phase_z2_mapper import split_source from phase_z2_mapper import split_source
units = split_source("top_bullets", content) units = split_source(source_shape, content)
objects: list[ContentObject] = [] objects: list[ContentObject] = []
for i, unit in enumerate(units): for i, unit in enumerate(units):
unit_text = unit if isinstance(unit, str) else str(unit) if source_shape == "h3_subsections" and isinstance(unit, tuple) and len(unit) == 2:
if not unit_text.strip(): title, body = unit
raw_payload = f"### {title}\n{body}".strip()
else:
raw_payload = (unit if isinstance(unit, str) else str(unit)).strip()
if not raw_payload:
continue continue
text_specific, line_count = _detect_text_block_specific(unit_text) text_specific, line_count = _detect_text_block_specific(raw_payload)
objects.append( objects.append(
ContentObject( ContentObject(
id=f"{section_id}.text-{i + 1}", id=f"{section_id}.text-{i + 1}",
type="text_block", type="text_block",
role="summary", role="summary",
raw_payload=unit_text.strip(), raw_payload=raw_payload,
size_estimate={"line_count": line_count}, size_estimate={"line_count": line_count},
type_specific=text_specific, type_specific=text_specific,
source_shape_index=i, source_shape_index=i,
source_shape_kind="top_bullets", source_shape_kind=source_shape,
) )
) )
return objects return objects
+25 -31
View File
@@ -1063,66 +1063,60 @@ def run_phase_z2_mvp1(mdx_path: Path, run_id: Optional[str] = None) -> Path:
# 결과 (PlacementPlan) = debug_zones[i].placement_trace 로 *기록만*. # 결과 (PlacementPlan) = debug_zones[i].placement_trace 로 *기록만*.
# render path / mapper output / final.html 모두 미변경 — B5 baseline SHA 유지. # render path / mapper output / final.html 모두 미변경 — B5 baseline SHA 유지.
# B4 frame selection = catalog declaration order (V4 evidence 미사용 — 별 axis). # B4 frame selection = catalog declaration order (V4 evidence 미사용 — 별 axis).
# Option 1 (PHASE_Z_B4_SOURCE_SHAPE_ENABLED, default OFF) : pilot = F13 top_bullets only. # Option 1 source_shape-aware path (default OFF) :
b4_source_shape_enabled = ( # PHASE_Z_B4_SOURCE_SHAPE_ENABLED → top_bullets (F13/F16 path, Option 1 pilot)
# PHASE_Z_B4_H3_SUBSECTIONS_ENABLED → h3_subsections (F29 path, Option 1 expansion)
# 두 flag 독립 — backward compat 보호 (기존 top_bullets flag 의미 = top_bullets only 보존).
top_bullets_enabled = (
os.environ.get("PHASE_Z_B4_SOURCE_SHAPE_ENABLED", "").strip().lower() os.environ.get("PHASE_Z_B4_SOURCE_SHAPE_ENABLED", "").strip().lower()
in {"1", "true", "yes"} in {"1", "true", "yes"}
) )
b1_source_shape = ( h3_subsections_enabled = (
contract.get("source_shape") os.environ.get("PHASE_Z_B4_H3_SUBSECTIONS_ENABLED", "").strip().lower()
if b4_source_shape_enabled and contract.get("source_shape") == "top_bullets" in {"1", "true", "yes"}
else None
) )
contract_source_shape = contract.get("source_shape")
if contract_source_shape == "top_bullets" and top_bullets_enabled:
b1_source_shape = "top_bullets"
elif contract_source_shape == "h3_subsections" and h3_subsections_enabled:
b1_source_shape = "h3_subsections"
else:
b1_source_shape = None
content_objects = extract_content_objects(synth_section, source_shape=b1_source_shape) content_objects = extract_content_objects(synth_section, source_shape=b1_source_shape)
placement_plan = plan_placement( placement_plan = plan_placement(
content_objects=content_objects, content_objects=content_objects,
frame_contracts=list(load_frame_contracts().values()), contract=contract,
section_id=synth_section.section_id, section_id=synth_section.section_id,
) )
mapper_frame_template_id = unit.frame_template_id mapper_frame_template_id = unit.frame_template_id
matches_mapper = (
placement_plan.selected_template_id == mapper_frame_template_id
)
match_note: Optional[str] = None
if not matches_mapper:
if placement_plan.selected_template_id is None:
match_note = "no_frame_covers_content_types"
else:
match_note = (
f"B4 selected '{placement_plan.selected_template_id}' but "
f"mapper uses '{mapper_frame_template_id}' (composition V4 rank-1)"
)
placement_trace = { placement_trace = {
**asdict(placement_plan), **asdict(placement_plan),
"mapper_frame_template_id": mapper_frame_template_id, "mapper_frame_template_id": mapper_frame_template_id,
"frame_selection_matches_mapper": matches_mapper,
"frame_selection_match_note": match_note,
} }
# ─── end trace-only runtime 연결 v0 ─── # ─── end trace-only runtime 연결 v0 ───
# ─── B4 gatekeeper (Q-V4B4 / PHASE_Z_B4_GATEKEEPER, default OFF) ─── # ─── Q-V4B4 reframe : slot-level validation gatekeeper (default OFF) ───
# B4 role reframe : V4/composition = frame authority / B4 = contract validation +
# slot assignment + rejection generation. rejection 발생 시 (contract content_type
# mismatch / cardinality 위반 / no compatible sub_zone 등) adapter_needed.
if ( if (
os.environ.get("PHASE_Z_B4_GATEKEEPER", "").strip().lower() os.environ.get("PHASE_Z_B4_GATEKEEPER", "").strip().lower()
in {"1", "true", "yes"} in {"1", "true", "yes"}
and not matches_mapper and placement_plan.rejection
): ):
adapter_record = { adapter_record = {
"position": position, "position": position,
"source_section_ids": unit.source_section_ids, "source_section_ids": unit.source_section_ids,
"merge_type": unit.merge_type, "merge_type": unit.merge_type,
"template_id": unit.frame_template_id, "template_id": unit.frame_template_id,
"reason": "v4_b4_mismatch", "reason": "slot_validation_failure",
"mismatch_detail": { "rejection_detail": list(placement_plan.rejection),
"v4_template_id": mapper_frame_template_id,
"b4_selected_template_id": placement_plan.selected_template_id,
"match_note": match_note,
},
} }
adapter_needed_units.append(adapter_record) adapter_needed_units.append(adapter_record)
print(f" adapter : zone--{position} {unit.source_section_ids}" print(f" adapter : zone--{position} {unit.source_section_ids}"
f"{unit.frame_template_id} v4_b4_mismatch → adapter_needed (skip render)") f"{unit.frame_template_id} slot validation rejection → adapter_needed (skip render)")
continue continue
# ─── end B4 gatekeeper ─── # ─── end Q-V4B4 reframe ───
# mapper 시도 — 실패 (FitError) 시 zone 을 adapter_needed 로 표시하고 skip # mapper 시도 — 실패 (FitError) 시 zone 을 adapter_needed 로 표시하고 skip
try: try:
+31 -55
View File
@@ -82,31 +82,6 @@ class PlacementPlan:
rejection: list[dict] = field(default_factory=list) rejection: list[dict] = field(default_factory=list)
# ─── Frame selection ─────────────────────────────────────────────
def _select_frame(
content_objects: list[ContentObject],
frame_contracts: list[dict[str, Any]],
) -> Optional[dict[str, Any]]:
"""frame_contracts 중 *content_type_set 을 모두 cover* 하는 첫 frame.
rule (B4 v0 lock) :
1. content_type_set = {obj.type for obj in content_objects}
2. frame_contract.accepted_content_types ⊇ content_type_set 인 후보 모음
3. frame_contracts 입력 순서 (= YAML declaration order) 첫 entry 선택
Returns :
frame_contract dict 또는 None (cover 가능 frame 없음)
"""
content_type_set = {obj.type for obj in content_objects}
for fc in frame_contracts:
accepted = set(fc.get("accepted_content_types") or [])
if content_type_set <= accepted: # ⊇ check
return fc
return None
# ─── Sub_zone assignment (Stage B) ─────────────────────────────── # ─── Sub_zone assignment (Stage B) ───────────────────────────────
@@ -163,28 +138,26 @@ def _assign_region_to_sub_zone(
def plan_placement( def plan_placement(
content_objects: list[ContentObject], content_objects: list[ContentObject],
frame_contracts: list[dict[str, Any]], contract: dict[str, Any],
section_id: str = "", section_id: str = "",
) -> PlacementPlan: ) -> PlacementPlan:
"""ContentObject[] + frame_contracts → PlacementPlan (Stage A + Stage B 통합). """ContentObject[] + selected contract → PlacementPlan.
B4 role (post role-reframe) :
- frame selection authority = V4/composition (caller); B4 receives the selected contract
- B4 책임 = contract validation + slot assignment + rejection generation
v0 algorithm : v0 algorithm :
1. Stage A = B2 plan_internal_regions() 호출 → internal_regions 획득 1. Stage A = plan_internal_regions() → internal_regions
2. frame 선택 : accepted_content_types cover + 입력 순서 first 2. Contract validation : accepted_content_types ⊇ content_type_set
- cover 실패 시 → rejection + early return - mismatch → rejection (selected_contract_content_type_mismatch) + early return
3. selected_frame 의 sub_zones 읽음 (B3 catalog) 3. contract 의 sub_zones 읽음 (B3 catalog 의 Frame Slot 선언)
4. Stage B (region 1:1 sub_zone 매핑) : 4. Stage B (region sub_zone 매핑) — Option 1 positional or legacy narrowest-first
- 각 region 마다 narrowest-accepts first + declaration order sub_zone 선택 5. display_strategy / overflow_buffer = v0 default
- region.content_unit_ids 를 sub_zone.cardinality 와 비교
- count > strict → rejection 추가 / SlotAssignment 미생성
- count ≤ strict → SlotAssignment 생성 (under-fill 허용)
- 매칭 sub_zone 없는 region → rejection 추가
5. display_strategy = inline_full 모두 / overflow_buffer = [] (v0)
Args : Args :
content_objects : list[ContentObject] — B1 v0 extractor 출력 content_objects : list[ContentObject] — B1 v0 extractor 출력
frame_contracts : list[dict] — frame_contracts.yaml 의 contract dict list contract : dict — V4/composition 결정 framecontract (frame_contracts.yaml entry)
(YAML declaration order = list 순서로 입력 권고)
section_id : region_id / 결과 식별자 prefix section_id : region_id / 결과 식별자 prefix
Returns : Returns :
@@ -195,29 +168,31 @@ def plan_placement(
if not content_objects: if not content_objects:
return plan return plan
# 1. Stage A — B2 호출 (logic 중복 X) # 1. Stage A — B2 호출
zone_plan: ZoneRegionPlan = plan_internal_regions( zone_plan: ZoneRegionPlan = plan_internal_regions(
content_objects=content_objects, content_objects=content_objects,
frame_contracts=frame_contracts,
section_id=section_id, section_id=section_id,
) )
plan.internal_regions = list(zone_plan.internal_regions) plan.internal_regions = list(zone_plan.internal_regions)
# 2. frame 선택 # 2. Contract validation — selected contract 가 content_type_set 을 수용?
selected_frame = _select_frame(content_objects, frame_contracts) content_type_set = {o.type for o in content_objects}
if selected_frame is None: accepted = set(contract.get("accepted_content_types") or [])
if not (content_type_set <= accepted):
plan.rejection.append({ plan.rejection.append({
"reason": "no_frame_covers_content_types", "reason": "selected_contract_content_type_mismatch",
"content_types": sorted({o.type for o in content_objects}), "content_types": sorted(content_type_set),
"accepted_content_types": sorted(accepted),
"template_id": contract.get("template_id"),
}) })
return plan return plan
plan.selected_template_id = selected_frame.get("template_id") plan.selected_template_id = contract.get("template_id")
fid = selected_frame.get("frame_id") fid = contract.get("frame_id")
plan.selected_frame_id = str(fid) if fid is not None else None plan.selected_frame_id = str(fid) if fid is not None else None
# 3. selected_frame 의 sub_zones (B3 catalog 의 Frame Slot 선언) # 3. contract 의 sub_zones (B3 catalog 의 Frame Slot 선언)
sub_zones = list(selected_frame.get("sub_zones") or []) sub_zones = list(contract.get("sub_zones") or [])
# 4. Stage B — region 1:1 sub_zone 매핑 # 4. Stage B — region 1:1 sub_zone 매핑
assigned_sub_zone_ids: set[str] = set() assigned_sub_zone_ids: set[str] = set()
@@ -279,8 +254,9 @@ def _run_self_test():
PROJECT_ROOT = Path(__file__).parent.parent PROJECT_ROOT = Path(__file__).parent.parent
catalog_path = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "frame_contracts.yaml" catalog_path = PROJECT_ROOT / "templates" / "phase_z2" / "catalog" / "frame_contracts.yaml"
catalog = yaml.safe_load(catalog_path.read_text(encoding="utf-8")) catalog = yaml.safe_load(catalog_path.read_text(encoding="utf-8"))
# YAML 의 top-level dict — Python 3.7+ insertion-order 보존. declaration order = F13/F29/F16. # B4 role reframe : caller (V4/composition) 가 frame 선택, B4 는 contract 받음.
frame_contracts = list(catalog.values()) f13_contract = catalog["three_parallel_requirements"]
f29_contract = catalog["process_product_two_way"]
# ─── Test 1 : 1 text_block → F13 → pillar_1 ───────────────── # ─── Test 1 : 1 text_block → F13 → pillar_1 ─────────────────
text_obj = ContentObject( text_obj = ContentObject(
@@ -294,7 +270,7 @@ def _run_self_test():
"max_indent_level": 0, "has_emphasis": False, "max_indent_level": 0, "has_emphasis": False,
}, },
) )
plan1 = plan_placement([text_obj], frame_contracts, section_id="t1") plan1 = plan_placement([text_obj], f13_contract, section_id="t1")
# frame 선택 = F13 (declaration order first, content_type_set={text_block} cover) # frame 선택 = F13 (declaration order first, content_type_set={text_block} cover)
assert plan1.selected_template_id == "three_parallel_requirements", \ assert plan1.selected_template_id == "three_parallel_requirements", \
@@ -346,7 +322,7 @@ def _run_self_test():
"rows": [{"from": "a", "arrow": "", "to": "b"}], "rows": [{"from": "a", "arrow": "", "to": "b"}],
}, },
) )
plan2 = plan_placement([text_obj2, transform_obj], frame_contracts, section_id="t2") plan2 = plan_placement([text_obj2, transform_obj], f29_contract, section_id="t2")
# frame 선택 = F29 (transform_table 수용 유일) # frame 선택 = F29 (transform_table 수용 유일)
assert plan2.selected_template_id == "process_product_two_way", \ assert plan2.selected_template_id == "process_product_two_way", \