Deliver explainability triples inline in retrieval response stream (#763)

Provenance triples are now included directly in explain messages from
GraphRAG, DocumentRAG, and Agent services, eliminating the need for
follow-up knowledge graph queries to retrieve explainability details.

Each explain message in the response stream now carries:
- explain_id: root URI for this provenance step (unchanged)
- explain_graph: named graph where triples are stored (unchanged)
- explain_triples: the actual provenance triples for this step (new)

Changes across the stack:
- Schema: added explain_triples field to GraphRagResponse,
  DocumentRagResponse, and AgentResponse
- Services: all explain message call sites pass triples through
  (graph_rag, document_rag, agent react, agent orchestrator)
- Translators: encode explain_triples via TripleTranslator for
  gateway wire format
- Python SDK: ProvenanceEvent now includes parsed ExplainEntity
  and raw triples; expanded event_type detection
- CLI: invoke_graph_rag, invoke_agent, invoke_document_rag use
  inline entity when available, fall back to graph query
- Tech specs updated

Additional explainability test
This commit is contained in:
cybermaggedon 2026-04-07 12:19:05 +01:00 committed by GitHub
parent 2f8d6a3ffb
commit ddd4bd7790
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 521 additions and 49 deletions

View file

@ -219,8 +219,8 @@ TG_ANSWER = TG + "answer"
| `trustgraph-base/trustgraph/provenance/triples.py` | Add TG types to GraphRAG triple builders, add Document RAG triple builders | | `trustgraph-base/trustgraph/provenance/triples.py` | Add TG types to GraphRAG triple builders, add Document RAG triple builders |
| `trustgraph-base/trustgraph/provenance/uris.py` | Add Document RAG URI generators | | `trustgraph-base/trustgraph/provenance/uris.py` | Add Document RAG URI generators |
| `trustgraph-base/trustgraph/provenance/__init__.py` | Export new types, predicates, and Document RAG functions | | `trustgraph-base/trustgraph/provenance/__init__.py` | Export new types, predicates, and Document RAG functions |
| `trustgraph-base/trustgraph/schema/services/retrieval.py` | Add explain_id and explain_graph to DocumentRagResponse | | `trustgraph-base/trustgraph/schema/services/retrieval.py` | Add explain_id, explain_graph, and explain_triples to DocumentRagResponse |
| `trustgraph-base/trustgraph/messaging/translators/retrieval.py` | Update DocumentRagResponseTranslator for explainability fields | | `trustgraph-base/trustgraph/messaging/translators/retrieval.py` | Update DocumentRagResponseTranslator for explainability fields including inline triples |
| `trustgraph-flow/trustgraph/agent/react/service.py` | Add explainability producer + recording logic | | `trustgraph-flow/trustgraph/agent/react/service.py` | Add explainability producer + recording logic |
| `trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py` | Add explainability callback and emit provenance triples | | `trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py` | Add explainability callback and emit provenance triples |
| `trustgraph-flow/trustgraph/retrieval/document_rag/rag.py` | Add explainability producer and wire up callback | | `trustgraph-flow/trustgraph/retrieval/document_rag/rag.py` | Add explainability producer and wire up callback |

View file

@ -63,7 +63,11 @@ Explainability events stream to client as the query executes:
3. Edges selected with reasoning → event emitted 3. Edges selected with reasoning → event emitted
4. Answer synthesized → event emitted 4. Answer synthesized → event emitted
Client receives `explain_id` and `explain_collection` to fetch full details. Client receives `explain_id`, `explain_graph`, and `explain_triples` inline
in each explain message. The triples contain the full provenance data for
that step — no follow-up graph query needed. The `explain_id` serves as
the root entity URI within the triples. Data is also written to the
knowledge graph for later audit/analysis.
## URI Structure ## URI Structure
@ -144,7 +148,8 @@ class GraphRagResponse:
response: str = "" response: str = ""
end_of_stream: bool = False end_of_stream: bool = False
explain_id: str | None = None explain_id: str | None = None
explain_collection: str | None = None explain_graph: str | None = None
explain_triples: list[Triple] = field(default_factory=list)
message_type: str = "" # "chunk" or "explain" message_type: str = "" # "chunk" or "explain"
end_of_session: bool = False end_of_session: bool = False
``` ```
@ -154,7 +159,7 @@ class GraphRagResponse:
| message_type | Purpose | | message_type | Purpose |
|--------------|---------| |--------------|---------|
| `chunk` | Response text (streaming or final) | | `chunk` | Response text (streaming or final) |
| `explain` | Explainability event with IRI reference | | `explain` | Explainability event with inline provenance triples |
### Session Lifecycle ### Session Lifecycle

View file

@ -0,0 +1,359 @@
"""
Tests for inline explainability triples in response translators
and ProvenanceEvent parsing.
"""
import pytest
from trustgraph.schema import (
GraphRagResponse, DocumentRagResponse, AgentResponse,
Term, Triple, IRI, LITERAL, Error,
)
from trustgraph.messaging.translators.retrieval import (
GraphRagResponseTranslator,
DocumentRagResponseTranslator,
)
from trustgraph.messaging.translators.agent import (
AgentResponseTranslator,
)
from trustgraph.api.types import ProvenanceEvent
# --- Helpers ---
def make_triple(s_iri, p_iri, o_value, o_type=LITERAL):
"""Create a Triple with IRI subject/predicate and typed object."""
o = Term(type=IRI, iri=o_value) if o_type == IRI else Term(type=LITERAL, value=o_value)
return Triple(
s=Term(type=IRI, iri=s_iri),
p=Term(type=IRI, iri=p_iri),
o=o,
)
def sample_triples():
"""A few provenance triples for a question entity."""
return [
make_triple(
"urn:trustgraph:question:abc123",
"http://www.w3.org/1999/02/22-rdf-syntax-ns#type",
"https://trustgraph.ai/ns/GraphRagQuestion",
o_type=IRI,
),
make_triple(
"urn:trustgraph:question:abc123",
"https://trustgraph.ai/ns/query",
"What is the internet?",
),
make_triple(
"urn:trustgraph:question:abc123",
"http://www.w3.org/ns/prov#startedAtTime",
"2026-04-07T09:00:00Z",
),
]
# --- GraphRag Translator ---
class TestGraphRagExplainTriples:
def test_explain_triples_encoded(self):
translator = GraphRagResponseTranslator()
triples = sample_triples()
response = GraphRagResponse(
message_type="explain",
explain_id="urn:trustgraph:question:abc123",
explain_graph="urn:graph:retrieval",
explain_triples=triples,
)
result = translator.encode(response)
assert "explain_triples" in result
assert len(result["explain_triples"]) == 3
# Check first triple is properly encoded
t = result["explain_triples"][0]
assert t["s"]["t"] == "i"
assert t["s"]["i"] == "urn:trustgraph:question:abc123"
assert t["p"]["t"] == "i"
def test_explain_triples_empty_not_included(self):
translator = GraphRagResponseTranslator()
response = GraphRagResponse(
message_type="chunk",
response="Some answer text",
)
result = translator.encode(response)
assert "explain_triples" not in result
def test_explain_with_completion_returns_not_final(self):
translator = GraphRagResponseTranslator()
response = GraphRagResponse(
message_type="explain",
explain_id="urn:trustgraph:question:abc123",
explain_triples=sample_triples(),
end_of_session=False,
)
result, is_final = translator.encode_with_completion(response)
assert is_final is False
def test_explain_id_and_graph_included(self):
translator = GraphRagResponseTranslator()
response = GraphRagResponse(
message_type="explain",
explain_id="urn:trustgraph:question:abc123",
explain_graph="urn:graph:retrieval",
explain_triples=sample_triples(),
)
result = translator.encode(response)
assert result["explain_id"] == "urn:trustgraph:question:abc123"
assert result["explain_graph"] == "urn:graph:retrieval"
# --- DocumentRag Translator ---
class TestDocumentRagExplainTriples:
def test_explain_triples_encoded(self):
translator = DocumentRagResponseTranslator()
response = DocumentRagResponse(
response=None,
message_type="explain",
explain_id="urn:trustgraph:docrag:abc123",
explain_graph="urn:graph:retrieval",
explain_triples=sample_triples(),
)
result = translator.encode(response)
assert "explain_triples" in result
assert len(result["explain_triples"]) == 3
def test_explain_triples_empty_not_included(self):
translator = DocumentRagResponseTranslator()
response = DocumentRagResponse(
response="Answer text",
message_type="chunk",
)
result = translator.encode(response)
assert "explain_triples" not in result
# --- Agent Translator ---
class TestAgentExplainTriples:
def test_explain_triples_encoded(self):
translator = AgentResponseTranslator()
response = AgentResponse(
chunk_type="explain",
content="",
explain_id="urn:trustgraph:agent:session:abc123",
explain_graph="urn:graph:retrieval",
explain_triples=sample_triples(),
)
result = translator.encode(response)
assert "explain_triples" in result
assert len(result["explain_triples"]) == 3
t = result["explain_triples"][1]
assert t["p"]["i"] == "https://trustgraph.ai/ns/query"
assert t["o"]["t"] == "l"
assert t["o"]["v"] == "What is the internet?"
def test_explain_triples_empty_not_included(self):
translator = AgentResponseTranslator()
response = AgentResponse(
chunk_type="thought",
content="I need to think...",
)
result = translator.encode(response)
assert "explain_triples" not in result
def test_explain_with_completion_not_final(self):
translator = AgentResponseTranslator()
response = AgentResponse(
chunk_type="explain",
explain_id="urn:trustgraph:agent:session:abc123",
explain_triples=sample_triples(),
end_of_dialog=False,
)
result, is_final = translator.encode_with_completion(response)
assert is_final is False
def test_explain_with_completion_final(self):
translator = AgentResponseTranslator()
response = AgentResponse(
chunk_type="answer",
content="The answer is...",
end_of_dialog=True,
)
result, is_final = translator.encode_with_completion(response)
assert is_final is True
# --- ProvenanceEvent ---
class TestProvenanceEvent:
def test_question_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:question:abc123",
)
assert event.event_type == "question"
def test_exploration_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:exploration:abc123",
)
assert event.event_type == "exploration"
def test_focus_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:focus:abc123",
)
assert event.event_type == "focus"
def test_synthesis_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:synthesis:abc123",
)
assert event.event_type == "synthesis"
def test_grounding_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:grounding:abc123",
)
assert event.event_type == "grounding"
def test_session_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:session:abc123",
)
assert event.event_type == "session"
def test_iteration_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:iteration:abc123:1",
)
assert event.event_type == "iteration"
def test_observation_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:observation:abc123:1",
)
assert event.event_type == "observation"
def test_conclusion_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:conclusion:abc123",
)
assert event.event_type == "conclusion"
def test_decomposition_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:decomposition:abc123",
)
assert event.event_type == "decomposition"
def test_finding_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:finding:abc123:0",
)
assert event.event_type == "finding"
def test_plan_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:plan:abc123",
)
assert event.event_type == "plan"
def test_step_result_event_type(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:agent:step-result:abc123:0",
)
assert event.event_type == "step-result"
def test_defaults(self):
event = ProvenanceEvent(
explain_id="urn:trustgraph:question:abc123",
)
assert event.entity is None
assert event.triples == []
assert event.explain_graph == ""
def test_with_triples(self):
raw = [{"s": {"t": "i", "i": "urn:x"}, "p": {"t": "i", "i": "urn:y"}, "o": {"t": "l", "v": "z"}}]
event = ProvenanceEvent(
explain_id="urn:trustgraph:question:abc123",
triples=raw,
)
assert len(event.triples) == 1
# --- Build ProvenanceEvent with entity parsing ---
class TestBuildProvenanceEvent:
def _make_client(self):
"""Create a minimal WebSocketClient-like object with _build_provenance_event."""
from trustgraph.api.socket_client import WebSocketClient
# We can't instantiate WebSocketClient easily, so test the method logic directly
return None
def test_entity_parsed_from_wire_triples(self):
"""Test that wire-format triples are parsed into an ExplainEntity."""
from trustgraph.api.explainability import ExplainEntity
wire_triples = [
{
"s": {"t": "i", "i": "urn:trustgraph:question:abc123"},
"p": {"t": "i", "i": "http://www.w3.org/1999/02/22-rdf-syntax-ns#type"},
"o": {"t": "i", "i": "https://trustgraph.ai/ns/GraphRagQuestion"},
},
{
"s": {"t": "i", "i": "urn:trustgraph:question:abc123"},
"p": {"t": "i", "i": "https://trustgraph.ai/ns/query"},
"o": {"t": "l", "v": "What is the internet?"},
},
]
# Parse triples the same way _build_provenance_event does
parsed = []
for t in wire_triples:
s = t.get("s", {}).get("i", "")
p = t.get("p", {}).get("i", "")
o_term = t.get("o", {})
if o_term.get("t") == "i":
o = o_term.get("i", "")
else:
o = o_term.get("v", "")
parsed.append((s, p, o))
entity = ExplainEntity.from_triples(
"urn:trustgraph:question:abc123", parsed
)
assert entity.entity_type == "question"
assert entity.query == "What is the internet?"
assert entity.question_type == "graph-rag"

