diff --git a/generate/graph_planner.py b/generate/graph_planner.py new file mode 100644 index 00000000..a701cf77 --- /dev/null +++ b/generate/graph_planner.py @@ -0,0 +1,237 @@ +"""Graph planner — converts a PropositionGraph into an ArticulationTarget. + +The planner walks the graph in topological order and emits an ordered +sequence of articulation steps that the downstream generation pipeline +can execute. Each step carries the proposition node ID, the rhetorical +move, and any constraints inherited from intent classification. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from enum import Enum, unique + +from generate.intent import DialogueIntent, IntentTag + + +@unique +class Relation(Enum): + ELABORATION = "elaboration" + CAUSE = "cause" + CONTRAST = "contrast" + SEQUENCE = "sequence" + CORRECTION = "correction" + + +@unique +class RhetoricalMove(Enum): + ASSERT = "assert" + ELABORATE = "elaborate" + CONTRAST = "contrast" + SEQUENCE = "sequence" + CORRECT = "correct" + + +@dataclass(frozen=True, slots=True) +class GraphEdge: + source: str + target: str + relation: Relation + + def as_dict(self) -> dict[str, str]: + return { + "source": self.source, + "target": self.target, + "relation": self.relation.value, + } + + +@dataclass(frozen=True, slots=True) +class GraphNode: + node_id: str + subject: str + predicate: str + obj: str + source_intent: IntentTag + + def as_dict(self) -> dict[str, str]: + return { + "node_id": self.node_id, + "subject": self.subject, + "predicate": self.predicate, + "object": self.obj, + "source_intent": self.source_intent.value, + } + + +@dataclass(frozen=True, slots=True) +class PropositionGraph: + nodes: tuple[GraphNode, ...] = () + edges: tuple[GraphEdge, ...] = () + + def add_node(self, node: GraphNode) -> PropositionGraph: + return PropositionGraph(nodes=(*self.nodes, node), edges=self.edges) + + def add_edge(self, edge: GraphEdge) -> PropositionGraph: + return PropositionGraph(nodes=self.nodes, edges=(*self.edges, edge)) + + def roots(self) -> tuple[str, ...]: + targets = frozenset(e.target for e in self.edges) + return tuple(n.node_id for n in self.nodes if n.node_id not in targets) + + def topo_order(self) -> tuple[str, ...]: + in_degree: dict[str, int] = {n.node_id: 0 for n in self.nodes} + for e in self.edges: + in_degree[e.target] = in_degree.get(e.target, 0) + 1 + queue = sorted(nid for nid, deg in in_degree.items() if deg == 0) + order: list[str] = [] + while queue: + nid = queue.pop(0) + order.append(nid) + for e in self.edges: + if e.source == nid: + in_degree[e.target] -= 1 + if in_degree[e.target] == 0: + queue.append(e.target) + return tuple(order) + + def as_dict(self) -> dict[str, object]: + return { + "nodes": tuple(n.as_dict() for n in self.nodes), + "edges": tuple(e.as_dict() for e in self.edges), + } + + def to_json(self) -> str: + import json + return json.dumps(self.as_dict(), sort_keys=True) + + +@dataclass(frozen=True, slots=True) +class ArticulationStep: + node_id: str + move: RhetoricalMove + predicate: str + subject: str + + def as_dict(self) -> dict[str, str]: + return { + "node_id": self.node_id, + "move": self.move.value, + "predicate": self.predicate, + "subject": self.subject, + } + + +@dataclass(frozen=True, slots=True) +class ArticulationTarget: + steps: tuple[ArticulationStep, ...] + source_intent: IntentTag + + def as_dict(self) -> dict[str, object]: + return { + "steps": tuple(s.as_dict() for s in self.steps), + "source_intent": self.source_intent.value, + } + + +_RELATION_TO_MOVE: dict[Relation, RhetoricalMove] = { + Relation.ELABORATION: RhetoricalMove.ELABORATE, + Relation.CAUSE: RhetoricalMove.ELABORATE, + Relation.CONTRAST: RhetoricalMove.CONTRAST, + Relation.SEQUENCE: RhetoricalMove.SEQUENCE, + Relation.CORRECTION: RhetoricalMove.CORRECT, +} + + +_INTENT_PREDICATES: dict[IntentTag, str] = { + IntentTag.DEFINITION: "is_defined_as", + IntentTag.CAUSE: "is_caused_by", + IntentTag.PROCEDURE: "has_steps", + IntentTag.COMPARISON: "contrasts_with", + IntentTag.CORRECTION: "corrects", + IntentTag.RECALL: "recalls", + IntentTag.VERIFICATION: "is_verified_as", +} + + +def graph_from_intent( + intent: DialogueIntent, + *, + prior_node_id: str | None = None, +) -> PropositionGraph: + """Build a minimal proposition graph from a classified intent.""" + predicate = _INTENT_PREDICATES.get(intent.tag, "addresses") + graph = PropositionGraph() + + if intent.tag is IntentTag.COMPARISON: + left = GraphNode( + node_id="p0", + subject=intent.subject, + predicate=predicate, + obj=intent.secondary_subject or "", + source_intent=intent.tag, + ) + right = GraphNode( + node_id="p1", + subject=intent.secondary_subject or "", + predicate=predicate, + obj=intent.subject, + source_intent=intent.tag, + ) + edge = GraphEdge(source="p0", target="p1", relation=Relation.CONTRAST) + return graph.add_node(left).add_node(right).add_edge(edge) + + if intent.tag is IntentTag.CORRECTION: + root = GraphNode( + node_id="p0", + subject=intent.subject, + predicate=predicate, + obj=prior_node_id or "", + source_intent=intent.tag, + ) + graph = graph.add_node(root) + if prior_node_id is not None: + graph = graph.add_edge( + GraphEdge(source="p0", target=prior_node_id, relation=Relation.CORRECTION) + ) + return graph + + root = GraphNode( + node_id="p0", + subject=intent.subject, + predicate=predicate, + obj="", + source_intent=intent.tag, + ) + return graph.add_node(root) + + +def plan_articulation(graph: PropositionGraph) -> ArticulationTarget: + """Walk *graph* in topological order and emit an articulation target.""" + node_map = {n.node_id: n for n in graph.nodes} + incoming: dict[str, Relation | None] = {n.node_id: None for n in graph.nodes} + for edge in graph.edges: + if edge.target in incoming: + incoming[edge.target] = edge.relation + + source_intent = IntentTag.UNKNOWN + if graph.nodes: + source_intent = graph.nodes[0].source_intent + + steps: list[ArticulationStep] = [] + for node_id in graph.topo_order(): + node = node_map.get(node_id) + if node is None: + continue + relation = incoming.get(node_id) + move = _RELATION_TO_MOVE.get(relation, RhetoricalMove.ASSERT) if relation is not None else RhetoricalMove.ASSERT + steps.append( + ArticulationStep( + node_id=node_id, + move=move, + predicate=node.predicate, + subject=node.subject, + ) + ) + + return ArticulationTarget(steps=tuple(steps), source_intent=source_intent) diff --git a/generate/intent.py b/generate/intent.py new file mode 100644 index 00000000..bff8321c --- /dev/null +++ b/generate/intent.py @@ -0,0 +1,73 @@ +"""Dialogue intent classification. + +Maps a raw prompt string to a typed intent tag. The classifier is rule-based +(prefix/pattern matching) — no ML dependency. Downstream, the intent selects +the proposition frame family and graph shape before generation begins. +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass +from enum import Enum, unique + + +@unique +class IntentTag(Enum): + DEFINITION = "definition" + CAUSE = "cause" + PROCEDURE = "procedure" + COMPARISON = "comparison" + CORRECTION = "correction" + RECALL = "recall" + VERIFICATION = "verification" + UNKNOWN = "unknown" + + +@dataclass(frozen=True, slots=True) +class DialogueIntent: + tag: IntentTag + subject: str + secondary_subject: str | None = None + + def requires_prior_turn(self) -> bool: + return self.tag is IntentTag.CORRECTION + + +_COMPARE_RE = re.compile( + r"^compare\s+(.+?)\s+(?:and|vs\.?|versus|with)\s+(.+)", + re.IGNORECASE, +) + +_RULES: tuple[tuple[re.Pattern[str], IntentTag], ...] = ( + (re.compile(r"^what\s+(?:is|are)\s+", re.IGNORECASE), IntentTag.DEFINITION), + (re.compile(r"^why\s+", re.IGNORECASE), IntentTag.CAUSE), + (re.compile(r"^how\s+(?:do|can|should|would)\s+(?:I|we|you)\s+", re.IGNORECASE), IntentTag.PROCEDURE), + (re.compile(r"^(?:is|are|does|do|can|could|would|should|was|were|has|have|will)\s+.+\??\s*$", re.IGNORECASE), IntentTag.VERIFICATION), + (re.compile(r"^(?:no|that'?s\s+(?:not|wrong)|incorrect|actually|correction)", re.IGNORECASE), IntentTag.CORRECTION), + (re.compile(r"^remember\s+", re.IGNORECASE), IntentTag.RECALL), +) + + +def classify_intent(prompt: str) -> DialogueIntent: + text = prompt.strip() + if not text: + return DialogueIntent(tag=IntentTag.UNKNOWN, subject="") + + compare_match = _COMPARE_RE.match(text) + if compare_match: + return DialogueIntent( + tag=IntentTag.COMPARISON, + subject=compare_match.group(1).strip(), + secondary_subject=compare_match.group(2).strip(), + ) + + for pattern, tag in _RULES: + match = pattern.match(text) + if match: + subject = text[match.end():].rstrip("?").strip() + if not subject: + subject = text + return DialogueIntent(tag=tag, subject=subject) + + return DialogueIntent(tag=IntentTag.UNKNOWN, subject=text) diff --git a/tests/test_intent_proposition_graph.py b/tests/test_intent_proposition_graph.py new file mode 100644 index 00000000..a37e878a --- /dev/null +++ b/tests/test_intent_proposition_graph.py @@ -0,0 +1,87 @@ +"""Tests for the intent -> proposition graph -> articulation target pipeline.""" + +from __future__ import annotations + +import json + +from generate.graph_planner import ( + Relation, + RhetoricalMove, + graph_from_intent, + plan_articulation, +) +from generate.intent import IntentTag, classify_intent + + +def test_what_is_definition_intent() -> None: + intent = classify_intent("What is a multivector?") + + assert intent.tag is IntentTag.DEFINITION + assert "multivector" in intent.subject.lower() + + graph = graph_from_intent(intent) + assert len(graph.nodes) == 1 + assert graph.nodes[0].predicate == "is_defined_as" + assert graph.nodes[0].source_intent is IntentTag.DEFINITION + + +def test_why_cause_intent() -> None: + intent = classify_intent("Why does the field diverge?") + + assert intent.tag is IntentTag.CAUSE + assert "field" in intent.subject.lower() + + graph = graph_from_intent(intent) + assert len(graph.nodes) == 1 + assert graph.nodes[0].predicate == "is_caused_by" + + +def test_compare_intent() -> None: + intent = classify_intent("Compare MLX and PyTorch") + + assert intent.tag is IntentTag.COMPARISON + assert intent.subject.lower() == "mlx" + assert intent.secondary_subject is not None + assert intent.secondary_subject.lower() == "pytorch" + + graph = graph_from_intent(intent) + assert len(graph.nodes) == 2 + assert len(graph.edges) == 1 + assert graph.edges[0].relation is Relation.CONTRAST + + target = plan_articulation(graph) + assert target.source_intent is IntentTag.COMPARISON + moves = [s.move for s in target.steps] + assert RhetoricalMove.CONTRAST in moves + + +def test_correction_intent_links_prior_turn() -> None: + intent = classify_intent("No, that's wrong — it should be grade 2") + + assert intent.tag is IntentTag.CORRECTION + assert intent.requires_prior_turn() + + prior_id = "prev_p0" + graph = graph_from_intent(intent, prior_node_id=prior_id) + assert len(graph.nodes) == 1 + assert graph.nodes[0].predicate == "corrects" + assert graph.nodes[0].obj == prior_id + + assert len(graph.edges) == 1 + assert graph.edges[0].relation is Relation.CORRECTION + assert graph.edges[0].target == prior_id + + +def test_graph_serialization_is_deterministic() -> None: + intent = classify_intent("Compare cats and dogs") + graph = graph_from_intent(intent) + + json_a = graph.to_json() + json_b = graph.to_json() + assert json_a == json_b + + parsed = json.loads(json_a) + assert "nodes" in parsed + assert "edges" in parsed + assert len(parsed["nodes"]) == 2 + assert len(parsed["edges"]) == 1