apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-10 05:42:12 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add agent explainability instrumentation and unify envelope field naming

Browse source 
Addresses recommendations from the UX developer's agent experience report.
Adds provenance predicates, DAG structure changes, error resilience, and
a published OWL ontology.

Explainability additions:

- Tool candidates: tg:toolCandidate on Analysis events lists the tools
  visible to the LLM for each iteration (names only, descriptions in config)
- Termination reason: tg:terminationReason on Conclusion/Synthesis events
  (final-answer, plan-complete, subagents-complete)
- Step counter: tg:stepNumber on iteration events
- Pattern decision: new tg:PatternDecision entity in the DAG between
  session and first iteration, carrying tg:pattern and tg:taskType
- Latency: tg:llmDurationMs on Analysis events, tg:toolDurationMs on
  Observation events
- Token counts on events: tg:inToken/tg:outToken/tg:llmModel on
  Grounding, Focus, Synthesis, and Analysis events
- Tool/parse errors: tg:toolError on Observation events with tg:Error
  mixin type. Parse failures return as error observations instead of
  crashing the agent, giving it a chance to retry.

Envelope unification:

- Rename chunk_type to message_type across AgentResponse schema,
  translator, SDK types, socket clients, CLI, and all tests.
  Agent and RAG services now both use message_type on the wire.

Ontology:

- specs/ontology/trustgraph.ttl — OWL vocabulary covering all 26 classes,
  7 object properties, and 36+ datatype properties including new predicates.

DAG structure tests:

- tests/unit/test_provenance/test_dag_structure.py verifies the
  wasDerivedFrom chain for GraphRAG, DocumentRAG, and all three agent
  patterns (react, plan, supervisor) including the pattern-decision link.
This commit is contained in:
Cyber MacGeddon 2026-04-13 14:55:36 +01:00
parent 14e49d83c7
commit 2e83413cbe
42 changed files with 1577 additions and 205 deletions
Download patch file Download diff file Expand all files Collapse all files

91
trustgraph-flow/trustgraph/agent/orchestrator/pattern_base.py
View file

