Updated explainability taxonomy:

GraphRAG: tg:Question → tg:Exploration → tg:Focus → tg:Synthesis

Agent: tg:Question → tg:Analysis(s) → tg:Conclusion

All entities also have their PROV-O type (prov:Activity or prov:Entity).

Updated commit message:

Add provenance recording to React agent loop

Enables agent sessions to be traced and debugged using the same
explainability infrastructure as GraphRAG.

Entity types follow human reasoning patterns:
- tg:Question - the user's query (shared with GraphRAG)
- tg:Analysis - each think/act/observe cycle
- tg:Conclusion - the final answer

Also adds explicit TG types to GraphRAG entities:
- tg:Question, tg:Exploration, tg:Focus, tg:Synthesis

All types retain their PROV-O base types (prov:Activity, prov:Entity).

Changes:
- Add session_id and collection fields to AgentRequest schema
- Add explainability entity types to namespaces.py
- Create agent provenance triple generators
- Register explainability producer in agent service
- Emit provenance triples during agent execution
- Update CLI tools to detect and render both trace types
This commit is contained in:
Cyber MacGeddon 2026-03-11 15:05:32 +00:00
parent 6895951d3f
commit 208c2d0cd9
7 changed files with 203 additions and 160 deletions

View file

@ -6,39 +6,58 @@ Add provenance recording to the React agent loop so agent sessions can be traced
**Design Decisions:**
- Write to `urn:graph:retrieval` (generic explainability graph)
- Linear dependency chain for now (iteration N → wasDerivedFrom → iteration N-1)
- Linear dependency chain for now (analysis N → wasDerivedFrom → analysis N-1)
- Tools are opaque black boxes (record input/output only)
- DAG support deferred to future iteration
## Entity Types
Both GraphRAG and Agent use PROV-O as the base ontology with TrustGraph-specific subtypes:
### GraphRAG Types
| Entity | PROV-O Type | TG Type | Description |
|--------|-------------|---------|-------------|
| Question | `prov:Activity` | `tg:Question` | The user's query |
| Exploration | `prov:Entity` | `tg:Exploration` | Edges retrieved from knowledge graph |
| Focus | `prov:Entity` | `tg:Focus` | Selected edges with reasoning |
| Synthesis | `prov:Entity` | `tg:Synthesis` | Final answer |
### Agent Types
| Entity | PROV-O Type | TG Type | Description |
|--------|-------------|---------|-------------|
| Question | `prov:Activity` | `tg:Question` | The user's query (same as GraphRAG) |
| Analysis | `prov:Entity` | `tg:Analysis` | Each think/act/observe cycle |
| Conclusion | `prov:Entity` | `tg:Conclusion` | Final answer |
## Provenance Model
```
AgentSession (urn:trustgraph:agent:{uuid})
Question (urn:trustgraph:agent:{uuid})
│ tg:query = "User's question"
│ prov:startedAtTime = timestamp
│ rdf:type = tg:AgentSession
│ rdf:type = prov:Activity, tg:Question
↓ prov:wasGeneratedBy
↓ prov:wasDerivedFrom
Iteration1 (urn:trustgraph:agent:{uuid}/i1)
Analysis1 (urn:trustgraph:agent:{uuid}/i1)
│ tg:thought = "I need to query the knowledge base..."
│ tg:action = "knowledge-query"
│ tg:arguments = {"question": "..."}
│ tg:observation = "Result from tool..."
│ rdf:type = tg:AgentIteration
│ rdf:type = prov:Entity, tg:Analysis
↓ prov:wasDerivedFrom
Iteration2 (urn:trustgraph:agent:{uuid}/i2)
Analysis2 (urn:trustgraph:agent:{uuid}/i2)
│ ...
↓ prov:wasDerivedFrom
FinalAnswer (urn:trustgraph:agent:{uuid}/final)
Conclusion (urn:trustgraph:agent:{uuid}/final)
│ tg:answer = "The final response..."
│ rdf:type = tg:AgentFinal
│ rdf:type = prov:Entity, tg:Conclusion
```
## Changes Required
@ -85,23 +104,24 @@ self.register_specification(
### 3. Provenance Triple Generation
**Option A:** Add to existing `trustgraph-base/trustgraph/provenance/` module
**Option B:** Create agent-specific provenance helpers in the agent module
**File:** `trustgraph-base/trustgraph/provenance/agent.py`
Create helper functions (similar to GraphRAG's `question_triples`, `exploration_triples`, etc.):
```python
def agent_session_triples(session_uri, query, timestamp):
"""Generate triples for agent session start."""
"""Generate triples for agent Question."""
return [
Triple(s=session_uri, p=RDF_TYPE, o=TG_AGENT_SESSION),
Triple(s=session_uri, p=RDF_TYPE, o=PROV_ACTIVITY),
Triple(s=session_uri, p=RDF_TYPE, o=TG_QUESTION),
Triple(s=session_uri, p=TG_QUERY, o=query),
Triple(s=session_uri, p=PROV_STARTED_AT_TIME, o=timestamp),
]
def agent_iteration_triples(iteration_uri, parent_uri, thought, action, arguments, observation):
"""Generate triples for one agent iteration."""
"""Generate triples for one Analysis step."""
return [
Triple(s=iteration_uri, p=RDF_TYPE, o=TG_AGENT_ITERATION),
Triple(s=iteration_uri, p=RDF_TYPE, o=PROV_ENTITY),
Triple(s=iteration_uri, p=RDF_TYPE, o=TG_ANALYSIS),
Triple(s=iteration_uri, p=TG_THOUGHT, o=thought),
Triple(s=iteration_uri, p=TG_ACTION, o=action),
Triple(s=iteration_uri, p=TG_ARGUMENTS, o=json.dumps(arguments)),
@ -110,127 +130,68 @@ def agent_iteration_triples(iteration_uri, parent_uri, thought, action, argument
]
def agent_final_triples(final_uri, parent_uri, answer):
"""Generate triples for agent final answer."""
"""Generate triples for Conclusion."""
return [
Triple(s=final_uri, p=RDF_TYPE, o=TG_AGENT_FINAL),
Triple(s=final_uri, p=RDF_TYPE, o=PROV_ENTITY),
Triple(s=final_uri, p=RDF_TYPE, o=TG_CONCLUSION),
Triple(s=final_uri, p=TG_ANSWER, o=answer),
Triple(s=final_uri, p=PROV_WAS_DERIVED_FROM, o=parent_uri),
]
```
### 4. Integration in service.py
**In `agent_request()` method:**
```python
async def agent_request(self, request, respond, next, flow):
# Generate or retrieve session ID
if not request.session_id:
session_id = str(uuid.uuid4())
else:
session_id = request.session_id
session_uri = f"urn:trustgraph:agent:{session_id}"
iteration_num = len(history) + 1
iteration_uri = f"{session_uri}/i{iteration_num}"
# On first iteration, emit session triples
if iteration_num == 1:
triples = agent_session_triples(session_uri, request.question, timestamp)
await flow("explainability").send(Triples(
metadata=Metadata(user=request.user, collection=..., id=session_uri),
triples=triples,
))
# ... existing react() call ...
if isinstance(act, Final):
# Emit final answer triples
final_uri = f"{session_uri}/final"
parent_uri = f"{session_uri}/i{iteration_num - 1}" if iteration_num > 1 else session_uri
triples = agent_final_triples(final_uri, parent_uri, act.final)
await flow("explainability").send(...)
else:
# Emit iteration triples
parent_uri = f"{session_uri}/i{iteration_num - 1}" if iteration_num > 1 else session_uri
triples = agent_iteration_triples(iteration_uri, parent_uri, act.thought, act.name, act.arguments, act.observation)
await flow("explainability").send(...)
# Pass session_id to next iteration
r = AgentRequest(
...,
session_id=session_id,
)
```
### 5. Predicate Definitions
### 4. Type Definitions
**File:** `trustgraph-base/trustgraph/provenance/namespaces.py`
Add new predicates:
Add explainability entity types and agent predicates:
```python
# Explainability entity types (used by both GraphRAG and Agent)
TG_QUESTION = TG + "Question"
TG_EXPLORATION = TG + "Exploration"
TG_FOCUS = TG + "Focus"
TG_SYNTHESIS = TG + "Synthesis"
TG_ANALYSIS = TG + "Analysis"
TG_CONCLUSION = TG + "Conclusion"
# Agent predicates
TG_THOUGHT = TG + "thought"
TG_ACTION = TG + "action"
TG_ARGUMENTS = TG + "arguments"
TG_OBSERVATION = TG + "observation"
TG_AGENT_SESSION = TG + "AgentSession"
TG_AGENT_ITERATION = TG + "AgentIteration"
TG_AGENT_FINAL = TG + "AgentFinal"
# TG_QUERY and TG_ANSWER likely already exist
TG_ANSWER = TG + "answer"
```
## Files to Modify
## Files Modified
| File | Change |
|------|--------|
| `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 agent predicates |
| `trustgraph-base/trustgraph/provenance/__init__.py` | Export new predicates |
| `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-flow/trustgraph/agent/react/service.py` | Add explainability producer + recording logic |
| `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 |
## Files to Potentially Create
## Files Created
| File | Purpose |
|------|---------|
| `trustgraph-base/trustgraph/provenance/agent.py` | Agent-specific triple generators (optional, could inline in service.py) |
| `trustgraph-base/trustgraph/provenance/agent.py` | Agent-specific triple generators |
## CLI Updates Detail
## CLI Updates
**Detection:** Both GraphRAG and Agent Questions have `tg:Question` type. Distinguished by:
1. URI pattern: `urn:trustgraph:agent:` vs `urn:trustgraph:question:`
2. Derived entities: `tg:Analysis` (agent) vs `tg:Exploration` (GraphRAG)
**`list_explain_traces.py`:**
- Currently queries for `tg:query` predicate to find questions
- Agent sessions also use `tg:query`, so should work automatically
- May want to add a type indicator column (GraphRAG vs Agent)
- Shows Type column (Agent vs GraphRAG)
**`show_explain_trace.py`:**
- Currently expects: question → exploration → focus → synthesis chain
- Agent traces have: session → iteration(s) → final chain
- Detection: check `rdf:type` of the root entity
- `tg:AgentSession` → agent trace rendering
- Otherwise → GraphRAG trace rendering (existing)
- Agent rendering shows:
- Session info (question, time)
- Each iteration: thought, action, args, observation
- Final answer
## Design Decisions
1. **Collection handling:** Add `collection` field to AgentRequest. Agent receives a collection parameter for provenance traces. Tools can access other collections per their config, but decision traces stay in the invoked collection.
2. **CLI tool updates:** Update existing `tg-show-explain-trace` to detect and handle both GraphRAG and agent trace types.
## Implementation Order
0. **Tech Spec:** Write this plan to `docs/tech-specs/agent-explainability.md`
1. **Schema:** Add `session_id` and `collection` to AgentRequest + translator
2. **Predicates:** Add agent predicates to `namespaces.py`
3. **Service:** Add explainability producer to agent service
4. **Provenance:** Create/add agent triple generators
5. **Integration:** Wire up provenance recording in `agent_request()`
6. **CLI:** Update `show_explain_trace.py` to detect and render agent traces
7. **Test:** Run agent query and verify trace is recorded and viewable
- Auto-detects trace type
- Agent rendering shows: Question → Analysis step(s) → Conclusion
## Backwards Compatibility
@ -244,19 +205,16 @@ TG_AGENT_FINAL = TG + "AgentFinal"
# Run an agent query
tg-invoke-agent -q "What is the capital of France?"
# Check triples were written
tg-query-graph -p "https://trustgraph.ai/ns/query" -g "urn:graph:retrieval"
# List traces (should show agent sessions)
# List traces (should show agent sessions with Type column)
tg-list-explain-traces -U trustgraph -C default
# Show trace (may need CLI updates for agent entity types)
# Show agent trace
tg-show-explain-trace "urn:trustgraph:agent:xxx"
```
## Future Work (Not This PR)
- DAG dependencies (when iteration N uses results from multiple prior iterations)
- DAG dependencies (when analysis N uses results from multiple prior analyses)
- Tool-specific provenance linking (KnowledgeQuery → its GraphRAG trace)
- CLI tool enhancements to better display agent traces
- Document RAG explainability
- Streaming provenance emission (emit as we go, not batch at end)

View file

@ -69,9 +69,11 @@ 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,
# Explainability entity types
TG_QUESTION, TG_EXPLORATION, TG_FOCUS, TG_SYNTHESIS,
TG_ANALYSIS, TG_CONCLUSION,
# 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,
)
@ -138,9 +140,11 @@ __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",
# Explainability entity types
"TG_QUESTION", "TG_EXPLORATION", "TG_FOCUS", "TG_SYNTHESIS",
"TG_ANALYSIS", "TG_CONCLUSION",
# 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

View file

@ -2,9 +2,9 @@
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
- Question: The root activity with query and timestamp
- Analysis: Each think/act/observe cycle
- Conclusion: The final answer
"""
import json
@ -15,9 +15,9 @@ from .. schema import Triple, Term, IRI, LITERAL
from . namespaces import (
RDF_TYPE, RDFS_LABEL,
PROV_ENTITY, PROV_WAS_DERIVED_FROM, PROV_STARTED_AT_TIME,
PROV_ACTIVITY, 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,
TG_QUESTION, TG_ANALYSIS, TG_CONCLUSION,
)
@ -42,10 +42,10 @@ def agent_session_triples(
timestamp: Optional[str] = None,
) -> List[Triple]:
"""
Build triples for an agent session start.
Build triples for an agent session start (Question).
Creates:
- Entity declaration for the session
- Activity declaration with tg:Question type
- Query text and timestamp
Args:
@ -60,8 +60,9 @@ def agent_session_triples(
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, RDF_TYPE, _iri(PROV_ACTIVITY)),
_triple(session_uri, RDF_TYPE, _iri(TG_QUESTION)),
_triple(session_uri, RDFS_LABEL, _literal("Agent Question")),
_triple(session_uri, PROV_STARTED_AT_TIME, _literal(timestamp)),
_triple(session_uri, TG_QUERY, _literal(query)),
]
@ -76,10 +77,10 @@ def agent_iteration_triples(
observation: str,
) -> List[Triple]:
"""
Build triples for one agent iteration (think/act/observe cycle).
Build triples for one agent iteration (Analysis - think/act/observe cycle).
Creates:
- Entity declaration for the iteration
- Entity declaration with tg:Analysis type
- wasDerivedFrom link to parent (previous iteration or session)
- Thought, action, arguments, and observation data
@ -95,8 +96,9 @@ def agent_iteration_triples(
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, RDF_TYPE, _iri(PROV_ENTITY)),
_triple(iteration_uri, RDF_TYPE, _iri(TG_ANALYSIS)),
_triple(iteration_uri, RDFS_LABEL, _literal(f"Analysis: {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)),
@ -113,10 +115,10 @@ def agent_final_triples(
answer: str,
) -> List[Triple]:
"""
Build triples for an agent final answer.
Build triples for an agent final answer (Conclusion).
Creates:
- Entity declaration for the final answer
- Entity declaration with tg:Conclusion type
- wasDerivedFrom link to parent (last iteration or session)
- The answer text
@ -129,8 +131,9 @@ def agent_final_triples(
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, RDF_TYPE, _iri(PROV_ENTITY)),
_triple(final_uri, RDF_TYPE, _iri(TG_CONCLUSION)),
_triple(final_uri, RDFS_LABEL, _literal("Conclusion")),
_triple(final_uri, PROV_WAS_DERIVED_FROM, _iri(parent_uri)),
_triple(final_uri, TG_ANSWER, _literal(answer)),
]

View file

@ -68,15 +68,20 @@ TG_REASONING = TG + "reasoning"
TG_CONTENT = TG + "content"
TG_DOCUMENT = TG + "document" # Reference to document in librarian
# Explainability entity types (used by both GraphRAG and Agent)
TG_QUESTION = TG + "Question"
TG_EXPLORATION = TG + "Exploration"
TG_FOCUS = TG + "Focus"
TG_SYNTHESIS = TG + "Synthesis"
TG_ANALYSIS = TG + "Analysis"
TG_CONCLUSION = TG + "Conclusion"
# 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

View file

@ -20,6 +20,8 @@ from . namespaces import (
# Query-time provenance predicates
TG_QUERY, TG_EDGE_COUNT, TG_SELECTED_EDGE, TG_EDGE, TG_REASONING, TG_CONTENT,
TG_DOCUMENT,
# Explainability entity types
TG_QUESTION, TG_EXPLORATION, TG_FOCUS, TG_SYNTHESIS,
)
from . uris import activity_uri, agent_uri, edge_selection_uri
@ -310,7 +312,8 @@ def question_triples(
return [
_triple(question_uri, RDF_TYPE, _iri(PROV_ACTIVITY)),
_triple(question_uri, RDFS_LABEL, _literal("GraphRAG question")),
_triple(question_uri, RDF_TYPE, _iri(TG_QUESTION)),
_triple(question_uri, RDFS_LABEL, _literal("GraphRAG Question")),
_triple(question_uri, PROV_STARTED_AT_TIME, _literal(timestamp)),
_triple(question_uri, TG_QUERY, _literal(query)),
]
@ -339,6 +342,7 @@ def exploration_triples(
"""
return [
_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_EDGE_COUNT, _literal(edge_count)),
@ -383,6 +387,7 @@ def focus_triples(
"""
triples = [
_triple(focus_uri, RDF_TYPE, _iri(PROV_ENTITY)),
_triple(focus_uri, RDF_TYPE, _iri(TG_FOCUS)),
_triple(focus_uri, RDFS_LABEL, _literal("Focus")),
_triple(focus_uri, PROV_WAS_DERIVED_FROM, _iri(exploration_uri)),
]
@ -443,6 +448,7 @@ def synthesis_triples(
"""
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(focus_uri)),
]

View file

@ -24,9 +24,13 @@ default_collection = 'default'
# Predicates
TG = "https://trustgraph.ai/ns/"
TG_QUERY = TG + "query"
TG_AGENT_SESSION = TG + "AgentSession"
TG_QUESTION = TG + "Question"
TG_ANALYSIS = TG + "Analysis"
TG_EXPLORATION = TG + "Exploration"
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"
# Retrieval graph
@ -120,14 +124,39 @@ def get_timestamp(socket, flow_id, user, collection, question_id):
def get_session_type(socket, flow_id, user, collection, session_id):
"""Get the type of session (Agent or GraphRAG)."""
triples = query_triples(
"""
Get the type of session (Agent or GraphRAG).
Both have tg:Question type, so we distinguish by URI pattern
or by checking what's derived from it.
"""
# Fast path: check URI pattern
if session_id.startswith("urn:trustgraph:agent:"):
return "Agent"
if session_id.startswith("urn:trustgraph:question:"):
return "GraphRAG"
# Check what's derived from this entity
derived = query_triples(
socket, flow_id, user, collection,
s=session_id, p=RDF_TYPE, g=RETRIEVAL_GRAPH
p=PROV_WAS_DERIVED_FROM, o=session_id, g=RETRIEVAL_GRAPH
)
for s, p, o in triples:
if o == TG_AGENT_SESSION:
return "Agent"
generated = query_triples(
socket, flow_id, user, collection,
p=PROV_WAS_GENERATED_BY, o=session_id, g=RETRIEVAL_GRAPH
)
for s, p, o in derived + generated:
child_types = query_triples(
socket, flow_id, user, collection,
s=s, p=RDF_TYPE, g=RETRIEVAL_GRAPH
)
for _, _, child_type in child_types:
if child_type == TG_ANALYSIS:
return "Agent"
if child_type == TG_EXPLORATION:
return "GraphRAG"
return "GraphRAG"

View file

@ -35,15 +35,20 @@ TG_REASONING = TG + "reasoning"
TG_CONTENT = TG + "content"
TG_DOCUMENT = TG + "document"
TG_REIFIES = TG + "reifies"
# Explainability entity types
TG_QUESTION = TG + "Question"
TG_EXPLORATION = TG + "Exploration"
TG_FOCUS = TG + "Focus"
TG_SYNTHESIS = TG + "Synthesis"
TG_ANALYSIS = TG + "Analysis"
TG_CONCLUSION = TG + "Conclusion"
# 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"
@ -296,19 +301,52 @@ def format_edge(edge, label_cache=None, socket=None, flow_id=None, user=None, co
def detect_trace_type(socket, flow_id, user, collection, entity_id):
"""
Detect whether an entity is an agent session or GraphRAG question.
Detect whether an entity is an agent Question or GraphRAG Question.
Both have rdf:type = tg:Question, so we distinguish by checking
what's derived from it:
- Agent: has tg:Analysis or tg:Conclusion derived
- GraphRAG: has tg:Exploration derived
Also checks URI pattern as fallback:
- urn:trustgraph:agent: -> agent
- urn:trustgraph:question: -> graphrag
Returns:
"agent" if entity has rdf:type = tg:AgentSession
"graphrag" otherwise (default)
"agent" or "graphrag"
"""
triples = query_triples(
# Check URI pattern first (fast path)
if entity_id.startswith("urn:trustgraph:agent:"):
return "agent"
if entity_id.startswith("urn:trustgraph:question:"):
return "graphrag"
# Check what's derived from this entity
derived = find_by_predicate_object(
socket, flow_id, user, collection,
s=entity_id, p=RDF_TYPE, g=RETRIEVAL_GRAPH
PROV_WAS_DERIVED_FROM, entity_id
)
for s, p, o in triples:
if o == TG_AGENT_SESSION:
return "agent"
# Also check wasGeneratedBy (GraphRAG exploration uses this)
generated = find_by_predicate_object(
socket, flow_id, user, collection,
PROV_WAS_GENERATED_BY, entity_id
)
all_children = derived + generated
for child_id in all_children:
child_types = query_triples(
socket, flow_id, user, collection,
s=child_id, p=RDF_TYPE, g=RETRIEVAL_GRAPH
)
for s, p, o in child_types:
if o == TG_ANALYSIS or o == TG_CONCLUSION:
return "agent"
if o == TG_EXPLORATION:
return "graphrag"
# Default to graphrag
return "graphrag"
@ -349,7 +387,7 @@ def build_agent_trace(socket, flow_id, user, collection, session_id, api=None, m
# Check type
types = derived_props.get(RDF_TYPE, [])
if TG_AGENT_ITERATION in types:
if TG_ANALYSIS in types:
iteration = {
"id": derived_id,
"iteration_num": iteration_num,
@ -362,7 +400,7 @@ def build_agent_trace(socket, flow_id, user, collection, session_id, api=None, m
current_uri = derived_id
iteration_num += 1
elif TG_AGENT_FINAL in types:
elif TG_CONCLUSION in types:
answer = derived_props.get(TG_ANSWER, [None])[0]
if answer and len(answer) > max_answer:
answer = answer[:max_answer] + "... [truncated]"
@ -390,12 +428,12 @@ def print_agent_text(trace):
print(f"Time: {trace['time']}")
print()
# Iterations
print("--- Iterations ---")
# Analysis steps
print("--- Analysis ---")
iterations = trace.get("iterations", [])
if iterations:
for iteration in iterations:
print(f"Iteration {iteration['iteration_num']}:")
print(f"Analysis {iteration['iteration_num']}:")
print(f" Thought: {iteration.get('thought', 'N/A')}")
print(f" Action: {iteration.get('action', 'N/A')}")
@ -422,18 +460,18 @@ def print_agent_text(trace):
print(f" Observation: {obs}")
print()
else:
print("No iterations recorded")
print("No analysis steps recorded")
print()
# Final answer
print("--- Final Answer ---")
# Conclusion
print("--- Conclusion ---")
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")
print("No conclusion recorded")
def print_agent_json(trace):