From 6895951d3f96ea0e3240ef6a92e27b0f06be88a3 Mon Sep 17 00:00:00 2001 From: Cyber MacGeddon Date: Wed, 11 Mar 2026 14:42:36 +0000 Subject: [PATCH] Add provenance recording to React agent loop Enables agent sessions to be traced and debugged using the same explainability infrastructure as GraphRAG. Agent traces record: - Session start with query and timestamp - Each iteration's thought, action, arguments, and observation - Final answer with derivation chain Changes: - Add session_id and collection fields to AgentRequest schema - Add agent predicates (TG_THOUGHT, TG_ACTION, etc.) to namespaces - Create agent provenance triple generators in provenance/agent.py - Register explainability producer in agent service - Emit provenance triples during agent execution - Update CLI tools to detect and render agent traces alongside GraphRAG --- .../trustgraph/messaging/translators/agent.py | 8 +- .../trustgraph/provenance/__init__.py | 25 +++ .../trustgraph/provenance/agent.py | 136 ++++++++++++ .../trustgraph/provenance/namespaces.py | 10 + trustgraph-base/trustgraph/provenance/uris.py | 46 ++++ .../trustgraph/schema/services/agent.py | 4 +- .../trustgraph/cli/list_explain_traces.py | 30 ++- .../trustgraph/cli/show_explain_trace.py | 210 +++++++++++++++++- .../trustgraph/agent/react/service.py | 104 ++++++++- 9 files changed, 552 insertions(+), 21 deletions(-) create mode 100644 trustgraph-base/trustgraph/provenance/agent.py diff --git a/trustgraph-base/trustgraph/messaging/translators/agent.py b/trustgraph-base/trustgraph/messaging/translators/agent.py index 4289df0a..4da0aec6 100644 --- a/trustgraph-base/trustgraph/messaging/translators/agent.py +++ b/trustgraph-base/trustgraph/messaging/translators/agent.py @@ -13,7 +13,9 @@ class AgentRequestTranslator(MessageTranslator): group=data.get("group", None), history=data.get("history", []), user=data.get("user", "trustgraph"), - streaming=data.get("streaming", False) + collection=data.get("collection", "default"), + streaming=data.get("streaming", False), + session_id=data.get("session_id", ""), ) def from_pulsar(self, obj: AgentRequest) -> Dict[str, Any]: @@ -23,7 +25,9 @@ class AgentRequestTranslator(MessageTranslator): "group": obj.group, "history": obj.history, "user": obj.user, - "streaming": getattr(obj, "streaming", False) + "collection": getattr(obj, "collection", "default"), + "streaming": getattr(obj, "streaming", False), + "session_id": getattr(obj, "session_id", ""), } diff --git a/trustgraph-base/trustgraph/provenance/__init__.py b/trustgraph-base/trustgraph/provenance/__init__.py index 5aa6d447..10a7f445 100644 --- a/trustgraph-base/trustgraph/provenance/__init__.py +++ b/trustgraph-base/trustgraph/provenance/__init__.py @@ -45,6 +45,10 @@ from . uris import ( exploration_uri, focus_uri, synthesis_uri, + # Agent provenance URIs + agent_session_uri, + agent_iteration_uri, + agent_final_uri, ) # Namespace constants @@ -65,6 +69,9 @@ from . namespaces import ( TG_SOURCE_TEXT, TG_SOURCE_CHAR_OFFSET, TG_SOURCE_CHAR_LENGTH, # Query-time provenance predicates TG_QUERY, TG_EDGE_COUNT, TG_SELECTED_EDGE, TG_REASONING, TG_CONTENT, + # Agent provenance predicates + TG_THOUGHT, TG_ACTION, TG_ARGUMENTS, TG_OBSERVATION, TG_ANSWER, + TG_AGENT_SESSION, TG_AGENT_ITERATION, TG_AGENT_FINAL, # Named graphs GRAPH_DEFAULT, GRAPH_SOURCE, GRAPH_RETRIEVAL, ) @@ -83,6 +90,13 @@ from . triples import ( set_graph, ) +# Agent provenance triple builders +from . agent import ( + agent_session_triples, + agent_iteration_triples, + agent_final_triples, +) + # Vocabulary bootstrap from . vocabulary import ( get_vocabulary_triples, @@ -107,6 +121,10 @@ __all__ = [ "exploration_uri", "focus_uri", "synthesis_uri", + # Agent provenance URIs + "agent_session_uri", + "agent_iteration_uri", + "agent_final_uri", # Namespaces "PROV", "PROV_ENTITY", "PROV_ACTIVITY", "PROV_AGENT", "PROV_WAS_DERIVED_FROM", "PROV_WAS_GENERATED_BY", @@ -120,6 +138,9 @@ __all__ = [ "TG_SOURCE_TEXT", "TG_SOURCE_CHAR_OFFSET", "TG_SOURCE_CHAR_LENGTH", # Query-time provenance predicates "TG_QUERY", "TG_EDGE_COUNT", "TG_SELECTED_EDGE", "TG_REASONING", "TG_CONTENT", + # Agent provenance predicates + "TG_THOUGHT", "TG_ACTION", "TG_ARGUMENTS", "TG_OBSERVATION", "TG_ANSWER", + "TG_AGENT_SESSION", "TG_AGENT_ITERATION", "TG_AGENT_FINAL", # Named graphs "GRAPH_DEFAULT", "GRAPH_SOURCE", "GRAPH_RETRIEVAL", # Triple builders @@ -131,6 +152,10 @@ __all__ = [ "exploration_triples", "focus_triples", "synthesis_triples", + # Agent provenance triple builders + "agent_session_triples", + "agent_iteration_triples", + "agent_final_triples", # Utility "set_graph", # Vocabulary diff --git a/trustgraph-base/trustgraph/provenance/agent.py b/trustgraph-base/trustgraph/provenance/agent.py new file mode 100644 index 00000000..bda979fa --- /dev/null +++ b/trustgraph-base/trustgraph/provenance/agent.py @@ -0,0 +1,136 @@ +""" +Helper functions to build PROV-O triples for agent provenance. + +Agent provenance tracks the reasoning trace of ReAct agent sessions: +- AgentSession: The root entity with query and timestamp +- AgentIteration: Each think/act/observe cycle +- AgentFinal: The final answer +""" + +import json +from datetime import datetime +from typing import List, Optional, Dict, Any + +from .. schema import Triple, Term, IRI, LITERAL + +from . namespaces import ( + RDF_TYPE, RDFS_LABEL, + PROV_ENTITY, PROV_WAS_DERIVED_FROM, PROV_STARTED_AT_TIME, + TG_QUERY, TG_THOUGHT, TG_ACTION, TG_ARGUMENTS, TG_OBSERVATION, TG_ANSWER, + TG_AGENT_SESSION, TG_AGENT_ITERATION, TG_AGENT_FINAL, +) + + +def _iri(uri: str) -> Term: + """Create an IRI term.""" + return Term(type=IRI, iri=uri) + + +def _literal(value) -> Term: + """Create a literal term.""" + return Term(type=LITERAL, value=str(value)) + + +def _triple(s: str, p: str, o_term: Term) -> Triple: + """Create a triple with IRI subject and predicate.""" + return Triple(s=_iri(s), p=_iri(p), o=o_term) + + +def agent_session_triples( + session_uri: str, + query: str, + timestamp: Optional[str] = None, +) -> List[Triple]: + """ + Build triples for an agent session start. + + Creates: + - Entity declaration for the session + - Query text and timestamp + + Args: + session_uri: URI of the session (from agent_session_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(session_uri, RDF_TYPE, _iri(TG_AGENT_SESSION)), + _triple(session_uri, RDFS_LABEL, _literal("Agent Session")), + _triple(session_uri, PROV_STARTED_AT_TIME, _literal(timestamp)), + _triple(session_uri, TG_QUERY, _literal(query)), + ] + + +def agent_iteration_triples( + iteration_uri: str, + parent_uri: str, + thought: str, + action: str, + arguments: Dict[str, Any], + observation: str, +) -> List[Triple]: + """ + Build triples for one agent iteration (think/act/observe cycle). + + Creates: + - Entity declaration for the iteration + - wasDerivedFrom link to parent (previous iteration or session) + - Thought, action, arguments, and observation data + + Args: + iteration_uri: URI of this iteration (from agent_iteration_uri) + parent_uri: URI of the parent (previous iteration or session) + thought: The agent's reasoning/thought + action: The tool/action name + arguments: Arguments passed to the tool (will be JSON-encoded) + observation: The result/observation from the tool + + Returns: + List of Triple objects + """ + triples = [ + _triple(iteration_uri, RDF_TYPE, _iri(TG_AGENT_ITERATION)), + _triple(iteration_uri, RDFS_LABEL, _literal(f"Iteration: {action}")), + _triple(iteration_uri, PROV_WAS_DERIVED_FROM, _iri(parent_uri)), + _triple(iteration_uri, TG_THOUGHT, _literal(thought)), + _triple(iteration_uri, TG_ACTION, _literal(action)), + _triple(iteration_uri, TG_ARGUMENTS, _literal(json.dumps(arguments))), + _triple(iteration_uri, TG_OBSERVATION, _literal(observation)), + ] + + return triples + + +def agent_final_triples( + final_uri: str, + parent_uri: str, + answer: str, +) -> List[Triple]: + """ + Build triples for an agent final answer. + + Creates: + - Entity declaration for the final answer + - wasDerivedFrom link to parent (last iteration or session) + - The answer text + + Args: + final_uri: URI of the final answer (from agent_final_uri) + parent_uri: URI of the parent (last iteration or session if no iterations) + answer: The final answer text + + Returns: + List of Triple objects + """ + return [ + _triple(final_uri, RDF_TYPE, _iri(TG_AGENT_FINAL)), + _triple(final_uri, RDFS_LABEL, _literal("Final Answer")), + _triple(final_uri, PROV_WAS_DERIVED_FROM, _iri(parent_uri)), + _triple(final_uri, TG_ANSWER, _literal(answer)), + ] diff --git a/trustgraph-base/trustgraph/provenance/namespaces.py b/trustgraph-base/trustgraph/provenance/namespaces.py index aaca70db..51947617 100644 --- a/trustgraph-base/trustgraph/provenance/namespaces.py +++ b/trustgraph-base/trustgraph/provenance/namespaces.py @@ -68,6 +68,16 @@ TG_REASONING = TG + "reasoning" TG_CONTENT = TG + "content" TG_DOCUMENT = TG + "document" # Reference to document in librarian +# Agent provenance predicates +TG_THOUGHT = TG + "thought" +TG_ACTION = TG + "action" +TG_ARGUMENTS = TG + "arguments" +TG_OBSERVATION = TG + "observation" +TG_ANSWER = TG + "answer" +TG_AGENT_SESSION = TG + "AgentSession" +TG_AGENT_ITERATION = TG + "AgentIteration" +TG_AGENT_FINAL = TG + "AgentFinal" + # Named graph URIs for RDF datasets # These separate different types of data while keeping them in the same collection GRAPH_DEFAULT = "" # Core knowledge facts (triples extracted from documents) diff --git a/trustgraph-base/trustgraph/provenance/uris.py b/trustgraph-base/trustgraph/provenance/uris.py index 6c7d0a7f..a5b4d42d 100644 --- a/trustgraph-base/trustgraph/provenance/uris.py +++ b/trustgraph-base/trustgraph/provenance/uris.py @@ -138,3 +138,49 @@ def edge_selection_uri(session_id: str, edge_index: int) -> str: URN in format: urn:trustgraph:prov:edge:{uuid}:{index} """ return f"urn:trustgraph:prov:edge:{session_id}:{edge_index}" + + +# Agent provenance URIs +# These URIs use the urn:trustgraph:agent: namespace to distinguish agent +# provenance from GraphRAG question provenance + +def agent_session_uri(session_id: str = None) -> str: + """ + Generate URI for an agent session. + + Args: + session_id: Optional UUID string. Auto-generates if not provided. + + Returns: + URN in format: urn:trustgraph:agent:{uuid} + """ + if session_id is None: + session_id = str(uuid.uuid4()) + return f"urn:trustgraph:agent:{session_id}" + + +def agent_iteration_uri(session_id: str, iteration_num: int) -> str: + """ + Generate URI for an agent iteration. + + Args: + session_id: The session UUID. + iteration_num: 1-based iteration number. + + Returns: + URN in format: urn:trustgraph:agent:{uuid}/i{num} + """ + return f"urn:trustgraph:agent:{session_id}/i{iteration_num}" + + +def agent_final_uri(session_id: str) -> str: + """ + Generate URI for an agent final answer. + + Args: + session_id: The session UUID. + + Returns: + URN in format: urn:trustgraph:agent:{uuid}/final + """ + return f"urn:trustgraph:agent:{session_id}/final" diff --git a/trustgraph-base/trustgraph/schema/services/agent.py b/trustgraph-base/trustgraph/schema/services/agent.py index 9f883ff2..35a387fc 100644 --- a/trustgraph-base/trustgraph/schema/services/agent.py +++ b/trustgraph-base/trustgraph/schema/services/agent.py @@ -23,7 +23,9 @@ class AgentRequest: group: list[str] | None = None history: list[AgentStep] = field(default_factory=list) user: str = "" # User context for multi-tenancy - streaming: bool = False # NEW: Enable streaming response delivery (default false) + collection: str = "default" # Collection for provenance traces + streaming: bool = False # Enable streaming response delivery (default false) + session_id: str = "" # For provenance tracking across iterations @dataclass class AgentResponse: diff --git a/trustgraph-cli/trustgraph/cli/list_explain_traces.py b/trustgraph-cli/trustgraph/cli/list_explain_traces.py index 2690159c..fd2e1419 100644 --- a/trustgraph-cli/trustgraph/cli/list_explain_traces.py +++ b/trustgraph-cli/trustgraph/cli/list_explain_traces.py @@ -1,8 +1,8 @@ """ -List all GraphRAG sessions (questions) in a collection. +List all explainability sessions (GraphRAG and Agent) in a collection. Queries for all questions stored in the retrieval graph and displays them -with their session IDs and timestamps. +with their session IDs, type (GraphRAG or Agent), and timestamps. Examples: tg-list-explain-traces -U trustgraph -C default @@ -24,8 +24,10 @@ default_collection = 'default' # Predicates TG = "https://trustgraph.ai/ns/" TG_QUERY = TG + "query" +TG_AGENT_SESSION = TG + "AgentSession" PROV = "http://www.w3.org/ns/prov#" PROV_STARTED_AT_TIME = PROV + "startedAtTime" +RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" # Retrieval graph RETRIEVAL_GRAPH = "urn:graph:retrieval" @@ -117,8 +119,20 @@ def get_timestamp(socket, flow_id, user, collection, question_id): return "" +def get_session_type(socket, flow_id, user, collection, session_id): + """Get the type of session (Agent or GraphRAG).""" + triples = query_triples( + socket, flow_id, user, collection, + s=session_id, p=RDF_TYPE, g=RETRIEVAL_GRAPH + ) + for s, p, o in triples: + if o == TG_AGENT_SESSION: + return "Agent" + return "GraphRAG" + + def list_sessions(socket, flow_id, user, collection, limit): - """List all GraphRAG sessions by finding questions.""" + """List all explainability sessions (GraphRAG and Agent) by finding questions.""" # Query for all triples with predicate = tg:query triples = query_triples( socket, flow_id, user, collection, @@ -129,9 +143,12 @@ def list_sessions(socket, flow_id, user, collection, limit): for question_id, _, query_text in triples: # Get timestamp if available timestamp = get_timestamp(socket, flow_id, user, collection, question_id) + # Get session type (Agent or GraphRAG) + session_type = get_session_type(socket, flow_id, user, collection, question_id) sessions.append({ "id": question_id, + "type": session_type, "question": query_text, "time": timestamp, }) @@ -154,18 +171,19 @@ def truncate_text(text, max_len=60): def print_table(sessions): """Print sessions as a table.""" if not sessions: - print("No GraphRAG sessions found.") + print("No explainability sessions found.") return rows = [] for session in sessions: rows.append([ session["id"], - truncate_text(session["question"], 50), + session.get("type", "Unknown"), + truncate_text(session["question"], 45), session.get("time", "") ]) - headers = ["Session ID", "Question", "Time"] + headers = ["Session ID", "Type", "Question", "Time"] print(tabulate(rows, headers=headers, tablefmt="simple")) diff --git a/trustgraph-cli/trustgraph/cli/show_explain_trace.py b/trustgraph-cli/trustgraph/cli/show_explain_trace.py index a678a180..13ef5394 100644 --- a/trustgraph-cli/trustgraph/cli/show_explain_trace.py +++ b/trustgraph-cli/trustgraph/cli/show_explain_trace.py @@ -1,11 +1,15 @@ """ -Show full explainability trace for a GraphRAG session. +Show full explainability trace for a GraphRAG or Agent session. -Given a question/session URI, displays the complete cascade: -Question -> Exploration -> Focus (edge selection) -> Synthesis (answer). +Given a question/session URI, displays the complete trace: +- GraphRAG: Question -> Exploration -> Focus (edge selection) -> Synthesis (answer) +- Agent: Session -> Iteration(s) (thought/action/observation) -> Final Answer + +The tool auto-detects the trace type based on rdf:type. Examples: tg-show-explain-trace -U trustgraph -C default "urn:trustgraph:question:abc123" + tg-show-explain-trace -U trustgraph -C default "urn:trustgraph:agent:abc123" tg-show-explain-trace --max-answer 1000 "urn:trustgraph:question:abc123" tg-show-explain-trace --show-provenance "urn:trustgraph:question:abc123" """ @@ -31,10 +35,20 @@ TG_REASONING = TG + "reasoning" TG_CONTENT = TG + "content" TG_DOCUMENT = TG + "document" TG_REIFIES = TG + "reifies" +# Agent predicates +TG_THOUGHT = TG + "thought" +TG_ACTION = TG + "action" +TG_ARGUMENTS = TG + "arguments" +TG_OBSERVATION = TG + "observation" +TG_ANSWER = TG + "answer" +TG_AGENT_SESSION = TG + "AgentSession" +TG_AGENT_ITERATION = TG + "AgentIteration" +TG_AGENT_FINAL = TG + "AgentFinal" PROV = "http://www.w3.org/ns/prov#" PROV_STARTED_AT_TIME = PROV + "startedAtTime" PROV_WAS_DERIVED_FROM = PROV + "wasDerivedFrom" PROV_WAS_GENERATED_BY = PROV + "wasGeneratedBy" +RDF_TYPE = "http://www.w3.org/1999/02/22-rdf-syntax-ns#type" RDFS_LABEL = "http://www.w3.org/2000/01/rdf-schema#label" # Graphs @@ -280,6 +294,153 @@ def format_edge(edge, label_cache=None, socket=None, flow_id=None, user=None, co return f"({s_label}, {p_label}, {o_label})" +def detect_trace_type(socket, flow_id, user, collection, entity_id): + """ + Detect whether an entity is an agent session or GraphRAG question. + + Returns: + "agent" if entity has rdf:type = tg:AgentSession + "graphrag" otherwise (default) + """ + triples = query_triples( + socket, flow_id, user, collection, + s=entity_id, p=RDF_TYPE, g=RETRIEVAL_GRAPH + ) + for s, p, o in triples: + if o == TG_AGENT_SESSION: + return "agent" + return "graphrag" + + +def build_agent_trace(socket, flow_id, user, collection, session_id, api=None, max_answer=500): + """Build the full explainability trace for an agent session.""" + trace = { + "session_id": session_id, + "type": "agent", + "question": None, + "time": None, + "iterations": [], + "final_answer": None, + } + + # Get session metadata + props = get_node_properties(socket, flow_id, user, collection, session_id) + trace["question"] = props.get(TG_QUERY, [None])[0] + trace["time"] = props.get(PROV_STARTED_AT_TIME, [None])[0] + + # Find all entities derived from this session (iterations and final) + # Start by looking for entities where prov:wasDerivedFrom = session_id + current_uri = session_id + iteration_num = 1 + + while True: + # Find entities derived from current + derived_ids = find_by_predicate_object( + socket, flow_id, user, collection, + PROV_WAS_DERIVED_FROM, current_uri + ) + + if not derived_ids: + break + + derived_id = derived_ids[0] + derived_props = get_node_properties(socket, flow_id, user, collection, derived_id) + + # Check type + types = derived_props.get(RDF_TYPE, []) + + if TG_AGENT_ITERATION in types: + iteration = { + "id": derived_id, + "iteration_num": iteration_num, + "thought": derived_props.get(TG_THOUGHT, [None])[0], + "action": derived_props.get(TG_ACTION, [None])[0], + "arguments": derived_props.get(TG_ARGUMENTS, [None])[0], + "observation": derived_props.get(TG_OBSERVATION, [None])[0], + } + trace["iterations"].append(iteration) + current_uri = derived_id + iteration_num += 1 + + elif TG_AGENT_FINAL in types: + answer = derived_props.get(TG_ANSWER, [None])[0] + if answer and len(answer) > max_answer: + answer = answer[:max_answer] + "... [truncated]" + trace["final_answer"] = { + "id": derived_id, + "answer": answer, + } + break + + else: + # Unknown type, stop traversal + break + + return trace + + +def print_agent_text(trace): + """Print agent trace in text format.""" + print(f"=== Agent Session: {trace['session_id']} ===") + print() + + if trace["question"]: + print(f"Question: {trace['question']}") + if trace["time"]: + print(f"Time: {trace['time']}") + print() + + # Iterations + print("--- Iterations ---") + iterations = trace.get("iterations", []) + if iterations: + for iteration in iterations: + print(f"Iteration {iteration['iteration_num']}:") + print(f" Thought: {iteration.get('thought', 'N/A')}") + print(f" Action: {iteration.get('action', 'N/A')}") + + args = iteration.get('arguments') + if args: + # Try to pretty-print JSON arguments + try: + import json + args_obj = json.loads(args) + args_str = json.dumps(args_obj, indent=4) + # Indent each line + args_lines = args_str.split('\n') + print(f" Arguments:") + for line in args_lines: + print(f" {line}") + except: + print(f" Arguments: {args}") + else: + print(f" Arguments: N/A") + + obs = iteration.get('observation', 'N/A') + if obs and len(obs) > 200: + obs = obs[:200] + "... [truncated]" + print(f" Observation: {obs}") + print() + else: + print("No iterations recorded") + print() + + # Final answer + print("--- Final Answer ---") + final = trace.get("final_answer") + if final and final.get("answer"): + print("Answer:") + for line in final["answer"].split("\n"): + print(f" {line}") + else: + print("No final answer recorded") + + +def print_agent_json(trace): + """Print agent trace as JSON.""" + print(json.dumps(trace, indent=2)) + + def build_trace(socket, flow_id, user, collection, question_id, api=None, show_provenance=False, max_answer=500): """Build the full explainability trace for a question.""" label_cache = {} @@ -530,21 +691,48 @@ def main(): socket = api.socket() try: - trace = build_trace( + # Detect trace type (agent vs graphrag) + trace_type = detect_trace_type( socket=socket, flow_id=args.flow_id, user=args.user, collection=args.collection, - question_id=args.question_id, - api=api, - show_provenance=args.show_provenance, - max_answer=args.max_answer, + entity_id=args.question_id, ) - if args.format == 'json': - print_json(trace) + if trace_type == "agent": + # Build and print agent trace + trace = build_agent_trace( + socket=socket, + flow_id=args.flow_id, + user=args.user, + collection=args.collection, + session_id=args.question_id, + api=api, + max_answer=args.max_answer, + ) + + if args.format == 'json': + print_agent_json(trace) + else: + print_agent_text(trace) else: - print_text(trace, show_provenance=args.show_provenance) + # Build and print GraphRAG trace (existing behavior) + trace = build_trace( + socket=socket, + flow_id=args.flow_id, + user=args.user, + collection=args.collection, + question_id=args.question_id, + api=api, + show_provenance=args.show_provenance, + max_answer=args.max_answer, + ) + + if args.format == 'json': + print_json(trace) + else: + print_text(trace, show_provenance=args.show_provenance) finally: socket.close() diff --git a/trustgraph-flow/trustgraph/agent/react/service.py b/trustgraph-flow/trustgraph/agent/react/service.py index 1c96adef..2ae93b05 100755 --- a/trustgraph-flow/trustgraph/agent/react/service.py +++ b/trustgraph-flow/trustgraph/agent/react/service.py @@ -7,6 +7,8 @@ import re import sys import functools import logging +import uuid +from datetime import datetime # Module logger logger = logging.getLogger(__name__) @@ -14,8 +16,22 @@ logger = logging.getLogger(__name__) from ... base import AgentService, TextCompletionClientSpec, PromptClientSpec from ... base import GraphRagClientSpec, ToolClientSpec, StructuredQueryClientSpec from ... base import RowEmbeddingsQueryClientSpec, EmbeddingsClientSpec +from ... base import ProducerSpec from ... schema import AgentRequest, AgentResponse, AgentStep, Error +from ... schema import Triples, Metadata + +# Provenance imports for agent explainability +from trustgraph.provenance import ( + agent_session_uri, + agent_iteration_uri, + agent_final_uri, + agent_session_triples, + agent_iteration_triples, + agent_final_triples, + set_graph, + GRAPH_RETRIEVAL, +) from . tools import KnowledgeQueryImpl, TextCompletionImpl, McpToolImpl, PromptImpl, StructuredQueryImpl, RowEmbeddingsQueryImpl, ToolServiceImpl from . agent_manager import AgentManager @@ -105,6 +121,14 @@ class Processor(AgentService): ) ) + # Explainability producer for agent provenance triples + self.register_specification( + ProducerSpec( + name = "explainability", + schema = Triples, + ) + ) + async def on_tools_config(self, config, version): logger.info(f"Loading configuration version {version}") @@ -285,6 +309,10 @@ class Processor(AgentService): # Check if streaming is enabled streaming = getattr(request, 'streaming', False) + # Generate or retrieve session ID for provenance tracking + session_id = getattr(request, 'session_id', '') or str(uuid.uuid4()) + collection = getattr(request, 'collection', 'default') + if request.history: history = [ Action( @@ -298,6 +326,27 @@ class Processor(AgentService): else: history = [] + # Calculate iteration number (1-based) + iteration_num = len(history) + 1 + session_uri = agent_session_uri(session_id) + + # On first iteration, emit session triples + if iteration_num == 1: + timestamp = datetime.utcnow().isoformat() + "Z" + triples = set_graph( + agent_session_triples(session_uri, request.question, timestamp), + GRAPH_RETRIEVAL + ) + await flow("explainability").send(Triples( + metadata=Metadata( + id=session_uri, + user=request.user, + collection=collection, + ), + triples=triples, + )) + logger.debug(f"Emitted session triples for {session_uri}") + logger.info(f"Question: {request.question}") if len(history) >= self.max_iterations: @@ -447,6 +496,28 @@ class Processor(AgentService): else: f = json.dumps(act.final) + # Emit final answer provenance triples + final_uri = agent_final_uri(session_id) + # Parent is last iteration, or session if no iterations + if iteration_num > 1: + parent_uri = agent_iteration_uri(session_id, iteration_num - 1) + else: + parent_uri = session_uri + + final_triples = set_graph( + agent_final_triples(final_uri, parent_uri, f), + GRAPH_RETRIEVAL + ) + await flow("explainability").send(Triples( + metadata=Metadata( + id=final_uri, + user=request.user, + collection=collection, + ), + triples=final_triples, + )) + logger.debug(f"Emitted final triples for {final_uri}") + if streaming: # Streaming format - send end-of-dialog marker # Answer chunks were already sent via answer() callback during parsing @@ -479,8 +550,37 @@ class Processor(AgentService): logger.debug("Send next...") + # Emit iteration provenance triples + iteration_uri = agent_iteration_uri(session_id, iteration_num) + # Parent is previous iteration, or session if this is first iteration + if iteration_num > 1: + parent_uri = agent_iteration_uri(session_id, iteration_num - 1) + else: + parent_uri = session_uri + + iter_triples = set_graph( + agent_iteration_triples( + iteration_uri, + parent_uri, + act.thought, + act.name, + act.arguments, + act.observation, + ), + GRAPH_RETRIEVAL + ) + await flow("explainability").send(Triples( + metadata=Metadata( + id=iteration_uri, + user=request.user, + collection=collection, + ), + triples=iter_triples, + )) + logger.debug(f"Emitted iteration triples for {iteration_uri}") + history.append(act) - + # Handle state transitions if tool execution was successful next_state = request.state if act.name in filtered_tools: @@ -501,7 +601,9 @@ class Processor(AgentService): for h in history ], user=request.user, + collection=collection, streaming=streaming, + session_id=session_id, # Pass session_id for provenance continuity ) await next(r)