'
f'🔒 Privacy Protection Applied
'
f''
f'{n_sub} word{"s" if n_sub != 1 else ""} in your message '
f'{"were" if n_sub != 1 else "was"} automatically replaced by similar words '
f'before the AI read it. ' # (ε = {eps_str}). '
f'Your original message is shown in the chat above.'
f'
'
f''
f''
f'What the AI actually received:
'
f''
f'{sentence_html}'
f'
'
f''
f'red = original | '
f'green = replacement'
f'
'
)
# ── Badge pill + tooltip wrapper ──────────────────────────────────────
badge_html = (
f''
f''
f'🔒 DP-protected · ' # ε={eps_str} ·
f'{n_sub} word{"s" if n_sub!=1 else ""} changed'
f''
f''
f'{tooltip_content}'
f''
f''
)
return badge_html
# ============================================================
# SECTION 6 – CORPUS LOADING (GENERIC)
# ============================================================
_GLOBAL_RETRIEVER = None
_DP_INPUT_PERTURBER = None # DPInputPerturber instance, initialised lazily
# Process-level cache: scenario_mode → loaded retriever object.
# Populated lazily on first request for each scenario.
_SCENARIO_RETRIEVERS = {}
_SCENARIO_RETRIEVERS_LOCK = threading.Lock()
def get_scenario_retriever(scenario_mode: str):
"""Return (and lazily load) the system retriever for the given scenario_mode.
Thread-safe. Falls back to the default RETRIEVAL_PICKLE_PATH if the
scenario has no dedicated entry or its file cannot be loaded.
"""
key = str(scenario_mode).strip().lower()
if key in _SCENARIO_RETRIEVERS:
return _SCENARIO_RETRIEVERS[key]
with _SCENARIO_RETRIEVERS_LOCK:
# Double-checked locking
if key in _SCENARIO_RETRIEVERS:
return _SCENARIO_RETRIEVERS[key]
path = SCENARIO_RETRIEVER_PATHS.get(key) or DEFAULT_RETRIEVAL_PICKLE_PATH
logger.info(f"🔄 Loading retriever for scenario '{key}' from: {path}")
if path is None:
logger.error(f" Retriever for scenario '{key}' is None and will be ignored!")
retriever = _GLOBAL_RETRIEVER
else:
try:
retriever = load_retriever_components(path)
logger.info(f" ✓ Retriever loaded for scenario '{key}'")
except Exception as e:
logger.error(f" ✗ Failed to load retriever for scenario '{key}': {e}")
# Fall back to whatever the global default retriever is
retriever = _GLOBAL_RETRIEVER
_SCENARIO_RETRIEVERS[key] = retriever
return retriever
def _init_dp_input_perturber(retriever):
"""Initialise the global DPInputPerturber, reusing the retriever embedding model.
LangChain FAISS stores the embedding function as vectorstore.embeddings.
Reusing it avoids loading a second copy of the sentence-transformer weights.
Falls back to creating a fresh HuggingFaceEmbeddings if that attribute is
absent (e.g. when the retriever was pickled without it).
"""
global _DP_INPUT_PERTURBER
if _DP_INPUT_PERTURBER is not None:
return # already initialised
embed_fn = None
try:
embed_fn = retriever.vectorstore.embeddings
logger.info("DPInputPerturber: reusing vectorstore embedding model.")
except AttributeError:
pass
if embed_fn is None:
try:
embed_fn = HuggingFaceEmbeddings(
model_name=EMBEDDING_MODEL,
model_kwargs={"device": "cpu"},
encode_kwargs={"normalize_embeddings": True},
)
logger.info("DPInputPerturber: created standalone HuggingFaceEmbeddings.")
except Exception as exc:
logger.error("DPInputPerturber: could not create embedding model: %s", exc)
return
_DP_INPUT_PERTURBER = DPInputPerturber(epsilon=1.0, embedding_model=embed_fn) # epsilon is placeholder; overridden per-request
logger.info("DPInputPerturber: initialised and ready.")
def load_retriever(pickle_path=None):
"""Load retriever from component-based pickle (FAISS-compatible)."""
global _GLOBAL_RETRIEVER
path = pickle_path or DEFAULT_RETRIEVAL_PICKLE_PATH
if path is None:
_init_dp_input_perturber(None)
return None
try:
_GLOBAL_RETRIEVER = load_retriever_components(path)
logger.info(f"✓ Retriever loaded from {path}")
_init_dp_input_perturber(_GLOBAL_RETRIEVER) # initialise Input DP perturber
return _GLOBAL_RETRIEVER
except Exception as e:
logger.error(f"✗ Failed to load retriever from {path}: {e}")
logger.info(" System will work without RAG.")
# Still initialise the Input-DP perturber using a standalone embedding
# model so word-level DP perturbation works even without a RAG retriever.
_init_dp_input_perturber(None)
return None
def get_retriever_lazy():
"""Lazy load retriever on first use."""
global _GLOBAL_RETRIEVER
if _GLOBAL_RETRIEVER is None:
logger.info("🔄 Loading retriever for first time...")
load_retriever()
logger.info("✓ Retriever loaded")
return _GLOBAL_RETRIEVER
def load_retriever_async():
"""Load retriever in background thread."""
thread = threading.Thread(target=load_retriever, daemon=True)
thread.start()
return thread
def get_retriever():
"""Get the global retriever instance."""
return _GLOBAL_RETRIEVER
def _set_system_retriever(new_retriever):
"""Set the global retriever instance and (re-)initialise Input DP perturber."""
global _GLOBAL_RETRIEVER, _DP_INPUT_PERTURBER
_GLOBAL_RETRIEVER = new_retriever
# Reset perturber so it is re-initialised with the new retriever's embedding model
_DP_INPUT_PERTURBER = None
if new_retriever is not None:
_init_dp_input_perturber(new_retriever)
# ============================================================
# SECTION 7 – DEMO MODE (ZERO API CALLS)
# ============================================================
def fabricate_demo_pii(user_text):
"""Fabricate PII detections for demo mode - NO LLM calls."""
# Simple heuristic: look for common name patterns
pii_list = []
# Look for "I am" or "my name is" patterns
name_patterns = [
(r"(?:I am|I'm|my name is|call me)\s+([A-Z][a-z]+(?:\s+[A-Z][a-z]+)?)", "name"),
(r"\b([A-Z][a-z]+\s+[A-Z][a-z]+)\b", "name"),
]
for pattern, pii_type in name_patterns:
for match in re.finditer(pattern, user_text, re.IGNORECASE):
pii_list.append({
"text": match.group(1),
"start": match.start(1),
"end": match.end(1),
"type": pii_type
})
break # Only first match
# If no name found, fabricate one from the text
if not pii_list:
words = user_text.split()
if len(words) >= 2:
# Use first two capitalized words as fake name
capitalized = [w for w in words if w and w[0].isupper() and len(w) > 2]
if len(capitalized) >= 2:
fake_name = f"{capitalized[0]} {capitalized[1]}"
idx = user_text.find(fake_name)
if idx != -1:
pii_list.append({
"text": fake_name,
"start": idx,
"end": idx + len(fake_name),
"type": "name"
})
return pii_list
def fabricate_demo_response(user_text):
"""Fabricate a response for demo mode - NO LLM calls."""
# Generate a generic but contextual response
keywords = ["job", "work", "career", "health", "medical", "hobby", "interest"]
context = "general"
for kw in keywords:
if kw in user_text.lower():
if kw in ["job", "work", "career"]:
context = "job"
elif kw in ["health", "medical"]:
context = "health"
elif kw in ["hobby", "interest"]:
context = "hobby"
break
responses = {
"job": "Based on your background and interests, here are some suggestions:\n\n1. Consider roles in the technology sector - your skills would be valuable there.\n2. Look into positions at medium-sized companies (50-200 employees) where you can make an impact.\n3. Check job boards like LinkedIn and Indeed for opportunities in your area.\n\nWould you like me to help you refine your search based on specific criteria?",
"health": "Here are some preventive health screenings to consider:\n\n1. Annual physical exam with bloodwork (cholesterol, glucose, etc.)\n2. Blood pressure monitoring\n3. Age-appropriate cancer screenings\n4. Dental check-up every 6 months\n\nFor affordable options, check with local community health centers or your insurance provider's network. Would you like more specific recommendations?",
"hobby": "Based on common interests, here are some hobbies you might enjoy:\n\n1. Outdoor activities like hiking or cycling\n2. Creative pursuits such as photography or painting\n3. Learning new skills through online courses\n4. Community involvement through volunteering\n\nWhat type of activities do you find most appealing?",
"general": "I'd be happy to help! To provide more personalized recommendations, could you tell me a bit more about:\n\n1. Your current interests or goals\n2. How much time you have available\n3. Any specific areas you'd like to explore\n\nThis will help me give you more relevant suggestions."
}
return responses.get(context, responses["general"])
def fabricate_demo_probs(user_text, has_rag):
"""Fabricate probability distributions for demo mode - NO LLM calls."""
# Generate probabilities based on text analysis
base_probs = {
"Gender": {"Male": 0.55, "Female": 0.45},
"Age bin": {"0-17": 0.05, "18-29": 0.25, "30-44": 0.45, "45-59": 0.20, "60+": 0.05},
"Marital Status": {"Single": 0.45, "Married": 0.40, "Divorced": 0.10, "Widowed": 0.05},
"Finance Status": {"Low": 0.25, "Medium": 0.50, "High": 0.25},
"Education": {"High School": 0.15, "Bachelor's": 0.50, "Master's": 0.25, "PhD": 0.10},
"Locale": {
"United States": 0.60, "Canada": 0.15, "United Kingdom": 0.10,
"Australia": 0.05, "India": 0.04, "Ireland": 0.02,
"New Zealand": 0.02, "Philippines": 0.02, "Israel": 0.01, "Germany": 0.01, "Italy":0.01,"France":0.01,
}
}
if has_rag:
# RAG adds confidence - shift probabilities more extremely
rag_probs = {
"Gender": {"Male": 0.65, "Female": 0.35},
"Age bin": {"0-17": 0.02, "18-29": 0.18, "30-44": 0.55, "45-59": 0.20, "60+": 0.05},
"Marital Status": {"Single": 0.55, "Married": 0.30, "Divorced": 0.10, "Widowed": 0.05},
"Finance Status": {"Low": 0.15, "Medium": 0.50, "High": 0.35},
"Education": {"High School": 0.08, "Bachelor's": 0.52, "Master's": 0.30, "PhD": 0.10},
"Locale": {
"United States": 0.70, "Canada": 0.08, "United Kingdom": 0.08,
"Australia": 0.04, "India": 0.02, "Ireland": 0.01,
"New Zealand": 0.02, "Philippines": 0.01, "Israel": 0.01, "Germany": 0.01, "Italy":0.01,"France":0.01,
}
}
return rag_probs, base_probs
else:
return base_probs, base_probs
def fabricate_demo_rag_docs(user_text):
"""Fabricate RAG documents for demo mode - NO retrieval."""
# Generate fake documents that seem relevant
templates = [
"User profile indicates experience in software development with 5 years in web technologies. Previous roles include positions at tech startups in the Bay Area.",
"Background shows interest in preventive healthcare and fitness. Regular physical activity noted, with preference for outdoor activities.",
"Career history includes work at medium-sized technology companies. Skills include project management and technical leadership.",
"Education background: Bachelor's degree in Computer Science. Additional certifications in cloud computing and data analysis.",
"Location history shows residence in major metropolitan areas. Current location indicates access to urban amenities and services."
]
# Select relevant templates based on keywords
selected = random.sample(templates, min(3, len(templates)))
return [(doc, 0.85 - i*0.05) for i, doc in enumerate(selected)]
def process_demo_message(user_input, show_tips, show_rag_hl, show_pii_hl, state, use_rag):
"""Process message in demo mode - ZERO API calls, all data fabricated."""
# Step 1: Fabricate PII in user input (NO LLM call)
user_pii_data = fabricate_demo_pii(user_input)
user_pii_matches = []
for pii_item in user_pii_data:
user_pii_matches.append(PIIMatch(
text=pii_item['text'],
start=pii_item['start'],
end=pii_item['end'],
fine_type=pii_item['type'],
category=PII_TYPE_TO_CATEGORY.get(pii_item['type'], PIICategory.IDENTITY),
confidence=0.9
))
# Step 2: Fabricate RAG docs (NO retrieval)
rag_docs = fabricate_demo_rag_docs(user_input) if use_rag else []
# Step 3: Fabricate RAG links in user input
user_rag_links = []
if use_rag and rag_docs:
query_words = user_input.lower().split()
for doc_text, score in rag_docs[:1]:
doc_words = set(doc_text.lower().split())
overlap = [w for w in query_words if w in doc_words and len(w) > 3]
if overlap:
for word in overlap[:1]:
idx = user_input.lower().find(word)
if idx != -1:
user_rag_links.append(RAGLink(
text=word,
start=idx,
end=idx + len(word),
corpus_snippets=[(doc_text[:RAG_LINKAGE_STORED_EXCERPT_LENGTH], score)],
top_similarity=score,
top_doc_text=doc_text[:RAG_LINKAGE_STORED_EXCERPT_LENGTH],
top_doc_score=score,
overlap_keywords=overlap[:3]
))
break
demo_evidence = {}
for attr in SENSITIVE_ATTRIBUTES:
demo_evidence[attr] = [
f"Based on context about {attr.lower()} from user input",
f"Relevant information suggesting {attr.lower()} from documents"
]
state.last_evidence_rag = demo_evidence
# Add user message
state.add("user", user_input, user_pii_matches, user_rag_links)
# Step 4: Fabricate response (NO LLM call)
llm_response = fabricate_demo_response(user_input)
# Step 5: Fabricate PII in response (NO LLM call)
response_pii_data = [] # Typically no PII in response
# Step 6: Fabricate RAG links in response
response_rag_links = []
if use_rag and rag_docs:
response_words = llm_response.lower().split()
for doc_text, score in rag_docs:
doc_words = set(doc_text.lower().split())
overlap = [w for w in response_words if w in doc_words and len(w) > 4]
if overlap:
for word in overlap[:1]:
idx = llm_response.lower().find(word)
if idx != -1:
response_rag_links.append(RAGLink(
text=word,
start=idx,
end=idx + len(word),
corpus_snippets=[(doc_text[:RAG_LINKAGE_STORED_EXCERPT_LENGTH], score)],
top_similarity=score,
top_doc_text=doc_text[:RAG_LINKAGE_STORED_EXCERPT_LENGTH],
top_doc_score=score,
overlap_keywords=overlap[:3]
))
break
if len(response_rag_links) >= 2:
break
# Add assistant message
state.add("assistant", llm_response, [], response_rag_links)
# Step 7: Fabricate probability distributions (NO LLM call)
probs_rag, probs_no_rag = fabricate_demo_probs(user_input, use_rag)
state.last_probs_rag = probs_rag
state.last_probs_no_rag = probs_no_rag
# Step 8: Calculate risk score
all_pii = user_pii_matches
all_rag = user_rag_links + response_rag_links
risk_score, risk_breakdown = calculate_privacy_risk(all_pii, all_rag, float('inf'))
# Build outputs
conv_html = fmt_conversation(state.messages, show_tips, show_rag_hl, show_pii_hl)
warning_html, demo_inference_metrics, demo_warning_shown = build_inference_warning(
user_input, probs_rag, probs_no_rag, use_rag, demo_evidence
)
# ── Log demo interaction ──────────────────────────────────
state._turn_count += 1
state._last_model = "demo"
state._last_epsilon = float("inf")
state._last_rag_enabled = use_rag
state._last_social_scraping = False
state._last_corpus_source = "demo"
state._demo_mode = True
append_interaction_log(
session_source=getattr(state, "_session_source", "demo"),
turn_number=state._turn_count,
demo_mode=True,
scenario_mode=getattr(state, "_scenario_mode", "real"),
persona_attributes={},
model="demo",
epsilon=float('inf'),
rag_enabled=use_rag,
show_risk=getattr(state, "_show_risk", True),
show_rag_highlights=show_rag_hl,
show_tips=show_tips,
show_pii_highlights=show_pii_hl,
show_settings=getattr(state, "_show_settings", False),
access_token=getattr(state, "_access_token", ""),
social_scraping_enabled=False,
corpus_source="demo",
uploaded_file_path=None,
user_prompt=user_input,
llm_response=llm_response,
risk_score=risk_score,
u_rag=user_rag_links,
r_rag=response_rag_links,
u_pii=user_pii_matches,
u_pii_perturbed=0,
r_pii=[],
inference_metrics=demo_inference_metrics,
inference_warning_shown=demo_warning_shown,
scraped_docs=None,
num_attributes_changed=0,
show_dp=getattr(state, "_show_dp", 1),
show_infr_attr_card=getattr(state, "_show_infr_attr_card", 1),
show_social_scraping=getattr(state, "_show_social_scraping", False),
show_upload_data=getattr(state, "_show_upload_data", False),
rag_corpus_path=getattr(state, "_rag_corpus_path", ""),
)
analysis = "### Demo Mode (NO API Calls)\n\nThis is a demonstration with fabricated data. "
analysis += f"**Detected {len(user_pii_matches)} PII(s)** in user prompt. "
analysis += f"**{len(user_rag_links)} RAG link(s)** in user text, "
analysis += f"**{len(response_rag_links)} RAG link(s)** in response.\n\n"
analysis += "⚠️ All data is fabricated for demonstration purposes.\n\n"
analysis += "Enable live mode by removing the `demo=1` parameter from the URL."
return conv_html, create_risk_display(risk_score, risk_breakdown), "", analysis, ""
# ============================================================
# SECTION 8 – ATTRIBUTE INFERENCE & METRICS
# ============================================================
class AttributeInferenceEngine:
"""Extract probabilities from LLM inference responses."""
# def extract_probabilities_from_response(self, response_text):
# """Extract probability distributions from JSON response."""
# try:
# # Try to extract JSON from the response
# json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
# if json_match:
# result = json.loads(json_match.group(0))
# return result
# except:
# pass
# return {}
def extract_probabilities_from_response(self, response_text):
"""Extract probability distributions AND evidence from JSON response.
Handles both evidence formats:
- new: [{"quote": "...", "type": "explicit"}, ...]
- legacy: ["...", ...] (plain strings, assumed "implicit")
"""
try:
# print("In extract_probabilities_from_response", response_text)
# Try to extract JSON from the response
json_match = re.search(r'\{.*\}', response_text, re.DOTALL)
# print("In extract_probabilities_from_response", json_match)
if json_match:
result = json.loads(json_match.group(0))
# print("********In extract_probabilities_from_response", result)
# Restructure to separate probabilities and evidence
probabilities = {}
evidence = {} # {attr: [{"quote": str, "type": "explicit"|"implicit"}]}
for attr, data in result.items():
if isinstance(data, dict):
# New format with probabilities and evidence
if 'probabilities' in data:
probabilities[attr] = data['probabilities']
raw_ev = data.get('evidence', [])
# Normalise evidence items to dicts
normalised = []
for item in raw_ev:
if isinstance(item, dict):
normalised.append({
"quote": str(item.get("quote", item.get("text", ""))).strip(),
"type": item.get("type", "implicit"),
})
else:
# Legacy plain-string evidence
normalised.append({
"quote": str(item).strip(),
"type": "implicit",
})
evidence[attr] = normalised
else:
# Old format (backward compatibility)
probabilities[attr] = data
evidence[attr] = []
else:
# Fallback for unexpected format
probabilities[attr] = {}
evidence[attr] = []
probabilities = _fill_missing_with_uniform(probabilities)
# print('probabilities', probabilities, 'evidence', evidence)
return {'probabilities': probabilities, 'evidence': evidence}
except:
pass
logger.info("Returning nothing after extracting probabilities from response.....")
return {'probabilities': _fill_missing_with_uniform({}), 'evidence': {}}
def get_predictions(self, probs):
"""Get argmax predictions."""
preds = {}
for attr, dist in probs.items():
if dist:
preds[attr] = max(dist, key=dist.get)
return preds
class InferentialPrivacyMetrics:
"""Compute privacy metrics."""
@staticmethod
def compute_z(p_rag, p_no_rag):
"""Compute log-ratio."""
if p_no_rag <= 0:
return float('inf')
return math.log(p_rag / p_no_rag) if p_rag > 0 else -float('inf')
@staticmethod
def compute_lift(z):
"""Compute lift from z."""
return math.exp(z) - 1.0 if z != float('inf') else float('inf')
@staticmethod
def compute_inferential_privacy(probs_rag, probs_no_rag, ground_truth):
"""Compute inferential privacy metrics with robust error handling."""
metrics = {}
# Validate inputs are dictionaries
if not isinstance(probs_rag, dict):
probs_rag = {}
if not isinstance(probs_no_rag, dict):
probs_no_rag = {}
if not isinstance(ground_truth, dict):
ground_truth = {}
for attr in SENSITIVE_ATTRIBUTES:
try:
# Check if attribute exists in both probability dictionaries
if attr not in probs_rag or attr not in probs_no_rag:
continue
# Validate that values are dictionaries
if not isinstance(probs_rag[attr], dict) or not isinstance(probs_no_rag[attr], dict):
continue
# Get ground truth value
gt_val = ground_truth.get(attr)
if not gt_val:
continue
# Get probabilities with default 0.0
p_rag = probs_rag[attr].get(gt_val, 0.0)
p_no_rag = probs_no_rag[attr].get(gt_val, 0.0)
# Validate probabilities are numeric
if not isinstance(p_rag, (int, float)) or not isinstance(p_no_rag, (int, float)):
continue
# Handle NaN or negative values
if math.isnan(p_rag) or math.isnan(p_no_rag) or p_rag < 0 or p_no_rag < 0:
continue
# Compute metrics
z = InferentialPrivacyMetrics.compute_z(p_rag, p_no_rag)
lift = InferentialPrivacyMetrics.compute_lift(z)
metrics[attr] = {
'z': z,
'lift': lift,
'p_rag': p_rag,
'p_no_rag': p_no_rag
}
except (TypeError, ValueError, KeyError, AttributeError):
# Silently skip this attribute if any error occurs
continue
return metrics
# ============================================================
# SECTION 9 – PROMPT BUILDERS
# ============================================================
# def build_inference_prompt(user_prompt, retrieved_docs=None):
# """Build prompt for attribute inference."""
# ctx_note = ""
# if retrieved_docs:
# doc_snippets = "\n".join(
# f" [{i}] {doc.page_content[:200]}..."
# for i, doc in enumerate(retrieved_docs[:5], 1)
# )
# ctx_note = (
# f"Note: Retrieved {len(retrieved_docs)} documents about the user:\n"
# f"{doc_snippets}\n\n"
# )
#
# return f"""As an AI safety researcher, analyze the following to estimate probability distributions for sensitive user attributes.
#
# {ctx_note}User Query: {user_prompt}
#
# Based on the {'retrieved documents and ' if retrieved_docs else ''}user query, estimate probability distributions for:
#
# {json.dumps(ATTRIBUTE_VALUES_MAP, indent=2)}
#
# Respond with ONLY a JSON object with probability estimates.
# Each attribute maps to {{value: probability}}, probabilities sum to 1.0.
#
# Example format:
# {{
# "Gender": {{"Male": 0.7, "Female": 0.3}},
# "Age bin": {{"18-29": 0.2, "30-44": 0.5, "45-59": 0.3, "60+": 0.0, "0-17": 0.0}},
# ...
# }}
#
# Your JSON response:"""
def build_inference_prompt(user_prompt, retrieved_docs=None, conversation_state=None):
"""Build prompt for attribute inference with evidence extraction.
Including prior conversation turns ensures that attribute lifts remain
stable across the session: even when the current turn is benign, the LLM
can still see sensitive information disclosed in earlier turns, so the
inferred probabilities (and therefore lifts) reflect the full accumulated
disclosure rather than only the isolated current message.
"""
ctx_note = ""
if retrieved_docs:
doc_snippets = "\n".join(
f" [{i}] {doc.page_content[:200]}..."
for i, doc in enumerate(retrieved_docs[:5], 1)
)
ctx_note = (
f"Note: Retrieved {len(retrieved_docs)} documents about the user:\n"
f"{doc_snippets}\n\n"
)
# Include prior turns so inference is cumulative across the conversation.
# Only user messages are included (assistant replies may introduce bias).
# IMPORTANT: when DP was active for a prior turn, use the PERTURBED text
# (what the LLM actually saw) rather than the original unperturbed text.
history_note = ""
if conversation_state and conversation_state.messages:
prior = []
for i, msg in enumerate(conversation_state.messages):
if msg.role != "user":
continue
dp_meta = getattr(msg, "dp_metadata", None)
if dp_meta and dp_meta.get("perturbed_text"):
turn_text = dp_meta["perturbed_text"] # what the LLM saw
else:
turn_text = msg.content # no DP, use as-is
prior.append(f" Turn {i+1}: {turn_text[:300]}")
if prior:
history_note = (
"Prior conversation turns (earlier user messages):\n"
+ "\n".join(prior)
+ "\n\n"
)
prior_best_note = ""
if conversation_state and conversation_state._best_probs_rag:
lines = []
for attr, dist in conversation_state._best_probs_rag.items():
if not dist:
continue
top_val = max(dist, key=dist.get)
top_p = dist[top_val]
evs = conversation_state._best_evidence_rag.get(attr, [])
ev_type = "explicit" if any(
(e.get("type") if isinstance(e, dict) else "implicit") == "explicit"
for e in evs
) else "implicit"
lines.append(f" {attr}: {top_val} (confidence {top_p:.0%}, {ev_type} evidence)")
if lines:
prior_best_note = (
"Previously inferred with highest confidence — treat as prior, "
"only revise if strong new contradicting evidence exists:\n"
+ "\n".join(lines)
+ "\n\n"
)
return f"""As an AI safety researcher, analyze the following to estimate probability distributions for sensitive user attributes.
{ctx_note}{history_note}{prior_best_note}Current User Query: {user_prompt}
Based on the {'retrieved documents, prior conversation, and ' if (retrieved_docs or history_note) else ''}current user query, estimate probability distributions for ALL of the following attributes:
{json.dumps(ATTRIBUTE_VALUES_MAP, indent=2)}
IMPORTANT: You MUST include every attribute listed above in your response - do not omit any.
For each attribute, also provide 1 evidence item: a short direct quotation (max 300 chars) supporting your prediction, classified as:
- "explicit": the attribute is directly stated (e.g., "I am 35 years old")
- "implicit": inferred from indirect hints (e.g., job or hobbies suggest age)
Respond with ONLY a valid JSON object. Each attribute must follow this structure:
{{
"
#
# """
def create_risk_display(risk_score, breakdown=None):
"""Create HTML for risk meter with shield badge and Why? breakdown tooltip.
Args:
risk_score – int 0-100
breakdown – optional dict from calculate_privacy_risk:
{"inference", "pii", "rag", "dp_discount"}
When provided the Why? anchor shows a data-driven explanation.
"""
if risk_score == 0:
label = "No Risk"
color = "#4CAF50"
icon = "✓"
elif risk_score < 30:
label = "Low Risk"
color = "#4CAF50"
icon = "✓"
elif risk_score < 70:
label = "Medium Risk"
color = "#FF9800"
icon = "⚠"
else:
label = "High Risk"
color = "#F44336"
icon = "⚠"
# ── Build the Why? tooltip content ────────────────────────────────────
if breakdown:
infer_pts = breakdown.get("inference", 0.0)
pii_pts = breakdown.get("pii", 0.0)
rag_pts = breakdown.get("rag", 0.0)
dp_pts = breakdown.get("dp_discount", 0.0)
use_rag = breakdown.get("use_rag", True)
infer_info = breakdown.get("inferential_score_info")
# Dominant driver sentence
drivers = sorted(
[("Inference risk", infer_pts),
("PII exposure", pii_pts),
("External data linkages", rag_pts)],
key=lambda x: x[1], reverse=True
)
top_driver, top_pts = drivers[0]
if top_pts > 0:
driver_sentence = (
f"The primary driver is {top_driver} "
f"({top_pts:.0f} pts)."
)
else:
driver_sentence = "No significant risk factors were detected yet."
# Component rows – only show non-zero values
rows = []
if infer_pts > 0:
rows.append(
f'{risk_score}
# {label}
#
# '
f'🔍 Inference risk'
f'+{infer_pts:.0f} pts'
f'
'
)
if pii_pts > 0:
rows.append(
f''
f'🏷 PII exposure'
f'+{pii_pts:.0f} pts'
f'
'
)
if rag_pts > 0:
rows.append(
f''
f'🔗 External data linkages'
f'+{rag_pts:.0f} pts'
f'
'
)
if dp_pts > 0:
rows.append(
f''
f'🔒 Differential Privacy'
f'−{dp_pts:.0f} pts'
f'
'
)
divider = ''
total_row = (
f''
f'Total'
f'{risk_score}/100'
f'
'
)
# ── Inferential privacy score section ──────────────────────────────────
inferential_section = ""
if infer_info:
i_score = infer_info["max_score"]
i_attr = infer_info["max_attr"]
i_p_rag = infer_info["p_rag"]
i_p_pop = infer_info.get("p_pop", infer_info.get("p_no_rag", 0.0))
if i_attr:
# We have at least one attribute with positive lift
attr_row = (
f''
f'Highest-lift attribute: {_esc(i_attr)}
'
f''
f'P (population prior)'
f'{i_p_pop:.1%}'
f'
'
f''
f'P (system estimate)'
f'{i_p_rag:.1%}'
f'
'
f''
f'Inferential score (max)'
f'{i_score:.3f}'
f'
'
f''
f'Inferential score (mean)'
f''
f'{infer_info.get("mean_score", 0.0):.3f}'
f'
'
)
else:
# No positive lift yet (first turn or no RAG)
attr_row = (
f''
f'No significant external-data boost detected yet. '
f'The score will update as more attributes are inferred.'
f'
'
)
inferential_section = (
f''
f'🔬 Inferential Privacy Score'
f' 0 else "#888"};">'
f'{i_score:.3f}
'
f'{attr_row}'
)
tooltip_body = (
f'Score breakdown
'
f'{"".join(rows)}'
f''
f'{total_row}'
f''
# f'{driver_sentence}'
f'{inferential_section}'
f'
'
)
else:
tooltip_body = (
'Score is calculated from PII exposure, external data linkages, '
'differential privacy settings, and inference risk from sensitive attributes.'
)
why_anchor = (
f''
f'Risk Score: {int(risk_score)}/100 — Why?'
f''
f'{tooltip_body}'
f''
f''
f'{tooltip_body}'
f''
f''
)
return f"""
{label}
{why_anchor}
']
for data in metrics_data:
attr = data['attr']
top_val = data['top_val']
p_rag = data['p_rag']
p_pop = data['p_pop']
lift = data['lift'] if not math.isinf(data['lift']) else -1
rows.append(
f'
')
return "\n".join(rows)
# def _build_analysis(pii, rag, risk, use_rag, eps, pr, pnr):
# """Build analysis markdown with improved visualizations."""
# parts = ["### Detailed Privacy Analysis\n\n**Detected PIIs:**\n"]
# if pii:
# for p in pii:
# parts.append(f"- {p.category.value}: `{p.text}` (confidence: {p.confidence:.0%})\n")
# else:
# parts.append("- None detected\n")
#
# parts.append("\n**RAG Linkages:**\n")
# if rag:
# # Group by document and create a more narrative format
# doc_groups = {}
# for r in rag:
# # Assume r has a doc_id or source attribute, fallback to grouping by similarity range
# doc_key = getattr(r, 'doc_id', f"doc_{hash(r.text) % 100}")
# if doc_key not in doc_groups:
# doc_groups[doc_key] = []
# doc_groups[doc_key].append(r)
#
# for doc_id, matches in doc_groups.items():
# avg_sim = sum(m.top_similarity for m in matches) / len(matches)
# match_count = len(matches)
# relevance = "High" if avg_sim > 0.8 else "Medium" if avg_sim > 0.6 else "Moderate"
# parts.append(f"- **Document {doc_id}**: {match_count} match{'es' if match_count > 1 else ''} "
# f"(relevance: {relevance}, avg similarity: {avg_sim:.2f})\n")
# else:
# parts.append("- None detected\n")
#
# parts.append(f"\n**Privacy Settings:**\n")
# parts.append(f"- RAG: {'Enabled' if use_rag else 'Disabled'}\n")
# parts.append(f"- ε (epsilon): {eps if eps != float('inf') else '∞ (no DP)'}\n")
#
# # Add privacy metrics plot if we have inference data
# if use_rag and pr and pnr:
# parts.append("\n**Inference Privacy Metrics:**\n")
# parts.append(_build_inference_plot(pr, pnr))
#
# return "".join(parts)
# def _build_analysis(pii, rag, risk, use_rag, eps, pr, pnr):
# """Build analysis markdown with improved visualizations."""
# parts = ["### Detailed Privacy Analysis\n\n**Detected PIIs:**\n"]
#
# # Count PIIs in prompts vs responses
# user_pii_count = sum(1 for p in pii if hasattr(p, 'start')) # Simplified check
# # Note: This assumes you have a way to distinguish user vs response PIIs
# # If pii is combined, you may need to track this separately
#
# if pii:
# # Group by category
# category_counts = {}
# for p in pii:
# cat_name = p.category.value
# category_counts[cat_name] = category_counts.get(cat_name, 0) + 1
#
# parts.append(f"- Total detected: {len(pii)}\n")
# for cat, count in category_counts.items():
# parts.append(f" - {cat}: {count}\n")
# else:
# parts.append("- None detected\n")
#
# parts.append(f"\n**Privacy Settings:**\n")
#
# # Add DP explanation
# if eps != float('inf'):
# parts.append(f"- **Differential Privacy**: ENABLED (ε = {eps:.1f})\n")
# parts.append(f" *Lower epsilon values provide stronger privacy protection "
# f"by adding noise to data retrieval, making it harder to infer sensitive "
# f"information about individuals.*\n")
# else:
# parts.append(f"- **Differential Privacy**: DISABLED (ε = ∞)\n")
# parts.append(f" *No privacy protection applied. Consider enabling DP (lower epsilon) "
# f"to add noise-based protection.*\n")
#
# # Add privacy metrics plot if we have inference data
# if use_rag and pr and pnr:
# parts.append("\n**Inference Privacy Metrics:**\n")
# parts.append(_build_inference_plot(pr, pnr))
#
# return "".join(parts) # last worked
# def _build_privacy_settings_html(eps):
# """Build Privacy Settings HTML for left panel with fixed tooltip positioning."""
#
# parts = [''
f'{attr}'
f'{top_val} ({p_rag:.0%})'
)
if lift > 0:
rows.append(
f''
f'+{lift:.0%} above base rate'
)
else:
rows.append('')
rows.append('
')
rows.append('']
# parts.append('
")
# return "".join(parts)
def _build_privacy_settings_html(eps):
"""Build Privacy Settings HTML for left panel with a CSS-native expandable DP explainer."""
if eps != float('inf'):
status_html = 'ENABLED'
# Translate epsilon to a human-readable protection level
if eps <= 30.0:
level_label = "High"
level_color = "#2E7D32"
level_desc = "Strong protection — the AI's view of your data is significantly randomised."
elif eps <= 60.0:
level_label = "Medium"
level_color = "#F57C00"
level_desc = "Moderate protection — some noise is added to limit profiling accuracy."
else:
level_label = "Low"
level_color = "#D32F2F"
level_desc = "Weak protection — only a small amount of noise is applied."
eps_line = (
f'Privacy Settings
') # # # Tooltip content - formatted as HTML with !important for font sizes # dp_tooltip = ( # 'What is Differential Privacy?
'
# ''
# ''
# 'Differential Privacy (DP) is a mathematical framework that adds calibrated noise '
# 'to data retrieval, ensuring individual records cannot be distinguished. '
# 'This protects sensitive information from being inferred when external data sources are used.'
# '
'
# ''
# 'Understanding Epsilon (ε):
'
# ''
# 'Lower values (0.1-1.0): Stronger privacy, more noise added.
' # 'Higher values (5.0-10.0): Less privacy, better data utility.
' # 'Epsilon = ∞: No privacy protection applied.' # '
'
# )
#
# # Unique IDs for this tooltip instance
# tooltip_id = f"dp_tooltip_{abs(hash(str(eps)))}"
# anchor_id = f"dp_anchor_{abs(hash(str(eps)))}"
#
# # JavaScript to position tooltip above anchor
# # JavaScript to position tooltip next to the current mouse cursor
# position_script = f"""
#
# """
#
# # Add DP status with fixed-position tooltip
# if eps != float('inf'):
# parts.append(
# f'{position_script}'
# '' # 'Higher values (5.0-10.0): Less privacy, better data utility.
' # 'Epsilon = ∞: No privacy protection applied.' # '
' # f'' # 'Differential Privacy' # ': ' # 'ENABLED' # '
' # f'' # f'Differential Privacy guarantee level (epsilon parameter) = {eps:.1f}' # f'
' # f'' # ) # else: # parts.append( # f'{position_script}' # '' # f'' # 'Differential Privacy' # ': ' # 'DISABLED' # '
' # f'' # f'Differential Privacy Guarantee = ∞' # f'
' # f'' # f'Consider enabling DP for privacy protection.' # f'
' # f'' # ) # # parts.append("' f'Protection level: ' f'{level_label}' f'
' f'' # f'{level_desc}
' ) else: status_html = 'DISABLED' eps_line = ( f'' f'Protection level: ' f'None' f'
' # f'' # f'Consider enabling privacy protection to reduce profiling risk.
' ) return ( f''
f'
' # end outer padding div
)
def _pii_severity(category):
"""Return (label, color) severity info for a PIICategory."""
if category == PIICategory.SENSITIVE:
return "High", "#b71c1c"
elif category == PIICategory.IDENTITY:
return "High", "#b71c1c"
elif category == PIICategory.CONTACT:
return "Medium", "#e65100"
else: # LOCATION
return "Low", "#1976D2"
def _load_avatar_image_b64(gender, age):
"""Load a profile photo from the avatars/ directory and return a base64 data-URI.
Expected filenames in Differential Privacy: {status_html}
' f'{eps_line}' # ── Pure-CSS expandable explainer ────────────────────────── f''
f'
'
f'' f'▼ What is this?' f'
' f''
f'
' # end inner content div
f''
f'When privacy protection is enabled, this tool automatically modifies your messages '
f'to make it harder for the AI to identify you, before any information '
f'leaves your device:'
f'
'
f''
f'✍️'
f'
'
f''
f''
f'Some words in your message. '
f'Some words in what you type are automatically swapped for similar-sounding '
f'alternatives before the AI reads them. '
f'A 🔒 badge on your message shows when this happened and which words changed.'
f'
'
f''
f'About the protection level:
'
f'🟢 High: '
f'Strong privacy — more noise applied, AI responses may be slightly less precise.
'
f'🟡 Medium: '
f'Balanced — moderate noise, reasonable trade-off between privacy and accuracy.
'
f'🔴 Low: '
f'Weak privacy — minimal noise, AI responses are more accurate but you are easier to profile.
'
f'Off: '
f'No protection — the AI sees your exact message and data.
'
f''
f'{initials}
'
)
# ── Per-attribute content extractor ──────────────────────────────────
# The '
f'{row_content}'
f'
'
)
attrs_html = "".join(rows) if rows else (
'' 'No attributes inferred yet.
' ) # ── Confidence badge ────────────────────────────────────────────────── conf = max(gender_p, age_p, finance_p) conf_pct = int(conf * 100) # ── Assemble card ───────────────────────────────────────────────────── return ( f''
# Header
f'
'
)
def _build_analysis(pii, rag, risk, use_rag, eps, pr, pnr, inference_warning_html="", is_warning=False, include_avatar=True, show_pii=True):
"""Build analysis HTML for the right panel.
Layout (top → bottom):
1. 'How the AI sees you' card (photo + inferred attrs + Why? tooltips)
— omitted when include_avatar=False (card rendered separately)
2. Divider
3. Detected PIIs (tags with severity)
"""
parts = [''
f''
f'{title_icon} How the AI sees you'
f''
f''
f'confidence: {conf_pct}%'
f''
f'
'
# Subtitle
f''
f'{subtitle}'
f'
'
# Photo centred
f'{photo_html}'
# Attribute rows — full card width
f''
f'{attrs_html}'
f'
'
# Footer disclaimer
f''
f'Photo is illustrative only — selected to match inferred profile. '
f'These are AI predictions and may be incorrect.'
f'
'
f'']
# ── Section 1: Combined avatar + inferable attributes ────────────────
if include_avatar:
avatar_html = _build_inferred_avatar(pr or {}, inference_warning_html, is_warning)
if avatar_html:
parts.append(avatar_html)
parts.append('')
# ── Section 2: Detected PIIs (last, as requested) ────────────────────
if show_pii:
parts.append('
")
return "".join(parts)
# def build_inference_warning(probs_rag, probs_no_rag, use_rag):
# ============================================================
# SECTION 14.5 – ROBUST EVIDENCE SOURCE MATCHING
# ============================================================
def tokenize_for_matching(text):
"""
Tokenize text for matching, removing punctuation and common words.
Returns:
set: Set of significant tokens (lowercased, >3 chars)
"""
import re
# Remove punctuation and split
tokens = re.findall(r'\b\w+\b', text.lower())
# Keep only significant tokens (>3 chars, not common stop words)
significant_tokens = {t for t in tokens if len(t) > 3 and t not in _DP_STOPWORDS}
return significant_tokens
def calculate_token_overlap(quote_tokens, doc_tokens, bidirectional=False):
"""Calculate token overlap ratio between two token sets.
Parameters
----------
quote_tokens, doc_tokens : set
bidirectional : bool
If True, return max(forward, reverse) overlap so that a short quote
with a high fraction of user-input words is also detected, not only
a quote whose words are mostly covered by the user input.
Use True when matching against user input; False (default) for docs.
Returns
-------
float: overlap ratio in [0, 1]
"""
if not quote_tokens or not doc_tokens:
return 0.0
intersection = len(quote_tokens & doc_tokens)
forward = intersection / len(quote_tokens)
if not bidirectional:
return forward
reverse = intersection / len(doc_tokens) # doc_tokens is user_tokens here
return max(forward, reverse)
def extract_doc_source_label(doc):
"""
Extract human-readable source label from document metadata.
Args:
doc: Document object with metadata
Returns:
str: Source label (e.g., "From uploaded data", "From Twitter (Internet)")
"""
if not hasattr(doc, 'metadata') or not isinstance(doc.metadata, dict):
return "From system data"
source_type = doc.metadata.get('source', '')
platform = doc.metadata.get('platform', '').strip().lower()
if source_type == 'uploaded_csv':
return "From uploaded data"
elif source_type.startswith('social_media_'):
platform_label = platform.capitalize() if platform else "Social Media"
return f"From {platform_label} (Internet)" if "web" not in platform_label.lower() else f"From the {platform_label} (Internet)"
elif source_type == 'fallback_tweet_data':
return "From Twitter (Internet)"
else:
# Documents from the default background corpus (e.g. PANORAMA synthetic profiles)
return DEFAULT_CORPUS_SOURCE_LABEL
def _doc_source_url(doc):
"""Extract the source URL from a document's metadata, or '' if absent."""
if not hasattr(doc, "metadata") or not isinstance(doc.metadata, dict):
return ""
return (
doc.metadata.get("source_url") or
doc.metadata.get("tweet_url") or
doc.metadata.get("url") or
""
)
def find_evidence_source(quote, user_input, retrieved_docs, min_overlap_ratio=0.4,
perturbed_input=None, prior_turns=None):
"""
Find the source of an evidence quote using robust multi-strategy matching.
Matching priority:
Strategy 1 – Exact substring: user turns first, then docs (high precision)
Strategy 2 – Token overlap: docs first, then user turns (avoids topic-word FPs)
Strategy 3 – Phrase windows: docs first, then user turns
Evidence quotes are LLM-generated paraphrases. Because the LLM is asked
to explain inferences drawn from retrieved documents, most evidence quotes
describe external-data content even when they share topic words with the
user's message. Checking docs before user turns in the fuzzy strategies
prevents those shared topic words from causing false "From your input" labels.
When no source can be matched the quote is considered LLM internal reasoning
(rationale) rather than a verbatim excerpt from any document.
Returns:
tuple: (is_from_user, is_from_docs, doc_source, match_confidence,
matched_turn_dp_active, matched_turn_original_text, doc_url)
doc_url – URL of the matched document, or "" if not from a doc / unknown.
"""
import re as _re
quote_lower = str(quote).lower().strip()
user_input_lower = user_input.lower() if user_input else ""
perturbed_lower = perturbed_input.lower() if perturbed_input else ""
# Build the full list of user turns to search
current_dp_active = bool(perturbed_lower)
turns_to_search = [(user_input_lower, perturbed_lower,
current_dp_active, user_input or "")]
if prior_turns:
for pt in prior_turns:
orig_l = (pt.get("original") or "").lower()
pert_l = (pt.get("perturbed") or "").lower()
dp_on = bool(pt.get("dp_active", False))
orig_d = pt.get("original") or ""
turns_to_search.append((orig_l, pert_l, dp_on, orig_d))
# ──────────────────────────────────────────────────────────────────────
# Strategy 1: Exact substring — user turns first, then docs
# Rationale: if the evidence is a verbatim copy of text the user typed,
# it genuinely comes from the user.
# ──────────────────────────────────────────────────────────────────────
for orig_l, pert_l, dp_on, orig_disp in turns_to_search:
if quote_lower in orig_l:
return True, False, "From your input", "exact", dp_on, orig_disp, ""
if len(quote_lower) > 50 and quote_lower[:50] in orig_l:
return True, False, "From your input", "exact", dp_on, orig_disp, ""
if pert_l and quote_lower in pert_l:
return True, False, "From your input", "exact", dp_on, orig_disp, ""
if pert_l and len(quote_lower) > 50 and quote_lower[:50] in pert_l:
return True, False, "From your input", "exact", dp_on, orig_disp, ""
if retrieved_docs:
for doc in retrieved_docs:
doc_text = doc.page_content if hasattr(doc, "page_content") else str(doc)
doc_lower = doc_text.lower()
if quote_lower in doc_lower or (
len(quote_lower) > 50 and quote_lower[:50] in doc_lower):
doc_source = extract_doc_source_label(doc)
return False, True, doc_source, "exact", False, "", _doc_source_url(doc)
# ──────────────────────────────────────────────────────────────────────
# Strategy 2: Token overlap — docs first, then user turns
# Rationale: LLM-generated evidence paraphrases retrieved docs. Both
# the evidence and the user message discuss the same topic, so shared
# topic tokens do NOT indicate the evidence came from the user.
# We check docs first; user turns are only a fallback with a high bar.
# ──────────────────────────────────────────────────────────────────────
quote_tokens = tokenize_for_matching(quote)
DOC_OVERLAP_THRESHOLD = 0.35 # forward overlap: fraction of quote tokens in doc
USER_OVERLAP_THRESHOLD = 0.65 # bidirectional: raised from 0.25 to avoid FPs
if len(quote_tokens) >= 3:
# 2a: Check retrieved docs first
if retrieved_docs:
best_overlap, best_doc = 0.0, None
for doc in retrieved_docs:
doc_text = doc.page_content if hasattr(doc, "page_content") else str(doc)
doc_tokens = tokenize_for_matching(doc_text)
ov = calculate_token_overlap(quote_tokens, doc_tokens)
if ov > best_overlap:
best_overlap, best_doc = ov, doc
if best_overlap >= DOC_OVERLAP_THRESHOLD and best_doc:
doc_source = extract_doc_source_label(best_doc)
logger.info(
f" 🔍 Token overlap match with doc ({best_overlap:.0%}): '{quote[:50]}...'")
return False, True, doc_source, "token_overlap", False, "", _doc_source_url(best_doc)
# 2b: Fall back to user turns only with a high threshold
for orig_l, pert_l, dp_on, orig_disp in turns_to_search:
turn_tokens = tokenize_for_matching(orig_l)
if calculate_token_overlap(quote_tokens, turn_tokens,
bidirectional=True) >= USER_OVERLAP_THRESHOLD:
logger.info(
f" 🔍 Token overlap match with user turn: '{quote[:50]}...'")
return True, False, "From your input", "token_overlap", dp_on, orig_disp, ""
if pert_l:
pert_tokens = tokenize_for_matching(pert_l)
if calculate_token_overlap(quote_tokens, pert_tokens,
bidirectional=True) >= USER_OVERLAP_THRESHOLD:
logger.info(
f" 🔍 Token overlap match with perturbed user turn: '{quote[:50]}...'")
return True, False, "From your input", "token_overlap", dp_on, orig_disp, ""
# ──────────────────────────────────────────────────────────────────────
# Strategy 3: Short-phrase windows — docs first, then user turns
# 3a: 3-word windows against docs
# 3b: 3-word windows against user turns (stricter guard: phrase must NOT
# appear in any retrieved doc first)
# ──────────────────────────────────────────────────────────────────────
quote_words = _re.findall(r'\b\w+\b', quote_lower)
if len(quote_words) >= 3 and retrieved_docs:
doc_texts_lower = [
(doc.page_content if hasattr(doc, "page_content") else str(doc)).lower()
for doc in retrieved_docs
]
for i in range(len(quote_words) - 2):
phrase = ' '.join(quote_words[i:i+3])
for doc, doc_lower in zip(retrieved_docs, doc_texts_lower):
if phrase in doc_lower:
doc_source = extract_doc_source_label(doc)
logger.info(
f" 🔍 3-word phrase match with doc: '{phrase}' → {doc_source}")
return False, True, doc_source, "partial", False, "", _doc_source_url(doc)
# 3b: 3-word phrase scan against user turns — only attribute to user when
# the phrase does NOT appear in any retrieved doc (disambiguates shared
# topic phrases).
if len(quote_words) >= 3:
doc_texts_lower_flat = [
(doc.page_content if hasattr(doc, "page_content") else str(doc)).lower()
for doc in (retrieved_docs or [])
]
for i in range(len(quote_words) - 2):
phrase = ' '.join(quote_words[i:i+3])
phrase_in_doc = any(phrase in dt for dt in doc_texts_lower_flat)
if phrase_in_doc:
continue # ambiguous — skip, will fall through to "General"
for orig_l, pert_l, dp_on, orig_disp in turns_to_search:
if phrase in orig_l:
logger.info(
f" 🔍 3-word phrase (user-only) match: '{phrase}'")
return True, False, "From your input", "partial", dp_on, orig_disp, ""
if pert_l and phrase in pert_l:
logger.info(
f" 🔍 3-word phrase (perturbed, user-only) match: '{phrase}'")
return True, False, "From your input", "partial", dp_on, orig_disp, ""
# No match found — the text is LLM-internal reasoning, not a source quote.
logger.info(f" ⚠️ No source match for evidence (treated as rationale): '{quote[:50]}...'")
return False, False, "General", "none", False, "", ""
# ============================================================
# SECTION 15 – BUILD INFERENCE WARNING
# ============================================================
# def build_inference_warning(probs_rag, probs_no_rag, use_rag):
# """Build yellow warning banner with lift threshold and tooltips."""
# if not probs_rag:
# return ""
#
# items = []
# max_lift = 0.0 # Track highest lift across all attributes
#
# for attr in SENSITIVE_ATTRIBUTES:
# dist = probs_rag.get(attr, {})
# if not dist:
# continue
# top_val = max(dist, key=dist.get)
# top_p = dist[top_val]
#
# lift = 0.0
# extra = ""
# if use_rag and probs_no_rag:
# p_nr = probs_no_rag.get(attr, {}).get(top_val, 0.0)
# z = InferentialPrivacyMetrics.compute_z(top_p, p_nr)
# lift = InferentialPrivacyMetrics.compute_lift(z)
#
# # Track maximum lift
# if lift > max_lift:
# max_lift = lift
#
# # Only show attributes that meet the lift threshold
# if lift > INFERENCE_LIFT_THRESHOLD and not math.isinf(lift):
# extra = (f' '
# f'(+{lift:.0%} risk from external data)')
# else:
# # Skip this attribute if lift is below threshold
# continue
#
# # Add attribute with tooltip - wrap in container with tooltip class
# items.append(
# f'Detected Personally Identifiable Information (PII) in user\'s messages:
') if pii: # De-duplicate by (text, fine_type) seen = set() unique_pii = [] for p in pii: key = (p.text.strip().lower(), p.fine_type.lower()) if key not in seen: seen.add(key) unique_pii.append(p) parts.append('')
for p in unique_pii:
col = PII_COLORS.get(p.category, {"bg": "#f5f5f5", "border": "#999"})
sev_label, sev_color = _pii_severity(p.category)
cat_label = p.category.value.capitalize()
fine = p.fine_type.title() if p.fine_type else cat_label
text_disp = p.text[:28] + ("…" if len(p.text) > 28 else "")
parts.append(
f''
f'{text_disp}'
f'[{fine}]'
f'{sev_label}'
f''
)
parts.append('
')
# Category legend
cat_counts = {}
for p in unique_pii:
cat_counts[p.category.value] = cat_counts.get(p.category.value, 0) + 1
legend_parts = [f'{v}× {k}' for k, v in cat_counts.items()]
parts.append(f''
f'Total: {len(unique_pii)} — {", ".join(legend_parts)}
')
else:
parts.append('None detected
')
parts.append("'
# ''
# '⚠ Privacy Warning – Inferable Attributes'
# ''
# '
' # f'Based on {source_note}, the AI system ' # f'could potentially infer the following about you:' # f'
')
def build_inference_warning(user_input, probs_rag, probs_no_rag, use_rag,
evidence_rag=None, retrieved_docs=None,
input_dp_metadata=None, perturbed_user_input=None,
conversation_state=None):
"""Build yellow warning banner with lift threshold and explainability tooltips.
Inference metrics are ALWAYS collected for every attribute so they can be
logged regardless of whether the banner is ultimately shown to the user.
The warning banner is only shown when at least one attribute exceeds
INFERENCE_LIFT_THRESHOLD, but inference_metrics always captures everything.
evidence_rag may contain either:
- New format: {attr: [{"quote": "...", "type": "explicit"|"implicit"}, ...]}
- Legacy format: {attr: ["...", ...]} (plain strings)
Returns:
tuple: (warning_html, inference_metrics_dict, warning_shown_bool)
warning_html – HTML string (empty string if threshold not met)
inference_metrics – dict with ALL attributes, for logging
warning_shown – True iff the banner will be visible to the user
"""
# print("In build_inference_warning")
if not probs_rag:
return "", {}, False
if evidence_rag is None:
evidence_rag = {}
# Build prior_turns list: all user messages with original + perturbed text.
prior_turns = []
if conversation_state and conversation_state.messages:
for msg in conversation_state.messages:
if msg.role != "user":
continue
dp_m = getattr(msg, "dp_metadata", None)
prior_turns.append({
"original": msg.content,
"perturbed": dp_m.get("perturbed_text", "") if dp_m else "",
"dp_active": bool(dp_m and dp_m.get("num_substitutions", 0) > 0),
})
items = [] # list items for the HTML banner (only threshold-passing attrs)
max_lift = 0.0
inference_metrics = {} # collected for ALL attributes (logging)
for attr in SENSITIVE_ATTRIBUTES:
dist = probs_rag.get(attr, {})
if not dist:
continue
top_val = max(dist, key=dist.get)
top_p = dist[top_val]
lift = 0.0
p_nr = 0.0
p_pop = POPULATION_PRIORS.get(attr, {}).get(top_val, 0.0)
# Lift is always computed against the population prior so that
# ALL experimental conditions (with or without RAG) produce a
# non-zero score when the posterior exceeds the base rate.
if p_pop > 0:
z = InferentialPrivacyMetrics.compute_z(top_p, p_pop)
lift = InferentialPrivacyMetrics.compute_lift(z)
if lift > max_lift:
max_lift = lift
# Still track the no-RAG probability for logging / comparison.
if probs_no_rag:
p_nr = probs_no_rag.get(attr, {}).get(top_val, 0.0)
# ── Determine dominant evidence_type for this attribute ──────────
# evidence_rag[attr] is a list of either:
# new: {"quote": str, "type": "explicit"|"implicit"}
# legacy: plain string → treated as "implicit"
raw_evidence = evidence_rag.get(attr, [])
evidence_type = "unknown"
if raw_evidence:
for item in raw_evidence:
if isinstance(item, dict):
t = item.get("type", "implicit")
else:
t = "implicit"
# If any evidence item is explicit, mark the whole attribute explicit
if t == "explicit":
evidence_type = "explicit"
break
else:
evidence_type = "implicit"
# ── Always record in inference_metrics (for logging) ─────────────
if not (math.isinf(lift) or math.isnan(lift)):
inference_metrics[attr] = {
'top_value': top_val,
'confidence': round(float(top_p), 4),
'lift': round(float(lift), 4),
'prob_rag': round(float(top_p), 4),
'prob_no_rag': round(float(p_nr), 4),
'p_pop': round(float(p_pop), 4),
'evidence_type': evidence_type,
}
# ── Only add to the visible banner if threshold is met ───────────
# print(lift, INFERENCE_LIFT_THRESHOLD)
above_threshold = lift > INFERENCE_LIFT_THRESHOLD and not math.isinf(lift)
# if not above_threshold:
# continue
extra = (
f' (+{lift:.0%} increase based on external data)'
if above_threshold and use_rag else ""
)
# ── Build evidence tooltip ────────────────────────────────────────
# Normalise evidence items to dicts so we can extract quote text
normalised_evidence = []
for item in raw_evidence:
if isinstance(item, dict):
quote_text = str(item.get("quote", item.get("text", ""))).strip()
ev_type = item.get("type", "implicit")
else:
quote_text = str(item).strip()
ev_type = "implicit"
if (quote_text and
'no evidence' not in quote_text.lower() and
'not found' not in quote_text.lower() and
'no specific' not in quote_text.lower() and
'unavailable' not in quote_text.lower() and
len(quote_text) > 5):
normalised_evidence.append({"quote": quote_text, "type": ev_type})
if normalised_evidence:
evidence_html = 'These attributes are AI predictions based on your input '
# 'and may not accurately reflect your actual information.
'
# ''
# ''
# 'These attributes are AI predictions based on your input '
# 'and may not accurately reflect your actual information.
'
# ''
# '' # f'Based on {source_note}, the AI system ' # f'could potentially infer the following about you:' # f'
- {"".join(items)}
Evidence for Prediction
'
evidence_html += 'The following evidence from your input or external data sources supports this prediction:
'
evidence_html += ''
for ev_item in normalised_evidence[:3]:
q = ev_item["quote"]
ev_badge = (
'explicit'
if ev_item["type"] == "explicit" else
'implicit'
)
q_esc = _esc(q)
(is_from_user, is_from_docs, doc_source, _conf,
matched_turn_dp, matched_turn_orig, doc_url) = find_evidence_source(
q, user_input, retrieved_docs, min_overlap_ratio=0.4,
perturbed_input=perturbed_user_input,
prior_turns=prior_turns,
)
# Determine whether this is a genuine source excerpt or LLM
# internal reasoning (rationale). When neither user input nor
# any retrieved document can be matched, the text is the model's
# own thinking and should NOT be displayed like a quoted source.
is_rationale = (not is_from_user) and (not is_from_docs)
if is_from_user:
source_label = "From your input"
source_color = "#a119d2"
elif is_from_docs:
source_label = doc_source
source_color = "#1976D2" if doc_source == "From uploaded data" else (
"#F57C00" if "(Internet)" in doc_source else "#666"
)
else:
source_label = "AI reasoning"
source_color = "#757575"
if len(q_esc) > ATTR_INFERENCE_EVIDENCE_STORED_EXCERPT_LENGTH:
q_esc = q_esc[:ATTR_INFERENCE_EVIDENCE_STORED_EXCERPT_LENGTH] + "..."
evidence_html += ''
evidence_html += f'
'
else:
evidence_html = ''
evidence_html += f'● {source_label}{ev_badge}'
evidence_html += '
'
if is_rationale:
# Display as plain italic reasoning text, not as a quoted excerpt.
evidence_html += (
f'{q_esc}
'
)
else:
# Display as a quoted source excerpt.
evidence_html += f'"{q_esc}"
'
# For doc sources, add a clickable URL if available.
if is_from_docs and doc_url:
url_esc = _esc(doc_url)
evidence_html += (
f''
f'Source link: '
f''
f'{url_esc}
'
)
# When DP was active for the matched turn, note that the AI saw
# a word-substituted version, not the original wording.
if is_from_user and matched_turn_dp and matched_turn_orig:
orig_display = matched_turn_orig[:400] + ("…" if len(matched_turn_orig) > 400 else "")
orig_esc = _esc(orig_display)
evidence_html += (
f''
f'🔒 Privacy protection was active for this message — the AI '
f'received a modified (word-substituted) version of your text, '
f'not your original wording.
' f'' f'Your original message: “{orig_esc}”' f'' f'
'
)
evidence_html += '' f'' f'Your original message: “{orig_esc}”' f'' f'
Evidence for Prediction
'
evidence_html += ''
evidence_html += (''
'No specific evidence from external sources was found to '
'support this prediction. The inference is based solely on the language '
"model's general knowledge and the user's input text.
")
items.append(
f'" + extra + "
" if extra else ""}'
f''
f''
f'{title_icon} {title_text}'
f''
f'
' f'{subtitle}' f'
'
)
return warning_html, inference_metrics, is_warning
# ============================================================
# SECTION 14 – CONVERSATION FORMATTING (WITH SMART TOOLTIPS)
# ============================================================
def _esc(text):
"""HTML escape."""
return (text.replace("&", "&").replace("<", "<")
.replace(">", ">").replace('"', """)
.replace("'", "'"))
def _merge_overlapping_annotations(anns):
"""
Split overlapping PII (pri=0) and RAG (pri=1) annotations into
non-overlapping sub-spans so that both highlights are always visible.
For each character range covered by both a PII and a RAG annotation a
"combined" sub-span is produced with:
• a diagonal gradient background blending both colours
• a box-shadow outline in the PII border colour
• a bottom-border in the RAG border colour
• a "P R" superscript label with each letter in its own colour
• both tooltips concatenated (PII first, divider, then RAG)
The non-overlapping tails of each annotation are preserved with their
original style.
"""
pii_list = [a for a in anns if a.get("pri", 1) == 0]
rag_list = [a for a in anns if a.get("pri", 1) == 1]
# If there is nothing to overlap, return as-is
if not pii_list or not rag_list:
return anns
out = []
# Track which (overlap_s, overlap_e) intervals belong to each source ann
used_pii = [] # list of (ov_s, ov_e, pii_ann_object)
used_rag = [] # list of (ov_s, ov_e, rag_ann_object)
for p in pii_list:
for r in rag_list:
ov_s = max(p["s"], r["s"])
ov_e = min(p["e"], r["e"])
if ov_s >= ov_e:
continue # no overlap
combined_tip = (
p.get("tip_html", "") +
''
'These attributes are AI predictions based on your input '
f'and may not accurately reflect your actual information.
'
f''
f'' f'{subtitle}' f'
- {"".join(items)}
Also linked to external data:
' +
r.get("tip_html", "")
)
out.append({
"s": ov_s,
"e": ov_e,
# Diagonal gradient: PII colour top-half, RAG colour bottom-half
"bg": f"linear-gradient(to bottom, {p['bg']} 52%, {r['bg']} 48%)",
"bd": r["bd"], # RAG border-bottom
"bd2": p["bd"], # PII outline via box-shadow
"lbl_p_c": p["bd"], # colour for "P" superscript
"lbl_r_c": r["bd"], # colour for "R" superscript
"tip_html": combined_tip,
"pri": 0,
"combined": True,
})
used_pii.append((ov_s, ov_e, p))
used_rag.append((ov_s, ov_e, r))
# Subtract already-emitted overlap intervals from an annotation and
# return the remaining (non-overlapping) sub-spans.
def _remainders(ann, used_list):
intervals = sorted(
(ov_s, ov_e) for ov_s, ov_e, src in used_list if src is ann
)
if not intervals:
return [ann]
# Merge adjacent/touching intervals
merged = [list(intervals[0])]
for iv_s, iv_e in intervals[1:]:
if iv_s <= merged[-1][1]:
merged[-1][1] = max(merged[-1][1], iv_e)
else:
merged.append([iv_s, iv_e])
# Gaps between merged intervals → individual sub-spans
parts = []
cur = ann["s"]
for iv_s, iv_e in merged:
if cur < iv_s:
parts.append({**ann, "s": cur, "e": iv_s})
cur = iv_e
if cur < ann["e"]:
parts.append({**ann, "s": cur, "e": ann["e"]})
return parts
for p in pii_list:
out.extend(_remainders(p, used_pii))
for r in rag_list:
out.extend(_remainders(r, used_rag))
return out
def render_highlighted(text, pii_matches, rag_links, show_tips=True, show_rag_hl=True, show_pii_hl=True):
"""Render text with colored highlight spans and smart-positioned tooltips."""
if text is None:
text = ""
if not isinstance(text, str):
text = str(text)
anns = []
if show_pii_hl and pii_matches:
for m in pii_matches:
c = PII_COLORS[m.category]
tip_html = (
f'{c["label"]}
'
f'Detected: {_esc(m.text)}
'
f''
f'Type: {m.fine_type} | Confidence: {m.confidence:.0%}
'
)
anns.append({
"s": m.start,
"e": m.end,
"bg": c["bg"],
"bd": c["border"],
"lbl": "P",
"tip_html": tip_html,
"pri": 0,
})
if show_rag_hl and rag_links:
for lk in rag_links:
kw_txt = ", ".join(lk.overlap_keywords or ["N/A"])
# Determine source color based on source type
source = getattr(lk, 'source', 'From system data')
if source == "From uploaded data":
source_color = "#1976D2" # Blue
elif "Internet" in source:
source_color = "#F57C00" # Orange
else:
source_color = "#666" # Gray
# Build optional URL line for social media / web sources
lk_url = getattr(lk, 'url', '')
url_line = ""
if lk_url and "Internet" in source:
display_url = lk_url # if len(lk_url) <= 80 else lk_url[:77] + "..."
url_line = (
f'Source link: '
f''
f''
f'{_esc(display_url)}'
f'
'
)
tip_html = (
'External Data Linkage
'
f'Text: {_esc(lk.text)}
'
''
''
'
'
'Top Retrieved Document:
'
f'{_esc(lk.top_doc_text)}
'
''
# f'
'
''
'Retrieval similarity: {float(lk.top_doc_score):.2f}
'
# f'Matched keywords: {_esc(kw_txt)}
'
f'Combined similarity: {float(lk.top_similarity):.2f}
'
f'Retrieval score: {float(lk.top_doc_score):.2f}
'
f'Matched keywords: {_esc(kw_txt)}
'
f'Source: {source}
'
f'{url_line}'
'This linkage increases the chance that sensitive attributes can be inferred by combining your text with external data.
'
)
anns.append({
"s": lk.start,
"e": lk.end,
"bg": RAG_LINK_COLOR["bg"],
"bd": RAG_LINK_COLOR["border"],
"lbl": "R",
"tip_html": tip_html,
"pri": 1, # RAG yields to PII in overlaps
})
if not anns:
return _esc(text)
# return f"{_esc(text)}
" # anns.sort(key=lambda a: (a["s"], -a["e"])) anns = _merge_overlapping_annotations(anns) anns.sort(key=lambda a: (a["s"], a.get("pri", 1), -a["e"])) parts, last = [], 0 for a in anns: if a["s"] < last: continue if a["s"] > last: parts.append(_esc(text[last:a["s"]])) st = _esc(text[a["s"] : a["e"]]) # CSS-based tooltip with smart positioning cls = "hl" tooltip_html = "" if show_tips and a.get("tip_html"): cls += " tt-anchor" # Add both above and below variants - JS will choose tooltip_html = ( f'{a["tip_html"]}' f'{a["tip_html"]}' f'{a["tip_html"]}' f'{a["tip_html"]}' f'{a["tip_html"]}' f'{a["tip_html"]}' ) if a.get("combined"): # Two-colour span: diagonal gradient bg + RAG bottom border + # PII outline via box-shadow. Label: "P" (PII colour) "R" (RAG colour). label_html = ( f'' f'P' f'R' f'' ) parts.append( f'{st}{label_html}{tooltip_html}' ) else: parts.append( f'{st}' f'{a["lbl"]}' f'{tooltip_html}' ) last = a["e"] if last < len(text): parts.append(_esc(text[last:])) # Wrap in a div to ensure proper rendering return "".join(parts) # if last < len(text): # parts.append(_esc(text[last:])) # return "".join(parts) # def fmt_conversation(messages, show_tips=True, show_rag_hl=True, show_pii_hl=True): # """Format conversation with highlights and smart tooltips.""" # css = ( # "" # # # JavaScript to detect position and add 'near-top' class # "" # ) # # h = [css, '']
# for m in messages:
# role_cls = "msg-user" if m.role == "user" else "msg-ai"
# role_label = "You" if m.role == "user" else "Assistant"
# # content_html = render_highlighted(m.content, m.pii_matches, m.rag_links, show_tips, show_rag_hl, show_pii_hl)
# if m.rag_links or m.pii_matches:
# content_html = render_highlighted(m.content, m.pii_matches, m.rag_links, show_tips, show_rag_hl,
# show_pii_hl)
# else:
# formatted_content = _format_response_text(m.content)
# content_html = render_highlighted(formatted_content, m.pii_matches, m.rag_links, show_tips, show_rag_hl,
# show_pii_hl)
# h.append(
# f'
')
# return "".join(h)
def fmt_conversation(messages, show_tips=True, show_rag_hl=True, show_pii_hl=True):
"""Format conversation with highlights and smart tooltips."""
css = (
""
)
h = [css, ''
# f'
'
# )
# h.append('{role_label}
'
# f'{content_html}']
for m in messages:
role_cls = "msg-user" if m.role == "user" else "msg-ai"
role_label = "You" if m.role == "user" else "Assistant"
# if m.rag_links or m.pii_matches:
# content_html = render_highlighted(m.content, m.pii_matches, m.rag_links, show_tips, show_rag_hl,
# show_pii_hl)
# else:
# formatted_content = _format_response_text(m.content)
# content_html = render_highlighted(formatted_content, m.pii_matches, m.rag_links, show_tips, show_rag_hl,
# show_pii_hl)
# Always format text first, regardless of annotations
content_html = render_highlighted(m.content, m.pii_matches, m.rag_links, show_tips, show_rag_hl, show_pii_hl)
# Append DP badge to user messages that were perturbed
badge_html = ""
if m.role == "user" and getattr(m, "dp_metadata", None):
badge_html = _dp_badge_html(m.dp_metadata)
if badge_html:
badge_row = (
f'
')
return "".join(h)
def clear_conversation(state, show_tips, show_rag_hl, show_pii_hl):
"""Clear conversation."""
state.clear()
return (
fmt_conversation([], show_tips, show_rag_hl, show_pii_hl),
create_risk_display(0),
"",
"", "", gr.update(visible=False), gr.update(visible=False)
)
# ============================================================
# SECTION 15 – LEGEND
# ============================================================
def create_legend_html(show_tips=True, show_rag_hl=True):
"""Create legend HTML."""
items = []
if show_tips:
for _c, col in PII_COLORS.items():
items.append(
f'{badge_html}
'
)
else:
badge_row = ""
h.append(
f''
f'
'
)
h.append('{role_label}
'
f'{content_html}{badge_row}'
f''
f'{col["label"]}
')
if show_rag_hl:
items.append(
f''
f''
f'{RAG_LINK_COLOR["label"]}
')
return "".join(items)
# ============================================================
# SECTION 16 – GRADIO UI
# ============================================================
def create_ui(
model=None,
rag="1",
epsilon="inf",
show_risk="0",
show_tips="0",
show_rag_highlights="0",
show_pii_highlights="1",
show_settings="0",
demo="0",
enable_social_scraping="0",
show_social_scraping="0",
show_upload_data = "1",
scenario_mode="real",
retriever=None,
show_dp="1",
show_infr_attr_card="1",
):
"""Create Gradio UI with settings panel."""
# state = ConversationState()
def _to_bool(v):
if v is None or v == "":
return False
if isinstance(v, (int, float)):
return bool(int(v))
s = str(v).strip().lower()
return s in {"1", "true", "t", "yes", "y", "on"}
def _to_float(v, default=float("inf")):
if v is None or v == "":
return default
s = str(v).strip().lower()
if s in {"inf", "infty", "infinite", "infinity"}:
return float("inf")
try:
return float(s)
except Exception:
return default
def _make_demo_prompt():
tpl = random.choice(list(DATASET_PROMPTS_PANORAMA.values()))
return tpl.format(name="Raymond Phillips")
# Defaults
use_rag_default = _to_bool(rag)
eps_default = _to_float(epsilon, default=float("inf"))
model_default = str(model or DEFAULT_MODEL_NAME)
show_risk_default = _to_bool(show_risk)
show_tips_default = _to_bool(show_tips)
show_raghl_default = _to_bool(show_rag_highlights)
# show_pii_hl_default = _to_bool(highlight_pii)
show_pii_hl_default = _to_bool(show_pii_highlights)
show_settings_default = _to_bool(show_settings)
show_social_scraping_default = _to_bool(show_social_scraping)
show_upload_data_default = _to_bool(show_upload_data)
demo_default = _to_bool(demo)
social_scraping_default = _to_bool(enable_social_scraping)
scenario_mode_default = str(scenario_mode or "real").strip().lower()
# show_dp: 0=never, 1=after each turn (default), 2=beginning only, 3=on End button
# show_infr_attr_card: 0=never, 1=after each turn (default), 2=on End button
show_dp_default = int(_to_float(show_dp, default=1.0))
show_infr_default = int(_to_float(show_infr_attr_card, default=1.0))
# Initialize social scraper
initialize_social_scraper(enabled=social_scraping_default)
# vs = get_retriever()
_get_system_retriever_for_scenario = get_scenario_retriever
with gr.Blocks(
# css=css,
title="What can LLMs infer about me?",
theme=gr.themes.Default(spacing_size=gr.themes.sizes.spacing_sm, radius_size=gr.themes.sizes.radius_none),
) as app:
access_control_css = gr.HTML("")
if show_settings_default:
gr.Markdown(
"# 🔒 What can LLMs infer about me?\n\n"
"Analyze privacy risks and understand what LLMs can infer from your data."
)
# Session-scoped configuration
rag_st = gr.State(use_rag_default)
eps_st = gr.State(eps_default)
mdl_st = gr.State(model_default)
tips_st = gr.State(show_tips_default)
raghl_st = gr.State(show_raghl_default)
piihl_st = gr.State(show_pii_hl_default)
riskpanel_st = gr.State(show_risk_default)
demo_st = gr.State(demo_default)
demo_prompt_st = gr.State(_make_demo_prompt() if demo_default else "")
vs_st = gr.State(None)
social_scraping_st = gr.State(False) # Social media scraping toggle
custom_corpus_loaded_st = gr.State(False) # Track if custom corpus is loaded
uploaded_file_path_st = gr.State(None)
state_st = gr.State(ConversationState())
# New: control panel visibility modes and conversation-ended flag
show_dp_st = gr.State(show_dp_default)
show_infr_st = gr.State(show_infr_default)
conv_ended_st = gr.State(False)
with gr.Row():
# LEFT: legend + settings panel
with gr.Column(min_width=220, scale=1) as legend_col:
with gr.Group(visible=False) as legend_group:
# with gr.Row():
gr.Markdown("### 🎨 Legend")
legend_html = gr.HTML("")
# Settings panel (toggled by URL parameter)
with gr.Group(visible=show_settings_default) as settings_panel:
gr.Markdown("### ⚙️ Settings")
# Model selector
model_dropdown = gr.Dropdown(
choices=[name for _, _, name in MODEL_CONFIGS],
value=model_default,
label="Model",
interactive=True
)
# Epsilon slider - Differential Privacy Guarantee (Noise Level)
epsilon_slider = gr.Slider(
minimum=0.1,
maximum=MAX_POSSIBLE_EPS,
value=min(eps_default, MAX_POSSIBLE_EPS) if eps_default != float('inf') else MAX_POSSIBLE_EPS,
label="Differential Privacy Guarantee (Noise Level)",
info="",#"Controls the amount of random noise (perturbation) added during linkage with external data. Lower values = more noise = stronger privacy protection. Higher values = less noise = better accuracy but weaker privacy guarantees.",
interactive=True
)
# Infinite epsilon checkbox
eps_inf_checkbox = gr.Checkbox(
value=(eps_default == float('inf')),
label="Disable Differential Privacy",
interactive=True
)
with gr.Group(visible=False, elem_id="privacy_settings_panel") as privacy_settings_group:
gr.Markdown("### 🔒 Privacy Settings")
privacy_settings_html = gr.HTML("")
social_scraping_group = gr.Group(visible=show_social_scraping_default)
with social_scraping_group:
# Social Media Scraping toggle
gr.Markdown("#### 🌐 Access to Social Media")
social_scraping_checkbox = gr.Checkbox(
value=social_scraping_default,
label="Allow access to my social media",
interactive=True
)
social_scraping_info = gr.HTML(
"""
Extract data from social media platforms in real-time.
This may take up to a minute.
"""
)
social_platforms = gr.CheckboxGroup(
choices=[
"Twitter",
"Facebook",
"LinkedIn",
"Web",
],
value=["Twitter","Facebook", "LinkedIn", "Web"],
label="Platforms",
info=(
"Twitter | "
"Facebook | "
"LinkedIn | "
"Web"
),
visible=social_scraping_default,
interactive=True
)
upload_data_group = gr.Group(visible=show_upload_data_default)
with upload_data_group:
# Custom Corpus Upload
gr.Markdown("#### 📁 Upload My Data")
corpus_file = gr.File(
label="Upload CSV file",
file_types=None, #[".csv"],
type="filepath",
interactive=True
)
corpus_upload_info = gr.HTML(
"""
This may take up to a minute.
Download your textual social media data from Facebook, LinkedIn, Twitter (X).
CSV should have columns: text (required), user id (if known), First Name, Last Name
"""
)
corpus_upload_btn = gr.Button("Upload", variant="secondary", size="sm")
corpus_status = gr.Markdown(
"",
elem_id="corpus_status",
elem_classes="corpus-status-box"
)
# CENTER: conversation
with gr.Column(scale=8):
conversation_panel_title = "### 💬 Conversation"
if show_raghl_default:
conversation_panel_title += " | Hover over highlighted text to see why it was flagged."
gr.Markdown(conversation_panel_title)
conv_html = gr.HTML(fmt_conversation([], show_tips_default, show_raghl_default, show_pii_hl_default))
warn_html = gr.HTML("")
persona_hint = gr.HTML("")
user_tb = gr.Textbox(
value="",
placeholder="Type your message here…",
label="Your Message",
lines=3,
)
with gr.Row():
send_btn = gr.Button("Send", variant="primary", scale=1)
# clear_btn = gr.Button("Clear", scale=1)
reset_btn = gr.Button("Reset conversation", scale=1)
end_btn = gr.Button("End conversation", variant="stop", scale=1)
# RIGHT: risk panel
right_col_needed = show_risk_default or show_pii_hl_default or (show_infr_default != 0)
with gr.Column(scale=3, visible=right_col_needed) as risk_col:
risk_title = gr.Markdown("", visible=bool(show_risk_default))
risk_html = gr.HTML("", visible=bool(show_risk_default))
analysis_title = gr.Markdown("### 📝 Analysis", visible=False)
infer_card_html = gr.HTML("", visible=(show_infr_default != 0))
analysis_md = gr.HTML("", visible=False)
# Page-load: apply URL query parameters
# def _apply_query_params(request: gr.Request):
# qp = getattr(request, "query_params", {}) or {}
#
# access_token = qp.get("token", "")
# if not validate_access_token(access_token):
# # Invalid or missing token - return error state
# error_msg = """
#
#
# """
# return (
# error_msg, # legend_html - show error
# gr.update(visible=False), # settings_panel
# gr.update(visible=False), # risk_col
# False, # rag_st
# 1.0, # eps_st
# DEFAULT_MODEL_NAME, # mdl_st
# False, # tips_st
# False, # raghl_st
# False, # piihl_st
# False, # riskpanel_st
# False, # demo_st
# "", # demo_prompt_st
# error_msg, # conv_html - show error
# "", # risk_html
# "", # warn_html
# "", # analysis_md
# "", # user_tb
# DEFAULT_MODEL_NAME, # model_dropdown
# 1.0, # epsilon_slider
# False, # eps_inf_checkbox
# )
#
# mdl = str(qp.get("model", model_default))
# use_rag = _to_bool(qp.get("rag", use_rag_default))
# eps = _to_float(qp.get("epsilon", eps_default), default=eps_default)
#
# show_risk_v = _to_bool(qp.get("show_risk", show_risk_default))
# show_tips_v = _to_bool(qp.get("show_tips", show_tips_default))
# show_raghl_v = _to_bool(qp.get("show_rag_highlights", show_raghl_default))
# show_pii_hl_v = _to_bool(qp.get("highlight_pii", show_pii_hl_default))
# show_settings_v = _to_bool(qp.get("show_settings", show_settings_default))
# demo_v = _to_bool(qp.get("demo", demo_default))
#
# legend_visible = bool(show_tips_v or show_raghl_v or show_pii_hl_v)
# # demo_prompt = _make_demo_prompt() if demo_v else ""
# demo_prompt = DEFAULT_USER_TEXT if DEFAULT_USER_TEXT else ""
#
# legend = create_legend_html(show_tips=show_tips_v, show_rag_hl=show_raghl_v)
#
# eps_is_inf = (eps == float('inf'))
# eps_slider_val = min(eps, 10.0) if not eps_is_inf else 10.0
#
# return (
# legend,
# gr.update(visible=show_settings_v),
# gr.update(visible=show_risk_v),
# use_rag,
# eps,
# mdl,
# show_tips_v,
# show_raghl_v,
# show_pii_hl_v,
# show_risk_v,
# demo_v,
# demo_prompt,
# fmt_conversation([], show_tips_v, show_raghl_v, show_pii_hl_v),
# create_risk_display(0),
# "",
# "",
# demo_prompt,
# mdl,
# eps_slider_val,
# eps_is_inf,
# )
def _persona_hint_html(scenario_mode, persona):
"""Return an HTML hint banner for the given persona, or '' for 'real' mode."""
if scenario_mode == "real" or not persona.get("description"):
return (
'⚠️ Access Denied
#Invalid or missing access token.
#Please contact the administrator for a valid access link.
#'
'💡 Tip: Describe what you need help with. '
'For example: "I am looking for job suggestions in my field" '
'or "What health screenings should I consider?"'
'
'
)
attrs = persona.get("attributes", {})
attr_str = ", ".join(f"{k}: {v}" for k, v in attrs.items())
return (
''
f'🎭 Your persona: {persona["description"]}
' # f'Profile — {attr_str}' '
'
)
def _apply_query_params(request: gr.Request):
qp = getattr(request, "query_params", {}) or {}
access_token = qp.get("token", "")
# ★ SIMPLE: If invalid token, inject CSS to hide everything + show error
if not validate_access_token(access_token):
hide_all_css = """
' # f'Profile — {attr_str}' '
🔒
Access Denied
Invalid or missing access token.
Please contact the administrator for a valid access link.
"
"❌ Unsupported file type: "
f"
" "Please upload a CSV file " "with at least a
"
), False, current_vs, None, state
return
# Immediate status update
yield "", False, current_vs, None, state
progress(0, desc="Starting upload...")
logger.info(f"Attempting to upload corpus from: {file_path}")
yield "", False, current_vs, None, state
progress(0.2, desc="Reading CSV file...")
time.sleep(0.1) # Brief pause to ensure UI updates
new_retriever = build_retriever_from_csv(file_path, progress)
if new_retriever is None:
yield "❌ **Failed to build retriever** from uploaded file", False, current_vs, None, state
return
time.sleep(0.1) # Brief pause to ensure UI updates
yield "", False, current_vs, None, state
progress(0.9, desc="Finalizing...")
time.sleep(1) # Brief pause to ensure UI updates
# Update the global retriever
# set_retriever(new_retriever)
state._uploaded_file_path = file_path
progress(1.0, desc="Complete!")
time.sleep(1) # Brief pause to ensure UI updates
yield "{ext or '(no extension)'}" "Please upload a CSV file " "with at least a
text column. "
"Other formats (PDF, DOCX, images, etc.) are not supported."
"✅ Custom data loaded successfully!
System is now ready to use.
", True, new_retriever, file_path, state
except Exception as e:
logger.error(f"Error uploading corpus: {str(e)}")
yield f"System is now ready to use.
❌ Error:
{str(e)}
", False, current_vs, None, state
corpus_upload_btn.click(
fn=_upload_corpus,
inputs=[corpus_file, vs_st, state_st],
outputs=[corpus_status, custom_corpus_loaded_st, vs_st, uploaded_file_path_st, state_st]
)
# Send message
def _send(msg, rag_v, eps_v, mdl_v, tips_v, raghl_v, pii_v, demo_v,
vs_v, social_scraping_v, social_platforms_v, uploaded_file_path_v,
show_dp_v, show_infr_v, state, request: gr.Request):
# Stamp session metadata on first (and every) call so it stays current
qp = getattr(request, "query_params", {}) or {}
ip = (
getattr(request, "client", None) and request.client.host
or qp.get("x-forwarded-for", "")
or "unknown"
)
state._session_source = ip
state._access_token = qp.get("token", "")
state._show_rag_highlights = raghl_v
state._show_tips = tips_v
state._show_pii_hl = pii_v
state._uploaded_file_path = uploaded_file_path_v
state._show_risk = _to_bool(qp.get("show_risk", True))
state._show_settings = _to_bool(qp.get("show_settings", False))
if demo_v:
result = process_demo_message(msg, tips_v, raghl_v, pii_v, state, rag_v)
else:
result = process_message(msg, rag_v, eps_v, mdl_v, tips_v, raghl_v,
pii_v, state, vs_v, social_scraping_v, social_platforms_v)
# Reveal panels now that a response exists, respecting each flag
# result[1] is the risk display value; None signals an LLM error.
llm_errored = (result[1] is None)
show_risk_now = getattr(state, "_show_risk", False) and not llm_errored
show_legend_now = (raghl_v or pii_v) and not llm_errored
# ── Privacy Settings panel visibility (show_dp) ───────────────
# 0=never, 1=after each turn, 2=beginning only (no update after turns),
# 3=on End button (hide during turns)
if llm_errored or show_dp_v == 0 or show_dp_v == 2 or show_dp_v == 3:
privacy_html_update = gr.update()
privacy_group_update = gr.update(visible=False)
else:
privacy_html_update = gr.update(value=result[4])
privacy_group_update = gr.update(visible=True)
# ── Inference card visibility (show_infr_attr_card) ───────────
# 0=never, 1=after each turn, 2=on End button
avatar_html = getattr(state, "_last_avatar_html", "")
if llm_errored or show_infr_v == 0 or show_infr_v == 2:
infer_card_update = gr.update(visible=False)
else: # show_infr_v == 1: show after each turn
infer_card_update = gr.update(value=avatar_html) # node already visible, just set content
# ── Per-card visibility ───────────────────────────────────────────
risk_html_val = result[1] if not llm_errored else ""
show_pii_now = bool(pii_v) and not llm_errored
# analysis_title is shown when at least one of its child cards is visible this turn
infer_visible_now = (not llm_errored) and show_infr_v == 1 and bool(avatar_html)
show_analysis_title_now = infer_visible_now or show_pii_now
return (
result[0], # conv_html
gr.update(value=(risk_html_val if show_risk_now else "")), # risk_html (value + visibility)
result[2], # warn_html
gr.update(value=result[3], visible=show_pii_now), # analysis_md (value + visibility)
privacy_html_update, # privacy_settings_html
privacy_group_update, # privacy_settings_group
gr.update(), # risk_col — visibility decided once at load; do not re-toggle the parent
gr.update(visible=show_legend_now), # legend_group
infer_card_update, # infer_card_html
gr.update(value=("### 🛡️ Privacy Risk" if show_risk_now else "")), # risk_title
gr.update(visible=show_analysis_title_now), # analysis_title
state,
)
send_btn.click(
fn=_send,
inputs=[user_tb, rag_st, eps_st, mdl_st, tips_st, raghl_st, piihl_st,
demo_st, vs_st, social_scraping_st, social_platforms, uploaded_file_path_st,
show_dp_st, show_infr_st, state_st],
outputs=[conv_html, risk_html, warn_html, analysis_md,
privacy_settings_html, privacy_settings_group, risk_col, legend_group, infer_card_html,
risk_title, analysis_title, state_st],
).then(fn=lambda: "", outputs=user_tb)
user_tb.submit(
fn=_send,
inputs=[user_tb, rag_st, eps_st, mdl_st, tips_st, raghl_st, piihl_st,
demo_st, vs_st, social_scraping_st, social_platforms, uploaded_file_path_st,
show_dp_st, show_infr_st, state_st],
outputs=[conv_html, risk_html, warn_html, analysis_md,
privacy_settings_html, privacy_settings_group, risk_col, legend_group, infer_card_html,
risk_title, analysis_title, state_st],
).then(fn=lambda: "", outputs=user_tb)
# clear_btn.click(
# fn=lambda tv, rv, pv: clear_conversation(state, tv, rv, pv),
# inputs=[tips_st, raghl_st, piihl_st],
# outputs=[conv_html, risk_html, warn_html, analysis_md, privacy_settings_html, risk_col, legend_group],
# )
# End conversation button – disables input, reveals panels per mode
def _end_conversation(show_dp_v, show_infr_v, state):
privacy_html = getattr(state, "_last_privacy_html", "")
avatar_html = getattr(state, "_last_avatar_html", "")
append_session_end_log(state)
# Privacy Settings panel: show only if show_dp == 3
if show_dp_v == 3 and privacy_html:
privacy_html_update = gr.update(value=privacy_html)
privacy_group_update = gr.update(visible=True)
else:
privacy_html_update = gr.update()
privacy_group_update = gr.update()
# Inference card: show only if show_infr == 2
if show_infr_v == 2 and avatar_html:
infer_update = gr.update(value=avatar_html, visible=True)
title_update = gr.update(visible=True)
else:
infer_update = gr.update()
title_update = gr.update()
return (
gr.update(interactive=False), # send_btn
gr.update(interactive=False), # reset_btn
gr.update(interactive=False), # end_btn
gr.update(interactive=False), # user_tb
privacy_html_update,
privacy_group_update,
infer_update,
title_update, # analysis_title
True,
)
end_btn.click(
fn=_end_conversation,
inputs=[show_dp_st, show_infr_st, state_st],
outputs=[send_btn, reset_btn, end_btn, user_tb,
privacy_settings_html, privacy_settings_group, infer_card_html, analysis_title, conv_ended_st],
)
def _reset(tv, rv, pv, state):
state.clear()
return (
fmt_conversation([], tv, rv, pv),
gr.update(value=create_risk_display(0), visible=False), # risk_html — hidden
"",
gr.update(value="", visible=False), # analysis_md — hidden
"",
gr.update(value=""), # privacy_settings_html
gr.update(visible=False), # privacy_settings_group
gr.update(visible=True), # risk_col — always visible container
gr.update(visible=False), # legend_group hidden
gr.update(value="", visible=False), # infer_card_html hidden
gr.update(visible=False), # risk_title hidden
gr.update(visible=False), # analysis_title hidden
gr.update(interactive=True), # send_btn re-enabled
gr.update(interactive=True), # reset_btn re-enabled
gr.update(interactive=True), # end_btn re-enabled
gr.update(value="", interactive=True), # user_tb re-enabled
False, # conv_ended_st reset
state,
)
reset_btn.click(
fn=_reset,
inputs=[tips_st, raghl_st, piihl_st, state_st],
outputs=[conv_html, risk_html, warn_html, analysis_md, user_tb,
privacy_settings_html, privacy_settings_group, risk_col, legend_group,
infer_card_html, risk_title, analysis_title,
send_btn, reset_btn, end_btn, user_tb,
conv_ended_st, state_st],
)
return app
# ============================================================
# SECTION 17 – MAIN ENTRY POINT
# ============================================================
if __name__ == "__main__":
import argparse
p = argparse.ArgumentParser()
p.add_argument("--model", default=DEFAULT_MODEL_NAME)
p.add_argument("--rag", default="1")
p.add_argument("--epsilon", default="inf")
p.add_argument("--show_risk", default="1")
p.add_argument("--show_tips", default="1")
p.add_argument("--show_rag_highlights", default="1")
p.add_argument("--show_pii_highlights", default="1")
p.add_argument("--show_settings", default="0")
p.add_argument("--demo", default="0")
p.add_argument("--enable_social_scraping", default="0", help="Enable social media scraping (experimental)")
p.add_argument("--show_social_scraping", default="0", help="Show social media scraping option")
p.add_argument("--show_upload_data", default="0", help="Show upload data option")
p.add_argument("--scenario_mode", default="real", help="Scenario mode: 'real' for free-style, or 'persona1', 'persona2', etc.")
p.add_argument("--show_dp", default="1")
p.add_argument("--show_infr_attr_card", default="1")
p.add_argument("--retriever_path", default=None)
p.add_argument("--port", default="7860")
args = p.parse_args()
css = """
/* --- COMPREHENSIVE LIGHT THEME OVERRIDE --- */
:root, body, .gradio-container, .dark, .dark * {
/* Layout Backgrounds */
--body-background-fill: #ffffff !important;
--background-fill-primary: #ffffff !important;
--background-fill-secondary: #f8f9fa !important;
--block-background-fill: #ffffff !important;
--panel-background-fill: #ffffff !important;
/* Text Colors */
--text-color: #000000 !important;
--body-text-color: #000000 !important;
--block-title-text-color: #000000 !important;
--block-label-text-color: #000000 !important;
/* General Inputs */
--input-background-fill: #ffffff !important;
--input-background-fill-focus: #ffffff !important;
--input-text-color: #000000 !important;
--border-color-primary: #aaaaaa !important;
/* Checkboxes & Radios */
--checkbox-background-color: #ffffff !important;
--checkbox-background-color-selected: #4A90D9 !important;
--checkbox-border-color: #aaaaaa !important;
--checkbox-border-color-selected: #4A90D9 !important;
--checkbox-border-color-focus: #4A90D9 !important;
--checkbox-label-background-fill: #ffffff !important;
--checkbox-label-background-fill-selected: #f0f7ff !important;
--checkbox-label-text-color: #000000 !important;
--checkbox-label-text-color-selected: #000000 !important;
--radio-background-color: #ffffff !important;
/* Sliders */
--slider-color: #4A90D9 !important;
/* Buttons */
--button-primary-background-fill: #f97316 !important;
--button-primary-text-color: #ffffff !important;
--button-secondary-background-fill: #ffffff !important;
--button-secondary-text-color: #000000 !important;
color-scheme: light !important;
}
/* Force Info Text to be smaller, italic, and solid black without altering line-height */
.gradio-container .info,
.dark .info,
.checkbox-container .info,
.dark .checkbox-container .info {
font-size: 0.85em !important;
font-style: italic !important;
color: #000000 !important;
opacity: 1 !important;
}
/* Force standard fallback for stubborn elements, EXCLUDING checkboxes/radios */
.dark textarea,
.dark input:not([type="checkbox"]):not([type="radio"]),
.dark select,
.dark .gr-box,
.dark .gr-panel,
.dark input[type="range"] {
background-color: #ffffff !important;
color: #000000 !important;
}
/* Hardcode the checkmark icon just in case Hugging Face wipes it */
.dark input[type="checkbox"]:checked {
background-color: #4A90D9 !important;
border-color: #4A90D9 !important;
background-image: url("data:image/svg+xml,%3csvg viewBox='0 0 16 16' fill='white' xmlns='http://www.w3.org/2000/svg'%3e%3cpath d='M12.207 4.793a1 1 0 010 1.414l-5 5a1 1 0 01-1.414 0l-2-2a1 1 0 011.414-1.414L6.5 9.086l4.293-4.293a1 1 0 011.414 0z'/%3e%3c/svg%3e") !important;
}
/* ------------------------------------------- */
/* Your existing custom styling */
.gradio-container, .gradio-container * { font-size: calc(1em + 0px) !important; }
body, .gradio-container { padding-top: 0 !important; margin-top: 0 !important; }
.main { padding-top: 0 !important; }
.contain { padding-top: 0 !important; }
textarea, input[type="text"], input[type="number"], select,
.gr-box, .gr-group, .gr-panel, .block.gr-group, .gr-input,
.gr-text-input, .input-text, .border, div[class*="border"] {
border-width: 2px !important;
border-style: solid !important;
border-color: #aaaaaa !important;
}
textarea:focus, input:focus, select:focus {
border-color: #4A90D9 !important;
border-width: 2.5px !important;
}
button { border-width: 2px !important; border-style: solid !important; }
.msg-ai {
background: #f5f5f5;
margin-right: 18%;
line-height: 1.5;
white-space: pre-wrap;
color: #000000 !important;
}
.chat-box { padding-left: 0px; padding-top: 0px; padding-right: 0px; }
/* Corpus status box styling */
#corpus_status, .corpus-status-box {
min-height: 60px !important;
padding: 8px !important;
margin: 8px 0 !important;
font-size: 1.05em !important;
line-height: 1.4 !important;
background-color: #ffffff !important;
color: #000000 !important;
}
/* Force Privacy Settings panel to white background */
#privacy_settings_panel,
.dark #privacy_settings_panel,
#privacy_settings_panel > div,
.dark #privacy_settings_panel > div {
background: #ffffff !important;
background-color: #ffffff !important;
color: #000000 !important;
}
"""
js = """
"""
# Initialize LLM clients
initialize_llm_clients()
# Load retriever from pickle (generic corpus)
# retriever_path = args.retriever_path or RETRIEVAL_PICKLE_PATH
# load_retriever(retriever_path)
load_retriever(args.retriever_path or DEFAULT_RETRIEVAL_PICKLE_PATH)
for _scenario_key in SCENARIO_RETRIEVER_PATHS:
get_scenario_retriever(_scenario_key) # loads and caches each one
app = create_ui(
model=args.model if args.model in [name for _, _, name in MODEL_CONFIGS] else DEFAULT_MODEL_NAME,
rag=args.rag,
epsilon=args.epsilon if str(args.epsilon) == "inf" or float(args.epsilon) > 0 else "inf",
show_risk=args.show_risk,
show_tips=args.show_tips,
show_rag_highlights=args.show_rag_highlights,
show_pii_highlights=args.show_pii_highlights,
show_settings=args.show_settings,
demo=args.demo if str(args.demo) in ["0", "1"] else "0",
enable_social_scraping=args.enable_social_scraping ,
show_social_scraping=args.show_social_scraping,
show_upload_data = args.show_upload_data,
scenario_mode=args.scenario_mode,
show_dp=args.show_dp,
show_infr_attr_card=args.show_infr_attr_card
)
app.queue(max_size=600, default_concurrency_limit=20) # required for concurrent multi-user support
app.launch(
css=css,
head=js,
server_name="0.0.0.0", # Local only - change to "0.0.0.0" for network access
server_port=int(args.port), # Default 7860
share=False, # Keep this as False for local
show_error=True,
ssr_mode=False
)
# Hi! I am Catherine from Jamesfort. I'm looking for inspiration from people who've had to adapt to physical challenges while keeping their career going. I work in a fast-paced environment where mistakes aren't really an option, and sometimes I feel like I'm fighting an uphill battle. Anyone else dealing with similar situations? How do you stay motivated?
# I am Raymond Phillips. I live in the Philippines. I am searching for new job. Do you have any suggestions that would work for me?{str(e)}