View file

@ -366,19 +366,13 @@ class SocketClient:
# Handle GraphRAG/DocRAG message format with message_type # Handle GraphRAG/DocRAG message format with message_type
if message_type == "explain": if message_type == "explain":
if include_provenance: if include_provenance:
return ProvenanceEvent( return self._build_provenance_event(resp)
explain_id=resp.get("explain_id", ""),
explain_graph=resp.get("explain_graph", "")
)
return None return None
# Handle Agent message format with chunk_type="explain" # Handle Agent message format with chunk_type="explain"
if chunk_type == "explain": if chunk_type == "explain":
if include_provenance: if include_provenance:
return ProvenanceEvent( return self._build_provenance_event(resp)
explain_id=resp.get("explain_id", ""),
explain_graph=resp.get("explain_graph", "")
)
return None return None
if chunk_type == "thought": if chunk_type == "thought":
@ -413,6 +407,42 @@ class SocketClient:
error=None error=None
) )
def _build_provenance_event(self, resp: Dict[str, Any]) -> ProvenanceEvent:
"""Build a ProvenanceEvent from a response dict, parsing inline triples
into an ExplainEntity if available."""
explain_id = resp.get("explain_id", "")
explain_graph = resp.get("explain_graph", "")
raw_triples = resp.get("explain_triples", [])
entity = None
if raw_triples:
try:
from .explainability import ExplainEntity
# Convert wire-format triple dicts to (s, p, o) tuples
parsed = []
for t in raw_triples:
s = t.get("s", {}).get("i", "") if t.get("s") else ""
p = t.get("p", {}).get("i", "") if t.get("p") else ""
o_term = t.get("o", {})
if o_term:
if o_term.get("t") == "i":
o = o_term.get("i", "")
else:
o = o_term.get("v", "")
else:
o = ""
parsed.append((s, p, o))
entity = ExplainEntity.from_triples(explain_id, parsed)
except Exception:
pass
return ProvenanceEvent(
explain_id=explain_id,
explain_graph=explain_graph,
entity=entity,
triples=raw_triples,
)
def close(self) -> None: def close(self) -> None:
"""Close the persistent WebSocket connection.""" """Close the persistent WebSocket connection."""
if self._loop and not self._loop.is_closed(): if self._loop and not self._loop.is_closed():

