Download app.py from RumleyRum/Deterministic-Governance-Mechanism: direct link, hf CLI and curl.
- Browser
- Download file 64.6 kB
-
https://huggingface.co/spaces/RumleyRum/Deterministic-Governance-Mechanism/resolve/main/app.py
- Command line
-
hf download hf://spaces/RumleyRum/Deterministic-Governance-Mechanism/app.py
-
curl -L -o app.py https://huggingface.co/spaces/RumleyRum/Deterministic-Governance-Mechanism/resolve/main/app.py
64.6 kB
| #!/usr/bin/env python3 | |
| """ | |
| Deterministic Exclusion Demo GUI (Development Build) | |
| FastAPI + Streamlit implementation for rapid prototyping | |
| Usage: | |
| streamlit run demo_gui_dev.py | |
| """ | |
| try: | |
| import streamlit as st | |
| import plotly.graph_objects as go | |
| import plotly.express as px | |
| except ModuleNotFoundError as exc: | |
| missing = str(exc).split("No module named ", 1)[-1].strip("'\"") | |
| print(f"Missing optional GUI dependency: {missing}") | |
| print("Install GUI deps: python -m pip install -r requirements-gui.txt") | |
| print("Run the GUI via: streamlit run demo_gui_dev.py") | |
| raise SystemExit(2) | |
| try: | |
| from streamlit.runtime.scriptrunner.script_run_context import get_script_run_ctx | |
| except Exception: | |
| get_script_run_ctx = None | |
| if get_script_run_ctx is not None and get_script_run_ctx() is None: | |
| print("This file is a Streamlit app and must be run with Streamlit:") | |
| print(" streamlit run demo_gui_dev.py") | |
| raise SystemExit(2) | |
| import time | |
| import hashlib | |
| import json | |
| from pathlib import Path | |
| from collections import deque | |
| from typing import List, Dict, Optional, Any, Tuple | |
| import os | |
| # Hugging Face Spaces detection | |
| IS_SPACES = os.getenv("SPACE_ID") is not None | |
| # Page Config MUST be the first Streamlit command | |
| st.set_page_config( | |
| page_title="Deterministic Exclusion Demo", | |
| layout="wide", | |
| initial_sidebar_state="expanded", | |
| menu_items={ | |
| 'Get Help': 'https://github.com/Rymley/Deterministic-Governance-Mechanism', | |
| 'Report a bug': 'https://github.com/Rymley/Deterministic-Governance-Mechanism/issues', | |
| 'About': """Deterministic Exclusion Demo GUI (Development Build) | |
| FastAPI + Streamlit implementation for rapid prototyping | |
| Usage: streamlit run demo_gui_dev.py""" | |
| } | |
| ) | |
| # Import engine components | |
| import sys | |
| sys.path.insert(0, str(Path(__file__).parent)) | |
| from material_field_engine import ( | |
| VerifiedSubstrate, Vector2D, MaterialFieldEngine, fp_from_float, fp_to_float, load_config | |
| ) | |
| from exclusion_demo import run_deterministic_exclusion_demo | |
| def compute_substrate_embeddings_2d(substrate_list: List[str]) -> List[List[float]]: | |
| """Cached 2D embedding for material field visualization.""" | |
| from llm_adapter import DeterministicHashEmbedderND | |
| embedder = DeterministicHashEmbedderND(dim=2) | |
| return [embedder.embed(t) for t in substrate_list] | |
| def compute_substrate_embeddings_highd(substrate_list: List[str]) -> List[List[float]]: | |
| """Cached 16D embedding for physics engine logic.""" | |
| from llm_adapter import DeterministicHashEmbedderND | |
| embedder = DeterministicHashEmbedderND(dim=16) | |
| return [embedder.embed(t) for t in substrate_list] | |
| # Custom CSS for monospace hash display | |
| st.markdown(""" | |
| <style> | |
| .hash-display { | |
| font-family: 'Courier New', monospace; | |
| font-size: 14px; | |
| background-color: #1e1e1e; | |
| color: #00ff00; | |
| padding: 10px; | |
| border-radius: 5px; | |
| word-break: break-all; | |
| } | |
| /* Verhash Brand Colors - Professional Slate & Blue */ | |
| .stApp { | |
| background-color: #020617; /* Slate 950 */ | |
| color: #f1f5f9; /* Slate 100 */ | |
| } | |
| /* Headers */ | |
| h1, h2, h3 { | |
| color: #f1f5f9 !important; /* Slate 100 */ | |
| font-weight: 700 !important; | |
| } | |
| /* Links */ | |
| a { | |
| color: #0059b8 !important; /* Muted Professional Blue */ | |
| text-decoration: none !important; | |
| } | |
| a:hover { | |
| color: #38bdf8 !important; /* Sky 400 */ | |
| text-decoration: underline !important; | |
| } | |
| /* Button styling */ | |
| .stButton > button { | |
| background-color: #0059b8; | |
| color: #f8fafc; | |
| border: none; | |
| font-weight: 600; | |
| border-radius: 6px; | |
| transition: all 0.2s ease; | |
| } | |
| .stButton > button:hover { | |
| background-color: #38bdf8; | |
| color: #f8fafc; | |
| box-shadow: 0 4px 12px rgba(0, 89, 184, 0.2); | |
| } | |
| /* Sidebar branding */ | |
| [data-testid="stSidebar"] { | |
| background-color: #0f172a; /* Slate 900 */ | |
| border-right: 1px solid #1e293b; | |
| } | |
| /* Metric styling */ | |
| [data-testid="stMetricValue"] { | |
| color: #0059b8; | |
| } | |
| .main-header { | |
| text-align: center; | |
| padding: 40px 0 20px 0; | |
| margin-bottom: 20px; | |
| } | |
| .company-name { | |
| color: #64748b; /* Slate 500 */ | |
| font-size: 16px; | |
| letter-spacing: 2px; | |
| text-transform: uppercase; | |
| margin-top: 8px; | |
| } | |
| .nav-links { | |
| text-align: center; | |
| padding: 15px 0; | |
| background: rgba(15, 23, 42, 0.3); | |
| border-radius: 8px; | |
| border: 1px solid #1e293b; | |
| margin-bottom: 30px; | |
| } | |
| .nav-links a { | |
| color: #0059b8; | |
| padding: 0 20px; | |
| font-size: 14px; | |
| } | |
| /* Tab Styling */ | |
| .stTabs [data-baseweb="tab-list"] { | |
| gap: 8px; | |
| background-color: transparent; | |
| } | |
| .stTabs [data-baseweb="tab"] { | |
| height: 40px; | |
| white-space: pre-wrap; | |
| background-color: #0f172a; /* Offset Color: Slate 900 */ | |
| border-radius: 4px 4px 0 0; | |
| gap: 1px; | |
| padding-top: 10px; | |
| padding-bottom: 10px; | |
| color: #94a3b8; /* Slate 400 */ | |
| border: 1px solid #1e293b; | |
| border-bottom: none; | |
| transition: all 0.2s ease; | |
| } | |
| .stTabs [data-baseweb="tab"]:hover { | |
| color: #f1f5f9; | |
| background-color: #1e293b; | |
| } | |
| .stTabs [data-baseweb="tab"][aria-selected="true"] { | |
| background-color: #0059b8 !important; /* Select Color: Sky 500 */ | |
| color: #ffffff !important; | |
| border: 1px solid #0059b8; | |
| } | |
| /* Hide default highlight bar */ | |
| [data-baseweb="tab-highlight"] { | |
| background-color: transparent !important; | |
| } | |
| /* Slider Styling */ | |
| .stSlider [data-baseweb="slider"] { | |
| background-color: transparent; | |
| } | |
| .stSlider [data-baseweb="slider"] > div > div { | |
| background: #0059b8 !important; | |
| } | |
| .stSlider [data-baseweb="slider"] [role="slider"] { | |
| background-color: #0059b8; | |
| border: 2px solid #020617; | |
| } | |
| /* Info Box Styling */ | |
| .stAlert { | |
| background-color: rgba(2, 132, 199, 0.05) !important; | |
| border: 1px solid rgba(2, 132, 199, 0.1) !important; | |
| color: #cbd5e1 !important; | |
| } | |
| </style> | |
| """, unsafe_allow_html=True) | |
| # ============================================================================ | |
| # Session State Initialization | |
| # ============================================================================ | |
| if 'results' not in st.session_state: | |
| st.session_state.results = None | |
| if 'config' not in st.session_state: | |
| st.session_state.config = { | |
| 'elastic_modulus_mode': 'multiplicative', | |
| 'elastic_modulus_sigma': 0.4, | |
| 'lambda_min': 0.35, | |
| 'lambda_max': 1.20, | |
| 'inference_steps': 8 | |
| } | |
| if 'audit_log' not in st.session_state: | |
| st.session_state.audit_log = deque(maxlen=100) | |
| # ============================================================================ | |
| # Header with Branding | |
| # ============================================================================ | |
| st.markdown(""" | |
| <div class='main-header'> | |
| <h1 style='color: #f8fafc; margin: 0; font-size: 2.8rem;'>Deterministic Governance Mechanism</h1> | |
| <p class='company-name'>Verhash LLC | Precision Substrate Computing</p> | |
| <p style='color: #64748b; font-size: 0.9rem; margin-top: 5px; font-style: italic;'> | |
| Development Build: FastAPI + Streamlit implementation for rapid prototyping | |
| </p> | |
| </div> | |
| <div class='nav-links'> | |
| <a href='https://verhash.com' target='_blank'>🌐 verhash.com</a> | |
| <a href='mailto:ryan@verhash.net'>📧 Contact</a> | |
| <a href='https://github.com/Rymley/Deterministic-Governance-Mechanism' target='_blank'>💻 GitHub</a> | |
| <a href='https://verhash.com/docs' target='_blank'>📚 Documentation</a> | |
| </div> | |
| """, unsafe_allow_html=True) | |
| st.markdown("---") | |
| # ============================================================================ | |
| # Sidebar: Configuration Panel | |
| # ============================================================================ | |
| st.sidebar.header("Configuration") | |
| # Elastic Modulus Mode | |
| mode = st.sidebar.selectbox( | |
| "Elastic Modulus Mode", | |
| options=['cosine', 'multiplicative', 'rbf'], | |
| index=1, # Default: multiplicative | |
| help="Cosine: direction only | Multiplicative: angle×proximity | RBF: proximity only" | |
| ) | |
| # Sigma parameter | |
| sigma = st.sidebar.slider( | |
| "Sigma (σ) - Field Extent", | |
| min_value=0.2, | |
| max_value=1.0, | |
| value=0.4, | |
| step=0.05, | |
| help="Lower = tighter binding, higher = looser binding" | |
| ) | |
| st.sidebar.markdown(f"**Current: σ={sigma:.2f}**") | |
| # Lambda max | |
| lambda_max = st.sidebar.slider( | |
| "Lambda Max (lambda_max) - Max Pressure", | |
| min_value=0.5, | |
| max_value=2.0, | |
| value=1.2, | |
| step=0.1, | |
| help="Higher = stricter exclusion" | |
| ) | |
| # Run button | |
| if st.sidebar.button("Run Deterministic Exclusion Demo", type="primary"): | |
| with st.spinner("Running inference..."): | |
| # Update config | |
| st.session_state.config['elastic_modulus_mode'] = mode | |
| st.session_state.config['elastic_modulus_sigma'] = sigma | |
| st.session_state.config['lambda_max'] = lambda_max | |
| # Run test | |
| results = run_deterministic_exclusion_demo( | |
| elastic_modulus_mode=mode, | |
| sigma=sigma, | |
| print_banner=False, | |
| ) | |
| st.session_state.results = results | |
| # Log event | |
| st.session_state.audit_log.append({ | |
| 'timestamp': time.time(), | |
| 'operation': 'run_inference', | |
| 'mode': mode, | |
| 'sigma': sigma, | |
| 'hash': results['hash'] | |
| }) | |
| st.success("Inference complete.") | |
| st.sidebar.markdown("---") | |
| st.sidebar.markdown("### 📞 Contact") | |
| st.sidebar.caption(""" | |
| **Verhash LLC** | [verhash.com](https://verhash.com) | |
| 📧 [ryan@verhash.net](mailto:ryan@verhash.net) | |
| *Patent Pending* | |
| """) | |
| # ============================================================================ | |
| # Main Content Area | |
| # ============================================================================ | |
| st.success("✨ **Live Demo** - Testing deterministic AI governance on real LLMs") | |
| st.markdown("---") | |
| # Create tabs | |
| tab1, tab2, tab3, tab4 = st.tabs(["Mechanism Demo", "LLM Guardrail", "Live LLM Testing", "Explain & Tune"]) | |
| # ---------------------------------------------------------------------------- | |
| # TAB 1: Mechanism Demo (Original) | |
| # ---------------------------------------------------------------------------- | |
| with tab1: | |
| if st.session_state.results is None: | |
| st.info("Configure parameters in the sidebar and click 'Run Deterministic Exclusion Demo'.") | |
| else: | |
| results = st.session_state.results | |
| # Row 1: Deterministic Audit Trail | |
| st.header("Deterministic Audit Trail") | |
| col1, col2 = st.columns([2, 1]) | |
| with col1: | |
| st.subheader("SHA-256") | |
| st.markdown(f'<div class="hash-display">{results["hash"]}</div>', unsafe_allow_html=True) | |
| with col2: | |
| st.metric("Total Excluded", results["excluded"]) | |
| # Row 2: Abstention Indicator | |
| st.header("Outcome Verification") | |
| if results.get("winner_index") is None: | |
| st.error("Abstained") | |
| else: | |
| st.success(f"Winner index: {results['winner_index']}") | |
| st.caption("Expected winner: Candidate 0 | Expected excluded: 3") | |
| # Row 3: Phase Log | |
| st.header("Phase Log") | |
| phase_log = results['phase_log'] | |
| import pandas as pd | |
| df = pd.DataFrame(phase_log) | |
| st.dataframe(df[['step', 'phase', 'pressure', 'survivors', 'excluded']], use_container_width=True, hide_index=True) | |
| # Visualization | |
| fig = go.Figure() | |
| fig.add_trace(go.Scatter( | |
| x=[e['step'] for e in phase_log], | |
| y=[e['pressure'] for e in phase_log], | |
| mode='lines+markers', | |
| name='Pressure lambda(t)', | |
| line=dict(color='cyan', width=3) | |
| )) | |
| st.plotly_chart(fig, use_container_width=True) | |
| # Row 5: Audit Log | |
| st.header("Run Log") | |
| if st.session_state.audit_log: | |
| log_df = pd.DataFrame(st.session_state.audit_log) | |
| log_df['timestamp'] = pd.to_datetime(log_df['timestamp'], unit='s') | |
| st.dataframe(log_df[['timestamp', 'operation', 'mode', 'sigma', 'hash']], use_container_width=True, hide_index=True) | |
| # ---------------------------------------------------------------------------- | |
| # TAB 2: LLM Guardrail Playground (New) | |
| # ---------------------------------------------------------------------------- | |
| with tab2: | |
| st.header("Deterministic LLM Filter") | |
| st.markdown(""" | |
| A model-agnostic post-processor that evaluates candidate outputs against a verified substrate. | |
| The mechanism deterministically accepts, rejects, or abstains based on explicit constraints. | |
| """) | |
| col_input1, col_input2 = st.columns(2) | |
| with col_input1: | |
| st.subheader("1. Verified Substrate") | |
| st.markdown("Approved facts (Ground Truth). One per line.") | |
| substrate_input = st.text_area( | |
| "Substrate", | |
| value="The sky is blue\nWater is wet\nParis is capital of France", | |
| height=200, | |
| key="llm_substrate" | |
| ) | |
| with col_input2: | |
| st.subheader("2. LLM Candidates") | |
| st.markdown("Generated responses (including hallucinations). One per line.") | |
| candidates_input = st.text_area( | |
| "Candidates", | |
| value="The sky is blue\nThe sky is green\nThe sky is made of cheese", | |
| height=200, | |
| key="llm_candidates" | |
| ) | |
| # Add full trace checkbox | |
| show_full_trace = st.checkbox( | |
| "Show full stress evolution (all steps for all candidates)", | |
| value=False, | |
| help="Disable for faster UI with large candidate sets" | |
| ) | |
| if st.button("Run Guardrail Filter", type="primary"): | |
| from llm_adapter import DeterministicGuardrail, DeterministicHashEmbedderND | |
| import math | |
| # Parse inputs | |
| substrate_list = [line.strip() for line in substrate_input.split('\n') if line.strip()] | |
| candidate_list = [line.strip() for line in candidates_input.split('\n') if line.strip()] | |
| if not substrate_list or not candidate_list: | |
| st.error("Please provide both substrate and candidates.") | |
| else: | |
| with st.spinner("Projecting to material field..."): | |
| # Initialize Guardrail | |
| # Use current sidebar config instead of hardcoded preset | |
| guard = DeterministicGuardrail( | |
| substrate_texts=substrate_list, | |
| config_preset='balanced' # Starting point | |
| ) | |
| # Overwrite with current sidebar settings for consistency | |
| guard.config['elastic_modulus_mode'] = st.session_state.config['elastic_modulus_mode'] | |
| guard.config['elastic_modulus_sigma'] = st.session_state.config['elastic_modulus_sigma'] | |
| guard.config['lambda_max'] = st.session_state.config['lambda_max'] | |
| # We want to INSPECT, not just filter | |
| inspection = guard.inspect(candidate_list) | |
| result_text = inspection['selected_text'] | |
| metrics = inspection['metrics'] | |
| candidate_metrics = metrics.get('candidates') | |
| if candidate_metrics is None: | |
| candidate_metrics = [ | |
| { | |
| 'phase_log': [], | |
| 'fractured': False, | |
| 'fractured_step': None, | |
| 'stress': 0.0, | |
| 'hash': 'N/A', | |
| } | |
| for _ in candidate_list | |
| ] | |
| # Build detailed numbers view (high-D physics + 2D projection for plot) | |
| with st.spinner("Computing embeddings..."): | |
| # Use specialized caches for different dimensionality needs | |
| sub_vecs = compute_substrate_embeddings_highd(substrate_list) | |
| sub_vecs_2d = compute_substrate_embeddings_2d(substrate_list) | |
| from llm_adapter import DeterministicHashEmbedderND | |
| embedder_highd = DeterministicHashEmbedderND(dim=16) | |
| cand_vecs = [embedder_highd.embed(t) for t in candidate_list] | |
| # Use 2D for visualization | |
| embedder_2d = DeterministicHashEmbedderND(dim=2) | |
| cand_vecs_2d = [embedder_2d.embed(t) for t in candidate_list] | |
| def cosine_similarity(v1, v2): | |
| """Cosine similarity between N-D vectors.""" | |
| dot = sum(a * b for a, b in zip(v1, v2)) | |
| mag1 = math.sqrt(sum(a * a for a in v1)) | |
| mag2 = math.sqrt(sum(b * b for b in v2)) | |
| if mag1 == 0 or mag2 == 0: | |
| return 0.0 | |
| return dot / (mag1 * mag2) | |
| def euclidean_distance(v1, v2): | |
| """Euclidean distance between N-D vectors.""" | |
| return math.sqrt(sum((a - b) ** 2 for a, b in zip(v1, v2))) | |
| numbers_view = { | |
| "embedder": { | |
| "name": "DeterministicHashEmbedderND", | |
| "definition": "sha256(text) -> 16D in [0,1], projected to 2D for plotting", | |
| }, | |
| "substrate": [ | |
| {"text": t, "vec2": [round(v[0], 8), round(v[1], 8)]} | |
| for t, v in zip(substrate_list, sub_vecs_2d) | |
| ], | |
| "candidates": [] | |
| } | |
| # For each candidate, compute detailed metrics | |
| for i, (cand_vec, cand_text) in enumerate(zip(cand_vecs, candidate_list)): | |
| # Compute alignment and distance to each substrate point | |
| per_substrate = [] | |
| best_alignment = -1 | |
| best_distance = float('inf') | |
| best_j = None | |
| for j, sub_vec in enumerate(sub_vecs): | |
| cos_sim = cosine_similarity(cand_vec, sub_vec) | |
| alignment = (cos_sim + 1.0) / 2.0 # Normalize to [0,1] | |
| dist = euclidean_distance(cand_vec, sub_vec) | |
| per_substrate.append({ | |
| "substrate_index": j, | |
| "substrate_text": substrate_list[j], | |
| "cosine_similarity": round(cos_sim, 8), | |
| "alignment_0_1": round(alignment, 8), | |
| "euclidean_distance": round(dist, 8), | |
| }) | |
| # Selection rule: highest alignment | |
| if alignment > best_alignment: | |
| best_alignment = alignment | |
| best_distance = dist | |
| best_j = j | |
| # Get engine results for this candidate | |
| cand_metrics = candidate_metrics[i] | |
| phase_log = cand_metrics.get('phase_log', []) | |
| # Build stress evolution - full or abbreviated | |
| if show_full_trace: | |
| stress_evolution = [ | |
| { | |
| "step": entry["step"], | |
| "phase": entry["phase"], | |
| "lambda": round(entry["pressure"], 8), | |
| "elastic_modulus_E": round(entry.get("elastic_modulus", 0.0), 8), | |
| "delta_stress": round(entry.get("delta_stress", 0.0), 8), | |
| "cumulative_stress": round(entry["stress"], 8), | |
| "fractured": entry["fractured"] | |
| } | |
| for entry in phase_log | |
| ] | |
| else: | |
| # Abbreviated: first 2 steps, fracture step (if any), last step | |
| abbreviated = [] | |
| if len(phase_log) > 0: | |
| abbreviated.append(phase_log[0]) | |
| if len(phase_log) > 1: | |
| abbreviated.append(phase_log[1]) | |
| fracture_step = cand_metrics.get('fractured_step') | |
| if fracture_step is not None and fracture_step > 1: | |
| abbreviated.append(phase_log[fracture_step]) | |
| elif len(phase_log) > 2: | |
| abbreviated.append(phase_log[-1]) | |
| stress_evolution = [ | |
| { | |
| "step": entry["step"], | |
| "phase": entry["phase"], | |
| "lambda": round(entry["pressure"], 8), | |
| "elastic_modulus_E": round(entry.get("elastic_modulus", 0.0), 8), | |
| "delta_stress": round(entry.get("delta_stress", 0.0), 8), | |
| "cumulative_stress": round(entry["stress"], 8), | |
| "fractured": entry["fractured"] | |
| } | |
| for entry in abbreviated | |
| ] | |
| if len(phase_log) > len(abbreviated): | |
| stress_evolution.append({ | |
| "note": f"...{len(phase_log) - len(abbreviated)} intermediate steps omitted (enable 'Show full stress evolution' to see all)" | |
| }) | |
| numbers_view["candidates"].append({ | |
| "candidate_index": i, | |
| "text": cand_text, | |
| "vec2": [round(cand_vecs_2d[i][0], 8), round(cand_vecs_2d[i][1], 8)], | |
| "comparisons": per_substrate, | |
| "selection_rule": "highest_alignment", | |
| "selected_by_alignment": { | |
| "substrate_index": best_j, | |
| "substrate_text": substrate_list[best_j] if best_j is not None else None, | |
| "alignment_0_1": round(best_alignment, 8), | |
| "euclidean_distance": round(float(best_distance), 8), | |
| }, | |
| "engine": { | |
| "fractured": cand_metrics.get('fractured', False), | |
| "fractured_step": cand_metrics.get('fractured_step'), | |
| "final_stress": round(float(cand_metrics['stress']), 8), | |
| "hash": cand_metrics.get('hash', 'N/A'), | |
| "stress_evolution": stress_evolution | |
| } | |
| }) | |
| # Display Result | |
| st.markdown("### Result") | |
| if result_text: | |
| st.success(f"**Selected:** {result_text}") | |
| else: | |
| st.warning("**Abstained**: No candidates met the yield strength requirements.") | |
| # Visualization of the Field | |
| st.markdown("### Material Field Projection") | |
| # Plot | |
| fig_map = go.Figure() | |
| # Plot Substrate (Green Squares) | |
| fig_map.add_trace(go.Scatter( | |
| x=[v[0] for v in sub_vecs_2d], | |
| y=[v[1] for v in sub_vecs_2d], | |
| mode='markers', | |
| name='Substrate (Facts)', | |
| text=substrate_list, | |
| marker=dict(symbol='square', size=12, color='green') | |
| )) | |
| # Plot Candidates (Red Circles) | |
| # Differentiate selected vs excluded | |
| selected_idx = metrics['final_output'].candidate_index if metrics['final_output'] else -1 | |
| colors = ['gold' if i == selected_idx else 'red' for i in range(len(cand_vecs))] | |
| sizes = [15 if i == selected_idx else 10 for i in range(len(cand_vecs))] | |
| fig_map.add_trace(go.Scatter( | |
| x=[v[0] for v in cand_vecs_2d], | |
| y=[v[1] for v in cand_vecs_2d], | |
| mode='markers+text', | |
| name='Candidates', | |
| text=candidate_list, | |
| textposition='top center', | |
| marker=dict(symbol='circle', size=sizes, color=colors) | |
| )) | |
| fig_map.update_layout( | |
| title="Semantic Material Field (2D Mock Projection)", | |
| xaxis_title="Dimension X", | |
| yaxis_title="Dimension Y", | |
| xaxis=dict(range=[0, 1]), | |
| yaxis=dict(range=[0, 1]), | |
| template="plotly_dark", | |
| height=500 | |
| ) | |
| st.plotly_chart(fig_map, use_container_width=True) | |
| # Metrics JSON | |
| st.markdown("### Metrics") | |
| metrics_json = { | |
| "survived": result_text is not None, | |
| "total_excluded": metrics['total_excluded'], | |
| "falsification_pressure": f"{metrics['phase_log'][-1]['pressure']:.2f} lambda" | |
| } | |
| st.json(metrics_json) | |
| # Complete Numerical Audit Trail | |
| st.markdown("### Complete Numerical Audit Trail") | |
| st.caption("Vectors, distances, selection rule, engine hash, stress evolution") | |
| st.json(numbers_view) | |
| # ---------------------------------------------------------------------------- | |
| # TAB 3: LLM Testing (HF Spaces Enhanced) | |
| # ---------------------------------------------------------------------------- | |
| def call_huggingface_inference(model, prompt, api_key, temperature=0.7, max_tokens=256): | |
| """Call HuggingFace Inference API directly""" | |
| import requests | |
| API_URL = f"https://api-inference.huggingface.co/models/{model}" | |
| headers = {"Authorization": f"Bearer {api_key}"} | |
| payload = { | |
| "inputs": prompt, | |
| "parameters": { | |
| "temperature": temperature, | |
| "max_new_tokens": max_tokens, | |
| "return_full_text": False | |
| } | |
| } | |
| response = requests.post(API_URL, headers=headers, json=payload, timeout=30) | |
| response.raise_for_status() | |
| result = response.json() | |
| if isinstance(result, list) and len(result) > 0: | |
| return result[0].get("generated_text", "") | |
| return result.get("generated_text", "") | |
| # ---------------------------------------------------------------------------- | |
| # TAB 3: LLM Testing (HF Spaces Enhanced) | |
| # ---------------------------------------------------------------------------- | |
| with tab3: | |
| st.header("LLM Testing") | |
| st.markdown(""" | |
| **Live API Testing** - Test any LLM with the Deterministic Guardrail in real-time. | |
| Supports: OpenAI, Anthropic, Google Gemini, local models (Ollama, llama.cpp, vLLM), and any OpenAI-compatible API. | |
| """) | |
| # HF Spaces detection | |
| IS_SPACES = os.getenv("SPACE_ID") is not None | |
| # Show banner if in Spaces | |
| if IS_SPACES: | |
| st.info("🚀 **Running on Hugging Face Spaces** - Configure API keys in Settings → Secrets (for admins) or enter below") | |
| # API Configuration | |
| st.subheader("1. LLM API Configuration") | |
| col_api1, col_api2 = st.columns(2) | |
| with col_api1: | |
| # Add Hugging Face Inference API option for Spaces | |
| provider_options = [ | |
| "Hugging Face Inference API (Free)", # NEW - great for Spaces demos | |
| "OpenAI", | |
| "Anthropic (Claude)", | |
| "Google (Gemini)", | |
| "Local (Ollama)", | |
| "Local (llama.cpp)", | |
| "Custom OpenAI-compatible" | |
| ] | |
| # Default to HF Inference if in Spaces | |
| default_index = 0 if IS_SPACES else 1 | |
| api_preset = st.selectbox( | |
| "Provider Preset", | |
| options=provider_options, | |
| index=default_index, | |
| help="Select a preset or use custom for any OpenAI-compatible endpoint" | |
| ) | |
| # Set defaults based on preset | |
| if api_preset == "Hugging Face Inference API (Free)": | |
| default_base_url = "https://api-inference.huggingface.co/models" | |
| default_model = "meta-llama/Llama-3.2-3B-Instruct" # Free tier model | |
| needs_key = True | |
| api_type = "huggingface" | |
| elif api_preset == "OpenAI": | |
| default_base_url = "https://api.openai.com/v1" | |
| default_model = "gpt-4o-mini" # Updated to current model | |
| needs_key = True | |
| api_type = "openai" | |
| elif api_preset == "Anthropic (Claude)": | |
| default_base_url = "https://api.anthropic.com/v1" | |
| default_model = "claude-3-5-sonnet-20241022" | |
| needs_key = True | |
| api_type = "anthropic" | |
| elif api_preset == "Google (Gemini)": | |
| default_base_url = "https://generativelanguage.googleapis.com/v1beta" | |
| default_model = "gemini-2.0-flash-exp" | |
| needs_key = True | |
| api_type = "google" | |
| elif api_preset == "Local (Ollama)": | |
| default_base_url = "http://localhost:11434/v1" | |
| default_model = "llama3.1" | |
| needs_key = False | |
| api_type = "openai" | |
| elif api_preset == "Local (llama.cpp)": | |
| default_base_url = "http://localhost:8080/v1" | |
| default_model = "local-model" | |
| needs_key = False | |
| api_type = "openai" | |
| else: | |
| default_base_url = "https://api.openai.com/v1" | |
| default_model = "gpt-4o-mini" | |
| needs_key = True | |
| api_type = "openai" | |
| # Disable local options if in Spaces | |
| if IS_SPACES and "Local" in api_preset: | |
| st.warning("⚠️ Local models not available in Spaces. Use cloud APIs or HF Inference API.") | |
| api_base_url = st.text_input( | |
| "Base URL", | |
| value=default_base_url, | |
| help="API endpoint base URL", | |
| disabled=IS_SPACES and "Local" in api_preset | |
| ) | |
| with col_api2: | |
| # Check for API key in environment (HF Secrets) | |
| env_key = None | |
| if IS_SPACES: | |
| if api_preset == "OpenAI": | |
| env_key = os.getenv("OPENAI_API_KEY") | |
| elif api_preset == "Anthropic (Claude)": | |
| env_key = os.getenv("ANTHROPIC_API_KEY") | |
| elif api_preset == "Google (Gemini)": | |
| env_key = os.getenv("GOOGLE_API_KEY") | |
| elif api_preset == "Hugging Face Inference API (Free)": | |
| env_key = os.getenv("HF_TOKEN") | |
| if env_key: | |
| st.success(f"✓ Using API key from Space secrets") | |
| api_key = env_key | |
| show_input = False | |
| else: | |
| show_input = True | |
| if show_input: | |
| api_key = st.text_input( | |
| "API Key" + (" (optional for local)" if not needs_key else ""), | |
| type="password", | |
| help="Your API key (not required for local models)", | |
| placeholder="sk-..." if needs_key else "not needed for local models" | |
| ) | |
| model_name = st.text_input( | |
| "Model Name", | |
| value=default_model, | |
| help="Model identifier" | |
| ) | |
| col_temp, col_num = st.columns(2) | |
| with col_temp: | |
| temperature = st.slider("Temperature", 0.0, 2.0, 0.7, 0.1, help="Higher = more creative") | |
| with col_num: | |
| # Limit responses in Spaces for performance | |
| max_responses = 5 if IS_SPACES else 10 | |
| default_responses = 2 if IS_SPACES else 3 | |
| num_responses = st.number_input( | |
| "Number of Responses", | |
| min_value=1, | |
| max_value=max_responses, | |
| value=default_responses, | |
| help=f"Generate multiple responses for comparison{' (limited in Spaces)' if IS_SPACES else ''}" | |
| ) | |
| col_timeout, col_retry = st.columns(2) | |
| with col_timeout: | |
| # Shorter timeout for Spaces | |
| default_timeout = 30 if IS_SPACES else 60 | |
| request_timeout = st.number_input( | |
| "Request Timeout (seconds)", | |
| min_value=5, | |
| max_value=600, | |
| value=default_timeout, | |
| step=5, | |
| help="Increase for slow or local models" | |
| ) | |
| with col_retry: | |
| max_retries = st.number_input( | |
| "Max Retries", | |
| min_value=0, | |
| max_value=5, | |
| value=2, | |
| step=1, | |
| help="Automatic retries on transient failures" | |
| ) | |
| # Substrate Configuration | |
| st.subheader("2. Verified Substrate (Ground Truth)") | |
| st.markdown("Enter verified facts that define what is correct. One per line.") | |
| substrate_input_llm = st.text_area( | |
| "Substrate Facts", | |
| value="The Eiffel Tower is in Paris\nWater boils at 100°C at sea level\nPython is a programming language", | |
| height=120, | |
| key="llm_test_substrate" | |
| ) | |
| # Prompt Configuration | |
| st.subheader("3. Prompt Configuration") | |
| col_prompt1, col_prompt2 = st.columns([3, 1]) | |
| with col_prompt1: | |
| user_prompt = st.text_area( | |
| "User Prompt", | |
| value="Tell me a fact about one of the topics mentioned above.", | |
| height=100, | |
| help="The prompt sent to the LLM" | |
| ) | |
| with col_prompt2: | |
| use_system_prompt = st.checkbox("Use System Prompt", value=False) | |
| if use_system_prompt: | |
| system_prompt = st.text_area( | |
| "System Prompt (optional)", | |
| value="You are a helpful assistant. Provide accurate, factual information.", | |
| height=100 | |
| ) | |
| else: | |
| system_prompt = None | |
| # Governance Configuration (Control Surface) | |
| st.subheader("4. Governance Controls") | |
| st.markdown("Adjust the physics strictness to see how the system responds to ambiguity vs. facts.") | |
| gov_col1, gov_col2 = st.columns(2) | |
| with gov_col1: | |
| gov_preset = st.selectbox( | |
| "Governance Preset", | |
| ["Forgiving", "Balanced", "Conservative", "Aggressive", "Mission Critical"], | |
| index=1, | |
| help="Sets physics parameters (Lambda/Sigma). 'Forgiving' tolerates ambiguity; 'Mission Critical' demands exact alignment." | |
| ) | |
| preset_map = { | |
| "Balanced": "balanced", | |
| "Conservative": "conservative", | |
| "Aggressive": "aggressive", | |
| "Mission Critical": "mission_critical", | |
| "Forgiving": "forgiving" | |
| } | |
| selected_gov_preset = preset_map[gov_preset] | |
| # Educational Display: Show the actual numbers | |
| from material_field_engine import load_config | |
| config_data = load_config(selected_gov_preset) | |
| st.markdown("---") | |
| st.markdown("**Physics Parameters (active settings):**") | |
| p_col1, p_col2, p_col3 = st.columns(3) | |
| with p_col1: | |
| st.metric( | |
| label="Sigma (σ) - Tolerance", | |
| value=config_data['elastic_modulus_sigma'], | |
| help="Field Extent. Higher (0.8) = Vague associations accepted. Lower (0.2) = Exact match required." | |
| ) | |
| with p_col2: | |
| st.metric( | |
| label="Lambda Max (λ) - Pressure", | |
| value=config_data['lambda_max'], | |
| help="Max Exclusion Pressure. Higher (1.5+) = Crushes weak bonds (High strictness). Lower (0.5) = Gentle." | |
| ) | |
| with p_col3: | |
| st.markdown(f"**Mode**: `{config_data['elastic_modulus_mode']}`") | |
| st.caption("Algorithm usually 'multiplicative' (Angle × Distance).") | |
| st.info(f"💡 **Teacher's Note**: To prevent valid answers from being blocked (fracturing), you would **increase Sigma** (widen the net) or **decrease Lambda** (reduce the pressure).") | |
| with gov_col2: | |
| st.write("**Gate Settings**") | |
| topic_gate_enabled = st.checkbox( | |
| "Enable Topic Gate", | |
| value=False, | |
| help="Fast pre-filter. If unchecked, physics runs on everything (good for demos)." | |
| ) | |
| ambiguity_detection = st.checkbox( | |
| "Allow Clarifications", | |
| value=True, | |
| help="If ON, clarifying questions ('Could you specify?') are marked CLARIFY instead of failing." | |
| ) | |
| # Run Button | |
| if st.button("🚀 Generate & Test Responses", type="primary"): | |
| # Parse substrate | |
| substrate_list = [line.strip() for line in substrate_input_llm.split('\n') if line.strip()] | |
| if not substrate_list: | |
| st.error("Please provide substrate facts") | |
| elif not user_prompt: | |
| st.error("Please provide a user prompt") | |
| elif not api_base_url.strip(): | |
| st.error("Please provide a base URL for the API") | |
| elif not model_name.strip(): | |
| st.error("Please provide a model name") | |
| else: | |
| st.info(f"📡 Generating {num_responses} response(s) from {api_preset}...") | |
| from urllib.parse import urlparse | |
| import socket | |
| parsed_url = urlparse(api_base_url) | |
| host = parsed_url.hostname | |
| port = parsed_url.port or (443 if parsed_url.scheme == "https" else 80) | |
| if not host: | |
| st.error("Invalid base URL. Please include scheme (http/https) and host.") | |
| st.stop() | |
| try: | |
| with socket.create_connection((host, port), timeout=3): | |
| pass | |
| except OSError as exc: | |
| st.error(f"Unable to connect to {api_base_url}: {exc}") | |
| if "localhost" in api_base_url or "127.0.0.1" in api_base_url: | |
| st.info("For local providers, ensure the server is running (e.g., `ollama serve`).") | |
| st.stop() | |
| try: | |
| import openai | |
| # Configure client | |
| api_key_value = api_key.strip() if api_key else "" | |
| if not needs_key and not api_key_value: | |
| api_key_value = "local" | |
| client = openai.OpenAI( | |
| api_key=api_key_value, | |
| base_url=api_base_url, | |
| timeout=request_timeout, | |
| max_retries=max_retries | |
| ) | |
| responses = [] | |
| with st.spinner(f"Calling LLM API ({num_responses} request(s))..."): | |
| for i in range(num_responses): | |
| messages = [] | |
| if system_prompt: | |
| messages.append({"role": "system", "content": system_prompt}) | |
| messages.append({"role": "user", "content": user_prompt}) | |
| try: | |
| response = client.chat.completions.create( | |
| model=model_name, | |
| messages=messages, | |
| temperature=temperature | |
| ) | |
| responses.append(response.choices[0].message.content) | |
| except openai.APITimeoutError as e: | |
| st.error(f"Request timed out on response {i+1}. Try increasing the timeout.") | |
| raise | |
| except openai.APIConnectionError as e: | |
| st.error(f"Connection error on response {i+1}. Check base URL and server status.") | |
| raise | |
| except openai.OpenAIError as e: | |
| st.error(f"OpenAI API error on response {i+1}: {str(e)}") | |
| raise | |
| except Exception as e: | |
| st.error(f"Error on response {i+1}: {str(e)}") | |
| if i == 0: # If first call fails, stop | |
| raise | |
| if not responses: | |
| st.error("No responses generated") | |
| else: | |
| st.success(f"✓ Generated {len(responses)} response(s)") | |
| # Now run the guardrail | |
| st.markdown("---") | |
| with st.spinner("Running Deterministic Guardrail..."): | |
| from llm_adapter import DeterministicGuardrail, DeterministicHashEmbedderND | |
| import math | |
| guard = DeterministicGuardrail( | |
| substrate_texts=substrate_list, | |
| config_preset=selected_gov_preset, | |
| topic_gate_enabled=topic_gate_enabled, | |
| ambiguity_detection_enabled=ambiguity_detection | |
| ) | |
| inspection = guard.inspect(responses) | |
| result_text = inspection['selected_text'] | |
| metrics = inspection['metrics'] | |
| candidate_metrics = metrics.get('candidates') | |
| if candidate_metrics is None: | |
| candidate_metrics = [ | |
| { | |
| 'phase_log': [], | |
| 'fractured': False, | |
| 'fractured_step': None, | |
| 'stress': 0.0, | |
| 'hash': 'N/A', | |
| } | |
| for _ in responses | |
| ] | |
| # Build detailed numbers view (high-D physics + 2D projection for plot) | |
| embedder = DeterministicHashEmbedderND() | |
| sub_vecs = [embedder.embed(t) for t in substrate_list] | |
| resp_vecs = [embedder.embed(t) for t in responses] | |
| sub_vecs_2d = [embedder.project_2d(v) for v in sub_vecs] | |
| resp_vecs_2d = [embedder.project_2d(v) for v in resp_vecs] | |
| def cosine_similarity(v1, v2): | |
| dot = sum(a * b for a, b in zip(v1, v2)) | |
| mag1 = math.sqrt(sum(a * a for a in v1)) | |
| mag2 = math.sqrt(sum(b * b for b in v2)) | |
| if mag1 == 0 or mag2 == 0: | |
| return 0.0 | |
| return dot / (mag1 * mag2) | |
| def euclidean_distance(v1, v2): | |
| return math.sqrt(sum((a - b) ** 2 for a, b in zip(v1, v2))) | |
| # Display result | |
| st.subheader("Guardrail Decision") | |
| if result_text: | |
| st.success("🟢 **SELECTED RESPONSE**") | |
| st.markdown(f"> {result_text}") | |
| else: | |
| st.warning("🔴 **ABSTAINED** - All responses fractured under stress") | |
| st.caption("The guardrail rejected all responses. None met the yield strength requirements.") | |
| # Show all responses with scores | |
| st.markdown("---") | |
| st.subheader("Detailed Scoring for All Responses") | |
| for i, (response, cand_vec) in enumerate(zip(responses, resp_vecs)): | |
| cand_metrics = candidate_metrics[i] | |
| is_selected = (result_text == response) | |
| # Compute alignment to best substrate | |
| best_alignment = -1 | |
| best_substrate = None | |
| best_cos_sim = 0 | |
| for j, sub_vec in enumerate(sub_vecs): | |
| cos_sim = cosine_similarity(cand_vec, sub_vec) | |
| alignment = (cos_sim + 1.0) / 2.0 | |
| if alignment > best_alignment: | |
| best_alignment = alignment | |
| best_substrate = substrate_list[j] | |
| best_cos_sim = cos_sim | |
| # Display card | |
| fractured = cand_metrics.get('fractured', False) | |
| out_of_domain = cand_metrics.get('out_of_domain', False) | |
| if is_selected: | |
| status = "SELECTED" | |
| elif out_of_domain: | |
| status = "EXCLUDED (Topic Gate)" | |
| elif fractured: | |
| status = "EXCLUDED (Fractured)" | |
| else: | |
| status = "SURVIVED (Not Selected)" | |
| with st.expander(f"**Response {i+1}** - {status}", expanded=is_selected): | |
| st.markdown(f"**Full Response:**") | |
| st.info(response) | |
| st.markdown("---") | |
| # Metrics | |
| col1, col2, col3, col4 = st.columns(4) | |
| with col1: | |
| st.metric("Alignment Score", f"{best_alignment:.4f}", help="Normalized cosine similarity: 0=opposite, 0.5=orthogonal, 1=identical") | |
| with col2: | |
| st.metric("Cosine Similarity", f"{best_cos_sim:.4f}", help="Raw cosine similarity: -1 to 1") | |
| with col3: | |
| st.metric("Final Stress σ", f"{cand_metrics['stress']:.4f}") | |
| with col4: | |
| if out_of_domain: | |
| st.metric("Status", "Topic-gated") | |
| else: | |
| st.metric("Status", "Intact" if not fractured else "Fractured") | |
| st.caption(f"**Best substrate match:** *\"{best_substrate}\"*") | |
| # Show stress evolution chart | |
| st.markdown("**Stress Evolution**") | |
| phase_log = cand_metrics.get('phase_log', []) | |
| stress_data = [entry['stress'] for entry in phase_log] | |
| steps = list(range(len(stress_data))) | |
| fig_stress = go.Figure() | |
| fig_stress.add_trace(go.Scatter( | |
| x=steps, | |
| y=stress_data, | |
| mode='lines+markers', | |
| name='Cumulative Stress', | |
| line=dict(color='red' if fractured else 'green', width=2), | |
| marker=dict(size=6) | |
| )) | |
| fig_stress.update_layout( | |
| title=f"Stress Accumulation - Response {i+1}", | |
| yaxis_title="Cumulative Stress σ(k)", | |
| xaxis_title="Inference Step k", | |
| height=300, | |
| template="plotly_dark", | |
| showlegend=True | |
| ) | |
| st.plotly_chart(fig_stress, use_container_width=True) | |
| # Summary metrics | |
| st.markdown("---") | |
| st.subheader("Summary Statistics") | |
| summary_cols = st.columns(4) | |
| with summary_cols[0]: | |
| st.metric("Total Responses", len(responses)) | |
| with summary_cols[1]: | |
| st.metric("Excluded", metrics['total_excluded']) | |
| with summary_cols[2]: | |
| st.metric("Survived", len(responses) - metrics['total_excluded']) | |
| with summary_cols[3]: | |
| selected = 1 if result_text else 0 | |
| st.metric("Selected", selected) | |
| # Detailed audit trail | |
| with st.expander("📊 Complete Numerical Audit Trail (JSON)", expanded=False): | |
| st.caption("Full vectors, distances, comparisons, and stress evolution for reproducibility") | |
| numbers_view = { | |
| "test_metadata": { | |
| "provider": api_preset, | |
| "base_url": api_base_url, | |
| "model": model_name, | |
| "temperature": temperature, | |
| "num_responses": len(responses), | |
| "num_substrate_facts": len(substrate_list), | |
| "prompt": user_prompt, | |
| "system_prompt": system_prompt if system_prompt else "None" | |
| }, | |
| "embedder": { | |
| "name": "DeterministicHashEmbedderND", | |
| "description": "Deterministic SHA-256 based 16D projection (2D shown for plotting)", | |
| "definition": "sha256(text) -> 16D in [0,1], projected to 2D" | |
| }, | |
| "substrate": [ | |
| {"index": idx, "text": t, "vec2": [round(v[0], 8), round(v[1], 8)]} | |
| for idx, (t, v) in enumerate(zip(substrate_list, sub_vecs_2d)) | |
| ], | |
| "responses": [] | |
| } | |
| for i, (response, resp_vec) in enumerate(zip(responses, resp_vecs)): | |
| cand_metrics = candidate_metrics[i] | |
| # Compute all substrate comparisons | |
| comparisons = [] | |
| for j, sub_vec in enumerate(sub_vecs): | |
| cos_sim = cosine_similarity(resp_vec, sub_vec) | |
| alignment = (cos_sim + 1.0) / 2.0 | |
| dist = euclidean_distance(resp_vec, sub_vec) | |
| comparisons.append({ | |
| "substrate_index": j, | |
| "substrate_text": substrate_list[j], | |
| "cosine_similarity": round(cos_sim, 8), | |
| "alignment_0_1": round(alignment, 8), | |
| "euclidean_distance": round(dist, 8), | |
| }) | |
| numbers_view["responses"].append({ | |
| "response_index": i, | |
| "text": response, | |
| "vec2": [round(resp_vecs_2d[i][0], 8), round(resp_vecs_2d[i][1], 8)], | |
| "substrate_comparisons": comparisons, | |
| "engine_results": { | |
| "fractured": cand_metrics.get('fractured', False), | |
| "fractured_at_step": cand_metrics.get('fractured_step'), | |
| "final_stress": round(float(cand_metrics['stress']), 8), | |
| "determinism_hash": cand_metrics.get('hash', 'N/A') | |
| } | |
| }) | |
| st.json(numbers_view) | |
| except ImportError: | |
| st.error("Missing `openai` library. Install with: `pip install openai`") | |
| except Exception as e: | |
| st.error(f"Error: {str(e)}") | |
| if "401" in str(e) or "authentication" in str(e).lower(): | |
| st.info("💡 Check your API key and make sure it's valid for the selected endpoint") | |
| elif "404" in str(e) or "not found" in str(e).lower(): | |
| st.info("💡 Check your model name and base URL. For local models, make sure the server is running.") | |
| st.exception(e) | |
| # ---------------------------------------------------------------------------- | |
| # TAB 4: Explain & Tune | |
| # ---------------------------------------------------------------------------- | |
| with tab4: | |
| st.header("🔧 Interactive Parameter Tuning") | |
| st.markdown("Select parameters to visualize their combined effect on the governance physics.") | |
| # 1. State Tracking for "Active Explanation" | |
| if 'last_params' not in st.session_state: | |
| st.session_state.last_params = { | |
| 'nuc': 0.4, 'quench': 0.75, 'lam': 1.2, 'yield': 1.5, 'align': 0.85, 'dist': 0.3 | |
| } | |
| if 'active_topic' not in st.session_state: | |
| st.session_state.active_topic = "General" | |
| # 1. Controls | |
| exp_col1, exp_col2 = st.columns([1, 2]) | |
| with exp_col1: | |
| st.subheader("Controls") | |
| st.caption("⚠️ **Educational Visualization**: This simulation uses the production physics engine but is intended for parameter intuition.") | |
| # Multi-select for visualization layers | |
| focused_params = st.multiselect( | |
| "Visualized Layers", | |
| options=["Nucleation Phase", "Quenching Phase", "Max Pressure (lambda)", "Yield Strength (sigma_y)"], | |
| default=["Max Pressure (lambda)"], | |
| help="Select multiple layers to see how they interact" | |
| ) | |
| st.markdown("---") | |
| st.markdown("**Physics Parameters**") | |
| # Sliders with callbacks to update active topic | |
| e_nuc = st.slider("Nucleation Fraction", 0.05, 0.9, 0.4, 0.05, key="slider_nuc") | |
| e_quench = st.slider("Quenching Fraction", 0.05, 0.95, 0.75, 0.05, key="slider_quench") | |
| e_lam = st.slider("Lambda Max (lambda)", 0.1, 4.0, 1.2, 0.1, key="slider_lam") | |
| e_yield = st.slider("Yield Strength (sigma_y)", 0.1, 5.0, 1.5, 0.1, key="slider_yield") | |
| st.markdown("**Theoretical Simulation**") | |
| sim_align = st.slider("Target Alignment", 0.0, 1.0, 0.85, 0.01, key="slider_align") | |
| sim_dist = st.slider("Target Distance", 0.0, 2.0, 0.3, 0.01, key="slider_dist") | |
| # ... (Detection logic remains same) | |
| current_params = { | |
| 'nuc': e_nuc, 'quench': e_quench, 'lam': e_lam, | |
| 'yield': e_yield, 'align': sim_align, 'dist': sim_dist | |
| } | |
| for param, val in current_params.items(): | |
| if val != st.session_state.last_params.get(param): | |
| st.session_state.active_topic = param | |
| st.session_state.last_params[param] = val | |
| break | |
| # 2. Simulation Logic (Production Backend) | |
| def run_simulation(steps, nuc, quench, lam_max, yld, align, dist): | |
| from material_field_engine import MaterialFieldEngine, VerifiedSubstrate, Vector2D, fp_from_float | |
| # To strictly follow the "one source of truth" principle, we drive the | |
| # visualization from the exact same engine code used in production. | |
| substrate = VerifiedSubstrate( | |
| elastic_modulus_mode=st.session_state.config.get('elastic_modulus_mode', 'multiplicative'), | |
| elastic_modulus_sigma=st.session_state.config.get('elastic_modulus_sigma', 0.5) | |
| ) | |
| # Add a reference point for alignment calculations | |
| substrate.add_verified_state(Vector2D(x=1.0, y=1.0, properties=None)) | |
| engine = MaterialFieldEngine( | |
| substrate, | |
| lambda_max=lam_max, | |
| inference_steps=steps | |
| ) | |
| engine.phase_controller.nucleation_threshold = nuc | |
| engine.phase_controller.quenching_threshold = quench | |
| # Initialize candidate at distance 'dist' from the reference | |
| engine.initialize_candidates([[1.0 + dist, 1.0]]) | |
| # Force yield strength to match slider for educational clarity | |
| if engine.candidate_vectors: | |
| engine.candidate_vectors[0].properties.yield_strength_q = fp_from_float(yld) | |
| res = engine.run_inference(collect_trace=True) | |
| history = [] | |
| if res.get('candidates'): | |
| cand_trace = res['candidates'][0] | |
| for entry in cand_trace['phase_log']: | |
| history.append({ | |
| "step": entry['step'], | |
| "stress": entry['stress'], | |
| "lambda": entry['pressure'], | |
| "phase": entry['phase'] | |
| }) | |
| n_end = int(steps * nuc) | |
| q_end = int(steps * quench) | |
| return history, n_end, q_end | |
| sim_data, sim_n_end, sim_q_end = run_simulation(20, e_nuc, e_quench, e_lam, e_yield, sim_align, sim_dist) | |
| # 3. Visualization (Plotly) | |
| with exp_col2: | |
| st.subheader("Effect Visualization") | |
| fig = go.Figure() | |
| steps = [d['step'] for d in sim_data] | |
| stress = [d['stress'] for d in sim_data] | |
| lams = [d['lambda'] for d in sim_data] | |
| # Base Curves | |
| fig.add_trace(go.Scatter(x=steps, y=stress, name='Stress σ', line=dict(color='blue', width=3))) | |
| fig.add_trace(go.Scatter(x=steps, y=lams, name='Pressure λ', line=dict(color='green', width=2, dash='dash'))) | |
| # Interactive Layers | |
| if "Nucleation Phase" in focused_params or st.session_state.active_topic == 'nuc': | |
| fig.add_vrect(x0=0, x1=sim_n_end, fillcolor="yellow", opacity=0.15, annotation_text="Nucleation", annotation_position="top left") | |
| if "Quenching Phase" in focused_params or st.session_state.active_topic == 'quench': | |
| fig.add_vrect(x0=sim_n_end, x1=sim_q_end, fillcolor="orange", opacity=0.15, annotation_text="Quenching", annotation_position="top left") | |
| if "Max Pressure (lambda)" in focused_params or st.session_state.active_topic == 'lam': | |
| fig.add_hline(y=e_lam, line_color="green", line_dash="dot", annotation_text=f"Max Pressure {e_lam}", annotation_position="bottom right") | |
| if "Yield Strength (sigma_y)" in focused_params or st.session_state.active_topic == 'yield': | |
| fig.add_hline(y=e_yield, line_color="red", line_width=3, annotation_text=f"Yield {e_yield}") | |
| fig.update_layout( | |
| title="Governance Physics Simulation", | |
| xaxis_title="Time Steps", | |
| yaxis_title="Magnitude", | |
| height=400, | |
| margin=dict(l=20, r=20, t=40, b=20), | |
| hovermode="x unified" | |
| ) | |
| st.plotly_chart(fig, use_container_width=True) | |
| # 4. Deep Dive Explanation (Dynamic) | |
| st.subheader(f"Deep Dive: {st.session_state.active_topic.upper()}") | |
| topic = st.session_state.active_topic | |
| if topic == 'nuc': | |
| st.info(""" | |
| **WHAT**: Nucleation Fraction (Time available for initial alignment). | |
| **HOW**: Defines the percentage of the timeline (steps 0 to N) where the system is "listening" before applying significant pressure. | |
| **WHY**: | |
| - **Too Short**: The system acts impulsively, fracturing valid ideas before they can stabilize. | |
| - **Too Long**: The system dithers, allowing hallucinations to persist too long. | |
| **WHO**: Tuned by governance architects to match the "patience" required for the domain (e.g., Creative writing needs long nucleation; Safety systems need short). | |
| """) | |
| elif topic == 'quench': | |
| st.info(""" | |
| **WHAT**: Quenching Fraction (The annealing window). | |
| **HOW**: Defines the period where pressure ramps up linearly to testing levels. | |
| **WHY**: This is the "soft filter" phase. Weak candidates (low alignment) are slowly crushed here, while strong candidates gain strength to survive the final crystallization. | |
| """) | |
| elif topic == 'lam': | |
| st.info(""" | |
| **WHAT**: Lambda Max (λ_max) - The Maximum Exclusion Pressure. | |
| **HOW**: Represents the "weight" of the governance mechanism. It is a multiplier on the error signal. | |
| **WHY**: | |
| - **High (1.5+)**: "Mission Critical" mode. Even minor deviations cause instant fracture. | |
| - **Low (0.5)**: "Forgiving" mode. Only egregious hallucinations are blocked. | |
| **RELATION**: Stress = λ * (1 - Alignment). If λ is huge, even 99% alignment might not be enough. | |
| """) | |
| elif topic == 'yield': | |
| st.info(""" | |
| **WHAT**: Yield Strength (σ_y) - The Breaking Point. | |
| **HOW**: A hard threshold. If accumulated Stress > Yield, the candidate is Rejected (Fractured). | |
| **WHY**: This defines the ultimate binary decision boundary. | |
| **IMPACT**: Raising this bar makes the system more "resilient" (harder to fracture). Lowering it makes it "brittle" (easy to fracture). | |
| """) | |
| elif topic in ['align', 'dist']: | |
| st.info(""" | |
| **WHAT**: Candidate Properties (Hypothetical Input). | |
| **HOW**: | |
| - **Alignment**: How semantically close the LLM output is to the Verified Substrate (1.0 = Perfect). | |
| - **Distance**: The spatial distance in the high-dimensional RBF field (0.0 = Perfect). | |
| **WHY**: Use these sliders to test *what if* scenarios. "What if the LLM produces a weak answer (Align=0.4)? Will it survive the current Lambda setting?" | |
| """) | |
| else: | |
| st.markdown("*Adjust any slider on the left to see a detailed breakdown of its function.*") | |
| # ============================================================================ | |
| # Compact Footer | |
| # ============================================================================ | |
| st.markdown("---") | |
| st.caption("© 2026 Verhash LLC | Deterministic Governance Reference Implementation") | |