diff --git a/docs/tech-specs/agent-explainability.md b/docs/tech-specs/agent-explainability.md index b8751d7d..cb3a1816 100644 --- a/docs/tech-specs/agent-explainability.md +++ b/docs/tech-specs/agent-explainability.md @@ -29,6 +29,15 @@ Both GraphRAG and Agent use PROV-O as the base ontology with TrustGraph-specific | Analysis | `prov:Entity` | `tg:Analysis` | Each think/act/observe cycle | | Conclusion | `prov:Entity` | `tg:Conclusion` | Final answer | +### Document RAG Types +| Entity | PROV-O Type | TG Type | Description | +|--------|-------------|---------|-------------| +| Question | `prov:Activity` | `tg:Question` | The user's query (same as GraphRAG) | +| Exploration | `prov:Entity` | `tg:Exploration` | Chunks retrieved from document store | +| Synthesis | `prov:Entity` | `tg:Synthesis` | Final answer | + +**Note:** Document RAG uses a subset of GraphRAG's types (no Focus step since there's no edge selection/reasoning phase). + ## Provenance Model ``` @@ -60,6 +69,33 @@ Conclusion (urn:trustgraph:agent:{uuid}/final) │ rdf:type = prov:Entity, tg:Conclusion ``` +### Document RAG Provenance Model + +``` +Question (urn:trustgraph:docrag:{uuid}) + │ + │ tg:query = "User's question" + │ prov:startedAtTime = timestamp + │ rdf:type = prov:Activity, tg:Question + │ + ↓ prov:wasGeneratedBy + │ +Exploration (urn:trustgraph:docrag:{uuid}/exploration) + │ + │ tg:chunkCount = 5 + │ tg:selectedChunk = "chunk-id-1" + │ tg:selectedChunk = "chunk-id-2" + │ ... + │ rdf:type = prov:Entity, tg:Exploration + │ + ↓ prov:wasDerivedFrom + │ +Synthesis (urn:trustgraph:docrag:{uuid}/synthesis) + │ + │ tg:content = "The synthesized answer..." + │ rdf:type = prov:Entity, tg:Synthesis +``` + ## Changes Required ### 1. Schema Changes @@ -167,10 +203,15 @@ TG_ANSWER = TG + "answer" |------|--------| | `trustgraph-base/trustgraph/schema/services/agent.py` | Add session_id and collection to AgentRequest | | `trustgraph-base/trustgraph/messaging/translators/agent.py` | Update translator for new fields | -| `trustgraph-base/trustgraph/provenance/namespaces.py` | Add entity types and agent predicates | -| `trustgraph-base/trustgraph/provenance/triples.py` | Add TG types to GraphRAG triple builders | -| `trustgraph-base/trustgraph/provenance/__init__.py` | Export new types and predicates | +| `trustgraph-base/trustgraph/provenance/namespaces.py` | Add entity types, agent predicates, and Document RAG predicates | +| `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-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 | | `trustgraph-cli/trustgraph/cli/show_explain_trace.py` | Handle agent trace types | | `trustgraph-cli/trustgraph/cli/list_explain_traces.py` | List agent sessions alongside GraphRAG | @@ -216,5 +257,4 @@ tg-show-explain-trace "urn:trustgraph:agent:xxx" - DAG dependencies (when analysis N uses results from multiple prior analyses) - Tool-specific provenance linking (KnowledgeQuery → its GraphRAG trace) -- Document RAG explainability - Streaming provenance emission (emit as we go, not batch at end) diff --git a/trustgraph-base/trustgraph/messaging/translators/retrieval.py b/trustgraph-base/trustgraph/messaging/translators/retrieval.py index f84ce103..9f102f9a 100644 --- a/trustgraph-base/trustgraph/messaging/translators/retrieval.py +++ b/trustgraph-base/trustgraph/messaging/translators/retrieval.py @@ -38,6 +38,16 @@ class DocumentRagResponseTranslator(MessageTranslator): if obj.response is not None: result["response"] = obj.response + # Include explain_id for explain messages + explain_id = getattr(obj, "explain_id", None) + if explain_id: + result["explain_id"] = explain_id + + # Include explain_graph for explain messages (named graph filter) + explain_graph = getattr(obj, "explain_graph", None) + if explain_graph is not None: + result["explain_graph"] = explain_graph + # Include end_of_stream flag result["end_of_stream"] = getattr(obj, "end_of_stream", False) diff --git a/trustgraph-base/trustgraph/provenance/__init__.py b/trustgraph-base/trustgraph/provenance/__init__.py index 06660016..5a2f04b0 100644 --- a/trustgraph-base/trustgraph/provenance/__init__.py +++ b/trustgraph-base/trustgraph/provenance/__init__.py @@ -40,7 +40,7 @@ from . uris import ( activity_uri, statement_uri, agent_uri, - # Query-time provenance URIs + # Query-time provenance URIs (GraphRAG) question_uri, exploration_uri, focus_uri, @@ -49,6 +49,10 @@ from . uris import ( agent_session_uri, agent_iteration_uri, agent_final_uri, + # Document RAG provenance URIs + docrag_question_uri, + docrag_exploration_uri, + docrag_synthesis_uri, ) # Namespace constants @@ -67,8 +71,10 @@ from . namespaces import ( TG_CHUNK_SIZE, TG_CHUNK_OVERLAP, TG_COMPONENT_VERSION, TG_LLM_MODEL, TG_ONTOLOGY, TG_EMBEDDING_MODEL, TG_SOURCE_TEXT, TG_SOURCE_CHAR_OFFSET, TG_SOURCE_CHAR_LENGTH, - # Query-time provenance predicates + # Query-time provenance predicates (GraphRAG) TG_QUERY, TG_EDGE_COUNT, TG_SELECTED_EDGE, TG_REASONING, TG_CONTENT, + # Query-time provenance predicates (DocumentRAG) + TG_CHUNK_COUNT, TG_SELECTED_CHUNK, # Explainability entity types TG_QUESTION, TG_EXPLORATION, TG_FOCUS, TG_SYNTHESIS, TG_ANALYSIS, TG_CONCLUSION, @@ -83,11 +89,15 @@ from . triples import ( document_triples, derived_entity_triples, triple_provenance_triples, - # Query-time provenance triple builders + # Query-time provenance triple builders (GraphRAG) question_triples, exploration_triples, focus_triples, synthesis_triples, + # Query-time provenance triple builders (DocumentRAG) + docrag_question_triples, + docrag_exploration_triples, + docrag_synthesis_triples, # Utility set_graph, ) @@ -127,6 +137,10 @@ __all__ = [ "agent_session_uri", "agent_iteration_uri", "agent_final_uri", + # Document RAG provenance URIs + "docrag_question_uri", + "docrag_exploration_uri", + "docrag_synthesis_uri", # Namespaces "PROV", "PROV_ENTITY", "PROV_ACTIVITY", "PROV_AGENT", "PROV_WAS_DERIVED_FROM", "PROV_WAS_GENERATED_BY", @@ -138,8 +152,10 @@ __all__ = [ "TG_CHUNK_SIZE", "TG_CHUNK_OVERLAP", "TG_COMPONENT_VERSION", "TG_LLM_MODEL", "TG_ONTOLOGY", "TG_EMBEDDING_MODEL", "TG_SOURCE_TEXT", "TG_SOURCE_CHAR_OFFSET", "TG_SOURCE_CHAR_LENGTH", - # Query-time provenance predicates + # Query-time provenance predicates (GraphRAG) "TG_QUERY", "TG_EDGE_COUNT", "TG_SELECTED_EDGE", "TG_REASONING", "TG_CONTENT", + # Query-time provenance predicates (DocumentRAG) + "TG_CHUNK_COUNT", "TG_SELECTED_CHUNK", # Explainability entity types "TG_QUESTION", "TG_EXPLORATION", "TG_FOCUS", "TG_SYNTHESIS", "TG_ANALYSIS", "TG_CONCLUSION", @@ -151,11 +167,15 @@ __all__ = [ "document_triples", "derived_entity_triples", "triple_provenance_triples", - # Query-time provenance triple builders + # Query-time provenance triple builders (GraphRAG) "question_triples", "exploration_triples", "focus_triples", "synthesis_triples", + # Query-time provenance triple builders (DocumentRAG) + "docrag_question_triples", + "docrag_exploration_triples", + "docrag_synthesis_triples", # Agent provenance triple builders "agent_session_triples", "agent_iteration_triples", diff --git a/trustgraph-base/trustgraph/provenance/namespaces.py b/trustgraph-base/trustgraph/provenance/namespaces.py index 673dcc1f..1375de94 100644 --- a/trustgraph-base/trustgraph/provenance/namespaces.py +++ b/trustgraph-base/trustgraph/provenance/namespaces.py @@ -59,7 +59,7 @@ TG_SOURCE_TEXT = TG + "sourceText" TG_SOURCE_CHAR_OFFSET = TG + "sourceCharOffset" TG_SOURCE_CHAR_LENGTH = TG + "sourceCharLength" -# Query-time provenance predicates +# Query-time provenance predicates (GraphRAG) TG_QUERY = TG + "query" TG_EDGE_COUNT = TG + "edgeCount" TG_SELECTED_EDGE = TG + "selectedEdge" @@ -68,6 +68,10 @@ TG_REASONING = TG + "reasoning" TG_CONTENT = TG + "content" TG_DOCUMENT = TG + "document" # Reference to document in librarian +# Query-time provenance predicates (DocumentRAG) +TG_CHUNK_COUNT = TG + "chunkCount" +TG_SELECTED_CHUNK = TG + "selectedChunk" + # Explainability entity types (used by both GraphRAG and Agent) TG_QUESTION = TG + "Question" TG_EXPLORATION = TG + "Exploration" diff --git a/trustgraph-base/trustgraph/provenance/triples.py b/trustgraph-base/trustgraph/provenance/triples.py index 66e7b6d3..8c21614e 100644 --- a/trustgraph-base/trustgraph/provenance/triples.py +++ b/trustgraph-base/trustgraph/provenance/triples.py @@ -17,9 +17,11 @@ from . namespaces import ( TG_CHUNK_INDEX, TG_CHAR_OFFSET, TG_CHAR_LENGTH, TG_CHUNK_SIZE, TG_CHUNK_OVERLAP, TG_COMPONENT_VERSION, TG_LLM_MODEL, TG_ONTOLOGY, TG_REIFIES, - # Query-time provenance predicates + # Query-time provenance predicates (GraphRAG) TG_QUERY, TG_EDGE_COUNT, TG_SELECTED_EDGE, TG_EDGE, TG_REASONING, TG_CONTENT, TG_DOCUMENT, + # Query-time provenance predicates (DocumentRAG) + TG_CHUNK_COUNT, TG_SELECTED_CHUNK, # Explainability entity types TG_QUESTION, TG_EXPLORATION, TG_FOCUS, TG_SYNTHESIS, ) @@ -461,3 +463,119 @@ def synthesis_triples( triples.append(_triple(synthesis_uri, TG_CONTENT, _literal(answer_text))) return triples + + +# Document RAG provenance triple builders +# +# Document RAG uses a subset of GraphRAG's model: +# Question - What was asked +# Exploration - Chunks retrieved from document store +# Synthesis - The final answer (no Focus step) + +def docrag_question_triples( + question_uri: str, + query: str, + timestamp: Optional[str] = None, +) -> List[Triple]: + """ + Build triples for a document RAG question activity. + + Creates: + - Activity declaration with tg:Question type + - Query text and timestamp + + Args: + question_uri: URI of the question (from docrag_question_uri) + query: The user's query text + timestamp: ISO timestamp (defaults to now) + + Returns: + List of Triple objects + """ + if timestamp is None: + timestamp = datetime.utcnow().isoformat() + "Z" + + return [ + _triple(question_uri, RDF_TYPE, _iri(PROV_ACTIVITY)), + _triple(question_uri, RDF_TYPE, _iri(TG_QUESTION)), + _triple(question_uri, RDFS_LABEL, _literal("DocumentRAG Question")), + _triple(question_uri, PROV_STARTED_AT_TIME, _literal(timestamp)), + _triple(question_uri, TG_QUERY, _literal(query)), + ] + + +def docrag_exploration_triples( + exploration_uri: str, + question_uri: str, + chunk_count: int, + chunk_ids: Optional[List[str]] = None, +) -> List[Triple]: + """ + Build triples for a document RAG exploration entity (chunks retrieved). + + Creates: + - Entity declaration with tg:Exploration type + - wasGeneratedBy link to question + - Chunk count and optional chunk references + + Args: + exploration_uri: URI of the exploration entity + question_uri: URI of the parent question + chunk_count: Number of chunks retrieved + chunk_ids: Optional list of chunk URIs/IDs + + Returns: + List of Triple objects + """ + triples = [ + _triple(exploration_uri, RDF_TYPE, _iri(PROV_ENTITY)), + _triple(exploration_uri, RDF_TYPE, _iri(TG_EXPLORATION)), + _triple(exploration_uri, RDFS_LABEL, _literal("Exploration")), + _triple(exploration_uri, PROV_WAS_GENERATED_BY, _iri(question_uri)), + _triple(exploration_uri, TG_CHUNK_COUNT, _literal(chunk_count)), + ] + + # Add references to selected chunks + if chunk_ids: + for chunk_id in chunk_ids: + triples.append(_triple(exploration_uri, TG_SELECTED_CHUNK, _iri(chunk_id))) + + return triples + + +def docrag_synthesis_triples( + synthesis_uri: str, + exploration_uri: str, + answer_text: str = "", + document_id: Optional[str] = None, +) -> List[Triple]: + """ + Build triples for a document RAG synthesis entity (final answer). + + Creates: + - Entity declaration with tg:Synthesis type + - wasDerivedFrom link to exploration (skips focus step) + - Either document reference or inline content + + Args: + synthesis_uri: URI of the synthesis entity + exploration_uri: URI of the parent exploration entity + answer_text: The synthesized answer text (used if no document_id) + document_id: Optional librarian document ID (preferred over inline content) + + Returns: + List of Triple objects + """ + triples = [ + _triple(synthesis_uri, RDF_TYPE, _iri(PROV_ENTITY)), + _triple(synthesis_uri, RDF_TYPE, _iri(TG_SYNTHESIS)), + _triple(synthesis_uri, RDFS_LABEL, _literal("Synthesis")), + _triple(synthesis_uri, PROV_WAS_DERIVED_FROM, _iri(exploration_uri)), + ] + + if document_id: + triples.append(_triple(synthesis_uri, TG_DOCUMENT, _iri(document_id))) + elif answer_text: + triples.append(_triple(synthesis_uri, TG_CONTENT, _literal(answer_text))) + + return triples diff --git a/trustgraph-base/trustgraph/provenance/uris.py b/trustgraph-base/trustgraph/provenance/uris.py index a5b4d42d..b14abe76 100644 --- a/trustgraph-base/trustgraph/provenance/uris.py +++ b/trustgraph-base/trustgraph/provenance/uris.py @@ -184,3 +184,48 @@ def agent_final_uri(session_id: str) -> str: URN in format: urn:trustgraph:agent:{uuid}/final """ return f"urn:trustgraph:agent:{session_id}/final" + + +# Document RAG provenance URIs +# These URIs use the urn:trustgraph:docrag: namespace to distinguish +# document RAG provenance from graph RAG provenance + +def docrag_question_uri(session_id: str = None) -> str: + """ + Generate URI for a document RAG question activity. + + Args: + session_id: Optional UUID string. Auto-generates if not provided. + + Returns: + URN in format: urn:trustgraph:docrag:{uuid} + """ + if session_id is None: + session_id = str(uuid.uuid4()) + return f"urn:trustgraph:docrag:{session_id}" + + +def docrag_exploration_uri(session_id: str) -> str: + """ + Generate URI for a document RAG exploration entity (chunks retrieved). + + Args: + session_id: The session UUID. + + Returns: + URN in format: urn:trustgraph:docrag:{uuid}/exploration + """ + return f"urn:trustgraph:docrag:{session_id}/exploration" + + +def docrag_synthesis_uri(session_id: str) -> str: + """ + Generate URI for a document RAG synthesis entity (final answer). + + Args: + session_id: The session UUID. + + Returns: + URN in format: urn:trustgraph:docrag:{uuid}/synthesis + """ + return f"urn:trustgraph:docrag:{session_id}/synthesis" diff --git a/trustgraph-base/trustgraph/schema/services/retrieval.py b/trustgraph-base/trustgraph/schema/services/retrieval.py index dd31444e..5b09b11e 100644 --- a/trustgraph-base/trustgraph/schema/services/retrieval.py +++ b/trustgraph-base/trustgraph/schema/services/retrieval.py @@ -42,5 +42,7 @@ class DocumentRagQuery: @dataclass class DocumentRagResponse: error: Error | None = None - response: str = "" + response: str | None = "" end_of_stream: bool = False + 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) diff --git a/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py b/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py index 5e77f733..7730ceac 100644 --- a/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py +++ b/trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py @@ -1,6 +1,20 @@ import asyncio import logging +import uuid +from datetime import datetime + +# Provenance imports +from trustgraph.provenance import ( + docrag_question_uri, + docrag_exploration_uri, + docrag_synthesis_uri, + docrag_question_triples, + docrag_exploration_triples, + docrag_synthesis_triples, + set_graph, + GRAPH_RETRIEVAL, +) # Module logger logger = logging.getLogger(__name__) @@ -33,7 +47,14 @@ class Query: return qembeds[0] if qembeds else [] async def get_docs(self, query): + """ + Get documents (chunks) matching the query. + Returns: + tuple: (docs, chunk_ids) where: + - docs: list of document content strings + - chunk_ids: list of chunk IDs that were successfully fetched + """ vectors = await self.get_vector(query) if self.verbose: @@ -50,11 +71,13 @@ class Query: # Fetch chunk content from Garage docs = [] + chunk_ids = [] for match in chunk_matches: if match.chunk_id: try: content = await self.rag.fetch_chunk(match.chunk_id, self.user) docs.append(content) + chunk_ids.append(match.chunk_id) except Exception as e: logger.warning(f"Failed to fetch chunk {match.chunk_id}: {e}") @@ -63,7 +86,7 @@ class Query: for doc in docs: logger.debug(f" {doc[:100]}...") - return docs + return docs, chunk_ids class DocumentRag: @@ -86,17 +109,56 @@ class DocumentRag: async def query( self, query, user="trustgraph", collection="default", doc_limit=20, streaming=False, chunk_callback=None, + explain_callback=None, ): + """ + Execute a Document RAG query with optional explainability tracking. + Args: + query: The query string + user: User identifier + collection: Collection identifier + doc_limit: Max chunks to retrieve + streaming: Enable streaming LLM response + chunk_callback: async def callback(chunk, end_of_stream) for streaming + explain_callback: async def callback(triples, explain_id) for explainability + + Returns: + str: The synthesized answer text + """ if self.verbose: logger.debug("Constructing prompt...") + # Generate explainability URIs upfront + session_id = str(uuid.uuid4()) + q_uri = docrag_question_uri(session_id) + exp_uri = docrag_exploration_uri(session_id) + syn_uri = docrag_synthesis_uri(session_id) + + timestamp = datetime.utcnow().isoformat() + "Z" + + # Emit question explainability immediately + if explain_callback: + q_triples = set_graph( + docrag_question_triples(q_uri, query, timestamp), + GRAPH_RETRIEVAL + ) + await explain_callback(q_triples, q_uri) + q = Query( rag=self, user=user, collection=collection, verbose=self.verbose, doc_limit=doc_limit ) - docs = await q.get_docs(query) + docs, chunk_ids = await q.get_docs(query) + + # Emit exploration explainability after chunks retrieved + if explain_callback: + exp_triples = set_graph( + docrag_exploration_triples(exp_uri, q_uri, len(chunk_ids), chunk_ids), + GRAPH_RETRIEVAL + ) + await explain_callback(exp_triples, exp_uri) if self.verbose: logger.debug("Invoking LLM...") @@ -104,12 +166,21 @@ class DocumentRag: logger.debug(f"Query: {query}") if streaming and chunk_callback: + # Accumulate chunks for answer storage while forwarding to callback + accumulated_chunks = [] + + async def accumulating_callback(chunk, end_of_stream): + accumulated_chunks.append(chunk) + await chunk_callback(chunk, end_of_stream) + resp = await self.prompt_client.document_prompt( query=query, documents=docs, streaming=True, - chunk_callback=chunk_callback + chunk_callback=accumulating_callback ) + # Combine all chunks into full response + resp = "".join(accumulated_chunks) else: resp = await self.prompt_client.document_prompt( query=query, @@ -119,5 +190,17 @@ class DocumentRag: if self.verbose: logger.debug("Query processing complete") + # Emit synthesis explainability after answer generated + if explain_callback: + answer_text = resp if resp else "" + syn_triples = set_graph( + docrag_synthesis_triples(syn_uri, exp_uri, answer_text), + GRAPH_RETRIEVAL + ) + await explain_callback(syn_triples, syn_uri) + + if self.verbose: + logger.debug(f"Emitted explain for session {session_id}") + return resp diff --git a/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py b/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py index 3bc7113a..c1d96260 100755 --- a/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py +++ b/trustgraph-flow/trustgraph/retrieval/document_rag/rag.py @@ -11,6 +11,8 @@ import logging from ... schema import DocumentRagQuery, DocumentRagResponse, Error from ... schema import LibrarianRequest, LibrarianResponse from ... schema import librarian_request_queue, librarian_response_queue +from ... schema import Triples, Metadata +from ... provenance import GRAPH_RETRIEVAL from . document_rag import DocumentRag from ... base import FlowProcessor, ConsumerSpec, ProducerSpec from ... base import PromptClientSpec, EmbeddingsClientSpec @@ -78,6 +80,13 @@ class Processor(FlowProcessor): ) ) + self.register_specification( + ProducerSpec( + name = "explainability", + schema = Triples, + ) + ) + # Librarian client for fetching chunk content from Garage librarian_request_q = params.get( "librarian_request_queue", default_librarian_request_queue @@ -194,6 +203,29 @@ class Processor(FlowProcessor): else: doc_limit = self.doc_limit + # Real-time explainability callback - emits triples and IDs as they're generated + # Triples are stored in the user's collection with a named graph (urn:graph:retrieval) + async def send_explainability(triples, explain_id): + # Send triples to explainability queue - stores in same collection with named graph + await flow("explainability").send(Triples( + metadata=Metadata( + id=explain_id, + user=v.user, + collection=v.collection, # Store in user's collection + ), + triples=triples, + )) + + # Send explain ID and graph to response queue + await flow("response").send( + DocumentRagResponse( + response=None, + explain_id=explain_id, + explain_graph=GRAPH_RETRIEVAL, + ), + properties={"id": id} + ) + # Check if streaming is requested if v.streaming: # Define async callback for streaming chunks @@ -217,6 +249,7 @@ class Processor(FlowProcessor): doc_limit=doc_limit, streaming=True, chunk_callback=send_chunk, + explain_callback=send_explainability, ) else: # Non-streaming path (existing behavior) @@ -224,7 +257,8 @@ class Processor(FlowProcessor): v.query, user=v.user, collection=v.collection, - doc_limit=doc_limit + doc_limit=doc_limit, + explain_callback=send_explainability, ) await flow("response").send(