File size: 3,564 Bytes
ea1618f 79a6369 655c300 79a6369 655c300 79a6369 ea1618f 79a6369 655c300 aedf4f8 79a6369 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 | """Resource registry loader and matcher for EmpathRAG Core support routing.
The filename remains `service_graph.py` for compatibility, but the paper/demo
concept is a resource registry of service objects, not a graph algorithm.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
from pathlib import Path
DEFAULT_SERVICE_GRAPH = Path("data/curated/service_graph.jsonl")
@dataclass(frozen=True)
class ServiceNode:
service_id: str
resource_name: str
description: str
urgency_level: str
safety_tiers: list[str]
route_types: list[str]
audience: list[str]
issue_types: list[str]
confidentiality_status: str
hours: str
contact_mode: list[str]
contact: str
location: str
source_url: str
source_authority: str
last_verified: str
usage_modes: list[str]
do_not_use_for: list[str]
notes: str
# Optional list of authoritative documents the user is encouraged to read
# directly (CPT form, OPT guide, RCL/Academic Difficulty policy, etc.).
# Each entry: {title, url, document_type, embeddable, requires_login}.
# Defaults empty so existing rows without the field still load.
documents: list[dict] = ()
@classmethod
def from_dict(cls, row: dict) -> "ServiceNode":
# Drop unknown keys so a future schema addition doesn't break older
# callers, and supply default for `documents` if missing.
allowed = {f.name for f in cls.__dataclass_fields__.values()}
kwargs = {k: v for k, v in row.items() if k in allowed}
kwargs.setdefault("documents", [])
return cls(**kwargs)
def as_source(self, why: str = "resource registry match") -> dict:
usage_mode = self.usage_modes[0] if self.usage_modes else "retrieval"
return {
"text": self.description,
"title": self.resource_name,
"source_name": self.resource_name,
"url": self.source_url,
"topic": ",".join(self.route_types),
"risk_level": self.urgency_level,
"usage_mode": usage_mode,
"source_type": self.source_authority,
"why_retrieved": why,
"documents": list(self.documents) if self.documents else [],
"last_verified": self.last_verified,
}
def load_service_graph(path: Path | str = DEFAULT_SERVICE_GRAPH) -> list[ServiceNode]:
graph_path = Path(path)
if not graph_path.exists():
return []
nodes: list[ServiceNode] = []
with graph_path.open("r", encoding="utf-8") as handle:
for line in handle:
if not line.strip():
continue
nodes.append(ServiceNode.from_dict(json.loads(line)))
return nodes
def match_services(
route: str,
safety_tier: str,
audience_mode: str = "student",
limit: int = 5,
path: Path | str = DEFAULT_SERVICE_GRAPH,
) -> list[ServiceNode]:
nodes = load_service_graph(path)
scored: list[tuple[int, ServiceNode]] = []
for node in nodes:
score = 0
if route in node.route_types:
score += 8
if safety_tier in node.safety_tiers:
score += 5
if audience_mode in node.audience:
score += 3
if audience_mode == "helping_friend" and "friend_peer" in node.audience:
score += 4
if score > 0 and route not in node.do_not_use_for:
scored.append((score, node))
scored.sort(key=lambda item: item[0], reverse=True)
return [node for _, node in scored[:limit]]
|