defmodule GEPA.Visualization do
@moduledoc """
Candidate-lineage visualization helpers.
This module ports upstream GEPA's candidate tree utilities to Elixir. It can
generate either Graphviz DOT or a self-contained HTML page from raw data,
`GEPA.State`, or `GEPA.Result`.
"""
alias GEPA.{Result, State}
alias GEPA.Utils.Pareto
@doc "Generate Graphviz DOT from a `GEPA.State` or `GEPA.Result`."
@spec candidate_tree_dot(State.t() | Result.t()) :: String.t()
def candidate_tree_dot(%State{} = state) do
candidate_tree_dot_from_data(
state.program_candidates,
state.parent_program_for_candidate,
aggregate_scores_from_state(state),
state.program_at_pareto_front_valset
)
end
def candidate_tree_dot(%Result{} = result) do
candidate_tree_dot_from_data(
result.candidates,
result.parents,
result.val_aggregate_scores,
result.per_val_instance_best_candidates || %{}
)
end
@doc "Generate a self-contained HTML visualization from a `GEPA.State` or `GEPA.Result`."
@spec candidate_tree_html(State.t() | Result.t()) :: String.t()
def candidate_tree_html(%State{} = state) do
candidate_tree_html_from_data(
state.program_candidates,
state.parent_program_for_candidate,
aggregate_scores_from_state(state),
state.program_at_pareto_front_valset
)
end
def candidate_tree_html(%Result{} = result) do
candidate_tree_html_from_data(
result.candidates,
result.parents,
result.val_aggregate_scores,
result.per_val_instance_best_candidates || %{}
)
end
@doc "Generate Graphviz DOT from raw optimization data."
@spec candidate_tree_dot_from_data([map()], list() | map(), [number()], map()) :: String.t()
def candidate_tree_dot_from_data(candidates, parents, val_scores, pareto_front_programs)
when is_list(candidates) and is_list(val_scores) do
n = length(candidates)
best_idx = best_index(val_scores)
dominator_ids = dominator_ids(pareto_front_programs, val_scores)
node_lines =
candidates
|> Enum.with_index()
|> Enum.map(fn {_candidate, idx} ->
score = Enum.at(val_scores, idx, 0.0)
role = node_role(idx, best_idx, dominator_ids)
color = node_color(role)
label = "#{idx}\\n(#{format_score(score, 2)})"
" #{idx} [label=\"#{escape_dot(label)}\", fillcolor=#{color}, tooltip=\" \"] ;"
end)
edge_lines =
0..max(n - 1, 0)//1
|> Enum.flat_map(fn child ->
parents_for(parents, child)
|> Enum.reject(&is_nil/1)
|> Enum.map(fn parent -> " #{parent} -> #{child};" end)
end)
[
"digraph G {",
" rankdir=TB;",
" node [style=filled, shape=circle, fontsize=14, width=0.6, height=0.6];"
| node_lines ++ edge_lines ++ ["}"]
]
|> Enum.join("\n")
end
@html_template """
GEPA Candidate Tree
"""
@doc "Generate a self-contained HTML page from raw optimization data."
@spec candidate_tree_html_from_data([map()], list() | map(), [number()], map()) :: String.t()
def candidate_tree_html_from_data(candidates, parents, val_scores, pareto_front_programs) do
best_idx = best_index(val_scores)
dominator_ids = dominator_ids(pareto_front_programs, val_scores)
nodes =
candidates
|> Enum.with_index()
|> Enum.map(fn {candidate, idx} ->
score = Enum.at(val_scores, idx, 0.0)
parent_ids = parents_for(parents, idx)
role = node_role(idx, best_idx, dominator_ids)
%{
idx: idx,
score: Float.round(score * 1.0, 4),
parents: parent_label(parent_ids),
role: role_label(role),
components: sorted_candidate(candidate)
}
end)
dot = candidate_tree_dot_from_data(candidates, parents, val_scores, pareto_front_programs)
nodes_json = Jason.encode!(nodes)
@html_template
|> String.replace("__DOT_STRING__", js_template_escape(dot))
|> String.replace("__NODES_JSON__", nodes_json)
end
defp aggregate_scores_from_state(%State{} = state) do
Enum.map(state.prog_candidate_val_subscores, fn scores ->
if map_size(scores) == 0 do
0.0
else
Enum.sum(Map.values(scores)) / map_size(scores)
end
end)
end
defp best_index([]), do: 0
defp best_index(scores) do
scores
|> Enum.with_index()
|> Enum.max_by(fn {score, _idx} -> score end, fn -> {0.0, 0} end)
|> elem(1)
end
defp dominator_ids(fronts, val_scores) when is_map(fronts) do
scores = val_scores |> Enum.with_index() |> Map.new(fn {score, idx} -> {idx, score} end)
ids = fronts |> normalize_fronts() |> Pareto.find_dominator_programs(scores)
Map.new(ids, &{&1, true})
end
defp dominator_ids(_fronts, _val_scores), do: %{}
defp normalize_fronts(fronts) do
Map.new(fronts, fn {key, value} ->
set =
cond do
is_struct(value, MapSet) -> value
is_list(value) -> MapSet.new(value)
true -> MapSet.new()
end
{key, set}
end)
end
defp parents_for(parents, idx) when is_map(parents), do: Map.get(parents, idx, [])
defp parents_for(parents, idx) when is_list(parents), do: Enum.at(parents, idx, [])
defp parents_for(_parents, _idx), do: []
defp node_role(idx, best_idx, dominator_ids) do
cond do
idx == best_idx -> :best
Map.has_key?(dominator_ids, idx) -> :pareto
idx == 0 -> :seed
true -> :other
end
end
defp node_color(:best), do: "cyan"
defp node_color(:pareto), do: "orange"
defp node_color(_role), do: "lightgray"
defp role_label(:best), do: "Best"
defp role_label(:pareto), do: "Pareto Front"
defp role_label(:seed), do: "Seed"
defp role_label(_role), do: ""
defp parent_label(parents) do
parents = Enum.reject(List.wrap(parents), &is_nil/1)
case parents do
[] -> "seed"
values -> Enum.map_join(values, ", ", &to_string/1)
end
end
defp sorted_candidate(candidate) when is_map(candidate) do
candidate
|> Enum.map(fn {key, value} -> {to_string(key), value} end)
|> Enum.sort_by(&elem(&1, 0))
|> Map.new()
end
defp sorted_candidate(candidate), do: %{"candidate" => inspect(candidate)}
defp format_score(score, decimals) when is_number(score) do
score |> Kernel.*(1.0) |> Float.round(decimals) |> :erlang.float_to_binary(decimals: decimals)
end
defp format_score(score, _decimals), do: to_string(score)
defp escape_dot(text) do
text
|> to_string()
|> String.replace("\\", "\\\\")
|> String.replace("\"", "\\\"")
|> String.replace("\n", "\\n")
end
defp js_template_escape(text) do
text
|> String.replace("\\", "\\\\")
|> String.replace("`", "\\`")
|> String.replace("${", "\\${")
end
end