| """Reproducible mechanical and CUDA evaluation for `.ppkg` v1.""" |
|
|
| from __future__ import annotations |
|
|
| from dataclasses import asdict |
| import gc |
| import hashlib |
| import json |
| import math |
| from pathlib import Path |
| import random |
| import tempfile |
| import time |
| import tracemalloc |
|
|
| import torch |
| import torch.nn.functional as F |
| from transformers import AutoModelForCausalLM, AutoTokenizer |
|
|
| from pcm.planner.canonical import CanonicalPConfig, CanonicalPStore |
| from pcm.planner.personality import ( |
| EvidenceAuthority, |
| EvidenceRecord, |
| FactorizedPersonalityCanonicalizer, |
| PersonalityPackage, |
| PersonalityQuery, |
| PersonalityRouter, |
| PersonalityStatus, |
| PersonalityTranslateSession, |
| PersonalityType, |
| PromotionPolicy, |
| merge_active_personality_with_p_cache, |
| synthetic_entry, |
| ) |
| from pcm.planner.pythia_split_translate import PythiaSplitTranslatedModel |
| from pcm.planner.representation import FactorizedStateRepresentation |
| from pcm.planner.compatibility import TensorTranslationLayer |
| from pcm.planner.split_translator import ByteEntityEncoder, CanonicalPRouter |
| from pcm.planner.canonical import CANONICAL_VALUE_LABELS as VALUE_LABELS |
|
|
|
|
| STAMP = "2026-06-01T00:00:00+00:00" |
| NATURAL_RP = ( |
| "Mira folded the map, squinted at the crooked ink, and laughed.\nCaptain:", |
| "Rain tapped the observatory windows while the old astronomer adjusted the brass lens.\nGuest:", |
| "The fox-eared courier set down her tea and nudged the sealed letter across the table.\nCourier:", |
| ) |
|
|
|
|
| def _file_sha256(path: Path) -> str: |
| digest = hashlib.sha256() |
| with path.open("rb") as handle: |
| for chunk in iter(lambda: handle.read(1024 * 1024), b""): |
| digest.update(chunk) |
| return digest.hexdigest() |
|
|
|
|
| def _evidence( |
| prefix: str, |
| index: int, |
| *, |
| value: str, |
| context: str, |
| scope: str, |
| authority: EvidenceAuthority = EvidenceAuthority.SINGLE_OBSERVED, |
| confidence: float = 0.8, |
| relation: str = "response_style", |
| subject: str = "user", |
| relationship: str | None = None, |
| connected: tuple[str, ...] = (), |
| ) -> EvidenceRecord: |
| return EvidenceRecord( |
| id=f"{prefix}-{index}", entry_type=PersonalityType.INTERACTION_STYLE.value, |
| subject=subject, relation=relation, value=value, context=context, |
| scope=scope, confidence=confidence, source_authority=authority.value, |
| timestamp=f"2026-06-{index + 1:02d}T00:00:00+00:00", |
| archive_reference=f"archive://heldout/{prefix}/{index}", |
| relationship=relationship, connected_evidence_ids=connected, |
| ) |
|
|
|
|
| def _promote( |
| package: PersonalityPackage, |
| prefix: str, |
| *, |
| value: str, |
| scope: str, |
| relation: str = "response_style", |
| subject: str = "user", |
| contexts: tuple[str, ...] = ("debugging", "planning", "explanation"), |
| relationship: str | None = None, |
| ): |
| decision = None |
| for index, context in enumerate(contexts): |
| decision = package.ingest(_evidence( |
| prefix, index, value=value, context=context, scope=scope, |
| relation=relation, subject=subject, relationship=relationship, |
| )) |
| if decision is None or not decision.promoted: |
| raise RuntimeError("synthetic durable personality did not promote") |
| return package.entry(decision.entry_id), decision |
|
|
|
|
| def build_proof_package(path: str | Path) -> dict[str, object]: |
| path = Path(path) |
| policy = PromotionPolicy() |
| with PersonalityPackage.create( |
| path, package_id="phase-b-personality-proof", created_at=STAMP, overwrite=True, |
| metadata={"purpose": "mechanical durable personality proof"}, |
| ) as package: |
| weak = package.ingest(_evidence( |
| "weak", 0, value="verbose", context="one-off", scope="global", |
| authority=EvidenceAuthority.SINGLE_OBSERVED, |
| )) |
| repeated, repeated_decision = _promote( |
| package, "repeat", value="concise", scope="global" |
| ) |
| narrow = [ |
| _evidence( |
| "narrow", index, value="structured", context="debugging", scope="global", |
| authority=EvidenceAuthority.SINGLE_OBSERVED, |
| ) |
| for index in range(5) |
| ] |
| connected = [ |
| _evidence( |
| "connected", index, value="direct", context=context, scope="global", |
| authority=EvidenceAuthority.SINGLE_OBSERVED, |
| connected=((f"connected-{index - 1}",) if index else ()), |
| ) |
| for index, context in enumerate(("debugging", "planning", "casual")) |
| ] |
| narrow_score = policy.score(narrow)["promotion_score"] |
| connected_score = policy.score(connected)["promotion_score"] |
| unsupported = None |
| for index in range(20): |
| unsupported = package.ingest(_evidence( |
| "unsupported", index, value="hostile", context=f"self-{index}", |
| scope="global", authority=EvidenceAuthority.MODEL_UNSUPPORTED, |
| )) |
| correction = package.ingest(_evidence( |
| "correction", 20, value="detailed", context="explicit-correction", |
| scope="global", authority=EvidenceAuthority.EXPLICIT_USER, |
| confidence=1.0, |
| )) |
| technical, _ = _promote( |
| package, "technical", value="concise", scope="technical" |
| ) |
| creative, _ = _promote( |
| package, "creative", value="detailed", scope="creative", |
| contexts=("fiction", "poetry", "worldbuilding"), |
| ) |
| relationship, _ = _promote( |
| package, "relationship", value="playful", scope="roleplay", |
| subject="assistant", relationship="captain-mira", |
| ) |
| persona, _ = _promote( |
| package, "persona", value="Alice", scope="technical", |
| relation="preferred_persona", |
| ) |
| router = PersonalityRouter() |
| technical_route = router.retrieve(package, PersonalityQuery( |
| subject="user", interaction_type="technical", domain="debugging", |
| relation="response_style", timestamp=STAMP, |
| ), top_k=1) |
| creative_route = router.retrieve(package, PersonalityQuery( |
| subject="user", interaction_type="creative", domain="fiction", |
| relation="response_style", timestamp=STAMP, |
| ), top_k=1) |
| relationship_route = router.retrieve(package, PersonalityQuery( |
| subject="assistant", interaction_type="roleplay", domain="chat", |
| relation="response_style", relationship="captain-mira", timestamp=STAMP, |
| ), top_k=1) |
| irrelevant = router.retrieve(package, PersonalityQuery( |
| subject="stranger", interaction_type="medical", domain="finance", |
| relation="unrelated", timestamp=STAMP, |
| ), top_k=8) |
| topk = {} |
| for count in (1, 4, 8): |
| result = router.retrieve(package, PersonalityQuery( |
| subject="user", interaction_type="technical", domain="debugging", |
| relation="response_style", timestamp=STAMP, |
| ), top_k=count) |
| topk[str(count)] = { |
| "loaded_entries": len(result.entries), |
| "target_recall": float(technical.id in result.route.entry_ids), |
| "logical_bytes_read": result.logical_bytes_read, |
| "latency_seconds": result.retrieval_latency_seconds, |
| } |
| package.checkpoint() |
| return { |
| "one_event_promoted": weak.promoted, |
| "repeated_promoted": repeated_decision.promoted, |
| "repeated_entry": asdict(repeated), |
| "narrow_context_score": narrow_score, |
| "connected_cross_context_score": connected_score, |
| "connectivity_outscores_narrow": connected_score > narrow_score, |
| "unsupported_model_claim_promoted": bool(unsupported and unsupported.promoted), |
| "explicit_correction_promoted": correction.promoted, |
| "old_conclusion_status": package.entry(repeated.id).status, |
| "technical_context_correct": technical_route.entries[0].id == technical.id, |
| "creative_context_correct": creative_route.entries[0].id == creative.id, |
| "relationship_context_correct": relationship_route.entries[0].id == relationship.id, |
| "relevance_accuracy": sum(( |
| technical_route.entries[0].id == technical.id, |
| creative_route.entries[0].id == creative.id, |
| relationship_route.entries[0].id == relationship.id, |
| len(irrelevant.entries) == 0, |
| )) / 4, |
| "irrelevant_loaded_entries": len(irrelevant.entries), |
| "top_k": topk, |
| "persona_entry_id": persona.id, |
| "entry_count": len(package.entries()), |
| "change_count": len(package.changes()), |
| "package_size_bytes": package.size_bytes, |
| "semantic_checksum": package.metadata()["content_sha256"], |
| } |
|
|
|
|
| def deterministic_serialization_proof(root: Path) -> dict[str, object]: |
| first = root / "deterministic-a.ppkg" |
| second = root / "deterministic-b.ppkg" |
| build_proof_package(first) |
| build_proof_package(second) |
| first_hash = _file_sha256(first) |
| second_hash = _file_sha256(second) |
| return { |
| "byte_identical": first.read_bytes() == second.read_bytes(), |
| "first_sha256": first_hash, |
| "second_sha256": second_hash, |
| } |
|
|
|
|
| def growth_benchmark(root: Path, counts=(100, 1_000, 10_000, 100_000)): |
| results = {} |
| router = PersonalityRouter() |
| for count in counts: |
| path = root / f"growth-{count}.ppkg" |
| with PersonalityPackage.create( |
| path, package_id=f"growth-{count}", created_at=STAMP, overwrite=True, |
| ) as package: |
| package.bulk_insert_entries( |
| (synthetic_entry(index, timestamp=STAMP) for index in range(count)), |
| updated_at=STAMP, |
| ) |
| tracemalloc.start() |
| validation_started = time.perf_counter() |
| with PersonalityPackage(path) as package: |
| validation_latency = time.perf_counter() - validation_started |
| before_current, before_peak = tracemalloc.get_traced_memory() |
| selection = router.retrieve(package, PersonalityQuery( |
| subject="subject-0", interaction_type="domain-0", domain="domain-0", |
| relation="trait-0", timestamp=STAMP, |
| ), top_k=4) |
| after_current, after_peak = tracemalloc.get_traced_memory() |
| size = package.size_bytes |
| tracemalloc.stop() |
| results[str(count)] = { |
| "disk_size_bytes": size, |
| "lookup_latency_seconds": selection.retrieval_latency_seconds, |
| "cold_checksum_validation_seconds": validation_latency, |
| "loaded_entries": len(selection.entries), |
| "logical_bytes_read": selection.logical_bytes_read, |
| "python_ram_peak_delta_bytes": max(0, after_peak - before_peak), |
| "inactive_vram_bytes": 0, |
| "active_canonical_bytes": len(selection.entries) * 1077, |
| "full_package_loaded": False, |
| } |
| path.unlink() |
| return results |
|
|
|
|
| def _entry_store_bytes(store: CanonicalPStore | None) -> int: |
| if store is None: |
| return 0 |
| tensors = ( |
| store.cache.values, store.cache.valid, store.cache.slot_type, |
| store.cache.confidence, store.cache.importance, store.cache.freshness, |
| store.cache.persistence, store.cache.last_updated, store.cache.source, |
| store.entity_id, store.relation_id, store.value_id, |
| store.canonical_metadata_id, |
| ) |
| return sum(value.numel() * value.element_size() for value in tensors) |
|
|
|
|
| def _make_persona_package(path: Path, value: str) -> None: |
| with PersonalityPackage.create( |
| path, package_id=f"persona-{value.casefold()}", created_at=STAMP, overwrite=True, |
| ) as package: |
| _promote( |
| package, "persona", value=value, scope="technical", |
| relation="preferred_persona", |
| ) |
|
|
|
|
| def cuda_ttl_benchmark( |
| *, |
| model_path: Path, |
| ttl_path: Path, |
| router_path: Path, |
| proof_package_path: Path, |
| representation: FactorizedStateRepresentation, |
| workdir: Path, |
| ) -> dict[str, object]: |
| if not torch.cuda.is_available(): |
| raise RuntimeError("CUDA is required for exact `.translate` personality proof") |
| tokenizer = AutoTokenizer.from_pretrained(model_path, local_files_only=True) |
| tokenizer.pad_token = tokenizer.eos_token |
| tokenizer.padding_side = "left" |
| base = AutoModelForCausalLM.from_pretrained( |
| model_path, local_files_only=True, dtype=torch.float16, low_cpu_mem_usage=True, |
| ).to("cuda").eval() |
| package = TensorTranslationLayer.load(ttl_path, device="cuda") |
| canonical_router = CanonicalPRouter.load(router_path, device="cuda") |
| wrapper = PythiaSplitTranslatedModel( |
| base, package, canonical_router, ByteEntityEncoder(128) |
| ).to("cuda").eval() |
| canonicalizer = FactorizedPersonalityCanonicalizer( |
| representation, value_labels=VALUE_LABELS |
| ) |
| disk_router = PersonalityRouter() |
| package_a = workdir / "persona-a.ppkg" |
| package_b = workdir / "persona-b.ppkg" |
| package_irrelevant = workdir / "persona-irrelevant.ppkg" |
| package_low = workdir / "persona-low-confidence.ppkg" |
| package_context = workdir / "persona-context.ppkg" |
| _make_persona_package(package_a, "Alice") |
| _make_persona_package(package_b, "Bob") |
| with PersonalityPackage.create( |
| package_irrelevant, package_id="persona-irrelevant", created_at=STAMP, |
| overwrite=True, |
| ) as irrelevant_package: |
| _promote( |
| irrelevant_package, "other", value="Alice", scope="creative", |
| relation="preferred_persona", subject="someone-else", |
| ) |
| with PersonalityPackage.create( |
| package_low, package_id="persona-low", created_at=STAMP, overwrite=True, |
| ) as low_package: |
| low_package.ingest(_evidence( |
| "low", 0, value="Alice", context="debugging", scope="technical", |
| relation="preferred_persona", authority=EvidenceAuthority.SINGLE_OBSERVED, |
| confidence=0.3, |
| )) |
| with PersonalityPackage.create( |
| package_context, package_id="persona-context", created_at=STAMP, overwrite=True, |
| ) as context_package: |
| _promote( |
| context_package, "context-technical", value="Alice", scope="technical", |
| relation="preferred_persona", |
| ) |
| _promote( |
| context_package, "context-creative", value="Bob", scope="creative", |
| relation="preferred_persona", contexts=("fiction", "poetry", "worldbuilding"), |
| ) |
| query = PersonalityQuery( |
| subject="user", interaction_type="technical", domain="debugging", |
| relation="preferred_persona", timestamp=STAMP, |
| ) |
| before_inactive = torch.cuda.memory_allocated() |
| with PersonalityPackage(proof_package_path) as disk_only: |
| inactive_entries = int(disk_only._connection.execute( |
| "SELECT COUNT(*) FROM entries WHERE status=?", |
| (PersonalityStatus.ACTIVE.value,), |
| ).fetchone()[0]) |
| after_inactive = torch.cuda.memory_allocated() |
| sessions = { |
| "package_a": PersonalityTranslateSession(package_a, disk_router, canonicalizer), |
| "package_b": PersonalityTranslateSession(package_b, disk_router, canonicalizer), |
| "irrelevant": PersonalityTranslateSession(package_irrelevant, disk_router, canonicalizer), |
| "low_confidence": PersonalityTranslateSession(package_low, disk_router, canonicalizer), |
| } |
| activations = { |
| label: session.activate(query, top_k=1, device="cuda") |
| for label, session in sessions.items() |
| } |
| context_session = PersonalityTranslateSession( |
| package_context, disk_router, canonicalizer |
| ) |
| activations["context_technical"] = context_session.activate( |
| query, top_k=1, device="cuda" |
| ) |
| activations["context_creative"] = context_session.activate( |
| PersonalityQuery( |
| subject="user", interaction_type="creative", domain="fiction", |
| relation="preferred_persona", timestamp=STAMP, |
| ), |
| top_k=1, device="cuda", |
| ) |
| prompt = "Following a fresh neutral technical exchange, The owner of the user is" |
| inputs = tokenizer(prompt, return_tensors="pt", add_special_tokens=False).to("cuda") |
| alice_id = tokenizer.encode(" Alice", add_special_tokens=False)[0] |
| bob_id = tokenizer.encode(" Bob", add_special_tokens=False)[0] |
|
|
| p_cache = CanonicalPStore(CanonicalPConfig( |
| slots=1, width=512, dtype=torch.float16, device="cuda", merge_similarity=1.0, |
| )) |
| other_entry = activations["package_a"].selection.entries[0] |
| other_entry = type(other_entry)(**{ |
| **asdict(other_entry), "subject": "current-task", "value": "Alice", |
| "supporting_evidence_ids": tuple(other_entry.supporting_evidence_ids), |
| "contradicting_evidence_ids": tuple(other_entry.contradicting_evidence_ids), |
| }) |
| vector, ids = canonicalizer.encode(other_entry) |
| p_cache.create( |
| vector, entity_id=ids[0], relation_id=ids[1], value_id=ids[2], |
| metadata_id=ids[3], label="current-task", |
| ) |
| combined = merge_active_personality_with_p_cache( |
| p_cache, activations["package_a"].store |
| ) |
|
|
| def altered_store(*, label="user", relation_id=None, metadata_id=None, invalidate=False): |
| source = activations["package_a"].store |
| source_slot = int(source.valid.nonzero(as_tuple=False)[0]) |
| result = CanonicalPStore(CanonicalPConfig( |
| slots=1, width=512, dtype=torch.float16, device="cuda", |
| merge_similarity=1.0, |
| )) |
| slot, _ = result.create( |
| source.canonical_values[source_slot], |
| entity_id=int(source.entity_id[source_slot]), |
| relation_id=( |
| int(source.relation_id[source_slot]) |
| if relation_id is None else relation_id |
| ), |
| value_id=int(source.value_id[source_slot]), |
| metadata_id=( |
| int(source.canonical_metadata_id[source_slot]) |
| if metadata_id is None else metadata_id |
| ), |
| label=label, |
| ) |
| if invalidate: |
| result.invalidate(slot) |
| return result |
|
|
| wrong_entity = altered_store(label="someone-else") |
| wrong_relation = altered_store(relation_id=1) |
| historical = altered_store(metadata_id=2) |
| invalidated = altered_store(invalidate=True) |
|
|
| conditions = { |
| "frozen_base": (None, {}), |
| "router_disabled": (None, {}), |
| "translator_disabled": ( |
| activations["package_a"].store, {"injection_enabled": False} |
| ), |
| "translator_oracle_route": ( |
| activations["package_a"].store, |
| {"oracle_indices": torch.tensor([0], device="cuda")}, |
| ), |
| "p_cache_only": (p_cache, {}), |
| "wrong_entity": (wrong_entity, {}), |
| "wrong_relation": (wrong_relation, {}), |
| "historical": (historical, {}), |
| "invalidated": (invalidated, {}), |
| "p_package_a_relevant": (activations["package_a"].store, {}), |
| "p_package_b_relevant": (activations["package_b"].store, {}), |
| "p_package_irrelevant": (activations["irrelevant"].store, {}), |
| "p_package_contradictory_low_confidence": ( |
| activations["low_confidence"].store, {} |
| ), |
| "p_package_context_technical": (activations["context_technical"].store, {}), |
| "p_package_context_creative": (activations["context_creative"].store, {}), |
| "p_cache_plus_p_package": (combined, {}), |
| } |
| causal = {} |
| latency = {} |
| with torch.inference_mode(): |
| base_logits = wrapper(**inputs, use_cache=False).logits[:, -1].float() |
| for label, (store, condition_kwargs) in conditions.items(): |
| output = wrapper( |
| **inputs, p_store=store, query_entity_surfaces=["user"], |
| collect_telemetry=True, use_cache=False, **condition_kwargs, |
| ).logits[:, -1].float() |
| probability = F.softmax(output, dim=-1) |
| route = wrapper.route_telemetry[-1] if wrapper.route_telemetry else None |
| selected_index = None if route is None else int(route.indices[0, -1, 0]) |
| causal[label] = { |
| "alice_logit": float(output[0, alice_id]), |
| "bob_logit": float(output[0, bob_id]), |
| "alice_probability": float(probability[0, alice_id]), |
| "bob_probability": float(probability[0, bob_id]), |
| "generated": tokenizer.decode([int(output.argmax(-1))]), |
| "kl_from_base": float(F.kl_div( |
| F.log_softmax(base_logits, dim=-1), |
| F.softmax(output, dim=-1), reduction="batchmean", |
| )), |
| "gate": 0.0 if not wrapper.gate_telemetry else float(torch.stack([ |
| gate[:, -1].float().mean() for gate in wrapper.gate_telemetry |
| ]).mean()), |
| "selected_index": selected_index, |
| "selected_state": ( |
| None if store is None or selected_index is None |
| else store.cache.labels[selected_index] |
| ), |
| "router_score": ( |
| None if route is None else float(route.scores[0, -1, 0]) |
| ), |
| "router_accepted": ( |
| False if route is None else bool(route.accepted[0, -1]) |
| ), |
| "active_state_vram_bytes": _entry_store_bytes(store), |
| } |
| for _ in range(2): |
| wrapper( |
| **inputs, p_store=store, query_entity_surfaces=["user"], |
| use_cache=False, **condition_kwargs, |
| ) |
| torch.cuda.synchronize() |
| started = time.perf_counter() |
| for _ in range(5): |
| wrapper( |
| **inputs, p_store=store, query_entity_surfaces=["user"], |
| use_cache=False, **condition_kwargs, |
| ) |
| torch.cuda.synchronize() |
| latency[label] = (time.perf_counter() - started) / 5 |
|
|
| rp_inputs = tokenizer( |
| list(NATURAL_RP), return_tensors="pt", padding=True, |
| add_special_tokens=False, |
| ).to("cuda") |
| frozen_rp = wrapper( |
| **rp_inputs, labels=rp_inputs.input_ids, use_cache=False |
| ) |
| natural = {} |
| for label, store in ( |
| ("frozen_base", None), |
| ("p_cache_only", p_cache), |
| ("p_package_relevant", activations["package_a"].store), |
| ("p_package_irrelevant", activations["irrelevant"].store), |
| ("p_package_contradictory_low_confidence", activations["low_confidence"].store), |
| ("p_cache_plus_p_package", combined), |
| ): |
| output = wrapper( |
| **rp_inputs, labels=rp_inputs.input_ids, p_store=store, |
| use_cache=False |
| ) |
| natural[label] = { |
| "loss": float(output.loss), |
| "kl_from_base": float(F.kl_div( |
| F.log_softmax(frozen_rp.logits[:, -1].float(), dim=-1), |
| F.softmax(output.logits[:, -1].float(), dim=-1), |
| reduction="batchmean", |
| )), |
| "samples": [ |
| tokenizer.decode([int(row.argmax())]) |
| for row in output.logits[:, -1] |
| ], |
| } |
|
|
| base_gradients = sum(parameter.grad is not None for parameter in base.parameters()) |
| activation_paths = { |
| "package_a": package_a, |
| "package_b": package_b, |
| "irrelevant": package_irrelevant, |
| "low_confidence": package_low, |
| "context_technical": package_context, |
| "context_creative": package_context, |
| } |
| active_memory = { |
| label: { |
| "loaded_entries": len(activation.selection.entries), |
| "canonical_store_bytes": _entry_store_bytes(activation.store), |
| "logical_disk_bytes_read": activation.selection.logical_bytes_read, |
| "package_disk_bytes": activation_paths[label].stat().st_size, |
| } |
| for label, activation in activations.items() |
| } |
| result = { |
| "causal": causal, |
| "natural_interaction": natural, |
| "latency_seconds": latency, |
| "active_memory": active_memory, |
| "inactive_package_entries": inactive_entries, |
| "inactive_vram_delta_bytes": after_inactive - before_inactive, |
| "base_parameters_with_grad": base_gradients, |
| "source_tokens_in_recent_kv": 0, |
| "extra_prompt_tokens": 0, |
| "full_package_uploaded_to_cuda": False, |
| "candidate_accuracy": { |
| "package_a_alice": float(causal["p_package_a_relevant"]["alice_logit"] > causal["p_package_a_relevant"]["bob_logit"]), |
| "package_b_bob": float(causal["p_package_b_relevant"]["bob_logit"] > causal["p_package_b_relevant"]["alice_logit"]), |
| "irrelevant_matches_base": float( |
| causal["p_package_irrelevant"]["alice_logit"] == causal["frozen_base"]["alice_logit"] |
| and causal["p_package_irrelevant"]["bob_logit"] == causal["frozen_base"]["bob_logit"] |
| ), |
| "context_technical_alice": float( |
| causal["p_package_context_technical"]["alice_logit"] |
| > causal["p_package_context_technical"]["bob_logit"] |
| ), |
| "context_creative_bob": float( |
| causal["p_package_context_creative"]["bob_logit"] |
| > causal["p_package_context_creative"]["alice_logit"] |
| ), |
| }, |
| "relevant_personality_chat": { |
| "target": "Alice", |
| "base_target_loss": float(F.cross_entropy( |
| base_logits, torch.tensor([alice_id], device="cuda") |
| )), |
| "package_target_loss": -math.log(max( |
| causal["p_package_a_relevant"]["alice_probability"], 1e-30 |
| )), |
| "target_accuracy": float( |
| causal["p_package_a_relevant"]["alice_logit"] |
| > causal["p_package_a_relevant"]["bob_logit"] |
| ), |
| "generated": causal["p_package_a_relevant"]["generated"], |
| }, |
| } |
| for session in sessions.values(): |
| session.close() |
| context_session.close() |
| wrapper.close() |
| del wrapper, base, package, canonical_router |
| gc.collect() |
| torch.cuda.empty_cache() |
| return result |
|
|
|
|
| def run_personality_package_experiment( |
| *, |
| output_package: Path, |
| model_path: Path | None = None, |
| ttl_path: Path | None = None, |
| router_path: Path | None = None, |
| representation: FactorizedStateRepresentation | None = None, |
| growth_counts=(100, 1_000, 10_000, 100_000), |
| ) -> dict[str, object]: |
| mechanical = build_proof_package(output_package) |
| with tempfile.TemporaryDirectory(prefix="pcm-ppkg-") as temporary: |
| root = Path(temporary) |
| deterministic = deterministic_serialization_proof(root) |
| growth = growth_benchmark(root, counts=growth_counts) |
| cuda = None |
| if model_path and ttl_path and router_path and representation is not None: |
| cuda = cuda_ttl_benchmark( |
| model_path=model_path, ttl_path=ttl_path, |
| router_path=router_path, proof_package_path=output_package, |
| representation=representation, workdir=root, |
| ) |
| durability_started = time.perf_counter() |
| with PersonalityPackage(output_package) as restored: |
| durability = { |
| "checksum_valid": True, |
| "active_entries_after_restart": len(restored.entries( |
| status=PersonalityStatus.ACTIVE.value |
| )), |
| "conversation_replay_required": False, |
| "cold_load_seconds": time.perf_counter() - durability_started, |
| } |
| return { |
| "experiment": "phase-b-personality-package-v1", |
| "format": "pcm-personality-package-v1", |
| "protocol": "pcm-canonical-personality-v1", |
| "phase_c_started": False, |
| "lora_used": False, |
| "base_weights_modified": False, |
| "mechanical": mechanical, |
| "deterministic_serialization": deterministic, |
| "growth": growth, |
| "durability": durability, |
| "cuda_ttl": cuda, |
| "package_path": str(output_package), |
| "package_file_sha256": _file_sha256(output_package), |
| } |
|
|