View file

@ -213,25 +213,47 @@ class ProvenanceEvent:
""" """
Provenance event for explainability. Provenance event for explainability.
Emitted during GraphRAG queries when explainable mode is enabled. Emitted during retrieval queries when explainable mode is enabled.
Each event represents a provenance node created during query processing. Each event represents a provenance node created during query processing.
Attributes: Attributes:
explain_id: URI of the provenance node (e.g., urn:trustgraph:question:abc123) explain_id: URI of the provenance node (e.g., urn:trustgraph:question:abc123)
explain_graph: Named graph where provenance triples are stored (e.g., urn:graph:retrieval) explain_graph: Named graph where provenance triples are stored (e.g., urn:graph:retrieval)
event_type: Type of provenance event (question, exploration, focus, synthesis) event_type: Type of provenance event (question, exploration, focus, synthesis, etc.)
entity: Parsed ExplainEntity from inline triples (if available)
triples: Raw triples from the response (wire format dicts)
""" """
explain_id: str explain_id: str
explain_graph: str = "" explain_graph: str = ""
event_type: str = "" # Derived from explain_id event_type: str = "" # Derived from explain_id
entity: object = None # ExplainEntity (parsed from triples)
triples: list = dataclasses.field(default_factory=list) # Raw wire-format triple dicts
def __post_init__(self): def __post_init__(self):
# Extract event type from explain_id # Extract event type from explain_id
if "question" in self.explain_id: if "question" in self.explain_id:
self.event_type = "question" self.event_type = "question"
elif "grounding" in self.explain_id:
self.event_type = "grounding"
elif "exploration" in self.explain_id: elif "exploration" in self.explain_id:
self.event_type = "exploration" self.event_type = "exploration"
elif "focus" in self.explain_id: elif "focus" in self.explain_id:
self.event_type = "focus" self.event_type = "focus"
elif "synthesis" in self.explain_id: elif "synthesis" in self.explain_id:
self.event_type = "synthesis" self.event_type = "synthesis"
elif "iteration" in self.explain_id:
self.event_type = "iteration"
elif "observation" in self.explain_id:
self.event_type = "observation"
elif "conclusion" in self.explain_id:
self.event_type = "conclusion"
elif "decomposition" in self.explain_id:
self.event_type = "decomposition"
elif "finding" in self.explain_id:
self.event_type = "finding"
elif "plan" in self.explain_id:
self.event_type = "plan"
elif "step-result" in self.explain_id:
self.event_type = "step-result"
elif "session" in self.explain_id:
self.event_type = "session"

