"""Helpers to serialize graph-derived evidence into a consistent payload."""
import json
from typing import Any, Dict
from lalandre_rag.scoring import (
build_evidence_score_payload,
build_relevance_score_payload,
clamp_unit_interval,
)
[docs]
def build_graph_node_source_item(
*,
node: Dict[str, Any],
source_id: str,
sequence_order: int,
) -> Dict[str, Any]:
"""Serialize a ranked graph node into a user-facing evidence item."""
act_id = int(node["id"])
celex = str(node.get("celex", f"ACT-{act_id}") or f"ACT-{act_id}")
title = str(node.get("title", "Unknown") or "Unknown")
normalized_score = clamp_unit_interval(node.get("_rank_score", 0.0))
return {
"source_id": source_id,
"source_kind": "graph_node",
"act_id": act_id,
"celex": celex,
"title": title,
"subdivision_id": act_id,
"subdivision_type": "GRAPH_ACT",
"sequence_order": sequence_order,
"content_used": f"{title} ({celex})\nPertinence graphe: {normalized_score:.2f}",
"content_preview": f"{title} ({celex})",
"content_truncated": False,
"trace": node.get("_rank_trace"),
**build_relevance_score_payload(
normalized_score,
rank_score=node.get("_rank_score"),
),
}
[docs]
def build_graph_edge_source_item(
*,
relationship: Dict[str, Any],
source_id: str,
sequence_order: int,
start_celex: str,
end_celex: str,
) -> Dict[str, Any]:
"""Serialize a ranked graph relationship into a user-facing evidence item."""
relation_type = str(relationship.get("type", "RELATED_TO"))
source_act_id = relationship.get("start_node")
target_act_id = relationship.get("end_node")
relation_excerpt = f"{start_celex} -[{relation_type}]-> {end_celex}"
description = str(relationship.get("description", "") or "").strip()
normalized_score = clamp_unit_interval(relationship.get("_rel_weight", 0.0))
return {
"source_id": source_id,
"source_kind": "graph_edge",
"title": relation_excerpt,
"subdivision_id": sequence_order,
"subdivision_type": "GRAPH_RELATION",
"sequence_order": sequence_order,
"relation_type": relation_type,
"start_act_id": int(source_act_id) if source_act_id is not None else None,
"end_act_id": int(target_act_id) if target_act_id is not None else None,
"start_celex": start_celex,
"end_celex": end_celex,
"content_used": f"{relation_excerpt}{f' | {description}' if description else ''}",
"content_preview": relation_excerpt,
"content_truncated": False,
"trace": relationship.get("_rel_trace"),
**build_relevance_score_payload(
normalized_score,
rank_score=relationship.get("_rel_weight"),
),
}
[docs]
def build_cypher_row_source_item(
*,
row: Dict[str, Any],
row_index: int,
include_full_content: bool,
content_preview_chars: int,
query_id: str,
graph_query_strategy: str,
generated_cypher: str | None,
) -> Dict[str, Any]:
"""Serialize a Cypher row into a concrete evidence item without fake scoring."""
row_serialized = json.dumps(row, ensure_ascii=False, default=str)
return {
"source_id": f"C{row_index}",
"source_kind": "cypher_row",
"title": f"Résultat Cypher {row_index}",
"subdivision_id": row_index,
"subdivision_type": "GRAPH_ROW",
"sequence_order": row_index,
"content": row_serialized if include_full_content else "",
"content_used": row_serialized,
"content_preview": row_serialized[:content_preview_chars],
"content_truncated": False,
"trace": {
"query_id": query_id,
"search_method": "nl_to_cypher",
"graph_query_strategy": graph_query_strategy,
"generated_cypher": generated_cypher,
"row_index": row_index,
"collection": "neo4j_graph",
},
**build_evidence_score_payload(),
}