Skip to main content

_Trace

Internal class for creating and managing traces in the NeuralTrust system. This class captures details about interactions or events, including metadata, timing information, and content.

Constructor​

_Trace(
    client: TraceClient,
    trace_id: Optional[str] = None,
    interaction_id: Optional[str] = None,
    conversation_id: Optional[str] = None,
    channel_id: Optional[str] = None,
    session_id: Optional[str] = None,
    user: Optional[User] = None,
    metadata: Optional[Metadata] = None,
    custom: Optional[dict] = None
)

Parameters:​

  • client (TraceClient): The TraceClient instance used to send traces
  • trace_id (Optional[str]): Unique identifier for this specific trace
  • interaction_id (Optional[str]): ID that can be shared across multiple related traces
  • conversation_id (Optional[str]): Unique identifier for the conversation
  • channel_id (Optional[str]): Optional identifier for the communication channel
  • session_id (Optional[str]): Optional identifier for the user session
  • user (Optional[User]): Optional User object containing user information
  • metadata (Optional[Metadata]): Optional metadata dictionary
  • custom (Optional[dict]): Optional custom data dictionary

Methods​

message()​

def message(input: str) -> "_Trace"

Start a message trace.

Parameters:​

  • input (str): The input message to trace

Returns:​

  • _Trace: A new Trace object initialized with the message event

tool()​

def tool(input: str, parent_id: Optional[str] = None) -> "_Trace"

Start a tool trace.

Parameters:​

  • input (str): The input for the tool event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized with the tool event

agent()​

def agent(input: str, parent_id: Optional[str] = None) -> "_Trace"

Start an agent trace.

Parameters:​

  • input (str): The input for the agent event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized with the agent event

retrieval()​

def retrieval(input: str, parent_id: Optional[str] = None) -> "_Trace"

Start a retrieval trace.

Parameters:​

  • input (str): The input for the retrieval event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized with the retrieval event

generation()​

def generation(input: str, parent_id: Optional[str] = None) -> "_Trace"

Start a generation trace.

Parameters:​

  • input (str): The input for the generation event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized with the generation event

router()​

def router(input: str, parent_id: Optional[str] = None) -> "_Trace"

Start and automatically end a router trace.

Parameters:​

  • input (str): The input for the router event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized and completed with the router event

event()​

def event(input: str, parent_id: Optional[str] = None) -> "_Trace"

Record a custom event trace.

Parameters:​

  • input (str): The input for the custom event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized and completed with the custom event

system()​

def system(prompt: Union[str, object], parent_id: Optional[str] = None) -> "_Trace"

Record a system trace.

Parameters:​

  • prompt (Union[str, object]): The input for the system event
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized and completed with the system event

feedback()​

def feedback(
    feedback_tag: FeedbackTag,
    feedback_text: str,
    parent_id: Optional[str] = None
) -> "_Trace"

Record a feedback trace.

Parameters:​

  • feedback_tag (FeedbackTag): The type/category of feedback being provided
  • feedback_text (str): The actual feedback content
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: A new Trace object initialized and completed with the feedback event

end()​

def end(output: Union[str, object]) -> "_Trace"

End the current trace and record its output.

Parameters:​

  • output (Union[str, object]): The output/result of the trace

Returns:​

  • _Trace: The current Trace object with completed trace data

send()​

def send(
    event_type: EventType,
    input: str,
    output: Optional[str] = None,
    latency: Optional[int] = None,
    start_timestamp: Optional[int] = None,
    end_timestamp: Optional[int] = None,
    parent_id: Optional[str] = None
) -> "_Trace"

Send an atomic event to NeuralTrust.

Parameters:​

  • event_type (EventType): The type of event being recorded
  • input (str): The input data for the event
  • output (Optional[str]): Optional output data for the event
  • latency (Optional[int]): Optional processing time in milliseconds
  • start_timestamp (Optional[int]): Optional event start time in milliseconds
  • end_timestamp (Optional[int]): Optional event end time in milliseconds
  • parent_id (Optional[str]): The ID of the parent trace

Returns:​

  • _Trace: The current Trace object with the recorded event data

Usage Example​

# Create a trace
trace = client.trace(conversation_id="conv123")

# Record a message event
message_trace = trace.message("User input")
response = message_trace.end("Bot response")

# Record a tool event
tool_trace = trace.tool("Processing data", parent_id=message_trace.trace_id)
result = tool_trace.end("Data processed")

# Send feedback
trace.feedback(FeedbackTag.POSITIVE, "Great response!", parent_id=message_trace.trace_id)