View file

@ -1,6 +1,7 @@
from typing import Dict, Any, Tuple from typing import Dict, Any, Tuple
from ...schema import AgentRequest, AgentResponse from ...schema import AgentRequest, AgentResponse
from .base import MessageTranslator from .base import MessageTranslator
from .primitives import TripleTranslator
class AgentRequestTranslator(MessageTranslator): class AgentRequestTranslator(MessageTranslator):
@ -50,6 +51,9 @@ class AgentRequestTranslator(MessageTranslator):
class AgentResponseTranslator(MessageTranslator): class AgentResponseTranslator(MessageTranslator):
"""Translator for AgentResponse schema objects""" """Translator for AgentResponse schema objects"""
def __init__(self):
self.triple_translator = TripleTranslator()
def decode(self, data: Dict[str, Any]) -> AgentResponse: def decode(self, data: Dict[str, Any]) -> AgentResponse:
raise NotImplementedError("Response translation to Pulsar not typically needed") raise NotImplementedError("Response translation to Pulsar not typically needed")
@ -75,6 +79,13 @@ class AgentResponseTranslator(MessageTranslator):
if explain_graph is not None: if explain_graph is not None:
result["explain_graph"] = explain_graph result["explain_graph"] = explain_graph
# Include explain_triples for explain messages
explain_triples = getattr(obj, "explain_triples", [])
if explain_triples:
result["explain_triples"] = [
self.triple_translator.encode(t) for t in explain_triples
]
# Always include error if present # Always include error if present
if hasattr(obj, 'error') and obj.error and obj.error.message: if hasattr(obj, 'error') and obj.error and obj.error.message:
result["error"] = {"message": obj.error.message, "code": obj.error.code} result["error"] = {"message": obj.error.message, "code": obj.error.code}

