| """v2.2 Serving Plane: compact identity/fact에서 직접 PUBLIC을 조립한다. |
| |
| legacy graph adapter, Event B3, Entity identity scorer는 이 경로의 입력이 아니다. |
| assembly가 하는 추가 의미 계산은 canonical temporal occurrence의 V2 규칙뿐이다. |
| """ |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import asdict, dataclass, field, replace |
| from hashlib import sha256 |
| import json |
| from pathlib import Path |
| import time |
| from typing import Any, Mapping |
|
|
| from runtime.eventframe.contracts import ArticleInput |
| from runtime.eventframe.resolved_contracts import ( |
| COMPACT_RUNTIME_CONTRACT_ID, OUTPUT_PROFILE_CONTRACT_ID, |
| PUBLIC_SCHEMA_VERSION, PUBLIC_SEMANTIC_PROPERTY_KEYS, |
| PUBLIC_SEMANTIC_REQUIRED_KEYS, |
| CanonicalStatementState, CanonicalTemporalState, |
| EventIdentityClosure, GroundedFact, GroundingRef, GroundingRegistry, |
| LocalEntityState, PublicIdentity, ResolvedGraphState, |
| ) |
| from runtime.temporal_identity import normalized_temporal_key |
|
|
| from .derivation.time_normalization_v2 import TimeNormalizerV2 |
| from .output_profiles import CompactOutputProfileProjector, OutputProfile |
|
|
|
|
| NODE_KINDS = ("ARTICLE", "EVENT", "STATEMENT", "ENTITY", "TIME") |
| EDGE_ENDPOINTS = { |
| "COVERS": ("ARTICLE", "EVENT"), |
| "CONTAINS_STATEMENT": ("ARTICLE", "STATEMENT"), |
| "MENTIONS": ("ARTICLE", "ENTITY"), |
| "ACTOR": ("EVENT", "ENTITY"), |
| "TARGET": ("EVENT", "ENTITY"), |
| "PLACE": ("EVENT", "ENTITY"), |
| "OCCURRED_ON": ("EVENT", "TIME"), |
| } |
| EDGE_TYPES = tuple(EDGE_ENDPOINTS) |
| COMPACT_ASSEMBLY_CONFIG_SCHEMA_VERSION = "articlelocal-compact-assembly-config-v22-v1" |
|
|
|
|
| def _stable_id(prefix: str, *parts: object) -> str: |
| material = "␟".join(str(part) for part in parts).encode("utf-8") |
| return prefix + "-" + sha256(material).hexdigest()[:24] |
|
|
|
|
| def _unique_refs(refs) -> tuple[GroundingRef, ...]: |
| return tuple(sorted(set(refs), key=lambda row: ( |
| row.article_version_id, row.char_start, row.char_end, |
| row.sentence_index if row.sentence_index is not None else -1, |
| ))) |
|
|
|
|
| def _grounding_registry(identities, facts, pending=()) -> GroundingRegistry: |
| refs = {} |
| for identity in identities: |
| for ref in _unique_refs( |
| ref for ref in (identity.representative, *identity.evidence) |
| if ref is not None |
| ): |
| refs[_stable_id("GREF", identity.identity_id, ref.article_version_id, |
| ref.char_start, ref.char_end)] = ref |
| for fact in (*facts, *pending): |
| for ref in _unique_refs(fact.supports): |
| refs[_stable_id("GREF", fact.fact_id, ref.article_version_id, |
| ref.char_start, ref.char_end)] = ref |
| return GroundingRegistry(refs) |
|
|
|
|
| def build_resolved_graph_state( |
| article: ArticleInput, *, event_closure: EventIdentityClosure, |
| local_entities: tuple[LocalEntityState, ...], |
| entity_identity_remap: Mapping[str, str], |
| statements: tuple[CanonicalStatementState, ...], |
| temporal_occurrences: tuple[CanonicalTemporalState, ...], |
| lane_statuses: Mapping[str, str], runtime_config_id: str, |
| assembly_config_id: str, active_policy_ids: tuple[str, ...], |
| component_provenance: Mapping[str, str], lane_reasons: Mapping[str, str] = None, |
| runtime_config_sha256: str | None = None, |
| assembly_config_sha256: str | None = None, |
| public_schema_sha256: str | None = None, |
| policy_config_sha256: Mapping[str, str] = None, |
| diagnostic_sink=None, |
| ) -> ResolvedGraphState: |
| """canonical producer들의 allowlisted scalar/ref만 graph state에 이관한다.""" |
| article_id = str(article.article_id) |
| article_version = str(article.article_version_id) |
| digest = ResolvedGraphState.content_digest(article.content) |
| identities = [PublicIdentity( |
| article_id, "ARTICLE", None, None, { |
| "article_id": article_id, "article_version_id": article_version, |
| "content_sha256": digest, "published_at": article.published_at, |
| "title": article.title, "source": article.source, |
| "article_local": True, |
| }, |
| )] |
| remap = dict(event_closure.identity_remap) |
| remap.update((str(key), str(value)) for key, value in entity_identity_remap.items()) |
| event_states = sorted(event_closure.local_events, key=lambda row: row.event_id) |
| for event in event_states: |
| trigger_refs = _unique_refs(event.triggers) |
| conflicts = event.conflict_summary |
| status = "UNRESOLVED" if ( |
| not conflicts.conflict_free or conflicts.missing_feature_reasons |
| ) else "EXECUTED" |
| identities.append(PublicIdentity( |
| event.event_id, "EVENT", event.representative, |
| event.canonical_text, { |
| "canonical_text": event.canonical_text, |
| "identity_confidence": event.identity_confidence, |
| "identity_confidence_source": event.identity_confidence_source, |
| "independent_support_count": event.independent_support_count, |
| "triggers": tuple({ |
| "text": ref.text(article_version_id=article_version, |
| source_text=article.content), |
| "sentence_index": ref.sentence_index, |
| "char_start": ref.char_start, "char_end": ref.char_end, |
| } for ref in trigger_refs), |
| "role_conflicts": conflicts.role_conflicts, |
| "time_conflict": conflicts.time_conflict, |
| "predicate_conflict": conflicts.predicate_conflict, |
| "modality_conflict": conflicts.modality_conflict, |
| "missing_feature_reasons": conflicts.missing_feature_reasons, |
| "status": status, |
| }, trigger_refs, |
| )) |
| for statement in sorted(statements, key=lambda row: row.statement_id): |
| identities.append(PublicIdentity( |
| statement.statement_id, "STATEMENT", statement.representative, |
| statement.canonical_text, { |
| "canonical_text": statement.canonical_text, |
| "statement_type_status": statement.statement_type_status, |
| "statement_type_value": statement.statement_type_value, |
| "status": ( |
| "UNRESOLVED" if statement.statement_type_status in ("ERROR", "UNRESOLVED") |
| else "EXECUTED" |
| ), |
| }, |
| )) |
| remap[statement.statement_id] = statement.statement_id |
| for row in sorted(local_entities, key=lambda item: item.entity_id): |
| if not row.grounding: |
| raise ValueError("compact LocalEntity lacks representative grounding") |
| ref = row.grounding[0] |
| identity_id = row.entity_id |
| name = row.canonical_name |
| if ref.text(article_version_id=article_version, source_text=article.content) != name: |
| raise ValueError("compact LocalEntity representative name differs from source") |
| identities.append(PublicIdentity( |
| identity_id, "ENTITY", ref, name, { |
| "canonical_name": name, |
| "entity_type": row.entity_type, |
| "identity_confidence": row.identity_confidence, |
| "identity_confidence_source": row.identity_confidence_source, |
| "article_local": True, "status": "EXECUTED", |
| }, |
| )) |
| time_refs: dict[str, list[GroundingRef]] = {} |
| time_keys = {} |
| for occurrence in temporal_occurrences: |
| if occurrence.representative.article_version_id != article_version: |
| raise ValueError("canonical Time occurrence crossed article version") |
| occurrence.representative.text(article_version_id=article_version, |
| source_text=article.content) |
| key = occurrence.normalized_key |
| if key is None: |
| continue |
| identity_id = _stable_id("LTIME", article_id, *key.parts()) |
| time_keys[identity_id] = key |
| time_refs.setdefault(identity_id, []).append(occurrence.representative) |
| remap[occurrence.prediction_id] = identity_id |
| remap[occurrence.temporal_occurrence_id] = identity_id |
| for identity_id, key in sorted(time_keys.items()): |
| refs = _unique_refs(time_refs[identity_id]) |
| identities.append(PublicIdentity( |
| identity_id, "TIME", refs[0], key.value, { |
| "normalized_value": key.value, |
| "granularity": key.granularity, |
| "normalization_status": "NORMALIZED", |
| "temporal_semantic_type": key.semantic_type, |
| "timezone": key.timezone, |
| "normalization_source": "V1_RUNTIME_COMPATIBILITY_VIEW", |
| "status": "EXECUTED", |
| }, refs, |
| )) |
| identity_ids = {row.identity_id for row in identities} |
| facts = [] |
| pending = [] |
| unresolved_count = 0 |
| unresolved_reasons = dict(lane_reasons or {}) |
| for event in event_states: |
| for fact in event.facts: |
| if fact.source_identity_id not in identity_ids: |
| raise ValueError("Event closure fact lacks canonical LocalEvent source") |
| if fact.target_identity_id in identity_ids: |
| facts.append(fact) |
| elif fact.relation == "OCCURRED_ON" and any( |
| item.temporal_occurrence_id == fact.target_identity_id |
| for item in temporal_occurrences |
| ): |
| pending.append(fact) |
| else: |
| unresolved_count += 1 |
| unresolved_reasons["fact_endpoint"] = "CANONICAL_TARGET_UNAVAILABLE" |
| if diagnostic_sink is not None: |
| diagnostic_sink.record("decision", { |
| "component": "compact_fact_endpoint", |
| "fact_id": fact.fact_id, |
| "relation": fact.relation, |
| "target_identity_id": fact.target_identity_id, |
| "decision": "UNRESOLVED_CANONICAL_TARGET", |
| }) |
| for identity in identities: |
| if identity.kind not in ("EVENT", "STATEMENT", "ENTITY"): |
| continue |
| relation = {"EVENT": "COVERS", "STATEMENT": "CONTAINS_STATEMENT", |
| "ENTITY": "MENTIONS"}[identity.kind] |
| facts.append(GroundedFact( |
| _stable_id("SFACT", article_id, relation, identity.identity_id), |
| relation, article_id, identity.identity_id, |
| (identity.representative,), 1.0, "EXECUTED", |
| )) |
| refs = _grounding_registry(identities, facts, pending) |
| statuses = {str(key): str(value) for key, value in lane_statuses.items()} |
| statuses["graph_assembly"] = "UNRESOLVED" if unresolved_count else "EXECUTED" |
| return ResolvedGraphState( |
| article_id, article_version, digest, article.published_at, |
| tuple(identities), tuple(facts), remap, refs, statuses, |
| article.title, article.source, runtime_config_id, assembly_config_id, |
| tuple(active_policy_ids), dict(component_provenance), |
| tuple(temporal_occurrences), tuple(pending), unresolved_count, |
| unresolved_reasons, |
| runtime_config_sha256, assembly_config_sha256, |
| public_schema_sha256, dict(policy_config_sha256 or {}), |
| ) |
|
|
|
|
| def derive_canonical_time_v2( |
| state: ResolvedGraphState, article: ArticleInput, |
| *, diagnostic_sink=None, |
| ) -> ResolvedGraphState: |
| """V1 미정규화 canonical occurrence에만 V2를 적용하고 pending fact를 remap한다.""" |
| if (state.article_version_id != article.article_version_id |
| or state.content_sha256 != ResolvedGraphState.content_digest(article.content)): |
| raise ValueError("direct Time derivation source differs from graph state") |
| identities = list(state.identities) |
| known_ids = {row.identity_id for row in identities} |
| remap = dict(state.identity_remap) |
| reasons = dict(state.lane_reasons or {}) |
| normalizer = TimeNormalizerV2() |
| time_derivation_issue = False |
| for occurrence in state.temporal_occurrences: |
| if occurrence.normalized_key is not None: |
| continue |
| text = occurrence.representative.text( |
| article_version_id=state.article_version_id, |
| source_text=article.content, |
| ) |
| if occurrence.temporal_semantic_type != "POINT": |
| result = None |
| reason = "NON_POINT_TIME_NOT_NORMALIZED" |
| else: |
| result = normalizer.normalize(text, state.published_at) |
| reason = result.rule_id |
| key = normalized_temporal_key( |
| result.to_dict(), semantic_type=occurrence.temporal_semantic_type, |
| ) if result is not None else None |
| if key is None: |
| reasons["time_v2"] = "UNRESOLVED_CANONICAL_TIME" |
| time_derivation_issue = True |
| decision = "NOT_DERIVED" |
| else: |
| identity_id = _stable_id("LTIME", state.article_id, *key.parts()) |
| if identity_id not in known_ids: |
| identities.append(PublicIdentity( |
| identity_id, "TIME", occurrence.representative, key.value, { |
| "normalized_value": key.value, "granularity": key.granularity, |
| "normalization_status": "NORMALIZED", |
| "temporal_semantic_type": key.semantic_type, |
| "timezone": key.timezone, |
| "normalization_source": "TIME_NORMALIZER_V2", |
| "status": "EXECUTED", |
| }, (occurrence.representative,), |
| )) |
| known_ids.add(identity_id) |
| remap[occurrence.prediction_id] = identity_id |
| remap[occurrence.temporal_occurrence_id] = identity_id |
| decision = "DERIVED" |
| if diagnostic_sink is not None: |
| diagnostic_sink.record("decision", { |
| "component": "compact_time_normalizer_v2", |
| "temporal_occurrence_id": occurrence.temporal_occurrence_id, |
| "decision": decision, "rule_id": reason, |
| "article_version_id": state.article_version_id, |
| "char_start": occurrence.representative.char_start, |
| "char_end": occurrence.representative.char_end, |
| }) |
| facts = list(state.facts) |
| unresolved_count = state.unresolved_fact_count |
| event_conflicts = { |
| row.identity_id: row.semantic_properties |
| for row in identities if row.kind == "EVENT" |
| } |
| for fact in state.pending_temporal_facts: |
| target = remap.get(fact.target_identity_id) |
| if target not in known_ids: |
| unresolved_count += 1 |
| time_derivation_issue = True |
| continue |
| conflict = event_conflicts[fact.source_identity_id] |
| conflict_status = bool( |
| conflict.get("time_conflict") or conflict.get("predicate_conflict") |
| or conflict.get("modality_conflict") |
| ) |
| facts.append(replace( |
| fact, fact_id=_stable_id("EFACT", fact.source_identity_id, |
| fact.relation, target), |
| target_identity_id=target, |
| status="UNRESOLVED" if conflict_status else "EXECUTED", |
| )) |
| |
| grouped: dict[tuple[str, str, str], list[GroundedFact]] = {} |
| for fact in facts: |
| grouped.setdefault((fact.source_identity_id, fact.relation, |
| fact.target_identity_id), []).append(fact) |
| compact_facts = tuple(GroundedFact( |
| _stable_id("CFACT", *key), key[1], key[0], key[2], |
| _unique_refs(ref for row in rows for ref in row.supports), |
| max((row.confidence for row in rows if row.confidence is not None), |
| default=None), |
| "UNRESOLVED" if any(row.status == "UNRESOLVED" for row in rows) |
| else "EXECUTED", |
| ) for key, rows in sorted(grouped.items())) |
| statuses = dict(state.lane_statuses) |
| if time_derivation_issue: |
| statuses["time_resolution"] = ( |
| "UNRESOLVED" if statuses.get("time_resolution") != "ERROR" else "ERROR" |
| ) |
| if unresolved_count: |
| statuses["graph_assembly"] = "UNRESOLVED" |
| return replace( |
| state, identities=tuple(identities), facts=compact_facts, |
| identity_remap=remap, |
| grounding_registry=_grounding_registry(identities, compact_facts), |
| pending_temporal_facts=(), temporal_occurrences=(), |
| unresolved_fact_count=unresolved_count, |
| lane_statuses=statuses, lane_reasons=reasons, |
| ) |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class CompactGraphEvidence: |
| owner_id: str |
| article_version_id: str |
| sentence_index: int | None |
| char_start: int |
| char_end: int |
| text: str |
| semantic_purpose: str |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class CompactGraphNode: |
| node_id: str |
| kind: str |
| properties: Mapping[str, Any] |
| evidence: tuple[CompactGraphEvidence, ...] |
| provenance: Mapping[str, Any] |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class CompactGraphEdge: |
| edge_id: str |
| edge_type: str |
| source_id: str |
| target_id: str |
| confidence: float | None |
| status: str |
| evidence: tuple[CompactGraphEvidence, ...] |
| provenance: Mapping[str, Any] |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class CompactKnowledgeGraphResult: |
| schema_version: str |
| release_version: str |
| compact_runtime_contract: str |
| output_profile_contract: str |
| output_profile: str |
| assembly_config_id: str |
| runtime_config_id: str |
| status: str |
| article: Mapping[str, Any] |
| nodes: tuple[CompactGraphNode, ...] |
| edges: tuple[CompactGraphEdge, ...] |
| coverage: Mapping[str, Any] |
| provenance: Mapping[str, Any] |
| diagnostics: Mapping[str, Any] |
| validation: Mapping[str, Any] = field(default_factory=dict) |
|
|
| def to_dict(self) -> dict[str, Any]: |
| return json.loads(json.dumps(asdict(self), ensure_ascii=False, allow_nan=False)) |
|
|
| def to_json(self, *, indent: int | None = 2) -> str: |
| return json.dumps(self.to_dict(), ensure_ascii=False, allow_nan=False, |
| indent=indent, sort_keys=True) |
|
|
| def pretty(self) -> str: |
| """PUBLIC canonical identities와 relations를 표시한다.""" |
| labels = { |
| "ARTICLE": "title", "EVENT": "canonical_text", |
| "STATEMENT": "canonical_text", "ENTITY": "canonical_name", |
| "TIME": "normalized_value", |
| } |
| lines = [f"ARTICLE {self.article['article_id']}: {self.article.get('title') or ''}"] |
| for node in self.nodes: |
| if node.kind == "ARTICLE": |
| continue |
| label = node.properties.get(labels[node.kind]) or ( |
| node.evidence[0].text if node.evidence else "UNRESOLVED" |
| ) |
| lines.append(f"{node.kind} {node.node_id}: {label}") |
| for edge in self.edges: |
| lines.append( |
| f"{edge.edge_type} {edge.source_id} -> {edge.target_id}" |
| f" [{edge.status}]" |
| ) |
| return "\n".join(lines) + "\n" |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class CompactAssemblyOutcome: |
| output: CompactKnowledgeGraphResult |
| resolved_state: ResolvedGraphState |
| diagnostic_records: tuple[Mapping[str, Any], ...] = () |
| diagnostic_artifact_reference: str | None = None |
|
|
|
|
| @dataclass(frozen=True, slots=True) |
| class CompactAssemblyConfig: |
| path: Path |
| payload: Mapping[str, Any] |
| schema_path: Path |
|
|
| @property |
| def assembly_config_id(self) -> str: |
| return str(self.payload["assembly_config_id"]) |
|
|
| @property |
| def runtime_config_id(self) -> str: |
| return str(self.payload["source_runtime_config_id"]) |
|
|
|
|
| def load_compact_assembly_config(path: str | Path) -> CompactAssemblyConfig: |
| config_path = Path(path).resolve() |
| value = json.loads(config_path.read_text(encoding="utf-8")) |
| if (value.get("schema_version") != COMPACT_ASSEMBLY_CONFIG_SCHEMA_VERSION |
| or value.get("public_schema_version") != PUBLIC_SCHEMA_VERSION |
| or value.get("compact_runtime_contract") != COMPACT_RUNTIME_CONTRACT_ID |
| or value.get("output_profile_contract") != OUTPUT_PROFILE_CONTRACT_ID |
| or value.get("release_version") != "2.2" |
| or value.get("default_profile") != "PUBLIC" |
| or value.get("node_kinds") != list(NODE_KINDS) |
| or value.get("edge_types") != list(EDGE_TYPES) |
| or value.get("default_legacy_graph_adapter") is not False |
| or not value.get("source_runtime_config_id") |
| or not value.get("assembly_config_id")): |
| raise ValueError("unsupported v2.2 compact assembly config") |
| relative = Path(str(value["public_schema"]["path"])) |
| if relative.is_absolute() or ".." in relative.parts: |
| raise ValueError("compact schema path must remain repository-relative") |
| root = Path(__file__).resolve().parents[2] |
| schema_path = (root / relative).resolve() |
| schema_path.relative_to(root) |
| if (not schema_path.is_file() |
| or sha256(schema_path.read_bytes()).hexdigest() |
| != str(value["public_schema"]["sha256"])): |
| raise ValueError("compact PUBLIC schema is absent or differs") |
| schema = json.loads(schema_path.read_text(encoding="utf-8")) |
| if (schema.get("schema_version") != PUBLIC_SCHEMA_VERSION |
| or schema.get("node_kinds") != list(NODE_KINDS) |
| or schema.get("edge_endpoints") != { |
| relation: list(kinds) for relation, kinds in EDGE_ENDPOINTS.items() |
| } |
| or { |
| kind: set(keys) for kind, keys in |
| schema.get("semantic_property_allowlist", {}).items() |
| } != { |
| kind: set(keys) for kind, keys in PUBLIC_SEMANTIC_PROPERTY_KEYS.items() |
| } |
| or { |
| kind: set(keys) for kind, keys in |
| schema.get("semantic_property_required", {}).items() |
| } != { |
| kind: set(keys) for kind, keys in PUBLIC_SEMANTIC_REQUIRED_KEYS.items() |
| }): |
| raise ValueError("compact PUBLIC schema ontology mismatch") |
| return CompactAssemblyConfig(config_path, value, schema_path) |
|
|
|
|
| def _evidence(owner: str, ref: GroundingRef, purpose: str, |
| state: ResolvedGraphState, article: ArticleInput) -> CompactGraphEvidence: |
| return CompactGraphEvidence( |
| owner, ref.article_version_id, ref.sentence_index, |
| ref.char_start, ref.char_end, |
| ref.text(article_version_id=state.article_version_id, |
| source_text=article.content), purpose, |
| ) |
|
|
|
|
| def _dedup_evidence(rows) -> tuple[CompactGraphEvidence, ...]: |
| keyed = { |
| (row.article_version_id, row.char_start, row.char_end, |
| row.semantic_purpose): row for row in rows |
| } |
| return tuple(keyed[key] for key in sorted(keyed)) |
|
|
|
|
| class CompactArticleLocalKGAssembler: |
| """ResolvedGraphState -> narrow V2 Time derivation -> direct materialization.""" |
|
|
| def __init__(self, config: CompactAssemblyConfig) -> None: |
| self.config = config |
| self.output_projector = CompactOutputProfileProjector() |
|
|
| @classmethod |
| def from_config(cls, path: str | Path) -> "CompactArticleLocalKGAssembler": |
| return cls(load_compact_assembly_config(path)) |
|
|
| def assemble(self, state: ResolvedGraphState, *, source_article: ArticleInput, |
| policy) -> CompactKnowledgeGraphResult: |
| return self.assemble_with_audit( |
| state, source_article=source_article, policy=policy, |
| ).output |
|
|
| def assemble_with_audit(self, state: ResolvedGraphState, *, |
| source_article: ArticleInput, policy) -> CompactAssemblyOutcome: |
| started = time.perf_counter() |
| if (not isinstance(state, ResolvedGraphState) |
| or state.runtime_config_id != self.config.runtime_config_id |
| or state.assembly_config_id != self.config.assembly_config_id): |
| raise ValueError("v2.2 direct assembly requires its compact state/config") |
| derived = derive_canonical_time_v2( |
| state, source_article, diagnostic_sink=policy.diagnostic_sink, |
| ) |
| derived = replace( |
| derived, lane_statuses={**derived.lane_statuses, |
| "graph_assembly": "EXECUTED"}, |
| ) |
| result = self._materialize(derived, source_article) |
| validation = self.validate(result, source_article) |
| result = replace(result, validation=validation) |
| policy.diagnostic_sink.record("stage", { |
| "stage": "⑧ 그래프 조립부", |
| "component": "Compact direct PUBLIC assembly", |
| "input_count": len(state.identities), |
| "candidate_count": 0, |
| "output_count": { |
| "nodes": len(result.nodes), "edges": len(result.edges), |
| }, |
| "drop_reason_counts": { |
| "UNRESOLVED_FACT": derived.unresolved_fact_count, |
| }, |
| "warnings": tuple(derived.lane_reasons or {}), |
| "elapsed_seconds": time.perf_counter() - started, |
| }) |
| result = self.output_projector.project( |
| result, policy=policy, |
| ) |
| capture = policy.diagnostic_sink.summary() |
| return CompactAssemblyOutcome( |
| result, derived, |
| tuple(policy.diagnostic_sink.records) |
| if policy.capture_level.value != "SUMMARY" else (), |
| capture["artifact_path"], |
| ) |
|
|
| def _materialize(self, state: ResolvedGraphState, |
| article: ArticleInput) -> CompactKnowledgeGraphResult: |
| public_ids = { |
| row.identity_id: _stable_id( |
| row.kind, state.article_id, state.article_version_id, row.identity_id, |
| ) for row in state.identities |
| } |
| nodes = [] |
| for identity in state.identities: |
| node_id = public_ids[identity.identity_id] |
| evidence = [] |
| if identity.representative is not None: |
| evidence.append(_evidence(node_id, identity.representative, |
| "REPRESENTATIVE", state, article)) |
| for ref in identity.evidence: |
| purpose = "TRIGGER" if identity.kind == "EVENT" else ( |
| "TIME_OCCURRENCE" if identity.kind == "TIME" else "SEMANTIC_SUPPORT" |
| ) |
| if ref != identity.representative: |
| evidence.append(_evidence(node_id, ref, purpose, state, article)) |
| nodes.append(CompactGraphNode( |
| node_id, identity.kind, dict(identity.semantic_properties), |
| _dedup_evidence(evidence), { |
| "identity_confidence_source": identity.semantic_properties.get( |
| "identity_confidence_source" |
| ), |
| "independent_support_count": identity.semantic_properties.get( |
| "independent_support_count" |
| ), |
| }, |
| )) |
| edges = [] |
| for fact in state.facts: |
| source, target = public_ids[fact.source_identity_id], public_ids[fact.target_identity_id] |
| edge_id = _stable_id("EDGE", state.article_id, state.article_version_id, |
| fact.relation, source, target) |
| edges.append(CompactGraphEdge( |
| edge_id, fact.relation, source, target, fact.confidence, |
| fact.status, |
| _dedup_evidence( |
| _evidence(edge_id, ref, fact.relation, state, article) |
| for ref in fact.supports |
| ), { |
| "fact_status": fact.status, |
| "confidence_source": ( |
| "LEARNED_RELATION_SCORE" if fact.relation in |
| ("ACTOR", "TARGET", "PLACE", "OCCURRED_ON") |
| else "DETERMINISTIC_ARTICLE_STRUCTURE" |
| ), |
| }, |
| )) |
| status = "PARTIAL" if ( |
| state.unresolved_fact_count |
| or state.lane_reasons |
| or any(value in ("UNRESOLVED", "ERROR") |
| for value in state.lane_statuses.values()) |
| ) else "READY" |
| return CompactKnowledgeGraphResult( |
| PUBLIC_SCHEMA_VERSION, "2.2", COMPACT_RUNTIME_CONTRACT_ID, |
| OUTPUT_PROFILE_CONTRACT_ID, OutputProfile.PUBLIC.value, |
| state.assembly_config_id, state.runtime_config_id, status, |
| { |
| "article_id": state.article_id, |
| "article_version_id": state.article_version_id, |
| "content_sha256": state.content_sha256, |
| "published_at": state.published_at, |
| "title": state.title, "source": state.source, |
| }, |
| tuple(sorted(nodes, key=lambda row: (NODE_KINDS.index(row.kind), row.node_id))), |
| tuple(sorted(edges, key=lambda row: (EDGE_TYPES.index(row.edge_type), row.edge_id))), |
| {"lane_statuses": dict(state.lane_statuses), |
| "unresolved_fact_count": state.unresolved_fact_count, |
| "lane_reasons": dict(state.lane_reasons or {})}, |
| { |
| "active_policy_ids": state.active_policy_ids, |
| "component_checkpoint_sha256": dict(state.component_provenance or {}), |
| "article_version_id": state.article_version_id, |
| "content_sha256": state.content_sha256, |
| "identity_scope": "ARTICLE_LOCAL_ONLY", |
| "runtime_config_sha256": state.runtime_config_sha256, |
| "assembly_config_sha256": state.assembly_config_sha256, |
| "public_schema_sha256": state.public_schema_sha256, |
| "policy_config_sha256": dict(state.policy_config_sha256 or {}), |
| }, |
| {}, |
| ) |
|
|
| @staticmethod |
| def validate(result: CompactKnowledgeGraphResult, |
| article: ArticleInput) -> Mapping[str, Any]: |
| """직접 node/edge를 검사한다; full result.to_dict()를 만들지 않는다.""" |
| if (result.schema_version != PUBLIC_SCHEMA_VERSION |
| or result.article["article_version_id"] != article.article_version_id |
| or result.article["content_sha256"] |
| != ResolvedGraphState.content_digest(article.content)): |
| raise ValueError("v2.2 PUBLIC source/schema mismatch") |
| node_by_id = {row.node_id: row for row in result.nodes} |
| if len(node_by_id) != len(result.nodes): |
| raise ValueError("duplicate v2.2 PUBLIC node") |
| if sum(row.kind == "ARTICLE" for row in result.nodes) != 1: |
| raise ValueError("v2.2 PUBLIC requires one ARTICLE") |
| if len({row.edge_id for row in result.edges}) != len(result.edges): |
| raise ValueError("duplicate v2.2 PUBLIC edge") |
| for node in result.nodes: |
| if node.kind not in NODE_KINDS: |
| raise ValueError("internal node entered v2.2 PUBLIC") |
| for ref in node.evidence: |
| if (ref.owner_id != node.node_id |
| or ref.article_version_id != article.article_version_id |
| or article.content[ref.char_start:ref.char_end] != ref.text): |
| raise ValueError("v2.2 node grounding/owner differs") |
| for edge in result.edges: |
| if (edge.edge_type not in EDGE_ENDPOINTS |
| or edge.source_id not in node_by_id |
| or edge.target_id not in node_by_id |
| or (node_by_id[edge.source_id].kind, |
| node_by_id[edge.target_id].kind) != EDGE_ENDPOINTS[edge.edge_type]): |
| raise ValueError("v2.2 PUBLIC dangling/invalid relation endpoint") |
| for ref in edge.evidence: |
| if (ref.owner_id != edge.edge_id |
| or ref.article_version_id != article.article_version_id |
| or article.content[ref.char_start:ref.char_end] != ref.text): |
| raise ValueError("v2.2 edge grounding/owner differs") |
| return { |
| "status": "PASS", "node_count": len(result.nodes), |
| "edge_count": len(result.edges), "dangling_edge_count": 0, |
| "invalid_endpoint_count": 0, "source_grounding_roundtrip": "PASS", |
| } |
|
|
