mirror of
https://github.com/trustgraph-ai/trustgraph.git
synced 2026-07-08 21:02:12 +02:00
8b4a4fac46
Refactor agent provenance so that the decision (thought + tool selection) and the result (observation) are separate DAG entities: Question ← Analysis+ToolUse ← Observation ← ... ← Conclusion Analysis gains tg:ToolUse as a mixin RDF type and is emitted before tool execution via an on_action callback in react(). This ensures sub-traces (e.g. GraphRAG) appear after their parent Analysis in the streaming event order. Observation becomes a standalone prov:Entity with tg:Observation type, emitted after tool execution. The linear DAG chain runs through Observation — subsequent iterations and the Conclusion derive from it, not from the Analysis. message_id is populated on streaming AgentResponse for thought and observation chunks, using the provenance URI of the entity being built. This lets clients group streamed chunks by entity. Wire changes: - provenance/agent.py: Add ToolUse type, new agent_observation_triples(), remove observation from iteration - agent_manager.py: Add on_action callback between reason() and tool execution - orchestrator/pattern_base.py: Split emit, wire message_id, chain through observation URIs - orchestrator/react_pattern.py: Emit Analysis via on_action before tool runs - agent/react/service.py: Same for non-orchestrator path - api/explainability.py: New Observation class, updated dispatch and chain walker - api/types.py: Add message_id to AgentThought/AgentObservation - cli: Render Observation separately, [analysis: tool] labels
73 lines
2.5 KiB
Python
73 lines
2.5 KiB
Python
|
|
from . request_response_spec import RequestResponse, RequestResponseSpec
|
|
from .. schema import GraphRagQuery, GraphRagResponse
|
|
|
|
class GraphRagClient(RequestResponse):
|
|
async def rag(self, query, user="trustgraph", collection="default",
|
|
chunk_callback=None, explain_callback=None,
|
|
parent_uri="",
|
|
timeout=600):
|
|
"""
|
|
Execute a graph RAG query with optional streaming callbacks.
|
|
|
|
Args:
|
|
query: The question to ask
|
|
user: User identifier
|
|
collection: Collection identifier
|
|
chunk_callback: Optional async callback(text, end_of_stream) for text chunks
|
|
explain_callback: Optional async callback(explain_id, explain_graph) for explain notifications
|
|
timeout: Request timeout in seconds
|
|
|
|
Returns:
|
|
Complete response text (accumulated from all chunks)
|
|
"""
|
|
accumulated_response = []
|
|
|
|
async def recipient(resp):
|
|
if resp.error:
|
|
raise RuntimeError(resp.error.message)
|
|
|
|
# Handle explain notifications
|
|
if resp.message_type == 'explain':
|
|
if explain_callback and resp.explain_id:
|
|
await explain_callback(resp.explain_id, resp.explain_graph)
|
|
return False # Continue receiving
|
|
|
|
# Handle text chunks
|
|
if resp.message_type == 'chunk':
|
|
if resp.response:
|
|
accumulated_response.append(resp.response)
|
|
if chunk_callback:
|
|
await chunk_callback(resp.response, resp.end_of_stream)
|
|
|
|
# Complete when session ends
|
|
if resp.end_of_session:
|
|
return True
|
|
|
|
return False # Continue receiving
|
|
|
|
await self.request(
|
|
GraphRagQuery(
|
|
query = query,
|
|
user = user,
|
|
collection = collection,
|
|
parent_uri = parent_uri,
|
|
),
|
|
timeout=timeout,
|
|
recipient=recipient,
|
|
)
|
|
|
|
return "".join(accumulated_response)
|
|
|
|
class GraphRagClientSpec(RequestResponseSpec):
|
|
def __init__(
|
|
self, request_name, response_name,
|
|
):
|
|
super(GraphRagClientSpec, self).__init__(
|
|
request_name = request_name,
|
|
request_schema = GraphRagQuery,
|
|
response_name = response_name,
|
|
response_schema = GraphRagResponse,
|
|
impl = GraphRagClient,
|
|
)
|
|
|