View file

@ -1,6 +1,7 @@
from typing import Dict, Any, Tuple from typing import Dict, Any, Tuple
from ...schema import DocumentRagQuery, DocumentRagResponse, GraphRagQuery, GraphRagResponse from ...schema import DocumentRagQuery, DocumentRagResponse, GraphRagQuery, GraphRagResponse
from .base import MessageTranslator from .base import MessageTranslator
from .primitives import TripleTranslator
class DocumentRagRequestTranslator(MessageTranslator): class DocumentRagRequestTranslator(MessageTranslator):
@ -28,6 +29,9 @@ class DocumentRagRequestTranslator(MessageTranslator):
class DocumentRagResponseTranslator(MessageTranslator): class DocumentRagResponseTranslator(MessageTranslator):
"""Translator for DocumentRagResponse schema objects""" """Translator for DocumentRagResponse schema objects"""
def __init__(self):
self.triple_translator = TripleTranslator()
def decode(self, data: Dict[str, Any]) -> DocumentRagResponse: def decode(self, data: Dict[str, Any]) -> DocumentRagResponse:
raise NotImplementedError("Response translation to Pulsar not typically needed") raise NotImplementedError("Response translation to Pulsar not typically needed")
@ -53,6 +57,13 @@ class DocumentRagResponseTranslator(MessageTranslator):
if explain_graph is not None: if explain_graph is not None:
result["explain_graph"] = explain_graph result["explain_graph"] = explain_graph
# Include explain_triples for explain messages
explain_triples = getattr(obj, "explain_triples", [])
if explain_triples:
result["explain_triples"] = [
self.triple_translator.encode(t) for t in explain_triples
]
# Include end_of_stream flag (LLM stream complete) # Include end_of_stream flag (LLM stream complete)
result["end_of_stream"] = getattr(obj, "end_of_stream", False) result["end_of_stream"] = getattr(obj, "end_of_stream", False)
@ -107,6 +118,9 @@ class GraphRagRequestTranslator(MessageTranslator):
class GraphRagResponseTranslator(MessageTranslator): class GraphRagResponseTranslator(MessageTranslator):
"""Translator for GraphRagResponse schema objects""" """Translator for GraphRagResponse schema objects"""
def __init__(self):
self.triple_translator = TripleTranslator()
def decode(self, data: Dict[str, Any]) -> GraphRagResponse: def decode(self, data: Dict[str, Any]) -> GraphRagResponse:
raise NotImplementedError("Response translation to Pulsar not typically needed") raise NotImplementedError("Response translation to Pulsar not typically needed")
@ -132,6 +146,13 @@ class GraphRagResponseTranslator(MessageTranslator):
if explain_graph is not None: if explain_graph is not None:
result["explain_graph"] = explain_graph result["explain_graph"] = explain_graph
# Include explain_triples for explain messages
explain_triples = getattr(obj, "explain_triples", [])
if explain_triples:
result["explain_triples"] = [
self.triple_translator.encode(t) for t in explain_triples
]
# Include end_of_stream flag (LLM stream complete) # Include end_of_stream flag (LLM stream complete)
result["end_of_stream"] = getattr(obj, "end_of_stream", False) result["end_of_stream"] = getattr(obj, "end_of_stream", False)

