"""RLM-on-KG agent for kinship reasoning.
Implements a ReAct-style agentic loop where a GRPO-trained Gemma 4B model
navigates a kinship knowledge graph via tool calls to answer inheritance
questions.
Architecture follows RLM-on-KG (arXiv:2604.17056):
- State = (current entity, visited set, hop index)
- Actions = 4-tool MDP (search_entities, get_entity, follow_entity_link, expand_neighbors)
- Dynamic Hop Control: model decides when to stop and submit answer
Output format follows SEOcrate:
- ... for step-by-step thought
- for tool calls
- ... for final submission
"""
from __future__ import annotations
import json
import re
import time
from dataclasses import dataclass, field
from typing import Any
from src.graph.tools import GraphToolsBase
# ---------------------------------------------------------------------------
# System prompt — the ontological rules the agent must follow
# ---------------------------------------------------------------------------
KINSHIP_SYSTEM_PROMPT = """\
You are a kinship reasoning agent. You navigate a knowledge graph to answer \
inheritance questions by making tool calls and reading the results.
## KINSHIP ONTOLOGY
The knowledge graph uses these classes and properties:
CLASSES:
- Person (has: fullName, age, isAlive)
- Asset (has: assetName, assetValue, belongsTo)
- WillClause (has: clauseText, excludesRelationType, requiresRelationType)
OBJECT PROPERTIES:
- isBiologicalParentOf / hasBiologicalParent → INVERSE PAIR. Directionality matters.
- isMarriedTo → SYMMETRIC. Marriage is mutual. A symmetric property can NEVER \
satisfy a "biological lineage" requirement.
- stipulatesCondition → WillClause conditions on a Person.
- belongsTo → Asset ownership.
- excludesRelationType → What property types the will FORBIDS.
- requiresRelationType → What property types the will REQUIRES.
## TOOLS
You have 4 tools to call via the knowledge graph:
1. search_entities(query) — Find entities by name. Start here. Takes a name string.
2. get_entity(iri) — Get all data properties for an entity (age, isAlive, type). REQUIRES an IRI.
3. follow_entity_link(iri, property?) — Follow outbound edges. Filter by property URI. REQUIRES an IRI.
4. expand_neighbors(iri) — Get ALL connected entities (inbound + outbound). REQUIRES an IRI.
IMPORTANT: get_entity, follow_entity_link, and expand_neighbors require entity IRIs (e.g. http://example.org/kinship#Person5).
If you only know a name, call search_entities first to get the IRI.
When a Tool result contains an "iri" or "target_iri" field, copy that exact value into the next action.
Never pass a person's full name as the entity argument to get_entity.
## CRITICAL ONTOLOGICAL RULES
1. READ the WillClause first. Check excludesRelationType and requiresRelationType.
2. NEVER traverse a property the will excludes (e.g., isMarriedTo when excluded).
3. isMarriedTo is a SymmetricProperty — it CANNOT establish biological lineage.
4. Always verify isAlive=true before declaring someone an heir.
5. Use the FEWEST hops possible (Dynamic Hop Control).
## OUTPUT FORMAT
For each step, output:
Your step-by-step ontological reasoning about what to do next
When you have enough evidence, output:
Your final synthesis of the evidence
Full name of the heir
Property URIs use the namespace http://example.org/kinship#
Example: http://example.org/kinship#isBiologicalParentOf
"""
# ---------------------------------------------------------------------------
# Agent state (MDP)
# ---------------------------------------------------------------------------
@dataclass
class AgentState:
"""MDP state for the RLM-on-KG agent.
Attributes:
current_entity: IRI of the entity currently in focus.
visited: Set of visited entity IRIs (loop avoidance).
hop_index: Current hop count.
hop_budget: Maximum allowed hops before forced termination.
evidence: List of evidence packets collected during traversal.
"""
current_entity: str | None = None
visited: set[str] = field(default_factory=set)
hop_index: int = 0
hop_budget: int = 8
evidence: list[dict] = field(default_factory=list)
# ---------------------------------------------------------------------------
# XML tag parsing
# ---------------------------------------------------------------------------
_ACTION_RE = re.compile(
r'',
re.IGNORECASE,
)
_ANSWER_RE = re.compile(
r'(.*?)',
re.IGNORECASE | re.DOTALL,
)
_REASONING_RE = re.compile(
r'(.*?)',
re.IGNORECASE | re.DOTALL,
)
def parse_action(text: str) -> dict | None:
"""Parse the first tag in *text*.
Returns:
Dict with keys 'tool', 'entity', 'property' (optional), or None.
"""
m = _ACTION_RE.search(text)
if not m:
return None
result: dict[str, str] = {
"tool": m.group(1),
"entity": m.group(2),
}
if m.group(3):
result["property"] = m.group(3)
return result
def parse_answer(text: str) -> str | None:
"""Extract the content of the first tag."""
m = _ANSWER_RE.search(text)
return m.group(1).strip() if m else None
def parse_reasoning(text: str) -> str | None:
"""Extract the content of the first tag."""
m = _REASONING_RE.search(text)
return m.group(1).strip() if m else None
# ---------------------------------------------------------------------------
# Tool executor
# ---------------------------------------------------------------------------
def execute_tool(tools: GraphToolsBase, action: dict) -> Any:
"""Dispatch a parsed action dict to the appropriate graph tool.
Args:
tools: A GraphToolsBase instance (RdflibTools or WoGraphTools).
action: Dict with 'tool', 'entity', and optionally 'property'.
Returns:
The tool's result (list or dict).
"""
tool_name = action["tool"]
entity = action.get("entity", "")
prop = action.get("property")
if tool_name == "search_entities":
# For search, the 'entity' field is used as the query string
return tools.search_entities(entity)
elif tool_name == "get_entity":
return tools.get_entity(entity)
elif tool_name == "follow_entity_link":
return tools.follow_entity_link(entity, prop)
elif tool_name == "expand_neighbors":
return tools.expand_neighbors(entity)
else:
return {"error": f"Unknown tool: {tool_name}"}
# ---------------------------------------------------------------------------
# Agent loop
# ---------------------------------------------------------------------------
@dataclass
class AgentResult:
"""Result from a completed agent run."""
answer: str | None
reasoning: str | None
trace: list[dict]
hops: int
elapsed_seconds: float
terminated_by: str # "answer" | "timeout" | "error"
def run_agent(
generate_fn,
tools: GraphToolsBase,
narrative: str,
question: str,
*,
hop_budget: int = 8,
verbose: bool = False,
) -> AgentResult:
"""Run the RLM-on-KG agentic loop.
Args:
generate_fn: A callable ``(messages: list[dict]) -> str`` that
generates a model completion given chat messages. This abstracts
away the specific model backend (transformers, vLLM, API, etc.).
tools: A GraphToolsBase instance for graph queries.
narrative: The text narrative (with semantic distractors).
question: The inheritance question.
hop_budget: Maximum hops before forced termination.
verbose: If True, print each step to stdout.
Returns:
An AgentResult with the answer, reasoning trace, and metadata.
"""
state = AgentState(hop_budget=hop_budget)
start_time = time.time()
messages: list[dict[str, str]] = [
{"role": "system", "content": KINSHIP_SYSTEM_PROMPT},
{
"role": "user",
"content": (
f"Narrative:\n{narrative}\n\n"
f"Question:\n{question}"
),
},
]
trace: list[dict] = []
while state.hop_index < state.hop_budget:
# Generate model response
response = generate_fn(messages)
if verbose:
print(f"\n--- Hop {state.hop_index} ---")
print(response)
# Check for final answer (Dynamic Hop Control)
answer = parse_answer(response)
if answer:
reasoning = parse_reasoning(response)
return AgentResult(
answer=answer,
reasoning=reasoning,
trace=trace,
hops=state.hop_index,
elapsed_seconds=time.time() - start_time,
terminated_by="answer",
)
# Parse and execute action
action = parse_action(response)
if not action:
# Model didn't produce a valid action or answer — add guidance
messages.append({"role": "assistant", "content": response})
messages.append({
"role": "user",
"content": (
"Please respond with either an tag to query the "
"graph, or an tag with your final answer."
),
})
state.hop_index += 1
continue
# Execute the tool call
try:
result = execute_tool(tools, action)
except Exception as e:
result = {"error": str(e)}
# Record in trace
trace_entry = {
"hop": state.hop_index,
"action": action,
"result": result,
"reasoning": parse_reasoning(response),
}
trace.append(trace_entry)
if verbose:
print(f" Action: {action}")
print(f" Result: {json.dumps(result, default=str)[:200]}")
# Update state
entity = action.get("entity", "")
state.visited.add(entity)
state.current_entity = entity
state.hop_index += 1
# Append to conversation
messages.append({"role": "assistant", "content": response})
messages.append({
"role": "user",
"content": f"Tool result:\n{json.dumps(result, default=str, indent=2)}",
})
# Budget exhausted
return AgentResult(
answer=None,
reasoning="Hop budget exhausted without reaching a conclusion.",
trace=trace,
hops=state.hop_index,
elapsed_seconds=time.time() - start_time,
terminated_by="timeout",
)
# ---------------------------------------------------------------------------
# Simple generate_fn for testing (uses a mock or Gemini API)
# ---------------------------------------------------------------------------
def make_gemini_generate_fn(model_name: str = "gemini-2.5-flash"):
"""Create a generate_fn that calls the Gemini API.
Requires the ``google-genai`` package and ``GOOGLE_API_KEY`` env var.
"""
import os
from google import genai
api_key = os.environ.get("GOOGLE_API_KEY") or os.environ.get("GEMINI_API_KEY")
client = genai.Client(api_key=api_key)
def generate(messages: list[dict[str, str]]) -> str:
# Convert our messages to Gemini format
contents = []
for msg in messages:
role = "model" if msg["role"] == "assistant" else "user"
if msg["role"] == "system":
# Prepend system as user message
contents.append({"role": "user", "parts": [{"text": msg["content"]}]})
contents.append({"role": "model", "parts": [{"text": "Understood. I will follow these instructions."}]})
else:
contents.append({"role": role, "parts": [{"text": msg["content"]}]})
response = client.models.generate_content(
model=model_name,
contents=contents,
)
return response.text
return generate
# ---------------------------------------------------------------------------
# Quick demo
# ---------------------------------------------------------------------------
if __name__ == "__main__":
import sys
sys.path.insert(0, ".")
from src.graph.graph_builder import get_arthur_scenario, build_rdflib_graph
from src.graph.tools import RdflibTools
from src.graph.scenario_generator import generate_narrative, generate_question
# Build the Arthur scenario
scenario = get_arthur_scenario()
graph = build_rdflib_graph(scenario)
tools = RdflibTools(graph)
narrative = generate_narrative(scenario)
question = generate_question(scenario)
print("=" * 60)
print("NARRATIVE:")
print(narrative)
print("\nQUESTION:", question)
print("GOLD ANSWER:", scenario["gold_answer"])
print("=" * 60)
# Try with Gemini if available
try:
generate_fn = make_gemini_generate_fn()
result = run_agent(generate_fn, tools, narrative, question, verbose=True)
print(f"\n{'=' * 60}")
print(f"ANSWER: {result.answer}")
print(f"CORRECT: {result.answer and 'Edward' in result.answer}")
print(f"HOPS: {result.hops}")
print(f"TIME: {result.elapsed_seconds:.2f}s")
print(f"TERMINATED BY: {result.terminated_by}")
except Exception as e:
print(f"\nCould not run with Gemini API: {e}")
print("Set GOOGLE_API_KEY to test with a real model.")