diff --git a/docs/tech-specs/agent-explainability.md b/docs/tech-specs/agent-explainability.md index f02a95b1..3dee0ac2 100644 --- a/docs/tech-specs/agent-explainability.md +++ b/docs/tech-specs/agent-explainability.md @@ -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/uris.py` | Add Document RAG URI generators | | `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/messaging/translators/retrieval.py` | Update DocumentRagResponseTranslator for explainability fields | +| `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 including inline triples | | `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/rag.py` | Add explainability producer and wire up callback | diff --git a/docs/tech-specs/query-time-explainability.md b/docs/tech-specs/query-time-explainability.md index e696745c..69cb45ac 100644 --- a/docs/tech-specs/query-time-explainability.md +++ b/docs/tech-specs/query-time-explainability.md @@ -63,7 +63,11 @@ Explainability events stream to client as the query executes: 3. Edges selected with reasoning → 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 @@ -144,7 +148,8 @@ class GraphRagResponse: response: str = "" end_of_stream: bool = False 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" end_of_session: bool = False ``` @@ -154,7 +159,7 @@ class GraphRagResponse: | message_type | Purpose | |--------------|---------| | `chunk` | Response text (streaming or final) | -| `explain` | Explainability event with IRI reference | +| `explain` | Explainability event with inline provenance triples | ### Session Lifecycle diff --git a/trustgraph-base/trustgraph/api/socket_client.py b/trustgraph-base/trustgraph/api/socket_client.py index 9c37a9b1..b6ceba00 100644 --- a/trustgraph-base/trustgraph/api/socket_client.py +++ b/trustgraph-base/trustgraph/api/socket_client.py @@ -366,19 +366,13 @@ class SocketClient: # Handle GraphRAG/DocRAG message format with message_type if message_type == "explain": if include_provenance: - return ProvenanceEvent( - explain_id=resp.get("explain_id", ""), - explain_graph=resp.get("explain_graph", "") - ) + return self._build_provenance_event(resp) return None # Handle Agent message format with chunk_type="explain" if chunk_type == "explain": if include_provenance: - return ProvenanceEvent( - explain_id=resp.get("explain_id", ""), - explain_graph=resp.get("explain_graph", "") - ) + return self._build_provenance_event(resp) return None if chunk_type == "thought": @@ -413,6 +407,42 @@ class SocketClient: 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: """Close the persistent WebSocket connection.""" if self._loop and not self._loop.is_closed(): diff --git a/trustgraph-base/trustgraph/api/types.py b/trustgraph-base/trustgraph/api/types.py index 0715293b..55635584 100644 --- a/trustgraph-base/trustgraph/api/types.py +++ b/trustgraph-base/trustgraph/api/types.py @@ -213,25 +213,47 @@ class ProvenanceEvent: """ 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. Attributes: 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) - 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_graph: str = "" 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): # Extract event type from explain_id if "question" in self.explain_id: self.event_type = "question" + elif "grounding" in self.explain_id: + self.event_type = "grounding" elif "exploration" in self.explain_id: self.event_type = "exploration" elif "focus" in self.explain_id: self.event_type = "focus" elif "synthesis" in self.explain_id: 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" diff --git a/trustgraph-base/trustgraph/messaging/translators/agent.py b/trustgraph-base/trustgraph/messaging/translators/agent.py index c2c00ac2..8cf525f5 100644 --- a/trustgraph-base/trustgraph/messaging/translators/agent.py +++ b/trustgraph-base/trustgraph/messaging/translators/agent.py @@ -1,6 +1,7 @@ from typing import Dict, Any, Tuple from ...schema import AgentRequest, AgentResponse from .base import MessageTranslator +from .primitives import TripleTranslator class AgentRequestTranslator(MessageTranslator): @@ -49,10 +50,13 @@ class AgentRequestTranslator(MessageTranslator): class AgentResponseTranslator(MessageTranslator): """Translator for AgentResponse schema objects""" - + + def __init__(self): + self.triple_translator = TripleTranslator() + def decode(self, data: Dict[str, Any]) -> AgentResponse: raise NotImplementedError("Response translation to Pulsar not typically needed") - + def encode(self, obj: AgentResponse) -> Dict[str, Any]: result = {} @@ -75,6 +79,13 @@ class AgentResponseTranslator(MessageTranslator): if explain_graph is not None: 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 if hasattr(obj, 'error') and obj.error and obj.error.message: result["error"] = {"message": obj.error.message, "code": obj.error.code} diff --git a/trustgraph-base/trustgraph/messaging/translators/retrieval.py b/trustgraph-base/trustgraph/messaging/translators/retrieval.py index 7e2abfa1..849bee94 100644 --- a/trustgraph-base/trustgraph/messaging/translators/retrieval.py +++ b/trustgraph-base/trustgraph/messaging/translators/retrieval.py @@ -1,6 +1,7 @@ from typing import Dict, Any, Tuple from ...schema import DocumentRagQuery, DocumentRagResponse, GraphRagQuery, GraphRagResponse from .base import MessageTranslator +from .primitives import TripleTranslator class DocumentRagRequestTranslator(MessageTranslator): @@ -28,6 +29,9 @@ class DocumentRagRequestTranslator(MessageTranslator): class DocumentRagResponseTranslator(MessageTranslator): """Translator for DocumentRagResponse schema objects""" + def __init__(self): + self.triple_translator = TripleTranslator() + def decode(self, data: Dict[str, Any]) -> DocumentRagResponse: raise NotImplementedError("Response translation to Pulsar not typically needed") @@ -53,6 +57,13 @@ class DocumentRagResponseTranslator(MessageTranslator): if explain_graph is not None: 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) result["end_of_stream"] = getattr(obj, "end_of_stream", False) @@ -107,6 +118,9 @@ class GraphRagRequestTranslator(MessageTranslator): class GraphRagResponseTranslator(MessageTranslator): """Translator for GraphRagResponse schema objects""" + def __init__(self): + self.triple_translator = TripleTranslator() + def decode(self, data: Dict[str, Any]) -> GraphRagResponse: raise NotImplementedError("Response translation to Pulsar not typically needed") @@ -132,6 +146,13 @@ class GraphRagResponseTranslator(MessageTranslator): if explain_graph is not None: 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) result["end_of_stream"] = getattr(obj, "end_of_stream", False) diff --git a/trustgraph-base/trustgraph/schema/services/agent.py b/trustgraph-base/trustgraph/schema/services/agent.py index 2a966dd4..fbc0101c 100644 --- a/trustgraph-base/trustgraph/schema/services/agent.py +++ b/trustgraph-base/trustgraph/schema/services/agent.py @@ -2,7 +2,7 @@ from dataclasses import dataclass, field 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 # Explainability fields - explain_id: str | None = None # Provenance URI (announced as created) - explain_graph: str | None = None # Named graph where explain was stored + explain_id: str | None = None # Root URI for this explain step + 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 message_id: str = "" # Unique ID for this response message diff --git a/trustgraph-base/trustgraph/schema/services/retrieval.py b/trustgraph-base/trustgraph/schema/services/retrieval.py index a4621549..4b17733d 100644 --- a/trustgraph-base/trustgraph/schema/services/retrieval.py +++ b/trustgraph-base/trustgraph/schema/services/retrieval.py @@ -1,5 +1,5 @@ -from dataclasses import dataclass -from ..core.primitives import Error, Term +from dataclasses import dataclass, field +from ..core.primitives import Error, Term, Triple ############################################################################ @@ -24,8 +24,9 @@ class GraphRagResponse: error: Error | None = None response: str = "" end_of_stream: bool = False # LLM response stream complete - explain_id: str | None = None # Single explain URI (announced as created) - explain_graph: str | None = None # Named graph where explain was stored (e.g., urn:graph:retrieval) + explain_id: str | None = None # Root URI for this explain step + 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" end_of_session: bool = False # Entire session complete @@ -46,7 +47,8 @@ class DocumentRagResponse: error: Error | None = None response: str | None = "" end_of_stream: bool = False # LLM response stream complete - explain_id: str | None = None # Single explain URI (announced as created) - explain_graph: str | None = None # Named graph where explain was stored (e.g., urn:graph:retrieval) + explain_id: str | None = None # Root URI for this explain step + 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" end_of_session: bool = False # Entire session complete diff --git a/trustgraph-cli/trustgraph/cli/invoke_agent.py b/trustgraph-cli/trustgraph/cli/invoke_agent.py index 1c4b757b..026286d0 100644 --- a/trustgraph-cli/trustgraph/cli/invoke_agent.py +++ b/trustgraph-cli/trustgraph/cli/invoke_agent.py @@ -182,16 +182,18 @@ def question_explainable( print(item.content, end="", flush=True) elif isinstance(item, ProvenanceEvent): - # Process provenance event immediately + # Use inline entity if available, otherwise fetch from graph prov_id = item.explain_id explain_graph = item.explain_graph or "urn:graph:retrieval" - entity = explain_client.fetch_entity( - prov_id, - graph=explain_graph, - user=user, - collection=collection - ) + entity = item.entity + if entity is None: + entity = explain_client.fetch_entity( + prov_id, + graph=explain_graph, + user=user, + collection=collection + ) if entity is None: if debug: diff --git a/trustgraph-cli/trustgraph/cli/invoke_document_rag.py b/trustgraph-cli/trustgraph/cli/invoke_document_rag.py index 7da9d779..066b92f4 100644 --- a/trustgraph-cli/trustgraph/cli/invoke_document_rag.py +++ b/trustgraph-cli/trustgraph/cli/invoke_document_rag.py @@ -45,16 +45,18 @@ def question_explainable( print(item.content, end="", flush=True) elif isinstance(item, ProvenanceEvent): - # Process provenance event immediately + # Use inline entity if available, otherwise fetch from graph prov_id = item.explain_id explain_graph = item.explain_graph or "urn:graph:retrieval" - entity = explain_client.fetch_entity( - prov_id, - graph=explain_graph, - user=user, - collection=collection - ) + entity = item.entity + if entity is None: + entity = explain_client.fetch_entity( + prov_id, + graph=explain_graph, + user=user, + collection=collection + ) if entity is None: if debug: diff --git a/trustgraph-cli/trustgraph/cli/invoke_graph_rag.py b/trustgraph-cli/trustgraph/cli/invoke_graph_rag.py index 76b8b158..230cc54b 100644 --- a/trustgraph-cli/trustgraph/cli/invoke_graph_rag.py +++ b/trustgraph-cli/trustgraph/cli/invoke_graph_rag.py @@ -667,16 +667,18 @@ def _question_explainable_api( print(item.content, end="", flush=True) elif isinstance(item, ProvenanceEvent): - # Process provenance event immediately + # Use inline entity if available, otherwise fetch from graph prov_id = item.explain_id explain_graph = item.explain_graph or "urn:graph:retrieval" - entity = explain_client.fetch_entity( - prov_id, - graph=explain_graph, - user=user, - collection=collection - ) + entity = item.entity + if entity is None: + entity = explain_client.fetch_entity( + prov_id, + graph=explain_graph, + user=user, + collection=collection + ) if entity is None: if debug: diff --git a/trustgraph-flow/trustgraph/agent/orchestrator/pattern_base.py b/trustgraph-flow/trustgraph/agent/orchestrator/pattern_base.py index 8849a206..c18c5bac 100644 --- a/trustgraph-flow/trustgraph/agent/orchestrator/pattern_base.py +++ b/trustgraph-flow/trustgraph/agent/orchestrator/pattern_base.py @@ -243,6 +243,7 @@ class PatternBase: content="", explain_id=session_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) async def emit_iteration_triples(self, flow, session_id, iteration_num, @@ -305,6 +306,7 @@ class PatternBase: content="", explain_id=iteration_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=iter_triples, )) async def emit_observation_triples(self, flow, session_id, iteration_num, @@ -360,6 +362,7 @@ class PatternBase: content="", explain_id=observation_entity_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=obs_triples, )) async def emit_final_triples(self, flow, session_id, iteration_num, @@ -416,6 +419,7 @@ class PatternBase: content="", explain_id=final_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=final_triples, )) # ---- Orchestrator provenance helpers ------------------------------------ @@ -437,6 +441,7 @@ class PatternBase: await respond(AgentResponse( chunk_type="explain", content="", explain_id=uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) async def emit_finding_triples( @@ -475,6 +480,7 @@ class PatternBase: await respond(AgentResponse( chunk_type="explain", content="", explain_id=uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) async def emit_plan_triples( @@ -494,6 +500,7 @@ class PatternBase: await respond(AgentResponse( chunk_type="explain", content="", explain_id=uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) async def emit_step_result_triples( @@ -526,6 +533,7 @@ class PatternBase: await respond(AgentResponse( chunk_type="explain", content="", explain_id=uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) async def emit_synthesis_triples( @@ -557,6 +565,7 @@ class PatternBase: await respond(AgentResponse( chunk_type="explain", content="", explain_id=uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) # ---- Response helpers --------------------------------------------------- diff --git a/trustgraph-flow/trustgraph/agent/react/service.py b/trustgraph-flow/trustgraph/agent/react/service.py index 0e783349..aca034a8 100755 --- a/trustgraph-flow/trustgraph/agent/react/service.py +++ b/trustgraph-flow/trustgraph/agent/react/service.py @@ -471,6 +471,7 @@ class Processor(AgentService): content="", explain_id=session_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, )) logger.info(f"Question: {request.question}") @@ -638,6 +639,7 @@ class Processor(AgentService): content="", explain_id=iter_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=iter_triples, )) user_context = UserAwareContext(flow, request.user) @@ -715,6 +717,7 @@ class Processor(AgentService): content="", explain_id=final_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=final_triples, )) if streaming: @@ -791,6 +794,7 @@ class Processor(AgentService): content="", explain_id=observation_entity_uri, explain_graph=GRAPH_RETRIEVAL, + explain_triples=obs_triples, )) history.append(act) diff --git a/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py b/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py index c0e55d84..3b281fe3 100755 --- a/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py +++ b/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py @@ -162,12 +162,13 @@ class Processor(FlowProcessor): triples=triples, )) - # Send explain ID and graph to response queue + # Send explain data to response queue await flow("response").send( DocumentRagResponse( response=None, explain_id=explain_id, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, message_type="explain", ), properties={"id": id} diff --git a/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py b/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py index 85a7491e..abf10e90 100755 --- a/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py +++ b/trustgraph-flow/trustgraph/retrieval/graph_rag/rag.py @@ -253,12 +253,13 @@ class Processor(FlowProcessor): triples=triples, )) - # Send explain ID and graph to response queue + # Send explain data to response queue await flow("response").send( GraphRagResponse( message_type="explain", explain_id=explain_id, explain_graph=GRAPH_RETRIEVAL, + explain_triples=triples, ), properties={"id": id} )