View file

@ -2,7 +2,7 @@
from dataclasses import dataclass, field from dataclasses import dataclass, field
from typing import Optional from typing import Optional
from ..core.primitives import Error from ..core.primitives import Error, Triple
############################################################################ ############################################################################
@ -57,8 +57,9 @@ class AgentResponse:
end_of_dialog: bool = False # Entire agent dialog is complete end_of_dialog: bool = False # Entire agent dialog is complete
# Explainability fields # Explainability fields
explain_id: str | None = None # Provenance URI (announced as created) explain_id: str | None = None # Root URI for this explain step
explain_graph: str | None = None # Named graph where explain was stored explain_graph: str | None = None # Named graph (e.g., urn:graph:retrieval)
explain_triples: list[Triple] = field(default_factory=list) # Provenance triples for this step
# Orchestration fields # Orchestration fields
message_id: str = "" # Unique ID for this response message message_id: str = "" # Unique ID for this response message

View file

@ -1,5 +1,5 @@
from dataclasses import dataclass from dataclasses import dataclass, field
from ..core.primitives import Error, Term from ..core.primitives import Error, Term, Triple
############################################################################ ############################################################################
@ -24,8 +24,9 @@ class GraphRagResponse:
error: Error | None = None error: Error | None = None
response: str = "" response: str = ""
end_of_stream: bool = False # LLM response stream complete end_of_stream: bool = False # LLM response stream complete
explain_id: str | None = None # Single explain URI (announced as created) explain_id: str | None = None # Root URI for this explain step
explain_graph: str | None = None # Named graph where explain was stored (e.g., urn:graph:retrieval) explain_graph: str | None = None # Named graph (e.g., urn:graph:retrieval)
explain_triples: list[Triple] = field(default_factory=list) # Provenance triples for this step
message_type: str = "" # "chunk" or "explain" message_type: str = "" # "chunk" or "explain"
end_of_session: bool = False # Entire session complete end_of_session: bool = False # Entire session complete
@ -46,7 +47,8 @@ class DocumentRagResponse:
error: Error | None = None error: Error | None = None
response: str | None = "" response: str | None = ""
end_of_stream: bool = False # LLM response stream complete end_of_stream: bool = False # LLM response stream complete
explain_id: str | None = None # Single explain URI (announced as created) explain_id: str | None = None # Root URI for this explain step
explain_graph: str | None = None # Named graph where explain was stored (e.g., urn:graph:retrieval) explain_graph: str | None = None # Named graph (e.g., urn:graph:retrieval)
explain_triples: list[Triple] = field(default_factory=list) # Provenance triples for this step
message_type: str = "" # "chunk" or "explain" message_type: str = "" # "chunk" or "explain"
end_of_session: bool = False # Entire session complete end_of_session: bool = False # Entire session complete