@ -25,6 +25,7 @@ from trustgraph.provenance import (
agent_plan_uri,
agent_step_result_uri,
agent_synthesis_uri,
agent_pattern_decision_uri,
agent_session_triples,
agent_iteration_triples,
agent_observation_triples,
@ -34,6 +35,7 @@ from trustgraph.provenance import (
agent_plan_triples,
agent_step_result_triples,
agent_synthesis_triples,
agent_pattern_decision_triples,
set_graph,
GRAPH_RETRIEVAL,
)
@ -182,7 +184,7 @@ class PatternBase:
logger.debug(f"Think: {x} (is_final={is_final})")
if streaming:
r = AgentResponse(
chunk_type="thought",
message_type="thought",
content=x,
end_of_message=is_final,
end_of_dialog=False,
@ -190,7 +192,7 @@ class PatternBase:
)
else:
r = AgentResponse(
chunk_type="thought",
message_type="thought",
content=x,
end_of_message=True,
end_of_dialog=False,
@ -205,7 +207,7 @@ class PatternBase:
logger.debug(f"Observe: {x} (is_final={is_final})")
if streaming:
r = AgentResponse(
chunk_type="observation",
message_type="observation",
content=x,
end_of_message=is_final,
end_of_dialog=False,
@ -213,7 +215,7 @@ class PatternBase:
)
else:
r = AgentResponse(
chunk_type="observation",
message_type="observation",
content=x,
end_of_message=True,
end_of_dialog=False,
@ -228,7 +230,7 @@ class PatternBase:
logger.debug(f"Answer: {x}")
if streaming:
r = AgentResponse(
chunk_type="answer",
message_type="answer",
content=x,
end_of_message=False,
end_of_dialog=False,
@ -236,7 +238,7 @@ class PatternBase:
)
else:
r = AgentResponse(
chunk_type="answer",
message_type="answer",
content=x,
end_of_message=True,
end_of_dialog=False,
@ -270,16 +272,43 @@ class PatternBase:
logger.debug(f"Emitted session triples for {session_uri}")
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=session_uri,
explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
async def emit_pattern_decision_triples(
self, flow, session_id, session_uri, pattern, task_type,
user, collection, respond,
):
"""Emit provenance triples for a meta-router pattern decision."""
uri = agent_pattern_decision_uri(session_id)
triples = set_graph(
agent_pattern_decision_triples(
uri, session_uri, pattern, task_type,
),
GRAPH_RETRIEVAL,
)
await flow("explainability").send(Triples(
metadata=Metadata(id=uri, user=user, collection=collection),
triples=triples,
))
await respond(AgentResponse(
message_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
return uri
async def emit_iteration_triples(self, flow, session_id, iteration_num,
session_uri, act, request, respond,
streaming):
streaming, tool_candidates=None,
step_number=None,
llm_duration_ms=None,
in_token=None, out_token=None,
model=None):
"""Emit provenance triples for an iteration (Analysis+ToolUse)."""
iteration_uri = agent_iteration_uri(session_id, iteration_num)
@ -319,6 +348,12 @@ class PatternBase:
arguments=act.arguments,
thought_uri=thought_entity_uri if thought_doc_id else None,
thought_document_id=thought_doc_id,
tool_candidates=tool_candidates,
step_number=step_number,
llm_duration_ms=llm_duration_ms,
in_token=in_token,
out_token=out_token,
model=model,
),
GRAPH_RETRIEVAL,
)
@ -333,7 +368,7 @@ class PatternBase:
logger.debug(f"Emitted iteration triples for {iteration_uri}")
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=iteration_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -342,7 +377,9 @@ class PatternBase:
async def emit_observation_triples(self, flow, session_id, iteration_num,
observation_text, request, respond,
context=None):
context=None,
tool_duration_ms=None,
tool_error=None):
"""Emit provenance triples for a standalone Observation entity."""
iteration_uri = agent_iteration_uri(session_id, iteration_num)
observation_entity_uri = agent_observation_uri(session_id, iteration_num)
@ -375,6 +412,8 @@ class PatternBase:
observation_entity_uri,
parent_uri,
document_id=observation_doc_id,
tool_duration_ms=tool_duration_ms,
tool_error=tool_error,
),
GRAPH_RETRIEVAL,
)
@ -389,7 +428,7 @@ class PatternBase:
logger.debug(f"Emitted observation triples for {observation_entity_uri}")
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=observation_entity_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -398,7 +437,7 @@ class PatternBase:
async def emit_final_triples(self, flow, session_id, iteration_num,
session_uri, answer_text, request, respond,
streaming):
streaming, termination_reason=None):
"""Emit provenance triples for the final answer and save to librarian."""
final_uri = agent_final_uri(session_id)
@ -432,6 +471,7 @@ class PatternBase:
question_uri=final_question_uri,
previous_uri=final_previous_uri,
document_id=answer_doc_id,
termination_reason=termination_reason,
),
GRAPH_RETRIEVAL,
)
@ -446,7 +486,7 @@ class PatternBase:
logger.debug(f"Emitted final triples for {final_uri}")
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=final_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -470,7 +510,7 @@ class PatternBase:
triples=triples,
))
await respond(AgentResponse(
chunk_type="explain", content="",
message_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
@ -509,7 +549,7 @@ class PatternBase:
triples=triples,
))
await respond(AgentResponse(
chunk_type="explain", content="",
message_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
@ -529,7 +569,7 @@ class PatternBase:
triples=triples,
))
await respond(AgentResponse(
chunk_type="explain", content="",
message_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
@ -562,14 +602,14 @@ class PatternBase:
triples=triples,
))
await respond(AgentResponse(
chunk_type="explain", content="",
message_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
async def emit_synthesis_triples(
self, flow, session_id, previous_uris, answer_text, user, collection,
respond, streaming,
respond, streaming, termination_reason=None,
):
"""Emit provenance for a synthesis answer."""
uri = agent_synthesis_uri(session_id)
@ -586,7 +626,10 @@ class PatternBase:
doc_id = None
triples = set_graph(
agent_synthesis_triples(uri, previous_uris, doc_id),
agent_synthesis_triples(
uri, previous_uris, doc_id,
termination_reason=termination_reason,
),
GRAPH_RETRIEVAL,
)
await flow("explainability").send(Triples(
@ -594,7 +637,7 @@ class PatternBase:
triples=triples,
))
await respond(AgentResponse(
chunk_type="explain", content="",
message_type="explain", content="",
explain_id=uri, explain_graph=GRAPH_RETRIEVAL,
explain_triples=triples,
))
@ -616,7 +659,7 @@ class PatternBase:
if text:
accumulated.append(text)
await respond(AgentResponse(
chunk_type="answer",
message_type="answer",
content=text,
end_of_message=False,
end_of_dialog=False,
@ -666,7 +709,7 @@ class PatternBase:
# Answer wasn't streamed yet — send it as a chunk first
if answer_text:
await respond(AgentResponse(
chunk_type="answer",
message_type="answer",
content=answer_text,
end_of_message=False,
end_of_dialog=False,
@ -675,7 +718,7 @@ class PatternBase:
if streaming:
# End-of-dialog marker with usage
await respond(AgentResponse(
chunk_type="answer",
message_type="answer",
content="",
end_of_message=True,
end_of_dialog=True,
@ -684,7 +727,7 @@ class PatternBase:
))
else:
await respond(AgentResponse(
chunk_type="answer",
message_type="answer",
content=answer_text,
end_of_message=True,
end_of_dialog=True,

10
trustgraph-flow/trustgraph/agent/orchestrator/plan_pattern.py
View file

@ -35,7 +35,8 @@ class PlanThenExecutePattern(PatternBase):
Subsequent calls execute the next pending plan step via ReACT.
"""
async def iterate(self, request, respond, next, flow, usage=None):
async def iterate(self, request, respond, next, flow, usage=None,
pattern_decision_uri=None):
if usage is None:
usage = UsageTracker()
@ -66,16 +67,18 @@ class PlanThenExecutePattern(PatternBase):
# Determine current phase by checking history for a plan step
plan = self._extract_plan(request.history)
derive_from_uri = pattern_decision_uri or session_uri
if plan is None:
await self._planning_iteration(
request, respond, next, flow,
session_id, collection, streaming, session_uri,
session_id, collection, streaming, derive_from_uri,
iteration_num, usage=usage,
)
else:
await self._execution_iteration(
request, respond, next, flow,
session_id, collection, streaming, session_uri,
session_id, collection, streaming, derive_from_uri,
iteration_num, plan, usage=usage,
)
@ -385,6 +388,7 @@ class PlanThenExecutePattern(PatternBase):
await self.emit_synthesis_triples(
flow, session_id, last_step_uri,
response_text, request.user, collection, respond, streaming,
termination_reason="plan-complete",
)
if self.is_subagent(request):

22
trustgraph-flow/trustgraph/agent/orchestrator/react_pattern.py
View file

@ -37,7 +37,8 @@ class ReactPattern(PatternBase):
result is appended to history and a next-request is emitted.
"""
async def iterate(self, request, respond, next, flow, usage=None):
async def iterate(self, request, respond, next, flow, usage=None,
pattern_decision_uri=None):
if usage is None:
usage = UsageTracker()
@ -108,11 +109,23 @@ class ReactPattern(PatternBase):
session_id, iteration_num,
)
# Tool names available to the LLM for this iteration
tool_candidates = [t.name for t in filtered_tools.values()]
# Use pattern decision as derivation source if available
derive_from_uri = pattern_decision_uri or session_uri
# Callback: emit Analysis+ToolUse triples before tool executes
async def on_action(act):
await self.emit_iteration_triples(
flow, session_id, iteration_num, session_uri,
flow, session_id, iteration_num, derive_from_uri,
act, request, respond, streaming,
tool_candidates=tool_candidates,
step_number=iteration_num,
llm_duration_ms=getattr(act, 'llm_duration_ms', None),
in_token=getattr(act, 'in_token', None),
out_token=getattr(act, 'out_token', None),
model=getattr(act, 'llm_model', None),
)
act = await temp_agent.react(
@ -138,8 +151,9 @@ class ReactPattern(PatternBase):
# Emit final provenance
await self.emit_final_triples(
flow, session_id, iteration_num, session_uri,
flow, session_id, iteration_num, derive_from_uri,
f, request, respond, streaming,
termination_reason="final-answer",
)
if self.is_subagent(request):
@ -157,6 +171,8 @@ class ReactPattern(PatternBase):
flow, session_id, iteration_num,
act.observation, request, respond,
context=context,
tool_duration_ms=getattr(act, 'tool_duration_ms', None),
tool_error=getattr(act, 'tool_error', None),
)
history.append(act)

38
trustgraph-flow/trustgraph/agent/orchestrator/service.py
View file

@ -23,7 +23,7 @@ from ... base import Consumer, Producer
from ... base import ConsumerMetrics, ProducerMetrics
from ... schema import AgentRequest, AgentResponse, AgentStep, Error
from ..orchestrator.pattern_base import UsageTracker
from ..orchestrator.pattern_base import UsageTracker, PatternBase
from ... schema import Triples, Metadata
from ... schema import LibrarianRequest, LibrarianResponse, DocumentMetadata
from ... schema import librarian_request_queue, librarian_response_queue
@ -537,19 +537,31 @@ class Processor(AgentService):
)
# Dispatch to the selected pattern
selected = self.react_pattern
if pattern == "plan-then-execute":
await self.plan_pattern.iterate(
request, respond, next, flow, usage=usage,
)
selected = self.plan_pattern
elif pattern == "supervisor":
await self.supervisor_pattern.iterate(
request, respond, next, flow, usage=usage,
)
else:
# Default to react
await self.react_pattern.iterate(
request, respond, next, flow, usage=usage,
)
selected = self.supervisor_pattern
# Emit pattern decision provenance on first iteration
pattern_decision_uri = None
if not request.history and pattern:
session_id = getattr(request, 'session_id', '')
if session_id:
session_uri = self.provenance_session_uri(session_id)
pattern_decision_uri = \
await selected.emit_pattern_decision_triples(
flow, session_id, session_uri,
pattern, getattr(request, 'task_type', ''),
request.user,
getattr(request, 'collection', 'default'),
respond,
)
await selected.iterate(
request, respond, next, flow, usage=usage,
pattern_decision_uri=pattern_decision_uri,
)
except Exception as e:
@ -565,7 +577,7 @@ class Processor(AgentService):
)
r = AgentResponse(
chunk_type="error",
message_type="error",
content=str(e),
end_of_message=True,
end_of_dialog=True,

10
trustgraph-flow/trustgraph/agent/orchestrator/supervisor_pattern.py
View file

@ -38,7 +38,8 @@ class SupervisorPattern(PatternBase):
- "synthesise": triggered by aggregator with results in subagent_results
"""
async def iterate(self, request, respond, next, flow, usage=None):
async def iterate(self, request, respond, next, flow, usage=None,
pattern_decision_uri=None):
if usage is None:
usage = UsageTracker()
@ -70,18 +71,20 @@ class SupervisorPattern(PatternBase):
)
)
derive_from_uri = pattern_decision_uri or session_uri
if has_results:
await self._synthesise(
request, respond, next, flow,
session_id, collection, streaming,
session_uri, iteration_num,
derive_from_uri, iteration_num,
usage=usage,
)
else:
await self._decompose_and_fanout(
request, respond, next, flow,
session_id, collection, streaming,
session_uri, iteration_num,
derive_from_uri, iteration_num,
usage=usage,
)
@ -235,6 +238,7 @@ class SupervisorPattern(PatternBase):
await self.emit_synthesis_triples(
flow, session_id, finding_uris,
response_text, request.user, collection, respond, streaming,
termination_reason="subagents-complete",
)
await self.send_final_response(

67
trustgraph-flow/trustgraph/agent/react/agent_manager.py
View file

@ -3,6 +3,7 @@ import logging
import json
import re
import asyncio
import time
from . types import Action, Final
@ -260,6 +261,7 @@ class AgentManager:
streaming=True,
chunk_callback=on_chunk
)
self._last_prompt_result = prompt_result
if usage:
usage.track(prompt_result)
@ -269,7 +271,13 @@ class AgentManager:
# Get result
result = parser.get_result()
if result is None:
raise RuntimeError("Parser failed to produce a result")
return Action(
thought="",
name="__parse_error__",
arguments={},
observation="",
tool_error="LLM response could not be parsed (streaming)",
)
return result
@ -281,6 +289,7 @@ class AgentManager:
variables=variables,
streaming=False
)
self._last_prompt_result = prompt_result
if usage:
usage.track(prompt_result)
response_text = prompt_result.text
@ -294,12 +303,19 @@ class AgentManager:
except ValueError as e:
logger.error(f"Failed to parse response: {e}")
logger.error(f"Response was: {response_text}")
raise RuntimeError(f"Failed to parse agent response: {e}")
return Action(
thought="",
name="__parse_error__",
arguments={},
observation="",
tool_error=f"LLM parse error: {e}",
)
async def react(self, question, history, think, observe, context,
streaming=False, answer=None, on_action=None,
usage=None):
t0 = time.monotonic()
act = await self.reason(
question = question,
history = history,
@ -310,6 +326,12 @@ class AgentManager:
answer = answer,
usage = usage,
)
act.llm_duration_ms = int((time.monotonic() - t0) * 1000)
pr = getattr(self, '_last_prompt_result', None)
if pr:
act.in_token = pr.in_token
act.out_token = pr.out_token
act.llm_model = pr.model
if isinstance(act, Final):
@ -328,24 +350,43 @@ class AgentManager:
logger.debug(f"ACTION: {act.name}")
# Notify caller before tool execution (for provenance)
if on_action:
await on_action(act)
# Handle parse errors — skip tool execution
if act.name == "__parse_error__":
resp = f"Error: {act.tool_error}"
act.tool_duration_ms = 0
await observe(resp, is_final=True)
act.observation = resp
return act
if act.name in self.tools:
action = self.tools[act.name]
else:
raise RuntimeError(f"No action for {act.name}!")
# Notify caller before tool execution (for provenance)
if on_action:
await on_action(act)
t0 = time.monotonic()
try:
resp = await action.implementation(context).invoke(
**act.arguments
)
resp = await action.implementation(context).invoke(
**act.arguments
)
if isinstance(resp, str):
resp = resp.strip()
else:
resp = str(resp)
resp = resp.strip()
if isinstance(resp, str):
resp = resp.strip()
else:
resp = str(resp)
resp = resp.strip()
act.tool_error = None
except Exception as e:
logger.error(f"Tool execution error ({act.name}): {e}")
resp = f"Error: {e}"
act.tool_error = str(e)
act.tool_duration_ms = int((time.monotonic() - t0) * 1000)
await observe(resp, is_final=True)

26
trustgraph-flow/trustgraph/agent/react/service.py
View file

@ -469,7 +469,7 @@ class Processor(AgentService):
# Send explain event for session
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=session_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -492,7 +492,7 @@ class Processor(AgentService):
if streaming:
r = AgentResponse(
chunk_type="thought",
message_type="thought",
content=x,
end_of_message=is_final,
end_of_dialog=False,
@ -500,7 +500,7 @@ class Processor(AgentService):
)
else:
r = AgentResponse(
chunk_type="thought",
message_type="thought",
content=x,
end_of_message=True,
end_of_dialog=False,
@ -515,7 +515,7 @@ class Processor(AgentService):
if streaming:
r = AgentResponse(
chunk_type="observation",
message_type="observation",
content=x,
end_of_message=is_final,
end_of_dialog=False,
@ -523,7 +523,7 @@ class Processor(AgentService):
)
else:
r = AgentResponse(
chunk_type="observation",
message_type="observation",
content=x,
end_of_message=True,
end_of_dialog=False,
@ -540,7 +540,7 @@ class Processor(AgentService):
if streaming:
r = AgentResponse(
chunk_type="answer",
message_type="answer",
content=x,
end_of_message=False,
end_of_dialog=False,
@ -548,7 +548,7 @@ class Processor(AgentService):
)
else:
r = AgentResponse(
chunk_type="answer",
message_type="answer",
content=x,
end_of_message=True,
end_of_dialog=False,
@ -637,7 +637,7 @@ class Processor(AgentService):
logger.debug(f"Emitted iteration triples for {iter_uri}")
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=iter_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -715,7 +715,7 @@ class Processor(AgentService):
# Send explain event for conclusion
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=final_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -725,7 +725,7 @@ class Processor(AgentService):
if streaming:
# End-of-dialog marker — answer chunks already sent via callback
r = AgentResponse(
chunk_type="answer",
message_type="answer",
content="",
end_of_message=True,
end_of_dialog=True,
@ -733,7 +733,7 @@ class Processor(AgentService):
)
else:
r = AgentResponse(
chunk_type="answer",
message_type="answer",
content=f,
end_of_message=True,
end_of_dialog=True,
@ -792,7 +792,7 @@ class Processor(AgentService):
# Send explain event for observation
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=observation_entity_uri,
explain_graph=GRAPH_RETRIEVAL,
@ -847,7 +847,7 @@ class Processor(AgentService):
streaming = getattr(request, 'streaming', False) if 'request' in locals() else False
r = AgentResponse(
chunk_type="error",
message_type="error",
content=str(e),
end_of_message=True,
end_of_dialog=True,

2
trustgraph-flow/trustgraph/agent/react/tools.py
View file

@ -42,7 +42,7 @@ class KnowledgeQueryImpl:
async def explain_callback(explain_id, explain_graph, explain_triples=None):
self.context.last_sub_explain_uri = explain_id
await respond(AgentResponse(
chunk_type="explain",
message_type="explain",
content="",
explain_id=explain_id,
explain_graph=explain_graph,

12
trustgraph-flow/trustgraph/agent/react/types.py
View file

@ -22,9 +22,19 @@ class Action:
name : str
arguments : dict
observation : str
llm_duration_ms : int = None
tool_duration_ms : int = None
tool_error : str = None
in_token : int = None
out_token : int = None
llm_model : str = None
@dataclasses.dataclass
class Final:
thought : str
final : str
llm_duration_ms : int = None
in_token : int = None
out_token : int = None
llm_model : str = None

13
trustgraph-flow/trustgraph/retrieval/document_rag/document_rag.py
View file

@ -56,6 +56,8 @@ class Query:
if not concepts:
concepts = [query]
self.concepts_usage = result
if self.verbose:
logger.debug(f"Extracted concepts: {concepts}")
@ -217,8 +219,14 @@ class DocumentRag:
# Emit grounding explainability after concept extraction
if explain_callback:
cu = getattr(q, 'concepts_usage', None)
gnd_triples = set_graph(
grounding_triples(gnd_uri, q_uri, concepts),
grounding_triples(
gnd_uri, q_uri, concepts,
in_token=cu.in_token if cu else None,
out_token=cu.out_token if cu else None,
model=cu.model if cu else None,
),
GRAPH_RETRIEVAL
)
await explain_callback(gnd_triples, gnd_uri)
@ -286,6 +294,9 @@ class DocumentRag:
docrag_synthesis_triples(
syn_uri, exp_uri,
document_id=synthesis_doc_id,
in_token=synthesis_result.in_token if synthesis_result else None,
out_token=synthesis_result.out_token if synthesis_result else None,
model=synthesis_result.model if synthesis_result else None,
),
GRAPH_RETRIEVAL
)

31
trustgraph-flow/trustgraph/retrieval/graph_rag/graph_rag.py
View file

@ -152,6 +152,8 @@ class Query:
if self.verbose:
logger.debug(f"Extracted concepts: {concepts}")
self.concepts_usage = result
# Fall back to raw query if extraction returns nothing
return concepts if concepts else [query]
@ -667,8 +669,14 @@ class GraphRag:
# Emit grounding explain after concept extraction
if explain_callback:
cu = getattr(q, 'concepts_usage', None)
gnd_triples = set_graph(
grounding_triples(gnd_uri, q_uri, concepts),
grounding_triples(
gnd_uri, q_uri, concepts,
in_token=cu.in_token if cu else None,
out_token=cu.out_token if cu else None,
model=cu.model if cu else None,
),
GRAPH_RETRIEVAL
)
await explain_callback(gnd_triples, gnd_uri)
@ -883,9 +891,25 @@ class GraphRag:
# Emit focus explain after edge selection completes
if explain_callback:
# Sum scoring + reasoning token usage for focus event
focus_in = 0
focus_out = 0
focus_model = None
for r in [scoring_result, reasoning_result]:
if r is not None:
if r.in_token is not None:
focus_in += r.in_token
if r.out_token is not None:
focus_out += r.out_token
if r.model is not None:
focus_model = r.model
foc_triples = set_graph(
focus_triples(
foc_uri, exp_uri, selected_edges_with_reasoning, session_id
foc_uri, exp_uri, selected_edges_with_reasoning, session_id,
in_token=focus_in or None,
out_token=focus_out or None,
model=focus_model,
),
GRAPH_RETRIEVAL
)
@ -956,6 +980,9 @@ class GraphRag:
synthesis_triples(
syn_uri, foc_uri,
document_id=synthesis_doc_id,
in_token=synthesis_result.in_token if synthesis_result else None,
out_token=synthesis_result.out_token if synthesis_result else None,
model=synthesis_result.model if synthesis_result else None,
),
GRAPH_RETRIEVAL
)