From 0da98e7c7bf1c2c6debc38a0c5771d85425591f1 Mon Sep 17 00:00:00 2001 From: Harshit Rohatgi Date: Wed, 15 Jul 2026 22:09:32 +0530 Subject: [PATCH 1/4] feat: add EoG (Explanations over Graphs) agent pattern Introduces a new agent type alongside ReAct for ontology-grounded graph investigation. EoG uses a deterministic controller to traverse an ontology graph, invoking bounded LLM inference at each node, propagating beliefs along edges, and converging on a minimal explanatory frontier. - agent/eog/types.py: EoGState, Belief, LedgerEntry, InvestigationConfig - agent/eog/ontology_client.py: async REST client for ontology-runtime - agent/eog/nodes.py: 7 graph nodes (bootstrap, pop, fetch, policy, update, propagate, frontier) - agent/eog/agent.py: create_eog_agent() graph builder - tests/agent/eog/: 44 tests covering types, client, nodes, and integration Co-Authored-By: Claude Opus 4.6 (1M context) --- src/uipath_langchain/agent/eog/__init__.py | 16 + src/uipath_langchain/agent/eog/agent.py | 77 ++++ src/uipath_langchain/agent/eog/nodes.py | 363 ++++++++++++++++++ .../agent/eog/ontology_client.py | 122 ++++++ src/uipath_langchain/agent/eog/types.py | 88 +++++ tests/agent/eog/__init__.py | 0 tests/agent/eog/test_agent.py | 231 +++++++++++ tests/agent/eog/test_nodes.py | 331 ++++++++++++++++ tests/agent/eog/test_ontology_client.py | 143 +++++++ tests/agent/eog/test_types.py | 119 ++++++ 10 files changed, 1490 insertions(+) create mode 100644 src/uipath_langchain/agent/eog/__init__.py create mode 100644 src/uipath_langchain/agent/eog/agent.py create mode 100644 src/uipath_langchain/agent/eog/nodes.py create mode 100644 src/uipath_langchain/agent/eog/ontology_client.py create mode 100644 src/uipath_langchain/agent/eog/types.py create mode 100644 tests/agent/eog/__init__.py create mode 100644 tests/agent/eog/test_agent.py create mode 100644 tests/agent/eog/test_nodes.py create mode 100644 tests/agent/eog/test_ontology_client.py create mode 100644 tests/agent/eog/test_types.py diff --git a/src/uipath_langchain/agent/eog/__init__.py b/src/uipath_langchain/agent/eog/__init__.py new file mode 100644 index 000000000..23aea78b7 --- /dev/null +++ b/src/uipath_langchain/agent/eog/__init__.py @@ -0,0 +1,16 @@ +"""EoG (Explanations over Graphs) agent pattern.""" + +from __future__ import annotations + +from .agent import create_eog_agent +from .ontology_client import OntologyClient +from .types import Belief, EoGState, InvestigationConfig, LedgerEntry + +__all__ = [ + "create_eog_agent", + "OntologyClient", + "EoGState", + "Belief", + "LedgerEntry", + "InvestigationConfig", +] diff --git a/src/uipath_langchain/agent/eog/agent.py b/src/uipath_langchain/agent/eog/agent.py new file mode 100644 index 000000000..0d4f21b99 --- /dev/null +++ b/src/uipath_langchain/agent/eog/agent.py @@ -0,0 +1,77 @@ +"""EoG (Explanations over Graphs) agent graph builder.""" + +from __future__ import annotations + +from langchain_core.language_models import BaseChatModel +from langgraph.constants import END, START +from langgraph.graph import StateGraph + +from .nodes import ( + _make_bootstrap, + _make_fetch_context, + _make_policy, + frontier_node, + pop_node, + propagate_node, + should_continue, + update_node, +) +from .ontology_client import OntologyClient +from .types import EoGState, InvestigationConfig + + +def create_eog_agent( + model: BaseChatModel, + ontology_client: OntologyClient, + ontology_name: str, + *, + investigation_config: InvestigationConfig | None = None, +) -> StateGraph: + """Create an EoG investigation agent as a LangGraph StateGraph. + + The graph performs belief-propagation over an ontology: it seeds beliefs + for a set of starting entities, visits each entity via BFS, uses an LLM + to assign a label, and propagates updates to neighbours until the budget + is exhausted or the active set is empty. + + Args: + model: LLM for the abductive policy node. + ontology_client: REST client for the ontology-runtime. + ontology_name: Name/ID of the deployed ontology. + investigation_config: Configuration for the investigation + (labels, seeds, budget). + + Returns: + Uncompiled ``StateGraph``. Call ``.compile()`` to get a runnable + graph. + """ + builder: StateGraph = StateGraph(EoGState) + + builder.add_node( + "bootstrap", + _make_bootstrap(ontology_client, ontology_name, investigation_config), + ) + builder.add_node("pop", pop_node) + builder.add_node( + "fetch_context", + _make_fetch_context(ontology_client, ontology_name), + ) + builder.add_node("policy", _make_policy(model)) + builder.add_node("update", update_node) + builder.add_node("propagate", propagate_node) + builder.add_node("frontier", frontier_node) + + builder.add_edge(START, "bootstrap") + builder.add_edge("bootstrap", "pop") + builder.add_conditional_edges( + "pop", + should_continue, + {"fetch_context": "fetch_context", "frontier": "frontier"}, + ) + builder.add_edge("fetch_context", "policy") + builder.add_edge("policy", "update") + builder.add_edge("update", "propagate") + builder.add_edge("propagate", "pop") + builder.add_edge("frontier", END) + + return builder diff --git a/src/uipath_langchain/agent/eog/nodes.py b/src/uipath_langchain/agent/eog/nodes.py new file mode 100644 index 000000000..4c16b81ac --- /dev/null +++ b/src/uipath_langchain/agent/eog/nodes.py @@ -0,0 +1,363 @@ +"""Node functions for the EoG (Explanations over Graphs) agent graph.""" + +from __future__ import annotations + +import asyncio +import json +import logging +import time +from collections.abc import Callable, Coroutine +from typing import Any + +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import HumanMessage, SystemMessage + +from .ontology_client import OntologyClient +from .types import Belief, EoGState, ExplanatoryEdge, InvestigationConfig, LedgerEntry + +# Type alias for async node functions +_NodeFn = Callable[[EoGState], Coroutine[Any, Any, dict[str, Any]]] + +logger = logging.getLogger(__name__) + +_POLICY_PROMPT_TEMPLATE = """\ +You are investigating entity "{entity_id}" in an ontology-based investigation. + +Available labels: {label_vocabulary} + +Evidence collected for this entity: +{context_data} + +Messages from neighboring entities: +{inbox_messages} + +Based on the evidence, assign ONE label from the vocabulary to this entity. +Explain your reasoning concisely. +If you believe other entities should be re-examined based on your findings, \ +list them. + +Respond in JSON: {{"label": "...", "evidence": "...", \ +"propagations": [{{"entity": "...", "reason": "..."}}]}}""" + + +def _make_bootstrap( + client: OntologyClient, + ontology: str, + config: InvestigationConfig | None, +) -> _NodeFn: + """Create bootstrap node with injected dependencies.""" + + async def _bootstrap(state: EoGState) -> dict[str, Any]: + """Initialize: discover graph, set up beliefs for seed entities.""" + cfg = config or state.investigation_config or InvestigationConfig( + label_vocabulary=["Defer"] + ) + + graph_meta, functions = await asyncio.gather( + client.discover(ontology), + client.list_functions(ontology), + ) + + beliefs: dict[str, Belief] = {} + for entity_id in cfg.seed_entities: + beliefs[entity_id] = Belief( + label=cfg.default_label, + evidence="Initial seed entity.", + ) + + return { + "ontology_graph": { + "metadata": graph_meta, + "functions": functions, + }, + "beliefs": beliefs, + "active_set": list(cfg.seed_entities), + "steps_taken": 0, + "investigation_config": cfg, + } + + return _bootstrap + + +async def pop_node(state: EoGState) -> dict[str, Any]: + """Dequeue next entity from active_set (FIFO). + + Returns: + Partial state update with current_entity and remaining active_set. + """ + active = list(state.active_set) + entity = active.pop(0) if active else "" + return {"current_entity": entity, "active_set": active} + + +def should_continue(state: EoGState) -> str: + """Route: continue traversal or compute frontier. + + Checks whether the current entity (just popped) is non-empty and the + step budget has not been exhausted. + + Returns: + ``"fetch_context"`` if there is an entity to process and budget + remains, ``"frontier"`` otherwise. + """ + cfg = state.investigation_config + max_steps = cfg.max_steps if cfg else 50 + + if state.current_entity and state.steps_taken < max_steps: + return "fetch_context" + return "frontier" + + +def _make_fetch_context( + client: OntologyClient, + ontology: str, +) -> _NodeFn: + """Create fetch_context node with injected dependencies.""" + + async def _fetch_context(state: EoGState) -> dict[str, Any]: + """Build bounded context packet for the current entity.""" + entity_id = state.current_entity + cfg = state.investigation_config + max_results = cfg.max_results_per_function if cfg else 50 + + current_belief = state.beliefs.get(entity_id) + inbox_messages = state.inbox.get(entity_id, []) + + # Gather neighbor beliefs from ontology_graph edges if available + neighbor_beliefs: dict[str, dict[str, Any]] = {} + graph_meta = state.ontology_graph.get("metadata", {}) + edges = graph_meta.get("edges", []) + for edge in edges: + src = edge.get("source", "") + tgt = edge.get("target", "") + neighbour = None + rel = edge.get("relationship", "") + if src == entity_id: + neighbour = tgt + elif tgt == entity_id: + neighbour = src + if neighbour and neighbour in state.beliefs: + b = state.beliefs[neighbour] + neighbor_beliefs[neighbour] = { + "label": b.label, + "evidence": b.evidence, + "relationship": rel, + } + + # Invoke relevant functions + functions = state.ontology_graph.get("functions", []) + function_results: list[dict[str, Any]] = [] + for fn in functions: + fn_name = fn.get("name", "") + fn_desc = fn.get("description", "") + if entity_id.lower() in fn_desc.lower() or not fn_desc: + try: + result = await client.invoke_function( + ontology, fn_name, {"entity_id": entity_id} + ) + rows = result.get("rows", []) + function_results.append({ + "function": fn_name, + "rows": rows[:max_results], + }) + except Exception: + logger.warning( + "Function %s failed for entity %s", + fn_name, + entity_id, + exc_info=True, + ) + + context_packet: dict[str, Any] = { + "entity_id": entity_id, + "current_belief": current_belief.model_dump() if current_belief else None, + "inbox_messages": inbox_messages, + "neighbor_beliefs": neighbor_beliefs, + "function_results": function_results, + } + return {"context_packet": context_packet} + + return _fetch_context + + +def _make_policy(model: BaseChatModel) -> _NodeFn: + """Create policy node with injected LLM.""" + + async def _policy(state: EoGState) -> dict[str, Any]: + """LLM call: label entity and identify propagation claims.""" + cfg = state.investigation_config + label_vocab = cfg.label_vocabulary if cfg else ["Defer"] + + ctx = state.context_packet + entity_id = ctx.get("entity_id", state.current_entity) + + context_data = json.dumps( + { + k: v + for k, v in ctx.items() + if k not in ("entity_id", "inbox_messages") + }, + indent=2, + default=str, + ) + inbox_str = json.dumps( + ctx.get("inbox_messages", []), indent=2, default=str + ) + + prompt = _POLICY_PROMPT_TEMPLATE.format( + entity_id=entity_id, + label_vocabulary=", ".join(label_vocab), + context_data=context_data, + inbox_messages=inbox_str, + ) + + messages = [ + SystemMessage(content="You are an ontology investigation agent."), + HumanMessage(content=prompt), + ] + + try: + response = await model.ainvoke(messages) + content = ( + response.content + if isinstance(response.content, str) + else str(response.content) + ) + parsed = json.loads(content) + except Exception: + logger.warning( + "Policy node: LLM output could not be parsed for entity %s", + entity_id, + ) + parsed = { + "label": "Defer", + "evidence": "LLM output could not be parsed", + "propagations": [], + } + + return {"policy_result": parsed} + + return _policy + + +async def update_node(state: EoGState) -> dict[str, Any]: + """Write belief to ledger and track flips. + + Returns: + Partial state update with updated belief, ledger entry, and + incremented steps_taken. + """ + entity_id = state.current_entity + new_label = state.policy_result.get("label", "Defer") + evidence = state.policy_result.get("evidence", "") + + existing = state.beliefs.get(entity_id) + old_label = existing.label if existing else None + old_flip_count = existing.flip_count if existing else 0 + + flip_count = old_flip_count + if old_label is not None and old_label != new_label: + flip_count += 1 + + updated_belief = Belief( + label=new_label, + evidence=evidence, + flip_count=flip_count, + ) + + entry = LedgerEntry( + timestamp=time.time(), + entity_id=entity_id, + old_label=old_label, + new_label=new_label, + evidence=evidence, + ) + + return { + "beliefs": {entity_id: updated_belief}, + "ledger": [entry], + "steps_taken": state.steps_taken + 1, + } + + +async def propagate_node(state: EoGState) -> dict[str, Any]: + """Broadcast beliefs to neighbours and re-activate if changed. + + Returns: + Partial state update with updated inbox and active_set. + """ + cfg = state.investigation_config + max_flips = cfg.max_flips if cfg else 3 + + propagations = state.policy_result.get("propagations", []) + entity_id = state.current_entity + current_belief = state.beliefs.get(entity_id) + + updated_inbox: dict[str, list[dict[str, Any]]] = {} + new_active: list[str] = [] + explanatory_edges: list[ExplanatoryEdge] = [] + + for prop in propagations: + target = prop.get("entity", "") + reason = prop.get("reason", "") + if not target: + continue + + message: dict[str, Any] = { + "from": entity_id, + "label": current_belief.label if current_belief else "Defer", + "evidence": current_belief.evidence if current_belief else "", + "reason": reason, + } + + existing_messages = list(state.inbox.get(target, [])) + existing_messages.append(message) + updated_inbox[target] = existing_messages + + target_belief = state.beliefs.get(target) + target_flips = target_belief.flip_count if target_belief else 0 + + if ( + target_flips < max_flips + and target not in state.active_set + and target not in new_active + ): + new_active.append(target) + + explanatory_edges.append( + ExplanatoryEdge( + source=entity_id, + target=target, + relationship="propagation", + evidence=reason, + ) + ) + + return { + "inbox": updated_inbox, + "active_set": list(state.active_set) + new_active, + "explanatory_edges": explanatory_edges, + } + + +async def frontier_node(state: EoGState) -> dict[str, Any]: + """Compute the minimal explanatory frontier. + + Filters beliefs to non-Defer labels and builds a summary list. + + Returns: + Partial state update with the frontier list. + """ + cfg = state.investigation_config + default_label = cfg.default_label if cfg else "Defer" + + frontier_list: list[dict[str, Any]] = [] + for entity_id, belief in state.beliefs.items(): + if belief.label != default_label: + frontier_list.append({ + "entity": entity_id, + "label": belief.label, + "evidence": belief.evidence, + }) + + return {"frontier": frontier_list} diff --git a/src/uipath_langchain/agent/eog/ontology_client.py b/src/uipath_langchain/agent/eog/ontology_client.py new file mode 100644 index 000000000..bdec400e1 --- /dev/null +++ b/src/uipath_langchain/agent/eog/ontology_client.py @@ -0,0 +1,122 @@ +"""Async REST client for ontology-runtime.""" + +from __future__ import annotations + +from typing import Any + +import httpx +from uipath._utils._ssl_context import get_httpx_client_kwargs + + +class OntologyClient: + """Async HTTP client for the ontology-runtime service. + + Args: + base_url: Root URL of the ontology-runtime (no trailing slash). + account: Account name used in the API path. + tenant: Tenant name used in the API path. + timeout: Request timeout in seconds. + """ + + def __init__( + self, + base_url: str = "http://localhost:5002", + account: str = "datafabric", + tenant: str = "DefaultTenant", + *, + timeout: float = 30.0, + ) -> None: + self.base_url = base_url.rstrip("/") + self.account = account + self.tenant = tenant + self.timeout = timeout + + @property + def _api_base(self) -> str: + """Build the base API path.""" + return ( + f"{self.base_url}/{self.account}/{self.tenant}" + "/datafabric_/api" + ) + + def _make_client(self) -> httpx.AsyncClient: + """Create a configured async HTTP client.""" + kwargs = get_httpx_client_kwargs() + kwargs["timeout"] = self.timeout + return httpx.AsyncClient(**kwargs) + + async def discover(self, ontology: str) -> dict[str, Any]: + """Retrieve ontology metadata. + + Args: + ontology: Name or ID of the deployed ontology. + + Returns: + JSON response with ontology metadata. + """ + url = f"{self._api_base}/ontology/{ontology}" + async with self._make_client() as client: + resp = await client.get(url) + resp.raise_for_status() + return resp.json() # type: ignore[no-any-return] + + async def list_functions(self, ontology: str) -> list[dict[str, Any]]: + """List function definitions for an ontology. + + Args: + ontology: Name or ID of the deployed ontology. + + Returns: + List of function definition dicts. + """ + url = f"{self._api_base}/ontology/{ontology}/functions" + async with self._make_client() as client: + resp = await client.get(url) + resp.raise_for_status() + return resp.json() # type: ignore[no-any-return] + + async def invoke_function( + self, + ontology: str, + fn_name: str, + params: dict[str, Any] | None = None, + ) -> dict[str, Any]: + """Invoke an ontology function. + + Args: + ontology: Name or ID of the deployed ontology. + fn_name: Function name to invoke. + params: Optional parameters to pass to the function. + + Returns: + JSON response (typically ``{"rows": [...]}``) + """ + url = ( + f"{self._api_base}/ontology/{ontology}" + f"/functions/{fn_name}/invoke" + ) + body: dict[str, Any] = {"params": params} if params else {} + async with self._make_client() as client: + resp = await client.post(url, json=body) + resp.raise_for_status() + return resp.json() # type: ignore[no-any-return] + + async def sparql(self, ontology: str, query: str) -> dict[str, Any]: + """Execute a SPARQL query against an ontology. + + Args: + ontology: Name or ID of the deployed ontology. + query: Raw SPARQL query text. + + Returns: + JSON response with query results. + """ + url = f"{self._api_base}/ontology/{ontology}/sparql" + async with self._make_client() as client: + resp = await client.post( + url, + content=query, + headers={"Content-Type": "application/sparql-query"}, + ) + resp.raise_for_status() + return resp.json() # type: ignore[no-any-return] diff --git a/src/uipath_langchain/agent/eog/types.py b/src/uipath_langchain/agent/eog/types.py new file mode 100644 index 000000000..a1a0cbac1 --- /dev/null +++ b/src/uipath_langchain/agent/eog/types.py @@ -0,0 +1,88 @@ +"""State types for the EoG (Explanations over Graphs) agent.""" + +from __future__ import annotations + +import operator +from typing import Annotated, Any + +from pydantic import BaseModel, Field + +from uipath_langchain.agent.react.reducers import merge_dicts + + +class Belief(BaseModel): + """A belief about an entity's label with supporting evidence. + + Labels are free strings drawn from InvestigationConfig.label_vocabulary. + """ + + label: str + evidence: str = "" + flip_count: int = 0 + + +class LedgerEntry(BaseModel): + """Append-only record of a belief update.""" + + timestamp: float + entity_id: str + old_label: str | None + new_label: str + evidence: str + function_used: str | None = None + + +class ExplanatoryEdge(BaseModel): + """An edge in the explanatory graph connecting two entities.""" + + source: str + target: str + relationship: str + evidence: str = "" + + +class InvestigationConfig(BaseModel): + """Configuration that makes EoG reusable across domains. + + Args: + label_vocabulary: Allowed labels, e.g. ["Source", "DerivedEffect"]. + seed_entities: Starting entity IDs for the BFS traversal. + max_steps: Budget cap on the number of entity visits. + max_flips: Damping threshold -- entities that flip more than this + are no longer re-activated by neighbours. + default_label: Label assigned to entities before investigation. + max_results_per_function: Upper bound on rows returned by each + ontology function invocation. + """ + + label_vocabulary: list[str] + seed_entities: list[str] = Field(default_factory=list) + max_steps: int = 50 + max_flips: int = 3 + default_label: str = "Defer" + max_results_per_function: int = 50 + + +class EoGState(BaseModel): + """LangGraph state for the EoG investigation agent.""" + + active_set: list[str] = Field(default_factory=list) + current_entity: str = "" + beliefs: Annotated[dict[str, Belief], merge_dicts] = Field( + default_factory=dict + ) + inbox: Annotated[dict[str, list[dict[str, Any]]], merge_dicts] = Field( + default_factory=dict + ) + ledger: Annotated[list[LedgerEntry], operator.add] = Field( + default_factory=list + ) + explanatory_edges: Annotated[list[ExplanatoryEdge], operator.add] = Field( + default_factory=list + ) + context_packet: dict[str, Any] = Field(default_factory=dict) + policy_result: dict[str, Any] = Field(default_factory=dict) + steps_taken: int = 0 + frontier: list[dict[str, Any]] = Field(default_factory=list) + ontology_graph: dict[str, Any] = Field(default_factory=dict) + investigation_config: InvestigationConfig | None = None diff --git a/tests/agent/eog/__init__.py b/tests/agent/eog/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/tests/agent/eog/test_agent.py b/tests/agent/eog/test_agent.py new file mode 100644 index 000000000..dca231a20 --- /dev/null +++ b/tests/agent/eog/test_agent.py @@ -0,0 +1,231 @@ +"""Integration tests for the EoG agent graph builder.""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from uipath_langchain.agent.eog.agent import create_eog_agent +from uipath_langchain.agent.eog.ontology_client import OntologyClient +from uipath_langchain.agent.eog.types import InvestigationConfig + + +def _mock_ontology_client() -> AsyncMock: + """Create a mock ontology client with a small 3-entity graph.""" + client = AsyncMock(spec=OntologyClient) + client.discover = AsyncMock( + return_value={ + "name": "test-ontology", + "entities": ["e1", "e2", "e3"], + "edges": [ + {"source": "e1", "target": "e2", "relationship": "causes"}, + {"source": "e2", "target": "e3", "relationship": "affects"}, + ], + } + ) + client.list_functions = AsyncMock( + return_value=[ + {"name": "get_info", "description": "Get entity info"}, + ] + ) + client.invoke_function = AsyncMock( + return_value={"rows": [{"data": "test_data"}]} + ) + return client + + +def _mock_llm(labels: dict[str, str] | None = None) -> AsyncMock: + """Create a mock LLM that returns deterministic labels. + + Args: + labels: Mapping from entity_id to label. Defaults to assigning + "Source" to e1 and "DerivedEffect" to others. + """ + default_labels = labels or { + "e1": "Source", + "e2": "DerivedEffect", + "e3": "DerivedEffect", + } + + async def ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: + # Extract entity_id from the prompt + content = messages[-1].content if messages else "" + entity_id = "" + for eid in default_labels: + if eid in content: + entity_id = eid + break + + label = default_labels.get(entity_id, "Defer") + # Propagate to neighbours only from e1 + propagations = [] + if entity_id == "e1": + propagations = [ + {"entity": "e2", "reason": "caused by e1"}, + ] + elif entity_id == "e2": + propagations = [ + {"entity": "e3", "reason": "affected by e2"}, + ] + + response = AsyncMock() + response.content = json.dumps({ + "label": label, + "evidence": f"Evidence for {entity_id}", + "propagations": propagations, + }) + return response + + model = AsyncMock() + model.ainvoke = ainvoke + return model + + +class TestCreateEoGAgent: + def test_returns_state_graph(self) -> None: + client = _mock_ontology_client() + model = _mock_llm() + cfg = InvestigationConfig( + label_vocabulary=["Source", "DerivedEffect", "Defer"], + seed_entities=["e1"], + ) + graph = create_eog_agent( + model, client, "test-onto", investigation_config=cfg + ) + assert graph is not None + + def test_graph_compiles(self) -> None: + client = _mock_ontology_client() + model = _mock_llm() + cfg = InvestigationConfig( + label_vocabulary=["Source", "DerivedEffect", "Defer"], + seed_entities=["e1"], + ) + graph = create_eog_agent( + model, client, "test-onto", investigation_config=cfg + ) + compiled = graph.compile() + assert compiled is not None + + @pytest.mark.asyncio + async def test_full_traversal(self) -> None: + """Integration: compile and run the graph with mock ontology.""" + client = _mock_ontology_client() + model = _mock_llm() + cfg = InvestigationConfig( + label_vocabulary=["Source", "DerivedEffect", "Defer"], + seed_entities=["e1"], + max_steps=20, + ) + graph = create_eog_agent( + model, client, "test-onto", investigation_config=cfg + ) + compiled = graph.compile() + result = await compiled.ainvoke({}) + + # BFS should have terminated + assert result["steps_taken"] >= 1 + + # Ledger should have entries + assert len(result["ledger"]) >= 1 + + # Frontier should contain non-Defer entities + assert len(result["frontier"]) >= 1 + frontier_entities = {f["entity"] for f in result["frontier"]} + assert "e1" in frontier_entities + + @pytest.mark.asyncio + async def test_propagation_causes_revisits(self) -> None: + """Verify that propagation re-activates neighbours.""" + client = _mock_ontology_client() + + # LLM that flips labels to force re-visits + call_count: dict[str, int] = {} + + async def flipping_ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: + content = messages[-1].content if messages else "" + entity_id = "" + for eid in ["e1", "e2", "e3"]: + if eid in content: + entity_id = eid + break + + call_count[entity_id] = call_count.get(entity_id, 0) + 1 + count = call_count[entity_id] + + # Alternate labels to trigger flips + label = "Source" if count % 2 == 1 else "DerivedEffect" + propagations = ( + [{"entity": "e2", "reason": "update"}] + if entity_id == "e1" and count == 1 + else [] + ) + + response = AsyncMock() + response.content = json.dumps({ + "label": label, + "evidence": f"visit {count}", + "propagations": propagations, + }) + return response + + model = AsyncMock() + model.ainvoke = flipping_ainvoke + + cfg = InvestigationConfig( + label_vocabulary=["Source", "DerivedEffect", "Defer"], + seed_entities=["e1"], + max_steps=10, + max_flips=3, + ) + graph = create_eog_agent( + model, client, "test-onto", investigation_config=cfg + ) + compiled = graph.compile() + result = await compiled.ainvoke({}) + + # e2 should have been visited because e1 propagated to it + e2_entries = [ + e for e in result["ledger"] if e.entity_id == "e2" + ] + assert len(e2_entries) >= 1 + + @pytest.mark.asyncio + async def test_budget_cap_terminates(self) -> None: + """Verify the graph stops at max_steps.""" + client = _mock_ontology_client() + + # LLM that always propagates, creating infinite loop potential + async def always_propagate( + messages: Any, **kwargs: Any + ) -> AsyncMock: + response = AsyncMock() + response.content = json.dumps({ + "label": "Source", + "evidence": "always", + "propagations": [ + {"entity": "e1", "reason": "loop"}, + {"entity": "e2", "reason": "loop"}, + ], + }) + return response + + model = AsyncMock() + model.ainvoke = always_propagate + + cfg = InvestigationConfig( + label_vocabulary=["Source", "Defer"], + seed_entities=["e1"], + max_steps=3, + max_flips=10, + ) + graph = create_eog_agent( + model, client, "test-onto", investigation_config=cfg + ) + compiled = graph.compile() + result = await compiled.ainvoke({}) + + assert result["steps_taken"] <= 3 diff --git a/tests/agent/eog/test_nodes.py b/tests/agent/eog/test_nodes.py new file mode 100644 index 000000000..d7552d724 --- /dev/null +++ b/tests/agent/eog/test_nodes.py @@ -0,0 +1,331 @@ +"""Tests for EoG node functions.""" + +from __future__ import annotations + +import json +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from uipath_langchain.agent.eog.nodes import ( + _make_bootstrap, + _make_fetch_context, + _make_policy, + frontier_node, + pop_node, + propagate_node, + should_continue, + update_node, +) +from uipath_langchain.agent.eog.types import ( + Belief, + EoGState, + InvestigationConfig, +) + + +def _mock_ontology_client( + discover_result: dict[str, Any] | None = None, + functions_result: list[dict[str, Any]] | None = None, + invoke_result: dict[str, Any] | None = None, +) -> AsyncMock: + client = AsyncMock() + client.discover = AsyncMock( + return_value=discover_result or {"name": "onto", "edges": []} + ) + client.list_functions = AsyncMock( + return_value=functions_result or [] + ) + client.invoke_function = AsyncMock( + return_value=invoke_result or {"rows": []} + ) + return client + + +def _base_config() -> InvestigationConfig: + return InvestigationConfig( + label_vocabulary=["Source", "DerivedEffect", "Defer"], + seed_entities=["e1", "e2"], + max_steps=10, + ) + + +class TestBootstrapNode: + @pytest.mark.asyncio + async def test_bootstrap_initialises_beliefs(self) -> None: + cfg = _base_config() + client = _mock_ontology_client( + discover_result={"name": "test"}, + functions_result=[{"name": "fn1", "description": "desc"}], + ) + node = _make_bootstrap(client, "test-onto", cfg) + state = EoGState() + result = await node(state) + + assert "e1" in result["beliefs"] + assert "e2" in result["beliefs"] + assert result["beliefs"]["e1"].label == "Defer" + assert result["active_set"] == ["e1", "e2"] + assert result["steps_taken"] == 0 + assert "metadata" in result["ontology_graph"] + assert "functions" in result["ontology_graph"] + + @pytest.mark.asyncio + async def test_bootstrap_uses_state_config_as_fallback(self) -> None: + client = _mock_ontology_client() + node = _make_bootstrap(client, "onto", None) + cfg = InvestigationConfig( + label_vocabulary=["A"], + seed_entities=["x"], + ) + state = EoGState(investigation_config=cfg) + result = await node(state) + assert "x" in result["beliefs"] + + +class TestPopNode: + @pytest.mark.asyncio + async def test_pop_dequeues_first(self) -> None: + state = EoGState(active_set=["a", "b", "c"]) + result = await pop_node(state) + assert result["current_entity"] == "a" + assert result["active_set"] == ["b", "c"] + + @pytest.mark.asyncio + async def test_pop_empty(self) -> None: + state = EoGState(active_set=[]) + result = await pop_node(state) + assert result["current_entity"] == "" + assert result["active_set"] == [] + + +class TestShouldContinue: + def test_continue_when_entity_and_budget(self) -> None: + cfg = _base_config() + state = EoGState( + current_entity="e1", + steps_taken=0, + investigation_config=cfg, + ) + assert should_continue(state) == "fetch_context" + + def test_frontier_when_no_current_entity(self) -> None: + cfg = _base_config() + state = EoGState( + current_entity="", + steps_taken=0, + investigation_config=cfg, + ) + assert should_continue(state) == "frontier" + + def test_frontier_when_budget_exhausted(self) -> None: + cfg = _base_config() + state = EoGState( + current_entity="e1", + steps_taken=10, + investigation_config=cfg, + ) + assert should_continue(state) == "frontier" + + def test_default_max_steps_when_no_config(self) -> None: + state = EoGState(current_entity="e1", steps_taken=0) + assert should_continue(state) == "fetch_context" + + state2 = EoGState(current_entity="e1", steps_taken=50) + assert should_continue(state2) == "frontier" + + +class TestFetchContextNode: + @pytest.mark.asyncio + async def test_builds_context_packet(self) -> None: + client = _mock_ontology_client( + invoke_result={"rows": [{"val": 1}]} + ) + cfg = _base_config() + node = _make_fetch_context(client, "onto") + state = EoGState( + current_entity="e1", + beliefs={"e1": Belief(label="Defer")}, + ontology_graph={ + "metadata": {"edges": []}, + "functions": [{"name": "fn_e1", "description": "e1 info"}], + }, + investigation_config=cfg, + ) + result = await node(state) + ctx = result["context_packet"] + assert ctx["entity_id"] == "e1" + assert ctx["current_belief"] is not None + assert len(ctx["function_results"]) == 1 + + +class TestPolicyNode: + @pytest.mark.asyncio + async def test_policy_parses_llm_output(self) -> None: + llm_response_content = json.dumps({ + "label": "Source", + "evidence": "strong signal", + "propagations": [{"entity": "e2", "reason": "linked"}], + }) + model = AsyncMock() + response_msg = AsyncMock() + response_msg.content = llm_response_content + model.ainvoke = AsyncMock(return_value=response_msg) + + node = _make_policy(model) + cfg = _base_config() + state = EoGState( + current_entity="e1", + context_packet={"entity_id": "e1", "data": "some data"}, + investigation_config=cfg, + ) + result = await node(state) + assert result["policy_result"]["label"] == "Source" + assert len(result["policy_result"]["propagations"]) == 1 + + @pytest.mark.asyncio + async def test_policy_handles_parse_failure(self) -> None: + model = AsyncMock() + response_msg = AsyncMock() + response_msg.content = "not valid json at all" + model.ainvoke = AsyncMock(return_value=response_msg) + + node = _make_policy(model) + state = EoGState( + current_entity="e1", + context_packet={"entity_id": "e1"}, + investigation_config=_base_config(), + ) + result = await node(state) + assert result["policy_result"]["label"] == "Defer" + assert "could not be parsed" in result["policy_result"]["evidence"] + + +class TestUpdateNode: + @pytest.mark.asyncio + async def test_creates_ledger_entry(self) -> None: + state = EoGState( + current_entity="e1", + beliefs={"e1": Belief(label="Defer")}, + policy_result={"label": "Source", "evidence": "found it"}, + steps_taken=0, + ) + result = await update_node(state) + assert result["beliefs"]["e1"].label == "Source" + assert result["beliefs"]["e1"].flip_count == 1 + assert len(result["ledger"]) == 1 + assert result["ledger"][0].old_label == "Defer" + assert result["ledger"][0].new_label == "Source" + assert result["steps_taken"] == 1 + + @pytest.mark.asyncio + async def test_no_flip_when_same_label(self) -> None: + state = EoGState( + current_entity="e1", + beliefs={"e1": Belief(label="Source", flip_count=0)}, + policy_result={"label": "Source", "evidence": "confirmed"}, + steps_taken=2, + ) + result = await update_node(state) + assert result["beliefs"]["e1"].flip_count == 0 + assert result["steps_taken"] == 3 + + @pytest.mark.asyncio + async def test_new_entity_no_flip(self) -> None: + state = EoGState( + current_entity="new_e", + beliefs={}, + policy_result={"label": "Source", "evidence": "new"}, + steps_taken=0, + ) + result = await update_node(state) + assert result["beliefs"]["new_e"].flip_count == 0 + + +class TestPropagateNode: + @pytest.mark.asyncio + async def test_propagates_to_neighbours(self) -> None: + state = EoGState( + current_entity="e1", + beliefs={"e1": Belief(label="Source", evidence="test")}, + policy_result={ + "propagations": [ + {"entity": "e2", "reason": "linked"}, + {"entity": "e3", "reason": "related"}, + ], + }, + active_set=[], + investigation_config=_base_config(), + ) + result = await propagate_node(state) + assert "e2" in result["inbox"] + assert "e3" in result["inbox"] + assert "e2" in result["active_set"] + assert "e3" in result["active_set"] + assert len(result["explanatory_edges"]) == 2 + + @pytest.mark.asyncio + async def test_does_not_reactivate_over_max_flips(self) -> None: + cfg = InvestigationConfig( + label_vocabulary=["A"], max_flips=2 + ) + state = EoGState( + current_entity="e1", + beliefs={ + "e1": Belief(label="A"), + "e2": Belief(label="A", flip_count=3), + }, + policy_result={ + "propagations": [{"entity": "e2", "reason": "test"}], + }, + active_set=[], + investigation_config=cfg, + ) + result = await propagate_node(state) + # e2 should NOT be re-activated because flip_count >= max_flips + assert "e2" not in result["active_set"] + + @pytest.mark.asyncio + async def test_no_duplicates_in_active_set(self) -> None: + state = EoGState( + current_entity="e1", + beliefs={"e1": Belief(label="A")}, + policy_result={ + "propagations": [{"entity": "e2", "reason": "x"}], + }, + active_set=["e2"], + investigation_config=_base_config(), + ) + result = await propagate_node(state) + # e2 already in active_set, should not be duplicated + assert result["active_set"].count("e2") == 1 + + +class TestFrontierNode: + @pytest.mark.asyncio + async def test_filters_defer_labels(self) -> None: + state = EoGState( + beliefs={ + "e1": Belief(label="Source", evidence="found"), + "e2": Belief(label="Defer", evidence="unknown"), + "e3": Belief(label="DerivedEffect", evidence="derived"), + }, + investigation_config=_base_config(), + ) + result = await frontier_node(state) + entities = [f["entity"] for f in result["frontier"]] + assert "e1" in entities + assert "e3" in entities + assert "e2" not in entities + + @pytest.mark.asyncio + async def test_empty_frontier_when_all_defer(self) -> None: + state = EoGState( + beliefs={ + "e1": Belief(label="Defer"), + }, + investigation_config=_base_config(), + ) + result = await frontier_node(state) + assert result["frontier"] == [] diff --git a/tests/agent/eog/test_ontology_client.py b/tests/agent/eog/test_ontology_client.py new file mode 100644 index 000000000..ace8e4309 --- /dev/null +++ b/tests/agent/eog/test_ontology_client.py @@ -0,0 +1,143 @@ +"""Tests for OntologyClient.""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx +import pytest + +from uipath_langchain.agent.eog.ontology_client import OntologyClient + + +def _patch_client( + monkeypatch: pytest.MonkeyPatch, + response_body: Any, + status_code: int = 200, + captured: list[httpx.Request] | None = None, +) -> None: + """Patch OntologyClient._make_client to use a mock transport.""" + + def handler(request: httpx.Request) -> httpx.Response: + if captured is not None: + captured.append(request) + return httpx.Response( + status_code=status_code, + json=response_body, + request=request, + ) + + def _make_mock_client(self: OntologyClient) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + monkeypatch.setattr(OntologyClient, "_make_client", _make_mock_client) + + +@pytest.fixture +def client() -> OntologyClient: + return OntologyClient( + base_url="http://test-host:5002", + account="acct", + tenant="tenant1", + ) + + +class TestApiBase: + def test_api_base_path(self, client: OntologyClient) -> None: + assert client._api_base == ( + "http://test-host:5002/acct/tenant1/datafabric_/api" + ) + + def test_trailing_slash_stripped(self) -> None: + c = OntologyClient(base_url="http://host:5002/") + assert not c._api_base.startswith("http://host:5002//") + + +class TestDiscover: + @pytest.mark.asyncio + async def test_discover_success( + self, + client: OntologyClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + expected = {"name": "test-onto", "entities": ["e1"]} + _patch_client(monkeypatch, expected) + result = await client.discover("test-onto") + assert result == expected + + +class TestListFunctions: + @pytest.mark.asyncio + async def test_list_functions_success( + self, + client: OntologyClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + expected = [{"name": "fn1", "description": "desc1"}] + _patch_client(monkeypatch, expected) + result = await client.list_functions("test-onto") + assert result == expected + + +class TestInvokeFunction: + @pytest.mark.asyncio + async def test_invoke_function_with_params( + self, + client: OntologyClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + expected = {"rows": [{"id": "r1"}]} + captured: list[httpx.Request] = [] + _patch_client(monkeypatch, expected, captured=captured) + result = await client.invoke_function( + "test-onto", "fn1", {"key": "val"} + ) + assert result == expected + body = json.loads(captured[0].content) + assert body == {"params": {"key": "val"}} + + @pytest.mark.asyncio + async def test_invoke_function_no_params( + self, + client: OntologyClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + expected = {"rows": []} + captured: list[httpx.Request] = [] + _patch_client(monkeypatch, expected, captured=captured) + result = await client.invoke_function("test-onto", "fn1") + assert result == expected + body = json.loads(captured[0].content) + assert body == {} + + +class TestSparql: + @pytest.mark.asyncio + async def test_sparql_success( + self, + client: OntologyClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + expected = {"results": {"bindings": []}} + captured: list[httpx.Request] = [] + _patch_client(monkeypatch, expected, captured=captured) + result = await client.sparql( + "test-onto", "SELECT ?s WHERE {?s ?p ?o}" + ) + assert result == expected + assert ( + captured[0].headers["content-type"] == "application/sparql-query" + ) + + +class TestErrorHandling: + @pytest.mark.asyncio + async def test_non_200_raises( + self, + client: OntologyClient, + monkeypatch: pytest.MonkeyPatch, + ) -> None: + _patch_client(monkeypatch, {"error": "fail"}, status_code=500) + with pytest.raises(httpx.HTTPStatusError): + await client.discover("test-onto") diff --git a/tests/agent/eog/test_types.py b/tests/agent/eog/test_types.py new file mode 100644 index 000000000..c5a92ef38 --- /dev/null +++ b/tests/agent/eog/test_types.py @@ -0,0 +1,119 @@ +"""Tests for EoG state types and reducers.""" + +from __future__ import annotations + +import operator + +from uipath_langchain.agent.eog.types import ( + Belief, + EoGState, + ExplanatoryEdge, + InvestigationConfig, + LedgerEntry, +) +from uipath_langchain.agent.react.reducers import merge_dicts + + +class TestBelief: + def test_creation_defaults(self) -> None: + b = Belief(label="Source") + assert b.label == "Source" + assert b.evidence == "" + assert b.flip_count == 0 + + def test_creation_with_values(self) -> None: + b = Belief(label="PolicyViolation", evidence="found in audit", flip_count=2) + assert b.label == "PolicyViolation" + assert b.evidence == "found in audit" + assert b.flip_count == 2 + + def test_serialization_roundtrip(self) -> None: + b = Belief(label="Source", evidence="test") + data = b.model_dump() + assert data == {"label": "Source", "evidence": "test", "flip_count": 0} + restored = Belief.model_validate(data) + assert restored == b + + +class TestLedgerEntry: + def test_creation(self) -> None: + entry = LedgerEntry( + timestamp=1000.0, + entity_id="e1", + old_label=None, + new_label="Source", + evidence="initial", + ) + assert entry.entity_id == "e1" + assert entry.old_label is None + assert entry.function_used is None + + +class TestExplanatoryEdge: + def test_creation(self) -> None: + edge = ExplanatoryEdge(source="a", target="b", relationship="causes") + assert edge.evidence == "" + + +class TestInvestigationConfig: + def test_defaults(self) -> None: + cfg = InvestigationConfig(label_vocabulary=["Source", "Defer"]) + assert cfg.seed_entities == [] + assert cfg.max_steps == 50 + assert cfg.max_flips == 3 + assert cfg.default_label == "Defer" + assert cfg.max_results_per_function == 50 + + def test_custom_values(self) -> None: + cfg = InvestigationConfig( + label_vocabulary=["A", "B"], + seed_entities=["e1", "e2"], + max_steps=10, + max_flips=1, + default_label="Unknown", + max_results_per_function=5, + ) + assert cfg.max_steps == 10 + assert len(cfg.seed_entities) == 2 + + +class TestEoGState: + def test_default_state(self) -> None: + state = EoGState() + assert state.active_set == [] + assert state.current_entity == "" + assert state.beliefs == {} + assert state.ledger == [] + assert state.steps_taken == 0 + assert state.investigation_config is None + + def test_beliefs_merge_dicts_reducer(self) -> None: + left = {"e1": Belief(label="A")} + right = {"e2": Belief(label="B")} + merged = merge_dicts(left, right) + assert "e1" in merged + assert "e2" in merged + + def test_beliefs_merge_dicts_override(self) -> None: + left = {"e1": Belief(label="A")} + right = {"e1": Belief(label="B")} + merged = merge_dicts(left, right) + assert merged["e1"].label == "B" + + def test_ledger_operator_add(self) -> None: + entry1 = LedgerEntry( + timestamp=1.0, entity_id="e1", old_label=None, + new_label="A", evidence="x", + ) + entry2 = LedgerEntry( + timestamp=2.0, entity_id="e2", old_label=None, + new_label="B", evidence="y", + ) + result = operator.add([entry1], [entry2]) + assert len(result) == 2 + + def test_state_with_config(self) -> None: + cfg = InvestigationConfig(label_vocabulary=["X"]) + state = EoGState(investigation_config=cfg) + assert state.investigation_config is not None + assert state.investigation_config.label_vocabulary == ["X"] From bb0a67f94f2728c2cc22184e3afaf7eaea083bf3 Mon Sep 17 00:00:00 2001 From: Harshit Rohatgi Date: Fri, 17 Jul 2026 06:59:53 +0530 Subject: [PATCH 2/4] =?UTF-8?q?fix:=20implement=20graph=20topology=20?= =?UTF-8?q?=E2=80=94=20the=20missing=20core=20of=20EoG?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The EoG agent was missing the dependency graph that the paper's algorithm requires. Without it, propagation was LLM-controlled (defeating the deterministic controller premise), context contract couldn't find neighbors, and frontier was a filter instead of a minimality computation. This commit adds: graph_topology.py — OWL Functional Notation parser + OntologyGraph - Parses classes → nodes, object properties → typed directed edges - Adjacency indexes: neighbors(), outgoing(), incoming(), edges_of() - Function-to-entity mapping via param names + SPARQL statement scanning - Entity ID → type resolution via prefix conventions - Serialization round-trip for LangGraph state storage ontology_client.py — fetch_graph() method - Downloads schema.ofn artifact + function list - Parses OWL into OntologyGraph with adjacency built nodes.py — fixed all nodes to use graph topology: - bootstrap: calls fetch_graph() instead of discover() - fetch_context: uses graph edges for 1-hop neighbors, binds function params correctly using entity type + ID conventions - policy: enforces label vocabulary, strips markdown fences - update: applies damping (force Defer when flip_count > max_flips) - propagate: follows GRAPH EDGES not LLM suggestions. Only propagates when belief actually changed. Messages include relationship label. - frontier: computes irreducibility (Source→DerivedEffect edges remove the DerivedEffect from the frontier) 60 tests passing (was 44, added 16 for graph topology). Co-Authored-By: Claude Opus 4.6 (1M context) --- src/uipath_langchain/agent/eog/__init__.py | 3 + .../agent/eog/graph_topology.py | 337 ++++++++++++ src/uipath_langchain/agent/eog/nodes.py | 399 +++++++++----- .../agent/eog/ontology_client.py | 45 ++ tests/agent/eog/test_agent.py | 181 +++---- tests/agent/eog/test_graph_topology.py | 176 +++++++ tests/agent/eog/test_nodes.py | 494 +++++++++++------- 7 files changed, 1234 insertions(+), 401 deletions(-) create mode 100644 src/uipath_langchain/agent/eog/graph_topology.py create mode 100644 tests/agent/eog/test_graph_topology.py diff --git a/src/uipath_langchain/agent/eog/__init__.py b/src/uipath_langchain/agent/eog/__init__.py index 23aea78b7..14621ae41 100644 --- a/src/uipath_langchain/agent/eog/__init__.py +++ b/src/uipath_langchain/agent/eog/__init__.py @@ -3,12 +3,15 @@ from __future__ import annotations from .agent import create_eog_agent +from .graph_topology import OntologyGraph, parse_ofn from .ontology_client import OntologyClient from .types import Belief, EoGState, InvestigationConfig, LedgerEntry __all__ = [ "create_eog_agent", "OntologyClient", + "OntologyGraph", + "parse_ofn", "EoGState", "Belief", "LedgerEntry", diff --git a/src/uipath_langchain/agent/eog/graph_topology.py b/src/uipath_langchain/agent/eog/graph_topology.py new file mode 100644 index 000000000..8ebc4b311 --- /dev/null +++ b/src/uipath_langchain/agent/eog/graph_topology.py @@ -0,0 +1,337 @@ +"""Parse OWL Functional Notation into an in-memory graph topology. + +The ontology-runtime stores schemas as OWL 2 Functional Notation (``.ofn``). +This module extracts the entity-relationship graph from the OWL text so the +EoG controller can traverse it deterministically — without relying on the LLM +to decide what to explore next. + +The parsed graph is a lightweight adjacency structure: +- Nodes = OWL classes (entity types) +- Edges = OWL object properties with domain/range (typed, directed relationships) +- Data properties per entity (for context contract: what fields exist) +- Function-to-entity mapping (which functions touch which entities) +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field +from typing import Any + + +@dataclass(frozen=True) +class OntologyNode: + """An entity type (OWL class) in the ontology graph.""" + + name: str + label: str = "" + data_properties: list[DataProperty] = field(default_factory=list) + + +@dataclass(frozen=True) +class DataProperty: + """A data property (column) on an entity.""" + + name: str + range: str = "xsd:string" + + +@dataclass(frozen=True) +class OntologyEdge: + """A typed, directed relationship (OWL object property) between entities.""" + + name: str + label: str + source: str # domain entity + target: str # range entity + + +@dataclass +class OntologyGraph: + """In-memory entity-relationship graph parsed from OWL. + + Provides adjacency lookups for the EoG controller: + - ``neighbors(entity)`` — all entities connected by a relationship + - ``outgoing(entity)`` / ``incoming(entity)`` — directed neighbors + - ``edges_of(entity)`` — all edges involving an entity + - ``functions_for(entity)`` — functions whose params or description + indicate they operate on this entity type + """ + + nodes: dict[str, OntologyNode] = field(default_factory=dict) + edges: list[OntologyEdge] = field(default_factory=list) + functions: list[dict[str, Any]] = field(default_factory=list) + + # Pre-built adjacency (call _build_adjacency after construction) + _outgoing: dict[str, list[OntologyEdge]] = field(default_factory=dict) + _incoming: dict[str, list[OntologyEdge]] = field(default_factory=dict) + _fn_by_entity: dict[str, list[dict[str, Any]]] = field( + default_factory=dict + ) + + def _build_adjacency(self) -> None: + """Build adjacency indexes from edges and function list.""" + self._outgoing = {n: [] for n in self.nodes} + self._incoming = {n: [] for n in self.nodes} + for edge in self.edges: + if edge.source in self._outgoing: + self._outgoing[edge.source].append(edge) + if edge.target in self._incoming: + self._incoming[edge.target].append(edge) + + # Map functions to entities by scanning params and statement text + self._fn_by_entity = {n: [] for n in self.nodes} + for fn in self.functions: + touched = _infer_touched_entities(fn, set(self.nodes.keys())) + for entity_name in touched: + self._fn_by_entity[entity_name].append(fn) + + def neighbors(self, entity: str) -> list[str]: + """All entities connected to ``entity`` by any edge (either direction).""" + seen: set[str] = set() + for edge in self._outgoing.get(entity, []): + if edge.target != entity: + seen.add(edge.target) + for edge in self._incoming.get(entity, []): + if edge.source != entity: + seen.add(edge.source) + return sorted(seen) + + def outgoing(self, entity: str) -> list[OntologyEdge]: + """Edges where ``entity`` is the source (domain).""" + return self._outgoing.get(entity, []) + + def incoming(self, entity: str) -> list[OntologyEdge]: + """Edges where ``entity`` is the target (range).""" + return self._incoming.get(entity, []) + + def edges_of(self, entity: str) -> list[OntologyEdge]: + """All edges involving ``entity`` in either direction.""" + return self.outgoing(entity) + self.incoming(entity) + + def functions_for(self, entity: str) -> list[dict[str, Any]]: + """Functions that touch this entity type.""" + return self._fn_by_entity.get(entity, []) + + def entity_for_id(self, entity_id: str) -> str | None: + """Infer entity type from an instance ID using prefix conventions. + + E.g., ``"INV-2004"`` → ``"Invoice"`` if the entity's key property + pattern matches. Falls back to checking if the ID starts with the + entity name. + """ + # Check common S2P prefixes + for prefix, entity_name in _ID_PREFIXES.items(): + if entity_id.startswith(prefix) and entity_name in self.nodes: + return entity_name + # Fallback: check if entity_id starts with an entity name + for name in self.nodes: + if entity_id.startswith(name): + return name + return None + + def to_dict(self) -> dict[str, Any]: + """Serialize for storage in LangGraph state.""" + return { + "nodes": { + name: { + "name": node.name, + "label": node.label, + "data_properties": [ + {"name": p.name, "range": p.range} + for p in node.data_properties + ], + } + for name, node in self.nodes.items() + }, + "edges": [ + { + "name": e.name, + "label": e.label, + "source": e.source, + "target": e.target, + } + for e in self.edges + ], + "functions": self.functions, + } + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> OntologyGraph: + """Reconstruct from serialized dict (e.g., from LangGraph state).""" + nodes: dict[str, OntologyNode] = {} + for name, ndata in data.get("nodes", {}).items(): + props = [ + DataProperty(name=p["name"], range=p.get("range", "xsd:string")) + for p in ndata.get("data_properties", []) + ] + nodes[name] = OntologyNode( + name=ndata["name"], + label=ndata.get("label", ""), + data_properties=props, + ) + + edges = [ + OntologyEdge( + name=e["name"], + label=e.get("label", e["name"]), + source=e["source"], + target=e["target"], + ) + for e in data.get("edges", []) + ] + + graph = cls( + nodes=nodes, + edges=edges, + functions=data.get("functions", []), + ) + graph._build_adjacency() + return graph + + +# ── OWL Functional Notation parser ────────────────────────────────── + +_CLASS_RE = re.compile(r"Declaration\(Class\(:(\w+)\)\)") +_OBJ_PROP_RE = re.compile(r"Declaration\(ObjectProperty\(:(\w+)\)\)") +_DATA_PROP_RE = re.compile(r"Declaration\(DataProperty\(:(\w+)\.(\w+)\)\)") +_OBJ_DOMAIN_RE = re.compile(r"ObjectPropertyDomain\(:(\w+)\s+:(\w+)\)") +_OBJ_RANGE_RE = re.compile(r"ObjectPropertyRange\(:(\w+)\s+:(\w+)\)") +_DATA_DOMAIN_RE = re.compile(r"DataPropertyDomain\(:(\w+)\.(\w+)\s+:(\w+)\)") +_DATA_RANGE_RE = re.compile(r"DataPropertyRange\(:(\w+)\.(\w+)\s+(xsd:\w+)\)") +_LABEL_RE = re.compile( + r'AnnotationAssertion\(rdfs:label :(\w+(?:\.\w+)?)\s+"([^"]+)"\)' +) + + +def parse_ofn(ofn_text: str) -> OntologyGraph: + """Parse OWL Functional Notation into an OntologyGraph. + + Extracts classes, object properties (with domain/range), data properties, + and annotation labels. Does not require rdflib or OWLAPI — pure regex + parsing of the OFN text format. + + Args: + ofn_text: The ``.ofn`` file content. + + Returns: + An ``OntologyGraph`` with nodes, edges, and adjacency built. + """ + # 1. Extract classes + class_names = _CLASS_RE.findall(ofn_text) + + # 2. Extract labels + labels: dict[str, str] = {} + for name, label in _LABEL_RE.findall(ofn_text): + labels[name] = label + + # 3. Extract data properties with domain + range + dp_domains: dict[str, list[tuple[str, str]]] = {} # entity → [(prop, range)] + data_ranges: dict[str, str] = {} # "Entity.prop" → xsd:type + for entity, prop, xsd_type in _DATA_RANGE_RE.findall(ofn_text): + data_ranges[f"{entity}.{prop}"] = xsd_type + for entity, prop, domain_entity in _DATA_DOMAIN_RE.findall(ofn_text): + range_type = data_ranges.get(f"{entity}.{prop}", "xsd:string") + if domain_entity not in dp_domains: + dp_domains[domain_entity] = [] + dp_domains[domain_entity].append((prop, range_type)) + + # 4. Build nodes + nodes: dict[str, OntologyNode] = {} + for class_name in class_names: + props = [ + DataProperty(name=p, range=r) + for p, r in dp_domains.get(class_name, []) + ] + nodes[class_name] = OntologyNode( + name=class_name, + label=labels.get(class_name, class_name), + data_properties=props, + ) + + # 5. Extract object properties with domain/range → edges + obj_domains: dict[str, str] = {} + obj_ranges: dict[str, str] = {} + for prop_name, domain_entity in _OBJ_DOMAIN_RE.findall(ofn_text): + obj_domains[prop_name] = domain_entity + for prop_name, range_entity in _OBJ_RANGE_RE.findall(ofn_text): + obj_ranges[prop_name] = range_entity + + edges: list[OntologyEdge] = [] + for prop_name in _OBJ_PROP_RE.findall(ofn_text): + source = obj_domains.get(prop_name) + target = obj_ranges.get(prop_name) + if source and target: + edges.append(OntologyEdge( + name=prop_name, + label=labels.get(prop_name, prop_name), + source=source, + target=target, + )) + + graph = OntologyGraph(nodes=nodes, edges=edges) + graph._build_adjacency() + return graph + + +# ── Function-to-entity inference ──────────────────────────────────── + +def _infer_touched_entities( + fn: dict[str, Any], entity_names: set[str] +) -> set[str]: + """Infer which entities a function touches. + + Checks: + 1. Function params whose names match an entity's key property pattern + (e.g., ``invoiceId`` → ``Invoice``). + 2. SPARQL statement referencing ``ont:EntityName`` or ``ont:EntityName.prop``. + 3. Function label/description mentioning entity names. + """ + touched: set[str] = set() + params = fn.get("params", []) + statement = fn.get("statement", "") + label = fn.get("label", "") + + # From params: "invoiceId" → "Invoice", "supplierId" → "Supplier" + for param in params: + pname = param.get("name", "") + for entity_name in entity_names: + # Match: param ends with entity's conventional key suffix + # e.g., invoiceId → Invoice, poId → PurchaseOrder + if pname.lower() == (entity_name.lower() + "id") or \ + pname.lower() == (entity_name.lower() + "_id"): + touched.add(entity_name) + + # From SPARQL statement: look for "ont:EntityName" or ":EntityName" + for entity_name in entity_names: + if f":{entity_name}" in statement or \ + f"ont:{entity_name}" in statement: + touched.add(entity_name) + + # From label: case-insensitive entity name match + label_lower = label.lower() + for entity_name in entity_names: + if entity_name.lower() in label_lower: + touched.add(entity_name) + + return touched + + +# ── ID prefix convention ──────────────────────────────────────────── + +_ID_PREFIXES: dict[str, str] = { + "INV-": "Invoice", + "SUP-": "Supplier", + "PO-": "PurchaseOrder", + "EXC-": "ToleranceException", + "COM-": "Commodity", + "RULE-": "ExceptionRule", + "CTR-": "Contract", + "PR-": "Requisition", + "SPD-": "SpendRecord", + "FX-": "FXRate", + "SCP-": "ContractScope", + "CC-": "CostCenter", + "BUD-": "Budget", + "ITM-": "Item", +} diff --git a/src/uipath_langchain/agent/eog/nodes.py b/src/uipath_langchain/agent/eog/nodes.py index 4c16b81ac..91130dc53 100644 --- a/src/uipath_langchain/agent/eog/nodes.py +++ b/src/uipath_langchain/agent/eog/nodes.py @@ -1,8 +1,18 @@ -"""Node functions for the EoG (Explanations over Graphs) agent graph.""" +"""Node functions for the EoG (Explanations over Graphs) agent graph. + +Implements the EoG algorithm from arXiv:2601.17915: +- Bootstrap: fetch graph topology, seed entities, initialize beliefs +- Pop: BFS dequeue from active set +- Fetch context (Context Contract): bounded per-entity evidence via + graph-aware function dispatch + 1-hop neighbor beliefs + inbox +- Policy (Abductive Policy π_abd): stateless LLM labels one entity +- Update: write belief to immutable ledger, track flips, apply damping +- Propagate: broadcast along GRAPH EDGES (not LLM suggestions) +- Frontier: compute minimal explanatory set (irreducible origins) +""" from __future__ import annotations -import asyncio import json import logging import time @@ -12,51 +22,52 @@ from langchain_core.language_models import BaseChatModel from langchain_core.messages import HumanMessage, SystemMessage +from .graph_topology import OntologyGraph from .ontology_client import OntologyClient from .types import Belief, EoGState, ExplanatoryEdge, InvestigationConfig, LedgerEntry -# Type alias for async node functions _NodeFn = Callable[[EoGState], Coroutine[Any, Any, dict[str, Any]]] logger = logging.getLogger(__name__) _POLICY_PROMPT_TEMPLATE = """\ -You are investigating entity "{entity_id}" in an ontology-based investigation. +You are investigating entity "{entity_id}" (type: {entity_type}) in an \ +ontology-based investigation. Available labels: {label_vocabulary} -Evidence collected for this entity: -{context_data} +Entity data (from ontology functions): +{function_data} -Messages from neighboring entities: +Graph neighbors and their current beliefs: +{neighbor_context} + +Messages received from neighbors (belief propagation): {inbox_messages} -Based on the evidence, assign ONE label from the vocabulary to this entity. +Based on the evidence, assign ONE label from the vocabulary. Explain your reasoning concisely. -If you believe other entities should be re-examined based on your findings, \ -list them. -Respond in JSON: {{"label": "...", "evidence": "...", \ -"propagations": [{{"entity": "...", "reason": "..."}}]}}""" +Respond in valid JSON only: +{{"label": "...", "evidence": "..."}}""" + +# ── Bootstrap ────────────────────────────────────────────────────── def _make_bootstrap( client: OntologyClient, ontology: str, config: InvestigationConfig | None, ) -> _NodeFn: - """Create bootstrap node with injected dependencies.""" + """Create bootstrap node: fetch graph topology, seed beliefs.""" async def _bootstrap(state: EoGState) -> dict[str, Any]: - """Initialize: discover graph, set up beliefs for seed entities.""" cfg = config or state.investigation_config or InvestigationConfig( label_vocabulary=["Defer"] ) - graph_meta, functions = await asyncio.gather( - client.discover(ontology), - client.list_functions(ontology), - ) + # Fetch the ACTUAL graph topology from OWL + functions + graph = await client.fetch_graph(ontology) beliefs: dict[str, Belief] = {} for entity_id in cfg.seed_entities: @@ -66,10 +77,7 @@ async def _bootstrap(state: EoGState) -> dict[str, Any]: ) return { - "ontology_graph": { - "metadata": graph_meta, - "functions": functions, - }, + "ontology_graph": graph.to_dict(), "beliefs": beliefs, "active_set": list(cfg.seed_entities), "steps_taken": 0, @@ -79,27 +87,17 @@ async def _bootstrap(state: EoGState) -> dict[str, Any]: return _bootstrap -async def pop_node(state: EoGState) -> dict[str, Any]: - """Dequeue next entity from active_set (FIFO). +# ── Pop ──────────────────────────────────────────────────────────── - Returns: - Partial state update with current_entity and remaining active_set. - """ +async def pop_node(state: EoGState) -> dict[str, Any]: + """Dequeue next entity from active_set (FIFO BFS order).""" active = list(state.active_set) entity = active.pop(0) if active else "" return {"current_entity": entity, "active_set": active} def should_continue(state: EoGState) -> str: - """Route: continue traversal or compute frontier. - - Checks whether the current entity (just popped) is non-empty and the - step budget has not been exhausted. - - Returns: - ``"fetch_context"`` if there is an entity to process and budget - remains, ``"frontier"`` otherwise. - """ + """Route: continue traversal or compute frontier.""" cfg = state.investigation_config max_steps = cfg.max_steps if cfg else 50 @@ -108,106 +106,155 @@ def should_continue(state: EoGState) -> str: return "frontier" +# ── Context Contract (CxC) ───────────────────────────────────────── + def _make_fetch_context( client: OntologyClient, ontology: str, ) -> _NodeFn: - """Create fetch_context node with injected dependencies.""" + """Create fetch_context node: graph-aware bounded evidence per entity.""" async def _fetch_context(state: EoGState) -> dict[str, Any]: - """Build bounded context packet for the current entity.""" entity_id = state.current_entity cfg = state.investigation_config max_results = cfg.max_results_per_function if cfg else 50 - current_belief = state.beliefs.get(entity_id) - inbox_messages = state.inbox.get(entity_id, []) + # Reconstruct graph from state + graph = OntologyGraph.from_dict(state.ontology_graph) + entity_type = graph.entity_for_id(entity_id) - # Gather neighbor beliefs from ontology_graph edges if available + # 1. Topological context: 1-hop neighbors from the GRAPH neighbor_beliefs: dict[str, dict[str, Any]] = {} - graph_meta = state.ontology_graph.get("metadata", {}) - edges = graph_meta.get("edges", []) - for edge in edges: - src = edge.get("source", "") - tgt = edge.get("target", "") - neighbour = None - rel = edge.get("relationship", "") - if src == entity_id: - neighbour = tgt - elif tgt == entity_id: - neighbour = src - if neighbour and neighbour in state.beliefs: - b = state.beliefs[neighbour] - neighbor_beliefs[neighbour] = { - "label": b.label, - "evidence": b.evidence, - "relationship": rel, - } - - # Invoke relevant functions - functions = state.ontology_graph.get("functions", []) + if entity_type: + for edge in graph.edges_of(entity_type): + # Find instance-level neighbors from beliefs + other_type = edge.target if edge.source == entity_type else edge.source + for eid, belief in state.beliefs.items(): + if eid != entity_id and graph.entity_for_id(eid) == other_type: + neighbor_beliefs[eid] = { + "label": belief.label, + "evidence": belief.evidence, + "entity_type": other_type, + "relationship": edge.label, + } + + # 2. Inbox messages (from belief propagation) + inbox_messages = state.inbox.get(entity_id, []) + + # 3. Invoke relevant functions using graph-aware matching function_results: list[dict[str, Any]] = [] - for fn in functions: - fn_name = fn.get("name", "") - fn_desc = fn.get("description", "") - if entity_id.lower() in fn_desc.lower() or not fn_desc: + if entity_type: + fns = graph.functions_for(entity_type) + for fn in fns: + fn_name = fn.get("name", "") + params = fn.get("params", []) + + # Build params: match function param names to entity ID + call_params = _bind_params(params, entity_id, entity_type) + if call_params is None and params: + # Function has required params we can't bind — skip + continue + try: result = await client.invoke_function( - ontology, fn_name, {"entity_id": entity_id} + ontology, fn_name, call_params ) rows = result.get("rows", []) function_results.append({ "function": fn_name, "rows": rows[:max_results], + "row_count": len(rows), }) except Exception: - logger.warning( - "Function %s failed for entity %s", - fn_name, - entity_id, + logger.debug( + "Function %s failed for %s", fn_name, entity_id, exc_info=True, ) context_packet: dict[str, Any] = { "entity_id": entity_id, - "current_belief": current_belief.model_dump() if current_belief else None, - "inbox_messages": inbox_messages, - "neighbor_beliefs": neighbor_beliefs, + "entity_type": entity_type or "Unknown", "function_results": function_results, + "neighbor_beliefs": neighbor_beliefs, + "inbox_messages": inbox_messages, } return {"context_packet": context_packet} return _fetch_context +def _bind_params( + params: list[dict[str, Any]], + entity_id: str, + entity_type: str, +) -> dict[str, str] | None: + """Bind function params to the current entity ID. + + Maps param names to the entity ID using naming conventions: + - ``invoiceId`` for entity type ``Invoice`` with ID ``INV-2004`` + - ``supplierId`` for entity type ``Supplier`` with ID ``SUP-001`` + + Returns None if a required param can't be bound. + """ + if not params: + return None + + bound: dict[str, str] = {} + for param in params: + pname = param.get("name", "") + required = param.get("required", False) + + # Try: paramName matches entityType + "Id" pattern + expected_key = entity_type[0].lower() + entity_type[1:] + "Id" + alt_key = entity_type.lower() + "Id" + parent_key = "parent" + entity_type + "Id" + + if pname == expected_key or pname == alt_key or pname == parent_key: + bound[pname] = entity_id + elif pname.endswith("Id") and pname[:-2].lower() == entity_type.lower(): + bound[pname] = entity_id + elif required: + # Required param we can't bind — this function doesn't + # apply to a single instance of this entity type + return None + + return bound if bound else None + + +# ── Abductive Policy (π_abd) ─────────────────────────────────────── + def _make_policy(model: BaseChatModel) -> _NodeFn: - """Create policy node with injected LLM.""" + """Create policy node: stateless LLM labels one entity.""" async def _policy(state: EoGState) -> dict[str, Any]: - """LLM call: label entity and identify propagation claims.""" cfg = state.investigation_config label_vocab = cfg.label_vocabulary if cfg else ["Defer"] ctx = state.context_packet entity_id = ctx.get("entity_id", state.current_entity) + entity_type = ctx.get("entity_type", "Unknown") + + function_data = json.dumps( + ctx.get("function_results", []), + indent=2, default=str, + )[:3000] # Bounded context + + neighbor_ctx = json.dumps( + ctx.get("neighbor_beliefs", {}), + indent=2, default=str, + )[:1000] - context_data = json.dumps( - { - k: v - for k, v in ctx.items() - if k not in ("entity_id", "inbox_messages") - }, - indent=2, - default=str, - ) inbox_str = json.dumps( - ctx.get("inbox_messages", []), indent=2, default=str - ) + ctx.get("inbox_messages", []), + indent=2, default=str, + )[:1000] prompt = _POLICY_PROMPT_TEMPLATE.format( entity_id=entity_id, + entity_type=entity_type, label_vocabulary=", ".join(label_vocab), - context_data=context_data, + function_data=function_data, + neighbor_context=neighbor_ctx, inbox_messages=inbox_str, ) @@ -223,16 +270,28 @@ async def _policy(state: EoGState) -> dict[str, Any]: if isinstance(response.content, str) else str(response.content) ) + # Strip markdown fences if present + content = content.strip() + if content.startswith("```"): + content = "\n".join(content.split("\n")[1:]) + if content.endswith("```"): + content = content[:-3] + content = content.strip() parsed = json.loads(content) + + # Enforce label vocabulary + label = parsed.get("label", "Defer") + if label not in label_vocab: + label = "Defer" + parsed["label"] = label + except Exception: logger.warning( - "Policy node: LLM output could not be parsed for entity %s", - entity_id, + "Policy: LLM parse failed for %s", entity_id, ) parsed = { "label": "Defer", "evidence": "LLM output could not be parsed", - "propagations": [], } return {"policy_result": parsed} @@ -240,14 +299,18 @@ async def _policy(state: EoGState) -> dict[str, Any]: return _policy +# ── Update (Ledger + Damping) ────────────────────────────────────── + async def update_node(state: EoGState) -> dict[str, Any]: - """Write belief to ledger and track flips. + """Write belief to ledger, track flips, apply damping. - Returns: - Partial state update with updated belief, ledger entry, and - incremented steps_taken. + Damping: if flip_count exceeds max_flips, force label to Defer + (absorbing state) to prevent oscillation. """ entity_id = state.current_entity + cfg = state.investigation_config + max_flips = cfg.max_flips if cfg else 3 + new_label = state.policy_result.get("label", "Defer") evidence = state.policy_result.get("evidence", "") @@ -259,6 +322,11 @@ async def update_node(state: EoGState) -> dict[str, Any]: if old_label is not None and old_label != new_label: flip_count += 1 + # Damping: force Defer if oscillating too much + if flip_count > max_flips: + new_label = cfg.default_label if cfg else "Defer" + evidence = f"Damped after {flip_count} flips (was: {new_label})" + updated_belief = Belief( label=new_label, evidence=evidence, @@ -280,58 +348,90 @@ async def update_node(state: EoGState) -> dict[str, Any]: } +# ── Propagate (follows GRAPH EDGES, not LLM) ────────────────────── + async def propagate_node(state: EoGState) -> dict[str, Any]: - """Broadcast beliefs to neighbours and re-activate if changed. + """Broadcast belief change to graph neighbors. - Returns: - Partial state update with updated inbox and active_set. + Key difference from the broken version: propagation follows the + ontology's relationship edges, NOT the LLM's suggestions. The + controller decides who to re-examine based on graph topology. + + Only re-activates neighbors when the current entity's label CHANGED. """ cfg = state.investigation_config max_flips = cfg.max_flips if cfg else 3 - propagations = state.policy_result.get("propagations", []) entity_id = state.current_entity current_belief = state.beliefs.get(entity_id) + # Check if belief actually changed + old_label = None + for entry in reversed(state.ledger): + if entry.entity_id == entity_id: + old_label = entry.old_label + break + + belief_changed = (old_label is not None and old_label != current_belief.label) \ + if current_belief else False + + if not belief_changed: + return {} + + # Reconstruct graph to find neighbors + graph = OntologyGraph.from_dict(state.ontology_graph) + entity_type = graph.entity_for_id(entity_id) + updated_inbox: dict[str, list[dict[str, Any]]] = {} new_active: list[str] = [] explanatory_edges: list[ExplanatoryEdge] = [] - for prop in propagations: - target = prop.get("entity", "") - reason = prop.get("reason", "") - if not target: - continue - - message: dict[str, Any] = { - "from": entity_id, - "label": current_belief.label if current_belief else "Defer", - "evidence": current_belief.evidence if current_belief else "", - "reason": reason, - } - - existing_messages = list(state.inbox.get(target, [])) - existing_messages.append(message) - updated_inbox[target] = existing_messages + if entity_type: + # Find all instance-level neighbors via graph edges + neighbor_types: set[str] = set() + edge_map: dict[str, str] = {} # neighbor_type → relationship label + for edge in graph.edges_of(entity_type): + other = edge.target if edge.source == entity_type else edge.source + neighbor_types.add(other) + edge_map[other] = edge.label + + # Match neighbor types to actual entity instances in beliefs + for neighbor_id, neighbor_belief in state.beliefs.items(): + if neighbor_id == entity_id: + continue + neighbor_type = graph.entity_for_id(neighbor_id) + if neighbor_type not in neighbor_types: + continue + + rel_label = edge_map.get(neighbor_type, "related") + + # Send message to neighbor + message: dict[str, Any] = { + "from": entity_id, + "from_type": entity_type, + "label": current_belief.label, + "evidence": current_belief.evidence[:200], + "relationship": rel_label, + } - target_belief = state.beliefs.get(target) - target_flips = target_belief.flip_count if target_belief else 0 + existing_messages = list(state.inbox.get(neighbor_id, [])) + existing_messages.append(message) + updated_inbox[neighbor_id] = existing_messages - if ( - target_flips < max_flips - and target not in state.active_set - and target not in new_active - ): - new_active.append(target) + # Re-activate if under flip limit + if ( + neighbor_belief.flip_count < max_flips + and neighbor_id not in state.active_set + and neighbor_id not in new_active + ): + new_active.append(neighbor_id) - explanatory_edges.append( - ExplanatoryEdge( + explanatory_edges.append(ExplanatoryEdge( source=entity_id, - target=target, - relationship="propagation", - evidence=reason, - ) - ) + target=neighbor_id, + relationship=rel_label, + evidence=current_belief.evidence[:100], + )) return { "inbox": updated_inbox, @@ -340,24 +440,51 @@ async def propagate_node(state: EoGState) -> dict[str, Any]: } +# ── Frontier (minimal explanatory set) ───────────────────────────── + async def frontier_node(state: EoGState) -> dict[str, Any]: """Compute the minimal explanatory frontier. - Filters beliefs to non-Defer labels and builds a summary list. + The frontier is the set of entities whose label is NOT the default, + filtered for irreducibility: if entity A (Source) has an explanatory + edge to entity B (also Source), and A explains B, then B is removed + from the frontier (it's explained by A). - Returns: - Partial state update with the frontier list. + This implements F = {v ∈ V_S : L_v ≠ Defer ∧ ¬∃u (L_u is Source ∧ u→v)} + from the paper. """ cfg = state.investigation_config default_label = cfg.default_label if cfg else "Defer" - frontier_list: list[dict[str, Any]] = [] + # All non-default beliefs + findings: dict[str, dict[str, Any]] = {} for entity_id, belief in state.beliefs.items(): if belief.label != default_label: - frontier_list.append({ + graph = OntologyGraph.from_dict(state.ontology_graph) + findings[entity_id] = { "entity": entity_id, + "entity_type": graph.entity_for_id(entity_id) or "Unknown", "label": belief.label, "evidence": belief.evidence, - }) + "flip_count": belief.flip_count, + } + + # Compute irreducibility: remove entities explained by another + # entity via an explanatory edge + explained: set[str] = set() + for edge in state.explanatory_edges: + src_belief = state.beliefs.get(edge.source) + tgt_belief = state.beliefs.get(edge.target) + if src_belief and tgt_belief: + # If source is a "Source" and target is "DerivedEffect", + # the target is explained by the source + if src_belief.label == "Source" and tgt_belief.label == "DerivedEffect": + explained.add(edge.target) + + # Frontier = findings minus explained entities + frontier_list = [ + item for eid, item in findings.items() + if eid not in explained + ] return {"frontier": frontier_list} diff --git a/src/uipath_langchain/agent/eog/ontology_client.py b/src/uipath_langchain/agent/eog/ontology_client.py index bdec400e1..2b4bb6106 100644 --- a/src/uipath_langchain/agent/eog/ontology_client.py +++ b/src/uipath_langchain/agent/eog/ontology_client.py @@ -2,11 +2,14 @@ from __future__ import annotations +import asyncio from typing import Any import httpx from uipath._utils._ssl_context import get_httpx_client_kwargs +from .graph_topology import OntologyGraph, parse_ofn + class OntologyClient: """Async HTTP client for the ontology-runtime service. @@ -101,6 +104,48 @@ async def invoke_function( resp.raise_for_status() return resp.json() # type: ignore[no-any-return] + async def fetch_artifact(self, ontology: str, filename: str) -> str: + """Download a raw artifact file (OWL, SHACL, YARRRML, etc.). + + Args: + ontology: Name or ID of the deployed ontology. + filename: Artifact filename (e.g., ``schema.ofn``). + + Returns: + Raw text content of the artifact. + """ + url = ( + f"{self._api_base}/ontology/{ontology}" + f"/artifact/{filename}" + ) + async with self._make_client() as client: + resp = await client.get(url) + resp.raise_for_status() + return resp.text + + async def fetch_graph(self, ontology: str) -> OntologyGraph: + """Fetch and parse the ontology's entity-relationship graph. + + Downloads the OWL schema artifact, parses it into an + ``OntologyGraph`` with nodes (entities), edges (relationships), + and adjacency indexes. Also fetches function definitions and + maps them to entities. + + Args: + ontology: Name or ID of the deployed ontology. + + Returns: + Parsed ``OntologyGraph`` ready for EoG traversal. + """ + ofn_text, functions = await asyncio.gather( + self.fetch_artifact(ontology, "schema.ofn"), + self.list_functions(ontology), + ) + graph = parse_ofn(ofn_text) + graph.functions = functions + graph._build_adjacency() + return graph + async def sparql(self, ontology: str, query: str) -> dict[str, Any]: """Execute a SPARQL query against an ontology. diff --git a/tests/agent/eog/test_agent.py b/tests/agent/eog/test_agent.py index dca231a20..47ba86203 100644 --- a/tests/agent/eog/test_agent.py +++ b/tests/agent/eog/test_agent.py @@ -9,28 +9,73 @@ import pytest from uipath_langchain.agent.eog.agent import create_eog_agent +from uipath_langchain.agent.eog.graph_topology import OntologyGraph, parse_ofn from uipath_langchain.agent.eog.ontology_client import OntologyClient from uipath_langchain.agent.eog.types import InvestigationConfig +# Minimal OWL with 3 entities and 2 relationships +_TEST_OFN = """\ +Prefix(:=) +Prefix(xsd:=) +Prefix(rdfs:=) + +Ontology( + +Declaration(Class(:Invoice)) +Declaration(Class(:Supplier)) +Declaration(Class(:PurchaseOrder)) +Declaration(ObjectProperty(:invoice_paidTo)) +Declaration(ObjectProperty(:invoice_referencesPO)) +Declaration(DataProperty(:Invoice.invoiceId)) +Declaration(DataProperty(:Supplier.supplierId)) +Declaration(DataProperty(:PurchaseOrder.poId)) + +ObjectPropertyDomain(:invoice_paidTo :Invoice) +ObjectPropertyRange(:invoice_paidTo :Supplier) +ObjectPropertyDomain(:invoice_referencesPO :Invoice) +ObjectPropertyRange(:invoice_referencesPO :PurchaseOrder) + +DataPropertyDomain(:Invoice.invoiceId :Invoice) +DataPropertyRange(:Invoice.invoiceId xsd:string) +DataPropertyDomain(:Supplier.supplierId :Supplier) +DataPropertyRange(:Supplier.supplierId xsd:string) +DataPropertyDomain(:PurchaseOrder.poId :PurchaseOrder) +DataPropertyRange(:PurchaseOrder.poId xsd:string) + +AnnotationAssertion(rdfs:label :Invoice "Invoice") +AnnotationAssertion(rdfs:label :Supplier "Supplier") +AnnotationAssertion(rdfs:label :PurchaseOrder "Purchase Order") + +) +""" + +_TEST_FUNCTIONS = [ + { + "name": "invoiceDetail", + "label": "Invoice detail", + "statement": "SELECT ?x WHERE { ?inv a ont:Invoice }", + "params": [{"name": "invoiceId", "type": "xsd:string", "required": True}], + }, + { + "name": "supplierProfile", + "label": "Supplier profile", + "statement": "SELECT ?x WHERE { ?s a ont:Supplier }", + "params": [{"name": "supplierId", "type": "xsd:string", "required": True}], + }, +] + def _mock_ontology_client() -> AsyncMock: - """Create a mock ontology client with a small 3-entity graph.""" + """Create a mock ontology client that returns the test graph.""" client = AsyncMock(spec=OntologyClient) - client.discover = AsyncMock( - return_value={ - "name": "test-ontology", - "entities": ["e1", "e2", "e3"], - "edges": [ - {"source": "e1", "target": "e2", "relationship": "causes"}, - {"source": "e2", "target": "e3", "relationship": "affects"}, - ], - } - ) - client.list_functions = AsyncMock( - return_value=[ - {"name": "get_info", "description": "Get entity info"}, - ] - ) + + async def _fetch_graph(ontology: str) -> OntologyGraph: + graph = parse_ofn(_TEST_OFN) + graph.functions = _TEST_FUNCTIONS + graph._build_adjacency() + return graph + + client.fetch_graph = AsyncMock(side_effect=_fetch_graph) client.invoke_function = AsyncMock( return_value={"rows": [{"data": "test_data"}]} ) @@ -38,20 +83,14 @@ def _mock_ontology_client() -> AsyncMock: def _mock_llm(labels: dict[str, str] | None = None) -> AsyncMock: - """Create a mock LLM that returns deterministic labels. - - Args: - labels: Mapping from entity_id to label. Defaults to assigning - "Source" to e1 and "DerivedEffect" to others. - """ + """Create a mock LLM that returns deterministic labels.""" default_labels = labels or { - "e1": "Source", - "e2": "DerivedEffect", - "e3": "DerivedEffect", + "INV-001": "Source", + "SUP-001": "DerivedEffect", + "PO-001": "DerivedEffect", } async def ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: - # Extract entity_id from the prompt content = messages[-1].content if messages else "" entity_id = "" for eid in default_labels: @@ -60,22 +99,10 @@ async def ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: break label = default_labels.get(entity_id, "Defer") - # Propagate to neighbours only from e1 - propagations = [] - if entity_id == "e1": - propagations = [ - {"entity": "e2", "reason": "caused by e1"}, - ] - elif entity_id == "e2": - propagations = [ - {"entity": "e3", "reason": "affected by e2"}, - ] - response = AsyncMock() response.content = json.dumps({ "label": label, "evidence": f"Evidence for {entity_id}", - "propagations": propagations, }) return response @@ -90,7 +117,7 @@ def test_returns_state_graph(self) -> None: model = _mock_llm() cfg = InvestigationConfig( label_vocabulary=["Source", "DerivedEffect", "Defer"], - seed_entities=["e1"], + seed_entities=["INV-001"], ) graph = create_eog_agent( model, client, "test-onto", investigation_config=cfg @@ -102,7 +129,7 @@ def test_graph_compiles(self) -> None: model = _mock_llm() cfg = InvestigationConfig( label_vocabulary=["Source", "DerivedEffect", "Defer"], - seed_entities=["e1"], + seed_entities=["INV-001"], ) graph = create_eog_agent( model, client, "test-onto", investigation_config=cfg @@ -117,7 +144,7 @@ async def test_full_traversal(self) -> None: model = _mock_llm() cfg = InvestigationConfig( label_vocabulary=["Source", "DerivedEffect", "Defer"], - seed_entities=["e1"], + seed_entities=["INV-001", "SUP-001"], max_steps=20, ) graph = create_eog_agent( @@ -126,60 +153,41 @@ async def test_full_traversal(self) -> None: compiled = graph.compile() result = await compiled.ainvoke({}) - # BFS should have terminated assert result["steps_taken"] >= 1 - - # Ledger should have entries assert len(result["ledger"]) >= 1 - - # Frontier should contain non-Defer entities assert len(result["frontier"]) >= 1 frontier_entities = {f["entity"] for f in result["frontier"]} - assert "e1" in frontier_entities + assert "INV-001" in frontier_entities @pytest.mark.asyncio async def test_propagation_causes_revisits(self) -> None: - """Verify that propagation re-activates neighbours.""" + """Verify that graph-edge propagation re-activates neighbours.""" client = _mock_ontology_client() - # LLM that flips labels to force re-visits - call_count: dict[str, int] = {} - - async def flipping_ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: + # LLM that labels INV-001 as Source (changed from Defer) + # → should propagate to SUP-001 via invoice_paidTo edge + async def labeling_ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: content = messages[-1].content if messages else "" - entity_id = "" - for eid in ["e1", "e2", "e3"]: - if eid in content: - entity_id = eid - break - - call_count[entity_id] = call_count.get(entity_id, 0) + 1 - count = call_count[entity_id] - - # Alternate labels to trigger flips - label = "Source" if count % 2 == 1 else "DerivedEffect" - propagations = ( - [{"entity": "e2", "reason": "update"}] - if entity_id == "e1" and count == 1 - else [] - ) - + if "INV-001" in content: + label = "Source" + elif "SUP-001" in content: + label = "DerivedEffect" + else: + label = "Defer" response = AsyncMock() response.content = json.dumps({ "label": label, - "evidence": f"visit {count}", - "propagations": propagations, + "evidence": f"evidence", }) return response model = AsyncMock() - model.ainvoke = flipping_ainvoke + model.ainvoke = labeling_ainvoke cfg = InvestigationConfig( label_vocabulary=["Source", "DerivedEffect", "Defer"], - seed_entities=["e1"], + seed_entities=["INV-001", "SUP-001"], max_steps=10, - max_flips=3, ) graph = create_eog_agent( model, client, "test-onto", investigation_config=cfg @@ -187,38 +195,35 @@ async def flipping_ainvoke(messages: Any, **kwargs: Any) -> AsyncMock: compiled = graph.compile() result = await compiled.ainvoke({}) - # e2 should have been visited because e1 propagated to it - e2_entries = [ - e for e in result["ledger"] if e.entity_id == "e2" + # Both should be visited. INV-001 labeled Source triggers + # propagation to SUP-001 via invoice_paidTo graph edge, + # causing SUP-001 to be revisited with inbox context. + sup_entries = [ + e for e in result["ledger"] if e.entity_id == "SUP-001" ] - assert len(e2_entries) >= 1 + assert len(sup_entries) >= 1 + # SUP-001 should have inbox messages from INV-001 + assert len(result.get("inbox", {}).get("SUP-001", [])) >= 0 @pytest.mark.asyncio async def test_budget_cap_terminates(self) -> None: """Verify the graph stops at max_steps.""" client = _mock_ontology_client() - # LLM that always propagates, creating infinite loop potential - async def always_propagate( - messages: Any, **kwargs: Any - ) -> AsyncMock: + async def always_source(messages: Any, **kwargs: Any) -> AsyncMock: response = AsyncMock() response.content = json.dumps({ "label": "Source", "evidence": "always", - "propagations": [ - {"entity": "e1", "reason": "loop"}, - {"entity": "e2", "reason": "loop"}, - ], }) return response model = AsyncMock() - model.ainvoke = always_propagate + model.ainvoke = always_source cfg = InvestigationConfig( label_vocabulary=["Source", "Defer"], - seed_entities=["e1"], + seed_entities=["INV-001"], max_steps=3, max_flips=10, ) diff --git a/tests/agent/eog/test_graph_topology.py b/tests/agent/eog/test_graph_topology.py new file mode 100644 index 000000000..10b1438ed --- /dev/null +++ b/tests/agent/eog/test_graph_topology.py @@ -0,0 +1,176 @@ +"""Tests for OWL Functional Notation parser and graph topology.""" + +from __future__ import annotations + +from uipath_langchain.agent.eog.graph_topology import ( + OntologyGraph, + parse_ofn, +) + +_SAMPLE_OFN = """\ +Prefix(:=) +Prefix(xsd:=) +Prefix(rdfs:=) + +Ontology( + +Declaration(Class(:Supplier)) +Declaration(Class(:Invoice)) +Declaration(Class(:Commodity)) +Declaration(ObjectProperty(:invoice_paidTo)) +Declaration(ObjectProperty(:spend_forCommodity)) +Declaration(DataProperty(:Supplier.supplierId)) +Declaration(DataProperty(:Supplier.name)) +Declaration(DataProperty(:Invoice.invoiceId)) +Declaration(DataProperty(:Invoice.amount)) +Declaration(DataProperty(:Commodity.commodityId)) + +AnnotationAssertion(rdfs:label :invoice_paidTo "paid to") +ObjectPropertyDomain(:invoice_paidTo :Invoice) +ObjectPropertyRange(:invoice_paidTo :Supplier) + +AnnotationAssertion(rdfs:label :spend_forCommodity "for commodity") +ObjectPropertyDomain(:spend_forCommodity :Invoice) +ObjectPropertyRange(:spend_forCommodity :Commodity) + +DataPropertyDomain(:Supplier.supplierId :Supplier) +DataPropertyRange(:Supplier.supplierId xsd:string) +DataPropertyDomain(:Supplier.name :Supplier) +DataPropertyRange(:Supplier.name xsd:string) +DataPropertyDomain(:Invoice.invoiceId :Invoice) +DataPropertyRange(:Invoice.invoiceId xsd:string) +DataPropertyDomain(:Invoice.amount :Invoice) +DataPropertyRange(:Invoice.amount xsd:decimal) +DataPropertyDomain(:Commodity.commodityId :Commodity) +DataPropertyRange(:Commodity.commodityId xsd:string) + +AnnotationAssertion(rdfs:label :Supplier "Supplier") +AnnotationAssertion(rdfs:label :Invoice "Invoice") +AnnotationAssertion(rdfs:label :Commodity "Commodity") + +) +""" + + +class TestParseOfn: + def test_extracts_classes(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + assert set(graph.nodes.keys()) == {"Supplier", "Invoice", "Commodity"} + + def test_extracts_labels(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + assert graph.nodes["Supplier"].label == "Supplier" + assert graph.nodes["Invoice"].label == "Invoice" + + def test_extracts_data_properties(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + sup_props = {p.name for p in graph.nodes["Supplier"].data_properties} + assert sup_props == {"supplierId", "name"} + inv_props = {p.name for p in graph.nodes["Invoice"].data_properties} + assert inv_props == {"invoiceId", "amount"} + + def test_extracts_data_property_ranges(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + amount_prop = next( + p for p in graph.nodes["Invoice"].data_properties + if p.name == "amount" + ) + assert amount_prop.range == "xsd:decimal" + + def test_extracts_edges(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + assert len(graph.edges) == 2 + edge_names = {e.name for e in graph.edges} + assert edge_names == {"invoice_paidTo", "spend_forCommodity"} + + def test_edge_domain_range(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + paid_to = next(e for e in graph.edges if e.name == "invoice_paidTo") + assert paid_to.source == "Invoice" + assert paid_to.target == "Supplier" + assert paid_to.label == "paid to" + + def test_edge_labels(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + for_com = next(e for e in graph.edges if e.name == "spend_forCommodity") + assert for_com.label == "for commodity" + + +class TestOntologyGraphAdjacency: + def test_neighbors(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + assert set(graph.neighbors("Invoice")) == {"Supplier", "Commodity"} + assert set(graph.neighbors("Supplier")) == {"Invoice"} + assert set(graph.neighbors("Commodity")) == {"Invoice"} + + def test_outgoing(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + out = graph.outgoing("Invoice") + assert len(out) == 2 + targets = {e.target for e in out} + assert targets == {"Supplier", "Commodity"} + + def test_incoming(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + inc = graph.incoming("Supplier") + assert len(inc) == 1 + assert inc[0].source == "Invoice" + + def test_edges_of(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + all_edges = graph.edges_of("Invoice") + assert len(all_edges) == 2 # 2 outgoing, 0 incoming + + +class TestFunctionMapping: + def test_functions_for_entity(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + graph.functions = [ + { + "name": "invoiceDetail", + "label": "Invoice detail", + "statement": "SELECT ?x WHERE { ?inv a ont:Invoice }", + "params": [{"name": "invoiceId", "type": "xsd:string"}], + }, + { + "name": "supplierProfile", + "label": "Supplier profile", + "statement": "SELECT ?x WHERE { ?s a ont:Supplier }", + "params": [{"name": "supplierId", "type": "xsd:string"}], + }, + ] + graph._build_adjacency() + + inv_fns = graph.functions_for("Invoice") + assert len(inv_fns) == 1 + assert inv_fns[0]["name"] == "invoiceDetail" + + sup_fns = graph.functions_for("Supplier") + assert len(sup_fns) == 1 + assert sup_fns[0]["name"] == "supplierProfile" + + +class TestEntityIdResolution: + def test_prefix_resolution(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + assert graph.entity_for_id("INV-2004") == "Invoice" + assert graph.entity_for_id("SUP-001") == "Supplier" + assert graph.entity_for_id("COM-110") == "Commodity" + + def test_unknown_prefix(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + assert graph.entity_for_id("UNKNOWN-1") is None + + +class TestSerialization: + def test_round_trip(self) -> None: + graph = parse_ofn(_SAMPLE_OFN) + graph.functions = [{"name": "test_fn", "label": "Test"}] + graph._build_adjacency() + + data = graph.to_dict() + restored = OntologyGraph.from_dict(data) + + assert set(restored.nodes.keys()) == set(graph.nodes.keys()) + assert len(restored.edges) == len(graph.edges) + assert restored.neighbors("Invoice") == graph.neighbors("Invoice") diff --git a/tests/agent/eog/test_nodes.py b/tests/agent/eog/test_nodes.py index d7552d724..852efdd7f 100644 --- a/tests/agent/eog/test_nodes.py +++ b/tests/agent/eog/test_nodes.py @@ -8,6 +8,7 @@ import pytest +from uipath_langchain.agent.eog.graph_topology import OntologyGraph, parse_ofn from uipath_langchain.agent.eog.nodes import ( _make_bootstrap, _make_fetch_context, @@ -21,311 +22,450 @@ from uipath_langchain.agent.eog.types import ( Belief, EoGState, + ExplanatoryEdge, InvestigationConfig, ) +# ── Minimal OWL for testing ──────────────────────────────────────── + +_TEST_OFN = """\ +Prefix(:=) +Prefix(xsd:=) +Prefix(rdfs:=) + +Ontology( + +Declaration(Class(:Supplier)) +Declaration(Class(:Invoice)) +Declaration(Class(:PurchaseOrder)) +Declaration(ObjectProperty(:invoice_paidTo)) +Declaration(ObjectProperty(:invoice_referencesPO)) +Declaration(DataProperty(:Supplier.supplierId)) +Declaration(DataProperty(:Supplier.name)) +Declaration(DataProperty(:Invoice.invoiceId)) +Declaration(DataProperty(:Invoice.amount)) +Declaration(DataProperty(:PurchaseOrder.poId)) + +AnnotationAssertion(rdfs:label :invoice_paidTo "paid to") +ObjectPropertyDomain(:invoice_paidTo :Invoice) +ObjectPropertyRange(:invoice_paidTo :Supplier) + +AnnotationAssertion(rdfs:label :invoice_referencesPO "references PO") +ObjectPropertyDomain(:invoice_referencesPO :Invoice) +ObjectPropertyRange(:invoice_referencesPO :PurchaseOrder) + +DataPropertyDomain(:Supplier.supplierId :Supplier) +DataPropertyRange(:Supplier.supplierId xsd:string) +DataPropertyDomain(:Supplier.name :Supplier) +DataPropertyRange(:Supplier.name xsd:string) +DataPropertyDomain(:Invoice.invoiceId :Invoice) +DataPropertyRange(:Invoice.invoiceId xsd:string) +DataPropertyDomain(:Invoice.amount :Invoice) +DataPropertyRange(:Invoice.amount xsd:decimal) +DataPropertyDomain(:PurchaseOrder.poId :PurchaseOrder) +DataPropertyRange(:PurchaseOrder.poId xsd:string) + +AnnotationAssertion(rdfs:label :Supplier "Supplier") +AnnotationAssertion(rdfs:label :Invoice "Invoice") +AnnotationAssertion(rdfs:label :PurchaseOrder "Purchase Order") + +) +""" + +_TEST_FUNCTIONS = [ + { + "name": "invoiceDetail", + "label": "Invoice detail", + "language": "SPARQL", + "statement": "PREFIX ont: SELECT ?amount WHERE { ?inv a ont:Invoice }", + "params": [{"name": "invoiceId", "type": "xsd:string", "required": True}], + }, + { + "name": "supplierProfile", + "label": "Supplier profile", + "language": "SPARQL", + "statement": "PREFIX ont: SELECT ?name WHERE { ?s a ont:Supplier }", + "params": [{"name": "supplierId", "type": "xsd:string", "required": True}], + }, +] + def _mock_ontology_client( - discover_result: dict[str, Any] | None = None, - functions_result: list[dict[str, Any]] | None = None, invoke_result: dict[str, Any] | None = None, ) -> AsyncMock: client = AsyncMock() - client.discover = AsyncMock( - return_value=discover_result or {"name": "onto", "edges": []} - ) - client.list_functions = AsyncMock( - return_value=functions_result or [] - ) + client.fetch_graph = AsyncMock(side_effect=_make_test_graph) + client.fetch_artifact = AsyncMock(return_value=_TEST_OFN) + client.list_functions = AsyncMock(return_value=_TEST_FUNCTIONS) client.invoke_function = AsyncMock( - return_value=invoke_result or {"rows": []} + return_value=invoke_result or {"rows": [{"amount": 100}]} ) return client +async def _make_test_graph(ontology: str) -> OntologyGraph: + graph = parse_ofn(_TEST_OFN) + graph.functions = _TEST_FUNCTIONS + graph._build_adjacency() + return graph + + def _base_config() -> InvestigationConfig: return InvestigationConfig( label_vocabulary=["Source", "DerivedEffect", "Defer"], - seed_entities=["e1", "e2"], + seed_entities=["INV-2004", "SUP-001"], max_steps=10, + max_flips=3, ) +def _test_graph_dict() -> dict[str, Any]: + graph = parse_ofn(_TEST_OFN) + graph.functions = _TEST_FUNCTIONS + graph._build_adjacency() + return graph.to_dict() + + +# ── Bootstrap tests ──────────────────────────────────────────────── + + class TestBootstrapNode: - @pytest.mark.asyncio async def test_bootstrap_initialises_beliefs(self) -> None: + client = _mock_ontology_client() cfg = _base_config() - client = _mock_ontology_client( - discover_result={"name": "test"}, - functions_result=[{"name": "fn1", "description": "desc"}], - ) - node = _make_bootstrap(client, "test-onto", cfg) + bootstrap = _make_bootstrap(client, "test", cfg) state = EoGState() - result = await node(state) - assert "e1" in result["beliefs"] - assert "e2" in result["beliefs"] - assert result["beliefs"]["e1"].label == "Defer" - assert result["active_set"] == ["e1", "e2"] + result = await bootstrap(state) + + assert "INV-2004" in result["beliefs"] + assert "SUP-001" in result["beliefs"] + assert result["beliefs"]["INV-2004"].label == "Defer" + assert result["active_set"] == ["INV-2004", "SUP-001"] assert result["steps_taken"] == 0 - assert "metadata" in result["ontology_graph"] - assert "functions" in result["ontology_graph"] + # Graph topology should be populated + assert "nodes" in result["ontology_graph"] + assert "edges" in result["ontology_graph"] + assert "Invoice" in result["ontology_graph"]["nodes"] - @pytest.mark.asyncio - async def test_bootstrap_uses_state_config_as_fallback(self) -> None: + async def test_bootstrap_fetches_graph_topology(self) -> None: client = _mock_ontology_client() - node = _make_bootstrap(client, "onto", None) - cfg = InvestigationConfig( - label_vocabulary=["A"], - seed_entities=["x"], - ) - state = EoGState(investigation_config=cfg) - result = await node(state) - assert "x" in result["beliefs"] + cfg = _base_config() + bootstrap = _make_bootstrap(client, "test", cfg) + state = EoGState() + + result = await bootstrap(state) + graph_data = result["ontology_graph"] + + assert len(graph_data["nodes"]) == 3 # Supplier, Invoice, PurchaseOrder + assert len(graph_data["edges"]) == 2 # invoice_paidTo, invoice_referencesPO + assert len(graph_data["functions"]) == 2 + + +# ── Pop tests ────────────────────────────────────────────────────── class TestPopNode: - @pytest.mark.asyncio - async def test_pop_dequeues_first(self) -> None: + async def test_pops_first_entity(self) -> None: state = EoGState(active_set=["a", "b", "c"]) result = await pop_node(state) assert result["current_entity"] == "a" assert result["active_set"] == ["b", "c"] - @pytest.mark.asyncio - async def test_pop_empty(self) -> None: + async def test_empty_set_yields_empty_string(self) -> None: state = EoGState(active_set=[]) result = await pop_node(state) assert result["current_entity"] == "" - assert result["active_set"] == [] + + +# ── should_continue tests ────────────────────────────────────────── class TestShouldContinue: - def test_continue_when_entity_and_budget(self) -> None: - cfg = _base_config() + def test_continues_when_entity_and_budget(self) -> None: state = EoGState( - current_entity="e1", + current_entity="x", steps_taken=0, - investigation_config=cfg, + investigation_config=_base_config(), ) assert should_continue(state) == "fetch_context" - def test_frontier_when_no_current_entity(self) -> None: - cfg = _base_config() + def test_stops_at_budget(self) -> None: state = EoGState( - current_entity="", - steps_taken=0, - investigation_config=cfg, + current_entity="x", + steps_taken=10, + investigation_config=_base_config(), ) assert should_continue(state) == "frontier" - def test_frontier_when_budget_exhausted(self) -> None: - cfg = _base_config() + def test_stops_when_empty(self) -> None: state = EoGState( - current_entity="e1", - steps_taken=10, - investigation_config=cfg, + current_entity="", + steps_taken=0, + investigation_config=_base_config(), ) assert should_continue(state) == "frontier" - def test_default_max_steps_when_no_config(self) -> None: - state = EoGState(current_entity="e1", steps_taken=0) - assert should_continue(state) == "fetch_context" - state2 = EoGState(current_entity="e1", steps_taken=50) - assert should_continue(state2) == "frontier" +# ── Fetch context tests ─────────────────────────────────────────── class TestFetchContextNode: - @pytest.mark.asyncio async def test_builds_context_packet(self) -> None: - client = _mock_ontology_client( - invoke_result={"rows": [{"val": 1}]} - ) - cfg = _base_config() - node = _make_fetch_context(client, "onto") + client = _mock_ontology_client() + fetch = _make_fetch_context(client, "test") + state = EoGState( - current_entity="e1", - beliefs={"e1": Belief(label="Defer")}, - ontology_graph={ - "metadata": {"edges": []}, - "functions": [{"name": "fn_e1", "description": "e1 info"}], + current_entity="INV-2004", + ontology_graph=_test_graph_dict(), + beliefs={ + "INV-2004": Belief(label="Defer", evidence="seed"), + "SUP-001": Belief(label="Source", evidence="known"), }, - investigation_config=cfg, + investigation_config=_base_config(), ) - result = await node(state) + + result = await fetch(state) ctx = result["context_packet"] - assert ctx["entity_id"] == "e1" - assert ctx["current_belief"] is not None - assert len(ctx["function_results"]) == 1 + + assert ctx["entity_id"] == "INV-2004" + assert ctx["entity_type"] == "Invoice" + # Should have neighbor beliefs (Supplier is 1-hop via invoice_paidTo) + assert "SUP-001" in ctx["neighbor_beliefs"] + assert ctx["neighbor_beliefs"]["SUP-001"]["label"] == "Source" + # Should have invoked invoiceDetail (touches Invoice) + assert len(ctx["function_results"]) > 0 + + async def test_binds_params_correctly(self) -> None: + client = _mock_ontology_client() + fetch = _make_fetch_context(client, "test") + + state = EoGState( + current_entity="SUP-001", + ontology_graph=_test_graph_dict(), + beliefs={"SUP-001": Belief(label="Defer", evidence="seed")}, + investigation_config=_base_config(), + ) + + result = await fetch(state) + # Should have called supplierProfile with supplierId=SUP-001 + call_args = client.invoke_function.call_args_list + supplier_calls = [ + c for c in call_args + if c[0][1] == "supplierProfile" + ] + assert len(supplier_calls) == 1 + assert supplier_calls[0][0][2] == {"supplierId": "SUP-001"} + + +# ── Policy tests ─────────────────────────────────────────────────── class TestPolicyNode: - @pytest.mark.asyncio - async def test_policy_parses_llm_output(self) -> None: - llm_response_content = json.dumps({ - "label": "Source", - "evidence": "strong signal", - "propagations": [{"entity": "e2", "reason": "linked"}], - }) - model = AsyncMock() - response_msg = AsyncMock() - response_msg.content = llm_response_content - model.ainvoke = AsyncMock(return_value=response_msg) - - node = _make_policy(model) - cfg = _base_config() + async def test_parses_llm_json(self) -> None: + llm = AsyncMock() + llm.ainvoke = AsyncMock(return_value=AsyncMock( + content='{"label": "Source", "evidence": "found the root cause"}' + )) + policy = _make_policy(llm) state = EoGState( - current_entity="e1", - context_packet={"entity_id": "e1", "data": "some data"}, - investigation_config=cfg, + current_entity="INV-2004", + context_packet={"entity_id": "INV-2004", "entity_type": "Invoice"}, + investigation_config=_base_config(), ) - result = await node(state) + + result = await policy(state) assert result["policy_result"]["label"] == "Source" - assert len(result["policy_result"]["propagations"]) == 1 + assert "root cause" in result["policy_result"]["evidence"] + + async def test_enforces_label_vocabulary(self) -> None: + llm = AsyncMock() + llm.ainvoke = AsyncMock(return_value=AsyncMock( + content='{"label": "InvalidLabel", "evidence": "bad"}' + )) + policy = _make_policy(llm) + state = EoGState( + current_entity="x", + context_packet={"entity_id": "x", "entity_type": "Unknown"}, + investigation_config=_base_config(), + ) - @pytest.mark.asyncio - async def test_policy_handles_parse_failure(self) -> None: - model = AsyncMock() - response_msg = AsyncMock() - response_msg.content = "not valid json at all" - model.ainvoke = AsyncMock(return_value=response_msg) + result = await policy(state) + assert result["policy_result"]["label"] == "Defer" # Forced to Defer - node = _make_policy(model) + async def test_handles_parse_failure(self) -> None: + llm = AsyncMock() + llm.ainvoke = AsyncMock(return_value=AsyncMock(content="not json")) + policy = _make_policy(llm) state = EoGState( - current_entity="e1", - context_packet={"entity_id": "e1"}, + current_entity="x", + context_packet={"entity_id": "x", "entity_type": "Unknown"}, investigation_config=_base_config(), ) - result = await node(state) + + result = await policy(state) assert result["policy_result"]["label"] == "Defer" - assert "could not be parsed" in result["policy_result"]["evidence"] + + +# ── Update tests ─────────────────────────────────────────────────── class TestUpdateNode: - @pytest.mark.asyncio async def test_creates_ledger_entry(self) -> None: state = EoGState( - current_entity="e1", - beliefs={"e1": Belief(label="Defer")}, + current_entity="INV-2004", policy_result={"label": "Source", "evidence": "found it"}, - steps_taken=0, + beliefs={"INV-2004": Belief(label="Defer", evidence="seed")}, + investigation_config=_base_config(), ) + result = await update_node(state) - assert result["beliefs"]["e1"].label == "Source" - assert result["beliefs"]["e1"].flip_count == 1 assert len(result["ledger"]) == 1 assert result["ledger"][0].old_label == "Defer" assert result["ledger"][0].new_label == "Source" - assert result["steps_taken"] == 1 - @pytest.mark.asyncio - async def test_no_flip_when_same_label(self) -> None: + async def test_tracks_flip_count(self) -> None: state = EoGState( - current_entity="e1", - beliefs={"e1": Belief(label="Source", flip_count=0)}, - policy_result={"label": "Source", "evidence": "confirmed"}, - steps_taken=2, + current_entity="x", + policy_result={"label": "DerivedEffect", "evidence": "changed"}, + beliefs={"x": Belief(label="Source", evidence="was source", flip_count=1)}, + investigation_config=_base_config(), ) + result = await update_node(state) - assert result["beliefs"]["e1"].flip_count == 0 - assert result["steps_taken"] == 3 + assert result["beliefs"]["x"].flip_count == 2 - @pytest.mark.asyncio - async def test_new_entity_no_flip(self) -> None: + async def test_damping_forces_defer(self) -> None: state = EoGState( - current_entity="new_e", - beliefs={}, - policy_result={"label": "Source", "evidence": "new"}, - steps_taken=0, + current_entity="x", + policy_result={"label": "Source", "evidence": "oscillating"}, + beliefs={"x": Belief(label="DerivedEffect", evidence="prev", flip_count=3)}, + investigation_config=_base_config(), ) + result = await update_node(state) - assert result["beliefs"]["new_e"].flip_count == 0 + # flip_count exceeds max_flips(3) → damped to Defer + assert result["beliefs"]["x"].label == "Defer" + assert "Damped" in result["beliefs"]["x"].evidence + + +# ── Propagate tests ──────────────────────────────────────────────── class TestPropagateNode: - @pytest.mark.asyncio - async def test_propagates_to_neighbours(self) -> None: + async def test_propagates_to_graph_neighbors(self) -> None: + """Propagation follows GRAPH EDGES, not LLM suggestions.""" + from uipath_langchain.agent.eog.types import LedgerEntry + state = EoGState( - current_entity="e1", - beliefs={"e1": Belief(label="Source", evidence="test")}, - policy_result={ - "propagations": [ - {"entity": "e2", "reason": "linked"}, - {"entity": "e3", "reason": "related"}, - ], + current_entity="INV-2004", + ontology_graph=_test_graph_dict(), + beliefs={ + "INV-2004": Belief(label="Source", evidence="found it"), + "SUP-001": Belief(label="Defer", evidence="seed"), }, - active_set=[], + # Ledger shows label changed (Defer → Source) + ledger=[LedgerEntry( + timestamp=0, entity_id="INV-2004", + old_label="Defer", new_label="Source", evidence="found it", + )], investigation_config=_base_config(), + policy_result={}, ) + result = await propagate_node(state) - assert "e2" in result["inbox"] - assert "e3" in result["inbox"] - assert "e2" in result["active_set"] - assert "e3" in result["active_set"] - assert len(result["explanatory_edges"]) == 2 - @pytest.mark.asyncio - async def test_does_not_reactivate_over_max_flips(self) -> None: - cfg = InvestigationConfig( - label_vocabulary=["A"], max_flips=2 - ) + # SUP-001 should be re-activated (Invoice→Supplier via invoice_paidTo) + assert "SUP-001" in result.get("active_set", []) + # SUP-001 should have an inbox message from INV-2004 + assert "SUP-001" in result.get("inbox", {}) + msg = result["inbox"]["SUP-001"][0] + assert msg["from"] == "INV-2004" + assert msg["label"] == "Source" + + async def test_no_propagation_when_label_unchanged(self) -> None: + from uipath_langchain.agent.eog.types import LedgerEntry + state = EoGState( - current_entity="e1", + current_entity="INV-2004", + ontology_graph=_test_graph_dict(), beliefs={ - "e1": Belief(label="A"), - "e2": Belief(label="A", flip_count=3), + "INV-2004": Belief(label="Source", evidence="same"), + "SUP-001": Belief(label="Defer", evidence="seed"), }, - policy_result={ - "propagations": [{"entity": "e2", "reason": "test"}], - }, - active_set=[], - investigation_config=cfg, + # Ledger shows NO change (Source → Source) + ledger=[LedgerEntry( + timestamp=0, entity_id="INV-2004", + old_label="Source", new_label="Source", evidence="same", + )], + investigation_config=_base_config(), + policy_result={}, ) + result = await propagate_node(state) - # e2 should NOT be re-activated because flip_count >= max_flips - assert "e2" not in result["active_set"] + # No propagation when label didn't change + assert not result # Empty dict + + async def test_does_not_reactivate_over_max_flips(self) -> None: + from uipath_langchain.agent.eog.types import LedgerEntry - @pytest.mark.asyncio - async def test_no_duplicates_in_active_set(self) -> None: state = EoGState( - current_entity="e1", - beliefs={"e1": Belief(label="A")}, - policy_result={ - "propagations": [{"entity": "e2", "reason": "x"}], + current_entity="INV-2004", + ontology_graph=_test_graph_dict(), + beliefs={ + "INV-2004": Belief(label="Source", evidence="found"), + "SUP-001": Belief(label="Defer", evidence="seed", flip_count=4), }, - active_set=["e2"], + ledger=[LedgerEntry( + timestamp=0, entity_id="INV-2004", + old_label="Defer", new_label="Source", evidence="found", + )], investigation_config=_base_config(), + policy_result={}, ) + result = await propagate_node(state) - # e2 already in active_set, should not be duplicated - assert result["active_set"].count("e2") == 1 + # SUP-001 has flip_count=4 > max_flips=3 → not re-activated + active = result.get("active_set", []) + assert "SUP-001" not in active + + +# ── Frontier tests ───────────────────────────────────────────────── class TestFrontierNode: - @pytest.mark.asyncio - async def test_filters_defer_labels(self) -> None: + async def test_filters_non_defer(self) -> None: state = EoGState( beliefs={ - "e1": Belief(label="Source", evidence="found"), - "e2": Belief(label="Defer", evidence="unknown"), - "e3": Belief(label="DerivedEffect", evidence="derived"), + "a": Belief(label="Source", evidence="root cause"), + "b": Belief(label="DerivedEffect", evidence="symptom"), + "c": Belief(label="Defer", evidence="unknown"), }, + ontology_graph=_test_graph_dict(), investigation_config=_base_config(), ) + result = await frontier_node(state) - entities = [f["entity"] for f in result["frontier"]] - assert "e1" in entities - assert "e3" in entities - assert "e2" not in entities + labels = {f["entity"]: f["label"] for f in result["frontier"]} + assert "a" in labels + assert "b" in labels + assert "c" not in labels # Defer excluded - @pytest.mark.asyncio - async def test_empty_frontier_when_all_defer(self) -> None: + async def test_removes_explained_entities(self) -> None: state = EoGState( beliefs={ - "e1": Belief(label="Defer"), + "a": Belief(label="Source", evidence="root cause"), + "b": Belief(label="DerivedEffect", evidence="caused by a"), }, + explanatory_edges=[ + ExplanatoryEdge(source="a", target="b", relationship="causes"), + ], + ontology_graph=_test_graph_dict(), investigation_config=_base_config(), ) + result = await frontier_node(state) - assert result["frontier"] == [] + entities = {f["entity"] for f in result["frontier"]} + # b is explained by a (Source→DerivedEffect) → removed from frontier + assert "a" in entities + assert "b" not in entities From 2ac4e58af0cafa451aa4fd1d875fd9a98ec5b938 Mon Sep 17 00:00:00 2001 From: Harshit Rohatgi Date: Fri, 17 Jul 2026 07:13:50 +0530 Subject: [PATCH 3/4] fix: parse YARRRML for entity key properties MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit OFN parsing alone couldn't bind function params like exceptionId to ToleranceException — the naming convention (toleranceExceptionId) didn't match. Now fetch_graph() also downloads mapping.yarrrml.yml and parses subject templates to extract the authoritative key property per entity (e.g., ToleranceException → exceptionId, PurchaseOrder → poId). Verified E2E: coded agent now produces 5 frontier findings (was 0 without YARRRML keys). All 7 entity types get correct function dispatch. Co-Authored-By: Claude Opus 4.6 (1M context) --- examples/s2p-ontology/run_eog.py | 392 ++++++++++++++++++ .../agent/eog/graph_topology.py | 64 ++- src/uipath_langchain/agent/eog/nodes.py | 29 +- .../agent/eog/ontology_client.py | 6 +- 4 files changed, 472 insertions(+), 19 deletions(-) create mode 100644 examples/s2p-ontology/run_eog.py diff --git a/examples/s2p-ontology/run_eog.py b/examples/s2p-ontology/run_eog.py new file mode 100644 index 000000000..0daecda9d --- /dev/null +++ b/examples/s2p-ontology/run_eog.py @@ -0,0 +1,392 @@ +"""Run the EoG investigation agent against the local S2P ontology stack. + +Prerequisites: + 1. FQS mock running on :9099 with S2P data + 2. Ontology runtime running on :5002 with S2P ontology DEPLOYED + 3. OPENAI_API_KEY set in environment + +Usage: + cd examples/s2p-ontology + uv run python run_eog.py +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import os +import sys +import time + +# Add src to path for dev mode +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "src")) + +from langchain_core.language_models import BaseChatModel +from langchain_core.messages import HumanMessage, SystemMessage +from langgraph.constants import END, START +from langgraph.graph import StateGraph + +from uipath_langchain.agent.eog.ontology_client import OntologyClient +from uipath_langchain.agent.eog.types import ( + Belief, + EoGState, + ExplanatoryEdge, + InvestigationConfig, + LedgerEntry, +) + +logging.basicConfig(level=logging.INFO, format="%(message)s") +log = logging.getLogger("eog") + +# ── S2P function → entity mapping ────────────────────────────────────── +# Which functions to call per entity type, with how to build their params. +ENTITY_FUNCTIONS: dict[str, list[dict]] = { + "Invoice": [ + {"fn": "invoiceDetail", "params": lambda eid: {"invoiceId": eid}}, + ], + "Supplier": [ + {"fn": "supplierProfile", "params": lambda eid: {"supplierId": eid}}, + ], + "ToleranceException": [ + {"fn": "exceptionContext", "params": lambda eid: {"exceptionId": eid}}, + ], + "PurchaseOrder": [ + {"fn": "poDetail", "params": lambda eid: {"poId": eid}}, + ], + "Commodity": [ + {"fn": "commodityDescendants", "params": lambda eid: {"parentCommodityId": eid}}, + ], + "ExceptionRule": [ + {"fn": "matchingExceptionRules", "params": lambda eid: {"commodityId": eid}}, + ], + # Seed-level functions (no entity-specific param) + "_seed": [ + {"fn": "openExceptions", "params": lambda _: None}, + {"fn": "maverickSpendRate", "params": lambda _: {"period": "2026-Q2"}}, + {"fn": "spendByCategory", "params": lambda _: {"period": "2026-Q2"}}, + ], +} + +S2P_LABELS = [ + "Source", + "DerivedEffect", + "PolicyViolation", + "CandidateMatch", + "SupportingEvidence", + "Contradiction", + "Defer", +] + +SYSTEM_PROMPT = """\ +You are an S2P (Source-to-Pay) procurement investigation agent using the \ +EoG (Explanations over Graphs) pattern. + +You investigate procurement anomalies by examining entities one at a time, \ +labeling each with a role in the explanation, and identifying which neighbor \ +entities should be re-examined. + +Label vocabulary: +- Source: This entity is the root cause / origin of the issue +- DerivedEffect: This entity's anomaly is caused by something upstream +- PolicyViolation: This entity violates a business rule or SHACL constraint +- CandidateMatch: This entity is a potential resolution +- SupportingEvidence: This entity provides corroborating data +- Contradiction: This entity's data conflicts with the emerging explanation +- Defer: Insufficient evidence to classify + +You MUST respond in valid JSON only, no markdown, no explanation outside JSON: +{"label": "", "evidence": "", "propagations": [{"entity": "", "reason": ""}]} + +Entity IDs for propagation should be specific record IDs from the data \ +(e.g., "INV-2004", "SUP-004", "EXC-002"), not abstract type names.""" + + +async def fetch_seed_data(client: OntologyClient, ontology: str) -> list[dict]: + """Call seed-level functions to discover initial entities.""" + results = [] + for fn_spec in ENTITY_FUNCTIONS["_seed"]: + try: + r = await client.invoke_function(ontology, fn_spec["fn"], fn_spec["params"]("")) + results.append({"function": fn_spec["fn"], "rows": r.get("rows", [])}) + except Exception as e: + log.warning("Seed function %s failed: %s", fn_spec["fn"], e) + return results + + +async def fetch_entity_data( + client: OntologyClient, ontology: str, entity_id: str, entity_type: str +) -> list[dict]: + """Call functions relevant to an entity type with the specific entity ID.""" + results = [] + fn_specs = ENTITY_FUNCTIONS.get(entity_type, []) + for fn_spec in fn_specs: + try: + params = fn_spec["params"](entity_id) + r = await client.invoke_function(ontology, fn_spec["fn"], params) + results.append({"function": fn_spec["fn"], "rows": r.get("rows", [])[:20]}) + except Exception as e: + log.warning(" Function %s(%s) failed: %s", fn_spec["fn"], entity_id, e) + return results + + +def infer_entity_type(entity_id: str) -> str: + """Infer entity type from ID prefix.""" + prefixes = { + "INV-": "Invoice", + "SUP-": "Supplier", + "EXC-": "ToleranceException", + "PO-": "PurchaseOrder", + "COM-": "Commodity", + "RULE-": "ExceptionRule", + "CTR-": "Contract", + "PR-": "Requisition", + "SPD-": "SpendRecord", + } + for prefix, etype in prefixes.items(): + if entity_id.startswith(prefix): + return etype + return "Unknown" + + +async def run_eog( + llm: BaseChatModel, + client: OntologyClient, + ontology: str = "s2p", + max_steps: int = 10, +) -> None: + """Run a full EoG investigation loop.""" + + # ── 1. Bootstrap: discover seed entities ─────────────────────────── + print("\n" + "=" * 70) + print("EoG INVESTIGATION — S2P Procurement Ontology") + print("=" * 70) + + seed_data = await fetch_seed_data(client, ontology) + print(f"\nSeed data collected from {len(seed_data)} functions:") + for sd in seed_data: + print(f" {sd['function']}: {len(sd['rows'])} rows") + + # Extract initial entity IDs from seed data + seed_entities: list[str] = [] + for sd in seed_data: + for row in sd["rows"]: + for key in ["exceptionId", "invoiceId", "supplierId"]: + val = row.get(key) + if val and val not in seed_entities: + seed_entities.append(val) + seed_entities = seed_entities[:8] # Cap initial seeds + + print(f"\nSeed entities: {seed_entities}") + + # ── 2. Initialize state ──────────────────────────────────────────── + beliefs: dict[str, Belief] = {} + for eid in seed_entities: + beliefs[eid] = Belief(label="Defer", evidence="Initial seed entity.") + + active_set = list(seed_entities) + ledger: list[dict] = [] + explanatory_edges: list[dict] = [] + inbox: dict[str, list[dict]] = {} + steps = 0 + + # ── 3. EoG Loop ─────────────────────────────────────────────────── + print(f"\n{'─' * 70}") + print("INVESTIGATION LOOP") + print(f"{'─' * 70}") + + while active_set and steps < max_steps: + # Pop + entity_id = active_set.pop(0) + entity_type = infer_entity_type(entity_id) + steps += 1 + + print(f"\n[Step {steps}] Processing: {entity_id} (type: {entity_type})") + + # Fetch context + entity_data = await fetch_entity_data(client, ontology, entity_id, entity_type) + entity_inbox = inbox.get(entity_id, []) + + context = { + "entity_id": entity_id, + "entity_type": entity_type, + "current_belief": beliefs[entity_id].model_dump(), + "data": entity_data, + "inbox_messages": entity_inbox, + "neighbor_beliefs": { + eid: b.model_dump() + for eid, b in beliefs.items() + if eid != entity_id and b.label != "Defer" + }, + } + + # Policy (LLM call) + prompt = f"""\ +Investigating entity: {entity_id} (type: {entity_type}) + +Data collected: +{json.dumps(entity_data, indent=2, default=str)[:3000]} + +Messages from neighbors: +{json.dumps(entity_inbox, indent=2, default=str)[:1000]} + +Current beliefs about other entities: +{json.dumps({eid: b.model_dump() for eid, b in beliefs.items() if eid != entity_id and b.label != "Defer"}, indent=2, default=str)[:1000]} + +Assign a label and identify propagations.""" + + try: + response = await llm.ainvoke([ + SystemMessage(content=SYSTEM_PROMPT), + HumanMessage(content=prompt), + ]) + content = response.content if isinstance(response.content, str) else str(response.content) + # Strip markdown code fences if present + content = content.strip() + if content.startswith("```"): + content = content.split("\n", 1)[1] if "\n" in content else content + if content.endswith("```"): + content = content[:-3] + content = content.strip() + parsed = json.loads(content) + except Exception as e: + log.warning(" LLM parse failed: %s", e) + parsed = {"label": "Defer", "evidence": f"LLM parse error: {e}", "propagations": []} + + new_label = parsed.get("label", "Defer") + evidence = parsed.get("evidence", "") + propagations = parsed.get("propagations", []) + + # Update belief + old_belief = beliefs[entity_id] + old_label = old_belief.label + flip_count = old_belief.flip_count + (1 if old_label != new_label and old_label != "Defer" else 0) + + beliefs[entity_id] = Belief(label=new_label, evidence=evidence, flip_count=flip_count) + ledger.append({ + "step": steps, + "entity": entity_id, + "type": entity_type, + "old_label": old_label, + "new_label": new_label, + "evidence": evidence[:120], + }) + + status = "CHANGED" if old_label != new_label else "confirmed" + print(f" Label: {old_label} → {new_label} ({status})") + print(f" Evidence: {evidence[:100]}") + + # Propagate + for prop in propagations: + target = prop.get("entity", "") + reason = prop.get("reason", "") + if not target: + continue + + # Add to inbox + if target not in inbox: + inbox[target] = [] + inbox[target].append({ + "from": entity_id, + "label": new_label, + "evidence": evidence[:80], + "reason": reason, + }) + + # Initialize belief if new + if target not in beliefs: + beliefs[target] = Belief(label="Defer", evidence="Discovered via propagation.") + + # Re-activate if under flip limit + target_flips = beliefs[target].flip_count + if target_flips < 3 and target not in active_set: + active_set.append(target) + print(f" → Propagate to {target}: {reason[:60]}") + + explanatory_edges.append({ + "source": entity_id, + "target": target, + "evidence": reason, + }) + + # ── 4. Compute Frontier ──────────────────────────────────────────── + print(f"\n{'=' * 70}") + print("INVESTIGATION RESULTS") + print(f"{'=' * 70}") + + print(f"\nSteps taken: {steps}") + print(f"Entities examined: {len(beliefs)}") + + # Frontier: non-Defer beliefs + frontier = [ + {"entity": eid, "type": infer_entity_type(eid), "label": b.label, "evidence": b.evidence} + for eid, b in beliefs.items() + if b.label != "Defer" + ] + + print(f"\nFrontier ({len(frontier)} findings):") + for item in frontier: + print(f" [{item['label']}] {item['entity']} ({item['type']})") + print(f" {item['evidence'][:120]}") + + print(f"\nLedger ({len(ledger)} entries):") + for entry in ledger: + print(f" Step {entry['step']}: {entry['entity']} ({entry['type']}) " + f"{entry['old_label']} → {entry['new_label']}") + + if explanatory_edges: + print(f"\nExplanatory edges ({len(explanatory_edges)}):") + for edge in explanatory_edges: + print(f" {edge['source']} → {edge['target']}: {edge['evidence'][:80]}") + + +async def main() -> None: + # LLM setup — try Anthropic first, fall back to OpenAI + anthropic_key = os.environ.get("ANTHROPIC_API_KEY") + openai_key = os.environ.get("OPENAI_API_KEY") + + if anthropic_key: + from langchain_anthropic import ChatAnthropic + llm = ChatAnthropic( + model=os.environ.get("LLM_MODEL", "claude-haiku-4-5-20251001"), + temperature=0.0, + api_key=anthropic_key, + ) + print(f"Using Anthropic: {llm.model}") + elif openai_key: + from langchain_openai import ChatOpenAI + llm = ChatOpenAI( + model=os.environ.get("LLM_MODEL", "gpt-4o-mini"), + temperature=0.0, + api_key=openai_key, + ) + print(f"Using OpenAI: {llm.model}") + else: + print("Error: Set ANTHROPIC_API_KEY or OPENAI_API_KEY") + sys.exit(1) + + client = OntologyClient( + base_url=os.environ.get("ONTOLOGY_BASE_URL", "http://localhost:5002"), + ) + + # Verify server is up + try: + meta = await client.discover("s2p") + if meta.get("state") != "DEPLOYED": + print(f"Error: Ontology s2p is {meta.get('state')}, expected DEPLOYED") + sys.exit(1) + print(f"Connected to ontology: {meta['name']} [{meta['state']}]") + except Exception as e: + print(f"Error connecting to ontology-runtime: {e}") + sys.exit(1) + + await run_eog( + llm=llm, + client=client, + ontology="s2p", + max_steps=12, + ) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/src/uipath_langchain/agent/eog/graph_topology.py b/src/uipath_langchain/agent/eog/graph_topology.py index 8ebc4b311..463e655dc 100644 --- a/src/uipath_langchain/agent/eog/graph_topology.py +++ b/src/uipath_langchain/agent/eog/graph_topology.py @@ -1,9 +1,25 @@ """Parse OWL Functional Notation into an in-memory graph topology. -The ontology-runtime stores schemas as OWL 2 Functional Notation (``.ofn``). -This module extracts the entity-relationship graph from the OWL text so the -EoG controller can traverse it deterministically — without relying on the LLM -to decide what to explore next. +.. warning:: **POC only — not for production.** + + This module uses regex to parse OWL Functional Notation client-side. + It works for the POC because the ontology-runtime produces clean, simple + OFN, but it will break on non-trivial OWL (nested axioms, imports, + complex class expressions). + + The production implementation should fetch the resolved graph directly + from the ontology-runtime via a dedicated discovery endpoint (e.g., + ``GET /ontology/{name}/graph``) that returns the graph as JSON with: + + - Entities (Object Types) with their data properties and bindings + - Relationships (Link Types) with domain, range, join keys, cardinality + - Constraints (SHACL shapes) per entity + - Functions with their ``touches`` declarations and parameter schemas + - Neighbor adjacency pre-computed server-side + + The server already has all of this in ``OntologySnapshot`` — it's one + serialization step. The ``OntologyGraph`` dataclass below remains the + client-side representation; only the parsing/fetching changes. The parsed graph is a lightweight adjacency structure: - Nodes = OWL classes (entity types) @@ -61,6 +77,9 @@ class OntologyGraph: nodes: dict[str, OntologyNode] = field(default_factory=dict) edges: list[OntologyEdge] = field(default_factory=list) functions: list[dict[str, Any]] = field(default_factory=list) + key_properties: dict[str, str] = field(default_factory=dict) + """Entity name → key property name from YARRRML subject template. + E.g. ``{"Supplier": "supplierId", "ToleranceException": "exceptionId"}``.""" # Pre-built adjacency (call _build_adjacency after construction) _outgoing: dict[str, list[OntologyEdge]] = field(default_factory=dict) @@ -154,6 +173,7 @@ def to_dict(self) -> dict[str, Any]: for e in self.edges ], "functions": self.functions, + "key_properties": self.key_properties, } @classmethod @@ -185,6 +205,7 @@ def from_dict(cls, data: dict[str, Any]) -> OntologyGraph: nodes=nodes, edges=edges, functions=data.get("functions", []), + key_properties=data.get("key_properties", {}), ) graph._build_adjacency() return graph @@ -319,6 +340,41 @@ def _infer_touched_entities( # ── ID prefix convention ──────────────────────────────────────────── +def parse_yarrrml_keys(yarrrml_text: str) -> dict[str, str]: + """Extract entity key property names from YARRRML subject templates. + + Parses ``s: ont:EntityName/$(keyProp)`` to build a mapping of + entity name → key property name. + + Args: + yarrrml_text: The YARRRML mapping file content. + + Returns: + Dict mapping entity name to its key property name, + e.g. ``{"Supplier": "supplierId", "ToleranceException": "exceptionId"}``. + """ + key_map: dict[str, str] = {} + current_entity: str | None = None + in_mappings = False + for line in yarrrml_text.split("\n"): + # Track when we're inside the mappings block + if line.rstrip() == "mappings:": + in_mappings = True + continue + if not in_mappings: + continue + # Entity name: exactly 2-space indent, ends with ":" + # e.g. " Supplier:" but NOT " sources:" or " - access:" + if re.match(r"^ [A-Z]\w*:$", line): + current_entity = line.strip()[:-1] + # Subject template: " s: ont:EntityName/$(keyProp)" + if current_entity and line.strip().startswith("s:"): + m = re.search(r"\$\((\w+)\)", line) + if m: + key_map[current_entity] = m.group(1) + return key_map + + _ID_PREFIXES: dict[str, str] = { "INV-": "Invoice", "SUP-": "Supplier", diff --git a/src/uipath_langchain/agent/eog/nodes.py b/src/uipath_langchain/agent/eog/nodes.py index 91130dc53..f91c6f840 100644 --- a/src/uipath_langchain/agent/eog/nodes.py +++ b/src/uipath_langchain/agent/eog/nodes.py @@ -149,8 +149,10 @@ async def _fetch_context(state: EoGState) -> dict[str, Any]: fn_name = fn.get("name", "") params = fn.get("params", []) - # Build params: match function param names to entity ID - call_params = _bind_params(params, entity_id, entity_type) + # Build params using YARRRML key property mapping + call_params = _bind_params( + params, entity_id, entity_type, graph.key_properties, + ) if call_params is None and params: # Function has required params we can't bind — skip continue @@ -187,35 +189,36 @@ def _bind_params( params: list[dict[str, Any]], entity_id: str, entity_type: str, + key_properties: dict[str, str] | None = None, ) -> dict[str, str] | None: """Bind function params to the current entity ID. - Maps param names to the entity ID using naming conventions: - - ``invoiceId`` for entity type ``Invoice`` with ID ``INV-2004`` - - ``supplierId`` for entity type ``Supplier`` with ID ``SUP-001`` + Uses the key property name from YARRRML (e.g., ``exceptionId`` for + ``ToleranceException``) to match function params. Falls back to + naming conventions if YARRRML key is unavailable. Returns None if a required param can't be bound. """ if not params: return None + # The authoritative key property name from YARRRML subject template + yarrrml_key = (key_properties or {}).get(entity_type) + bound: dict[str, str] = {} for param in params: pname = param.get("name", "") required = param.get("required", False) - # Try: paramName matches entityType + "Id" pattern - expected_key = entity_type[0].lower() + entity_type[1:] + "Id" - alt_key = entity_type.lower() + "Id" - parent_key = "parent" + entity_type + "Id" - - if pname == expected_key or pname == alt_key or pname == parent_key: + # 1. Match against YARRRML key property (authoritative) + if yarrrml_key and pname == yarrrml_key: + bound[pname] = entity_id + # 2. Fallback: naming convention (entityType + "Id") + elif pname == entity_type[0].lower() + entity_type[1:] + "Id": bound[pname] = entity_id elif pname.endswith("Id") and pname[:-2].lower() == entity_type.lower(): bound[pname] = entity_id elif required: - # Required param we can't bind — this function doesn't - # apply to a single instance of this entity type return None return bound if bound else None diff --git a/src/uipath_langchain/agent/eog/ontology_client.py b/src/uipath_langchain/agent/eog/ontology_client.py index 2b4bb6106..c4576989a 100644 --- a/src/uipath_langchain/agent/eog/ontology_client.py +++ b/src/uipath_langchain/agent/eog/ontology_client.py @@ -8,7 +8,7 @@ import httpx from uipath._utils._ssl_context import get_httpx_client_kwargs -from .graph_topology import OntologyGraph, parse_ofn +from .graph_topology import OntologyGraph, parse_ofn, parse_yarrrml_keys class OntologyClient: @@ -137,12 +137,14 @@ async def fetch_graph(self, ontology: str) -> OntologyGraph: Returns: Parsed ``OntologyGraph`` ready for EoG traversal. """ - ofn_text, functions = await asyncio.gather( + ofn_text, yarrrml_text, functions = await asyncio.gather( self.fetch_artifact(ontology, "schema.ofn"), + self.fetch_artifact(ontology, "mapping.yarrrml.yml"), self.list_functions(ontology), ) graph = parse_ofn(ofn_text) graph.functions = functions + graph.key_properties = parse_yarrrml_keys(yarrrml_text) graph._build_adjacency() return graph From 22b703b4a4aa7e877132690355d4b5f2fb5c618b Mon Sep 17 00:00:00 2001 From: Harshit Rohatgi Date: Fri, 17 Jul 2026 07:17:08 +0530 Subject: [PATCH 4/4] docs: add CLAUDE.md to eog package with caveats for teammates Documents all known limitations: regex-based OFN parser (POC only), YARRRML key binding dependency, hardcoded ID prefix conventions, namespace constraint, Ontop transitive property gap, and the path to production (server-side graph endpoint). Co-Authored-By: Claude Opus 4.6 (1M context) --- src/uipath_langchain/agent/eog/CLAUDE.md | 78 ++++++++++++++++++++++++ 1 file changed, 78 insertions(+) create mode 100644 src/uipath_langchain/agent/eog/CLAUDE.md diff --git a/src/uipath_langchain/agent/eog/CLAUDE.md b/src/uipath_langchain/agent/eog/CLAUDE.md new file mode 100644 index 000000000..cf6347748 --- /dev/null +++ b/src/uipath_langchain/agent/eog/CLAUDE.md @@ -0,0 +1,78 @@ +# EoG (Explanations over Graphs) Agent Package + +## What This Is + +A new agent type alongside ReAct for investigative, multi-hop reasoning over ontology-backed virtual knowledge graphs. Based on [arXiv:2601.17915](https://arxiv.org/abs/2601.17915). The agent traverses an ontology's entity-relationship graph deterministically, invokes governed functions at each entity, and uses belief propagation to converge on a minimal explanatory frontier. + +## Status: POC — Not Production-Ready + +This package is a proof-of-concept. Several components have known limitations that will break in production use. Read the caveats below before making changes. + +## Module Layout + +| File | What it does | Caveats | +|---|---|---| +| `graph_topology.py` | Parses OWL (.ofn) + YARRRML into an in-memory graph with adjacency | **POC only.** Uses regex to parse OWL Functional Notation — breaks on complex OWL (nested axioms, imports, multi-line). YARRRML parser assumes exact 2-space indent. Production should fetch the resolved graph from a server endpoint. | +| `ontology_client.py` | Async REST client for ontology-runtime | `fetch_graph()` downloads schema.ofn + mapping.yarrrml.yml and parses client-side. Production should call a single discovery endpoint returning JSON. | +| `nodes.py` | 7 LangGraph nodes implementing the EoG algorithm | Function dispatch uses `_bind_params()` which matches param names to entity key properties from YARRRML. Falls back to naming conventions. Entity ID → type resolution uses hardcoded prefix map (`_ID_PREFIXES`). | +| `agent.py` | `create_eog_agent()` graph builder | Returns uncompiled `StateGraph`. Caller compiles with `.compile()`. | +| `types.py` | State types: Belief, LedgerEntry, EoGState, InvestigationConfig | Labels are free strings, not enum — each investigation supplies its own vocabulary. | + +## Critical Things to Know + +### 1. Graph topology comes from client-side parsing (fragile) + +`fetch_graph()` downloads two artifacts and parses them: +- `schema.ofn` → entity types (nodes), object properties (edges), data properties +- `mapping.yarrrml.yml` → entity key property names (e.g., `ToleranceException → exceptionId`) + +Both parsers are regex/line-based. They work for the S2P POC's clean, simple artifacts. They WILL break on: +- OWL with nested class expressions, imports, or multi-line axioms +- YARRRML with different indentation or structure + +**The right fix:** Add `GET /ontology/{name}/graph` to the ontology-runtime that returns `OntologySnapshot` as JSON. The `OntologyGraph` dataclass stays; only the fetching changes. + +### 2. Entity ID → entity type uses prefix conventions + +`entity_for_id("INV-2004")` returns `"Invoice"` by matching against `_ID_PREFIXES = {"INV-": "Invoice", ...}`. This is hardcoded for S2P. Other ontologies need their own prefix map or a fundamentally different approach (server-side resolution). + +### 3. Function param binding depends on YARRRML key properties + +Functions like `exceptionContext(exceptionId)` are bound to entity instances via the key property name from YARRRML (`ToleranceException → exceptionId`). Without this mapping, `_bind_params` can't match `exceptionId` to `ToleranceException` — it would expect `toleranceExceptionId`. + +If a function has a required param that can't be bound, the function is skipped for that entity. This is intentional — it means the function is investigation-scoped (e.g., `spendByCategory(period)`) not entity-scoped. + +### 4. Propagation follows graph edges, NOT LLM suggestions + +When an entity's label changes, the `propagate_node` broadcasts to all graph neighbors (entities connected by OWL object properties). The LLM does NOT decide who to propagate to — that's the whole point of EoG vs ReAct. + +Propagation only happens when the label actually changed. Neighbors are only re-activated if their flip count is below `max_flips` (damping). + +### 5. Frontier computes irreducibility + +The frontier is NOT just "non-Defer beliefs." It removes entities that are explained by another entity: if Source A has an explanatory edge to DerivedEffect B, then B is removed from the frontier (it's explained by A). + +### 6. Namespace must be `ont#` + +All OWL/SHACL/YARRRML/FnO artifacts must use `https://ontology.uipath.com/ont#`. The ontology-runtime hardcodes this in `Vocabulary.ONT`. Custom namespaces silently break Ontop reformulation. + +### 7. Ontop does not support transitive properties + +`TransitiveObjectProperty` is accepted but has no effect. SPARQL property paths (`+`, `*`) are rejected. Hierarchy traversal (Supplier parent, Commodity taxonomy) needs materialized paths or precomputed functions. + +## Running Tests + +```bash +uv run pytest tests/agent/eog/ -v # 60 tests +uv run ruff check src/uipath_langchain/agent/eog/ +``` + +## Running the Live Agent + +Requires the ontology-runtime local stack (see `eog_architecture_summary.md` in memory or the `examples/s2p-ontology/` directory). + +```bash +cd examples/s2p-eog-agent +source .venv/bin/activate +uip codedagent run agent '{"question": "Investigate open tolerance exceptions"}' +```