View file

@ -182,16 +182,18 @@ def question_explainable(
print(item.content, end="", flush=True) print(item.content, end="", flush=True)
elif isinstance(item, ProvenanceEvent): elif isinstance(item, ProvenanceEvent):
# Process provenance event immediately # Use inline entity if available, otherwise fetch from graph
prov_id = item.explain_id prov_id = item.explain_id
explain_graph = item.explain_graph or "urn:graph:retrieval" explain_graph = item.explain_graph or "urn:graph:retrieval"
entity = explain_client.fetch_entity( entity = item.entity
prov_id, if entity is None:
graph=explain_graph, entity = explain_client.fetch_entity(
user=user, prov_id,
collection=collection graph=explain_graph,
) user=user,
collection=collection
)
if entity is None: if entity is None:
if debug: if debug:

View file

@ -45,16 +45,18 @@ def question_explainable(
print(item.content, end="", flush=True) print(item.content, end="", flush=True)
elif isinstance(item, ProvenanceEvent): elif isinstance(item, ProvenanceEvent):
# Process provenance event immediately # Use inline entity if available, otherwise fetch from graph
prov_id = item.explain_id prov_id = item.explain_id
explain_graph = item.explain_graph or "urn:graph:retrieval" explain_graph = item.explain_graph or "urn:graph:retrieval"
entity = explain_client.fetch_entity( entity = item.entity
prov_id, if entity is None:
graph=explain_graph, entity = explain_client.fetch_entity(
user=user, prov_id,
collection=collection graph=explain_graph,
) user=user,
collection=collection
)
if entity is None: if entity is None:
if debug: if debug:

View file

@ -667,16 +667,18 @@ def _question_explainable_api(
print(item.content, end="", flush=True) print(item.content, end="", flush=True)
elif isinstance(item, ProvenanceEvent): elif isinstance(item, ProvenanceEvent):
# Process provenance event immediately # Use inline entity if available, otherwise fetch from graph
prov_id = item.explain_id prov_id = item.explain_id
explain_graph = item.explain_graph or "urn:graph:retrieval" explain_graph = item.explain_graph or "urn:graph:retrieval"
entity = explain_client.fetch_entity( entity = item.entity
prov_id, if entity is None:
graph=explain_graph, entity = explain_client.fetch_entity(
user=user, prov_id,
collection=collection graph=explain_graph,
) user=user,
collection=collection
)
if entity is None: if entity is None:
if debug: if debug:

View file

@ -243,6 +243,7 @@ class PatternBase:
content="", content="",
explain_id=session_uri, explain_id=session_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
async def emit_iteration_triples(self, flow, session_id, iteration_num, async def emit_iteration_triples(self, flow, session_id, iteration_num,
@ -305,6 +306,7 @@ class PatternBase:
content="", content="",
explain_id=iteration_uri, explain_id=iteration_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=iter_triples,
)) ))
async def emit_observation_triples(self, flow, session_id, iteration_num, async def emit_observation_triples(self, flow, session_id, iteration_num,
@ -360,6 +362,7 @@ class PatternBase:
content="", content="",
explain_id=observation_entity_uri, explain_id=observation_entity_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=obs_triples,
)) ))
async def emit_final_triples(self, flow, session_id, iteration_num, async def emit_final_triples(self, flow, session_id, iteration_num,
@ -416,6 +419,7 @@ class PatternBase:
content="", content="",
explain_id=final_uri, explain_id=final_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=final_triples,
)) ))
# ---- Orchestrator provenance helpers ------------------------------------ # ---- Orchestrator provenance helpers ------------------------------------
@ -437,6 +441,7 @@ class PatternBase:
await respond(AgentResponse( await respond(AgentResponse(
chunk_type="explain", content="", chunk_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL, explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
async def emit_finding_triples( async def emit_finding_triples(
@ -475,6 +480,7 @@ class PatternBase:
await respond(AgentResponse( await respond(AgentResponse(
chunk_type="explain", content="", chunk_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL, explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
async def emit_plan_triples( async def emit_plan_triples(
@ -494,6 +500,7 @@ class PatternBase:
await respond(AgentResponse( await respond(AgentResponse(
chunk_type="explain", content="", chunk_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL, explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
async def emit_step_result_triples( async def emit_step_result_triples(
@ -526,6 +533,7 @@ class PatternBase:
await respond(AgentResponse( await respond(AgentResponse(
chunk_type="explain", content="", chunk_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL, explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
async def emit_synthesis_triples( async def emit_synthesis_triples(
@ -557,6 +565,7 @@ class PatternBase:
await respond(AgentResponse( await respond(AgentResponse(
chunk_type="explain", content="", chunk_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL, explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
# ---- Response helpers --------------------------------------------------- # ---- Response helpers ---------------------------------------------------

View file

@ -473,6 +473,7 @@ class Processor(AgentService):
content="", content="",
explain_id=session_uri, explain_id=session_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
)) ))
logger.info(f"Question: {request.question}") logger.info(f"Question: {request.question}")
@ -640,6 +641,7 @@ class Processor(AgentService):
content="", content="",
explain_id=iter_uri, explain_id=iter_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=iter_triples,
)) ))
user_context = UserAwareContext(flow, request.user) user_context = UserAwareContext(flow, request.user)
@ -717,6 +719,7 @@ class Processor(AgentService):
content="", content="",
explain_id=final_uri, explain_id=final_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=final_triples,
)) ))
if streaming: if streaming:
@ -793,6 +796,7 @@ class Processor(AgentService):
content="", content="",
explain_id=observation_entity_uri, explain_id=observation_entity_uri,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=obs_triples,
)) ))
history.append(act) history.append(act)

View file

@ -162,12 +162,13 @@ class Processor(FlowProcessor):
triples=triples, triples=triples,
)) ))
# Send explain ID and graph to response queue # Send explain data to response queue
await flow("response").send( await flow("response").send(
DocumentRagResponse( DocumentRagResponse(
response=None, response=None,
explain_id=explain_id, explain_id=explain_id,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
message_type="explain", message_type="explain",
), ),
properties={"id": id} properties={"id": id}

View file

@ -253,12 +253,13 @@ class Processor(FlowProcessor):
triples=triples, triples=triples,
)) ))
# Send explain ID and graph to response queue # Send explain data to response queue
await flow("response").send( await flow("response").send(
GraphRagResponse( GraphRagResponse(
message_type="explain", message_type="explain",
explain_id=explain_id, explain_id=explain_id,
explain_graph=GRAPH_RETRIEVAL, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
), ),
properties={"id": id} properties={"id": id}
) )