"""
Comparia Dataset Explorer - Gradio UI
Usage:
python app.py
"""
import base64
import json
from pathlib import Path
import gradio as gr
from gradio.themes.base import Base
from gradio.themes.utils import colors, fonts, sizes
import plotly.graph_objects as go
from config import SHOW_CONVERSATIONS
from search import parse_conversation
from topic_map import build_topic_map, get_highlighted_figure, get_topic_map_record, get_topic_map_stats, search_topic_map
from votes_reactions import (
enrich_message_with_reactions,
format_quality_flags,
format_vote_banner,
get_reactions,
get_vote,
load_votes_reactions,
)
# ---- Theme ----
_accent = colors.Color(
c50="#f0f0fe",
c100="#e0e0fd",
c200="#aeaef9",
c300="#9494f7",
c400="#7f7ff5",
c500="#6a6af4",
c600="#5555d4",
c700="#4040b4",
c800="#2e2e8a",
c900="#1e1e60",
c950="#121240",
name="accent",
)
_neutral = colors.Color(
c50="#fcfcfd",
c100="#f5f5f6",
c200="#ebebed",
c300="#d6d6da",
c400="#b0b0b8",
c500="#8a8a96",
c600="#666674",
c700="#4d4d58",
c800="#35353d",
c900="#1e1e24",
c950="#12121a",
name="neutral",
)
class CompariaTheme(Base):
def __init__(self):
super().__init__(
primary_hue=_accent,
secondary_hue=_accent,
neutral_hue=_neutral,
font=("Marianne", "system-ui", "sans-serif"),
font_mono=("Marianne", "ui-monospace", "monospace"),
)
self.set(
body_background_fill="#fcfcfd",
body_background_fill_dark="#1e1e24",
button_primary_background_fill="#6a6af4",
button_primary_background_fill_hover="#5555d4",
button_primary_text_color="white",
)
# ---- Helpers ----
def to_chatbot_messages(
messages: list[dict],
reactions: list[dict] | None = None,
model_name: str = "",
) -> list[dict]:
"""Convert parsed conversation to Gradio Chatbot messages format.
If reactions are provided, enriches assistant messages with inline reaction footers.
"""
chatbot_msgs = []
assistant_idx = 0
for msg in messages:
role = msg.get("role", "user")
content = msg.get("content", "")
if role in ("user", "human"):
chatbot_msgs.append({"role": "user", "content": content})
else:
if reactions and model_name:
content = enrich_message_with_reactions(
content, reactions, model_name, assistant_idx,
)
chatbot_msgs.append({"role": "assistant", "content": content})
assistant_idx += 1
return chatbot_msgs
def display_conversation(record: dict) -> tuple:
"""Extract and format conversation data from a record.
Returns (conv_a_msgs, conv_b_msgs, model_a_label, model_b_label, info_text).
"""
model_a_name = record.get("model_a_name", "Unknown")
model_b_name = record.get("model_b_name", "Unknown")
conv_id = record.get("id", "")
# Look up vote and reactions
vote = get_vote(conv_id)
reactions = get_reactions(conv_id)
keywords = []
try:
keywords = json.loads(record.get("keywords_str", "[]"))
except (json.JSONDecodeError, TypeError):
pass
# Build info text with vote banner
vote_banner = format_vote_banner(vote, model_a_name, model_b_name)
if SHOW_CONVERSATIONS:
info_parts = [
f"**Turns:** {record.get('conv_turns', '?')}",
f"**Language:** {record.get('primary_language', '?')}",
]
if keywords:
info_parts.append(f"**Keywords:** {', '.join(keywords[:5])}")
info_text = " | ".join(info_parts)
if vote_banner:
info_text = vote_banner + "\n\n" + info_text
else:
info_text = vote_banner or ""
# Build model labels with quality flags
label_a = f"**Model A: {model_a_name}**"
flags_a = format_quality_flags(vote, "a")
if flags_a:
label_a += f"\n\n{flags_a}"
label_b = f"**Model B: {model_b_name}**"
flags_b = format_quality_flags(vote, "b")
if flags_b:
label_b += f"\n\n{flags_b}"
if SHOW_CONVERSATIONS:
conv_a = parse_conversation(record.get("conversation_a_json", "[]"))
conv_b = parse_conversation(record.get("conversation_b_json", "[]"))
msgs_a = to_chatbot_messages(conv_a, reactions, model_a_name)
msgs_b = to_chatbot_messages(conv_b, reactions, model_b_name)
else:
msgs_a = []
msgs_b = []
return (msgs_a, msgs_b, label_a, label_b, info_text)
# ---- App ----
def create_app():
# Build topic map at startup (loads from pre-computed cache, zero computation)
print("Loading topic map from cache...")
topic_fig = build_topic_map()
n_display, n_total = get_topic_map_stats()
print("Topic map ready.")
# Load votes and reactions data (only needed when showing conversations)
if SHOW_CONVERSATIONS:
print("Loading votes and reactions...")
load_votes_reactions()
# Head HTML: load Marianne font + Plotly click bridge JS
head_html = """
"""
with gr.Blocks(title="Compar:IA Dataset Explorer") as app:
_logo_b64 = base64.b64encode(Path("data/comparia-logo.png").read_bytes()).decode()
gr.HTML(
f''
)
_intro_action = (
"or click any point to read the full conversation."
if SHOW_CONVERSATIONS
else "or click any point to see its details."
)
_count_text = (
f"The map below shows {n_display:,} conversations"
+ (f" (sampled from {n_total:,} total)" if n_display < n_total else "")
)
gr.Markdown(
"[Compar:IA](https://comparia.beta.gouv.fr/) is an open-source LLM arena "
"created by the French government, where users blindly compare AI models "
"in real conversations. The resulting data is continuously published as three "
"open-source datasets "
"([conversations](https://huggingface.co/datasets/ministere-culture/comparia-conversations), "
"[votes](https://huggingface.co/datasets/ministere-culture/comparia-votes), "
"[reactions](https://huggingface.co/datasets/ministere-culture/comparia-reactions)) "
"available for research and model training.\n\n"
f"{_count_text}, positioned by semantic similarity "
f"and colored by topic. Search to highlight matching points, "
f"{_intro_action}\n\n"
"Contact: [contact@comparia.beta.gouv.fr](mailto:contact@comparia.beta.gouv.fr)"
)
_n_total_str = f"{n_total:,}"
_methodology_steps = (
f"**1. Data collection.** {_n_total_str} conversations are randomly sampled from the full "
"[comparia-conversations](https://huggingface.co/datasets/ministere-culture/comparia-conversations) "
"dataset using reservoir sampling, ensuring uniform coverage across the entire dataset. "
"Each conversation is a blind A/B comparison between two LLMs on "
"[Compar:IA](https://comparia.beta.gouv.fr/).\n\n"
"**2. Embedding.** Each conversation is summarized into a single text (combining the "
"summary, keywords, and opening message), then embedded into a 384-dimensional vector "
"using [multilingual-e5-small](https://huggingface.co/intfloat/multilingual-e5-small). "
"Vectors are stored in [LanceDB](https://lancedb.com/) for fast retrieval.\n\n"
"**3. Clustering.** K-Means groups the embeddings into topic clusters. "
"TF-IDF extracts the most distinctive terms per cluster, and an LLM "
"([Mistral Small](https://mistral.ai/news/mistral-small-3-1)) generates a short human-readable "
"label for each topic.\n\n"
"**4. Projection.** A two-stage [UMAP](https://umap-learn.readthedocs.io/) projection reduces "
"the dimensions to 2D for visualization (preserving structure), "
"giving cleaner cluster separation than a single-pass projection.\n\n"
"**5. Search.** Queries are embedded with the same model and matched against stored vectors "
"using cosine similarity. Results are highlighted on the map with size and opacity proportional "
"to relevance."
)
if SHOW_CONVERSATIONS:
_methodology_steps += (
"\n\n**6. Votes & reactions.** Two companion datasets record user preferences: "
"[votes](https://huggingface.co/datasets/ministere-culture/comparia-votes) (which model was "
"preferred after a full conversation) and "
"[reactions](https://huggingface.co/datasets/ministere-culture/comparia-reactions) (per-message "
"thumbs up/down and quality tags). These are shown alongside each conversation."
)
with gr.Accordion("How this works", open=False):
gr.Markdown(_methodology_steps)
with gr.Row():
topic_search_input = gr.Textbox(
label="Search in map",
placeholder="Type to highlight matching conversations...",
scale=5,
)
topic_search_btn = gr.Button("Search", variant="primary", scale=1)
topic_match_count = gr.Markdown("")
topic_plot = gr.Plot(value=topic_fig, elem_id="topic-plot")
# Hidden textbox to receive clicked point index from JS
# Note: use visible=True + CSS hide so elem_id is rendered in DOM
clicked_point_idx = gr.Textbox(elem_id="clicked-point-idx")
# CSS to hide the bridge textbox
gr.HTML("")
topic_conv_info = gr.Markdown("")
with gr.Row(visible=SHOW_CONVERSATIONS):
with gr.Column():
topic_conv_a_label = gr.Markdown("**Model A**")
topic_conv_a_display = gr.Chatbot(label="Model A", height=400)
with gr.Column():
topic_conv_b_label = gr.Markdown("**Model B**")
topic_conv_b_display = gr.Chatbot(label="Model B", height=400)
def do_topic_search(query):
fig, count_text = search_topic_map(query)
return (
fig, count_text,
[], [], "**Model A**", "**Model B**", "",
)
topic_search_outputs = [
topic_plot, topic_match_count,
topic_conv_a_display, topic_conv_b_display,
topic_conv_a_label, topic_conv_b_label, topic_conv_info,
]
topic_search_btn.click(
fn=do_topic_search,
inputs=[topic_search_input],
outputs=topic_search_outputs,
)
topic_search_input.submit(
fn=do_topic_search,
inputs=[topic_search_input],
outputs=topic_search_outputs,
)
# Shared outputs for selection handlers (figure + conversation)
selection_outputs = [
topic_plot,
topic_conv_a_display, topic_conv_b_display,
topic_conv_a_label, topic_conv_b_label, topic_conv_info,
]
empty_selection = (
go.Figure(), [], [], "**Model A**", "**Model B**", "",
)
# Click a point on the Plotly chart (via JS bridge)
def on_point_click(idx_str):
if not idx_str:
return empty_selection
try:
idx = int(idx_str)
record = get_topic_map_record(idx)
except (ValueError, TypeError):
return empty_selection
if record is None:
return empty_selection
fig = get_highlighted_figure(idx)
return (fig,) + display_conversation(record)
clicked_point_idx.change(
fn=on_point_click,
inputs=[clicked_point_idx],
outputs=selection_outputs,
)
gr.HTML(
''
)
return app, head_html
if __name__ == "__main__":
app, head = create_app()
app.launch(theme=CompariaTheme(), head=head, allowed_paths=["data"])
'
f'