from __future__ import annotations
from typing import Any
def _format_path_chain(path: Any) -> str:
"""Format a ScoredPathResponse into a readable chain string.
Each triplet becomes: [src] -> predicate -> [tgt]: context [Time: ...]
Multiple triplets are joined with ↳ to show traversal direction.
"""
triplets = getattr(path, "triplets", None) or []
parts = []
for t in triplets:
src = t.source.name if t.source else ""
tgt = t.target.name if t.target else ""
rel = t.relation
pred = rel.canonical_predicate if rel else ""
line = f"[{src}] -> {pred} -> [{tgt}]"
ctx = rel.context if rel else None
if ctx:
line += f": {ctx}"
temporal = rel.temporal_details if rel else None
if temporal:
line += f" [Time: {temporal}]"
parts.append(line)
return "\n ↳ ".join(parts)
def build_context_string(result) -> str:
"""
Build a formatted context string from a HydraDB full_recall response.
Combines entity paths, chunk content, graph relations, and extra context
into a single readable string ready to pass into an LLM prompt.
"""
lines: list[str] = []
gc = result.graph_context
# --- Entity paths -----------------------------------------------------------
query_paths = gc.query_paths if gc else []
if query_paths:
lines.append("=== ENTITY PATHS ===")
for path in query_paths:
lines.append(_format_path_chain(path))
lines.append("")
# --- Chunks -----------------------------------------------------------------
chunks = result.chunks or []
additional_context = result.additional_context or {}
chunk_id_to_group_ids = gc.chunk_id_to_group_ids if gc else {}
chunk_relations = gc.chunk_relations if gc else []
if chunks:
lines.append("=== CONTEXT ===")
for i, chunk in enumerate(chunks):
lines.append(f"Chunk {i + 1}")
source = chunk.source_title or ""
if source:
lines.append(f"Source: {source}")
lines.append(chunk.chunk_content or "")
# Graph relations for this chunk
chunk_uuid = chunk.chunk_uuid or ""
if chunk_uuid and chunk_id_to_group_ids and chunk_relations:
group_ids = chunk_id_to_group_ids.get(chunk_uuid, [])
relevant_relations = [
r for r in chunk_relations
if r.group_id in group_ids
]
if relevant_relations:
lines.append("Graph Relations:")
for rel in relevant_relations:
for triplet in (rel.triplets or []):
src = triplet.source.name if triplet.source else ""
tgt = triplet.target.name if triplet.target else ""
pred = triplet.relation.canonical_predicate if triplet.relation else ""
ctx = triplet.relation.context if triplet.relation else ""
line = f" [{src}] -> {pred} -> [{tgt}]: {ctx}"
temporal = triplet.relation.temporal_details if triplet.relation else None
if temporal:
line += f" [Time: {temporal}]"
lines.append(line)
# Extra context
extra_ids = getattr(chunk, "extra_context_ids", None) or []
if extra_ids and additional_context:
extras = [
additional_context[eid]
for eid in extra_ids
if eid in additional_context
]
if extras:
lines.append("Extra Context:")
for extra in extras:
extra_source = extra.source_title or ""
extra_content = extra.chunk_content or ""
lines.append(f" Related Context ({extra_source}): {extra_content}")
lines.append("---")
lines.append("")
return "\n".join(lines)