feat: add intent-proposition graph comprehension layer
Implements the dialogue understanding pipeline:
prompt -> dialogue intent -> proposition graph -> articulation target
New modules:
- generate/intent.py: rule-based classifier (7 intent tags + UNKNOWN)
- generate/graph_planner.py: immutable PropositionGraph DAG, topological
walk to ArticulationTarget with rhetorical moves
Tests cover definition, cause, comparison, correction with prior-turn
linking, and deterministic serialization.
This commit is contained in:
Shay
parent
8ed6793a03
commit
8dcc26581a
3 changed files with 397 additions and 0 deletions
237
generate/graph_planner.py
Normal file
237
generate/graph_planner.py
Normal file
|
|
@ -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 "<pending>",
|
||||
source_intent=intent.tag,
|
||||
)
|
||||
right = GraphNode(
|
||||
node_id="p1",
|
||||
subject=intent.secondary_subject or "<pending>",
|
||||
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 "<prior>",
|
||||
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="<pending>",
|
||||
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)
|
||||
73
generate/intent.py
Normal file
73
generate/intent.py
Normal file
|
|
@ -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)
|
||||
87
tests/test_intent_proposition_graph.py
Normal file
87
tests/test_intent_proposition_graph.py
Normal file
|
|
@ -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
|
||||
Loading…
Reference in a new issue