Compare commits

...

8 Commits

Author SHA1 Message Date
nityanandagohain
8de75d73ec fix: add system message only after checking if it doesn't exists 2026-10-15 10:36:36 +05:30
nityanandagohain
584790e1c6 feat: apply gen_ai.system_instructions and finish_reasons to the thread 2026-10-15 10:03:54 +05:30
nityanandagohain
bb1adce04a fix: update references 2026-10-15 09:45:56 +05:30
nityanandagohain
3de24379d8 fix: cleanup 2026-10-15 00:47:19 +05:30
nityanandagohain
7fa3cc883b fix: keep minimal changes 2026-10-15 00:09:52 +05:30
nityanandagohain
e1c7c0a3ad fix: remove formatting changes 2026-10-14 17:06:29 +05:30
nityanandagohain
15784b2060 Merge remote-tracking branch 'origin/main' into feat/ai-trace-thread-normalize-1 2026-10-14 13:15:18 +05:30
nityanandagohain
5fa85064e9 feat: add more thread normalizers 2026-10-13 17:37:52 +05:30
17 changed files with 596 additions and 133 deletions

View File

@@ -16440,12 +16440,12 @@ paths:
/api/v1/traces/{traceID}/thread:
get:
deprecated: false
description: Returns the spans carrying gen_ai input or output messages, or
a tool execution, in timestamp order. Messages already in the OTel GenAI shape
are decoded into formatted_input and formatted_output, with formatter naming
the converter and formatter_warnings what it could not resolve. Pass nextCursor
as after or prevCursor as before to page, or spanId to open the page around
a span.
description: Returns the gen_ai spans of a trace, model calls, tool calls and
agent spans, in timestamp order. The messages a span carries are decoded into
formatted_input and formatted_output, with formatter naming the converter
and formatter_warnings what it could not resolve; a span without messages
has both empty. Pass nextCursor as after or prevCursor as before to page,
or spanId to open the page around a span.
operationId: GetTraceThread
parameters:
- description: Page size, at most 100. 0 means 20.

View File

@@ -261,7 +261,7 @@ export const invalidateGetTraceSummary = async (
};
/**
* Returns the spans carrying gen_ai input or output messages, or a tool execution, in timestamp order. Messages already in the OTel GenAI shape are decoded into formatted_input and formatted_output, with formatter naming the converter and formatter_warnings what it could not resolve. Pass nextCursor as after or prevCursor as before to page, or spanId to open the page around a span.
* Returns the gen_ai spans of a trace, model calls, tool calls and agent spans, in timestamp order. The messages a span carries are decoded into formatted_input and formatted_output, with formatter naming the converter and formatter_warnings what it could not resolve; a span without messages has both empty. Pass nextCursor as after or prevCursor as before to page, or spanId to open the page around a span.
* @summary Get thread view for a trace
*/
export const getTraceThread = (

View File

@@ -90,7 +90,7 @@ func (provider *provider) addTraceDetailRoutes(router *mux.Router) error {
ID: "GetTraceThread",
Tags: []string{"tracedetail"},
Summary: "Get thread view for a trace",
Description: "Returns the spans carrying gen_ai input or output messages, or a tool execution, in timestamp order. Messages already in the OTel GenAI shape are decoded into formatted_input and formatted_output, with formatter naming the converter and formatter_warnings what it could not resolve. Pass nextCursor as after or prevCursor as before to page, or spanId to open the page around a span.",
Description: "Returns the gen_ai spans of a trace, model calls, tool calls and agent spans, in timestamp order. The messages a span carries are decoded into formatted_input and formatted_output, with formatter naming the converter and formatter_warnings what it could not resolve; a span without messages has both empty. Pass nextCursor as after or prevCursor as before to page, or spanId to open the page around a span.",
RequestQuery: new(spantypes.GetTraceThreadParams),
Response: new(spantypes.GettableTraceThread),
ResponseContentType: "application/json",

View File

@@ -297,11 +297,11 @@ func (s *traceStore) GetThreadSpans(ctx context.Context, orgID valuer.UUID, trac
sb.SelectMore("attributes")
}
sb.From(fmt.Sprintf("%s.%s", spantypes.TraceDB, spantypes.TraceTable))
keys, err := s.genAIFieldKeys(ctx, orgID, aiobservabilitytypes.GenAIThreadKeys)
keys, err := s.genAIFieldKeys(ctx, orgID, aiobservabilitytypes.GenAISpanGateKeys)
if err != nil {
return nil, err
}
isThreadSpan, err := s.anyExistsCondition(ctx, q, aiobservabilitytypes.GenAIThreadKeys, keys, sb)
isThreadSpan, err := s.anyExistsCondition(ctx, q, aiobservabilitytypes.GenAISpanGateKeys, keys, sb)
if err != nil {
return nil, err
}

View File

@@ -220,14 +220,14 @@ func TestGetThreadSpans(t *testing.T) {
name: "FlagOff_ReadsAndFiltersLegacyMaps",
jsonRelease: &jsonInsideTrace,
selectSQL: selectSQL,
whereSQL: "(mapContains(attributes_string, 'gen_ai.input.messages') OR mapContains(attributes_string, 'gen_ai.output.messages') OR mapContains(attributes_string, 'gen_ai.tool.name'))",
whereSQL: "(mapContains(attributes_string, 'gen_ai.request.model') OR mapContains(attributes_string, 'gen_ai.tool.name') OR mapContains(attributes_string, 'gen_ai.agent.name'))",
},
{
name: "FlagOn_ReleasedDuringTrace_ReadsJSONFiltersJSONThenMaps",
jsonOn: true,
jsonRelease: &jsonInsideTrace,
selectSQL: selectSQL + ", attributes",
whereSQL: "(multiIf(attributes.`gen_ai.input.messages` IS NOT NULL, attributes.`gen_ai.input.messages`::String, mapContains(attributes_string, 'gen_ai.input.messages'), attributes_string['gen_ai.input.messages'], NULL) IS NOT NULL OR multiIf(attributes.`gen_ai.output.messages` IS NOT NULL, attributes.`gen_ai.output.messages`::String, mapContains(attributes_string, 'gen_ai.output.messages'), attributes_string['gen_ai.output.messages'], NULL) IS NOT NULL OR multiIf(attributes.`gen_ai.tool.name` IS NOT NULL, attributes.`gen_ai.tool.name`::String, mapContains(attributes_string, 'gen_ai.tool.name'), attributes_string['gen_ai.tool.name'], NULL) IS NOT NULL)",
whereSQL: "(multiIf(attributes.`gen_ai.request.model` IS NOT NULL, attributes.`gen_ai.request.model`::String, mapContains(attributes_string, 'gen_ai.request.model'), attributes_string['gen_ai.request.model'], NULL) IS NOT NULL OR multiIf(attributes.`gen_ai.tool.name` IS NOT NULL, attributes.`gen_ai.tool.name`::String, mapContains(attributes_string, 'gen_ai.tool.name'), attributes_string['gen_ai.tool.name'], NULL) IS NOT NULL OR multiIf(attributes.`gen_ai.agent.name` IS NOT NULL, attributes.`gen_ai.agent.name`::String, mapContains(attributes_string, 'gen_ai.agent.name'), attributes_string['gen_ai.agent.name'], NULL) IS NOT NULL)",
},
}

View File

@@ -17,8 +17,14 @@ const (
GenAIUsageCacheCreationInputTokens = "gen_ai.usage.cache_creation.input_tokens"
GenAIUsageReasoningOutputTokens = "gen_ai.usage.reasoning.output_tokens"
GenAIInputMessages = "gen_ai.input.messages"
GenAIOutputMessages = "gen_ai.output.messages"
GenAIInputMessages = "gen_ai.input.messages"
GenAIOutputMessages = "gen_ai.output.messages"
GenAISystemInstructions = "gen_ai.system_instructions"
GenAIResponseFinishReasons = "gen_ai.response.finish_reasons"
GenAIToolCallID = "gen_ai.tool.call.id"
GenAIToolCallArguments = "gen_ai.tool.call.arguments"
GenAIToolCallResult = "gen_ai.tool.call.result"
)
// Per-span costs the SigNoz LLM pricing processor attaches; SigNoz semconv
@@ -32,12 +38,10 @@ const (
)
// GenAISpanGateKeys mark a span as gen_ai: an LLM call, a tool call, or an
// agent span. A trace belongs to the AI explorer when any span carries one.
// agent span. A trace belongs to the AI explorer when any span carries one,
// and the thread of a trace is these spans.
var GenAISpanGateKeys = []string{GenAIRequestModel, GenAIToolName, GenAIAgentName}
// GenAIThreadKeys mark a span as a thread row: a model call with its messages, or a tool execution.
var GenAIThreadKeys = []string{GenAIInputMessages, GenAIOutputMessages, GenAIToolName}
// GenAISpanFilterExpression renders the gate as a query-builder filter
// expression: each gate key ORed on EXISTS.
func GenAISpanFilterExpression() string {

View File

@@ -1,5 +1,6 @@
// The message types mirror the OpenTelemetry GenAI message schemas, one type per schema definition:
// open-telemetry/semantic-conventions-genai@06ec68e, model/gen-ai/gen-ai-{input,output}-messages.json.
// https://github.com/open-telemetry/semantic-conventions-genai/blob/06ec68e722c45a7218e23ea1bc1339fe4e21ecae/model/gen-ai/gen-ai-input-messages.json
// https://github.com/open-telemetry/semantic-conventions-genai/blob/06ec68e722c45a7218e23ea1bc1339fe4e21ecae/model/gen-ai/gen-ai-output-messages.json
// Keys the schema does not name are dropped; the raw attribute on the span still has them.
package aiobservabilitytypes

View File

@@ -11,6 +11,7 @@ import (
const (
FormatterSemconv = "semconv"
FormatterTool = "tool"
FormatterOpenAIChat = "openai.chat"
FormatterText = "text"
FormatterGeneric = "generic"
@@ -18,9 +19,13 @@ const (
var dataURL = regexp.MustCompile(`^data:([^;,]+);base64,(.+)$`)
// Provider spellings of the finish reasons; the enum values themselves pass through.
// Provider spellings of the finish reasons; the enum values themselves pass through. tool_calls and
// function_call are OpenAI's, end_turn, stop_sequence and max_tokens Anthropic's, tool-calls the Vercel AI SDK's:
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L44782
// https://platform.claude.com/docs/en/api/messages
// https://github.com/vercel/ai/blob/a5e92fe3c692f8e152fe1eb9ef50a4e734ad0ffb/packages/provider/src/language-model/v2/language-model-v2-finish-reason.ts
var finishReasonAliases = map[string]FinishReason{
"tool_calls": FinishReasonToolCall, "function_call": FinishReasonToolCall,
"tool_calls": FinishReasonToolCall, "function_call": FinishReasonToolCall, "tool-calls": FinishReasonToolCall,
"end_turn": FinishReasonStop, "stop_sequence": FinishReasonStop,
"max_tokens": FinishReasonLength,
}
@@ -34,19 +39,45 @@ type FormattedMessages struct {
Warnings []string
}
// FormatMessages builds the OTel GenAI view of gen_ai.input.messages and gen_ai.output.messages from the
// formats SDKs write. Each value is a JSON string or a structured value; nil means absent.
func FormatMessages(input, output any) FormattedMessages {
f := &formatting{out: FormattedMessages{Input: InputMessages{}, Output: OutputMessages{}, Warnings: []string{}}}
// FormatMessages builds the OTel GenAI view of a span's messages from the formats SDKs write. attributes
// is the span's JSON attribute document, nested by key segment, so an object value is whole. Each
// message value is a JSON string or a structured value. A tool call and its result take precedence over
// messages beside them. gen_ai.system_instructions leads the input and gen_ai.response.finish_reasons
// overrides the finish reasons the converter found, whichever converter ran.
func FormatMessages(attributes map[string]any) FormattedMessages {
f := &formatting{attributes: attributes, out: FormattedMessages{Input: InputMessages{}, Output: OutputMessages{}, Warnings: []string{}}}
switch {
case attribute(f.attributes, GenAIToolCallArguments) != nil || attribute(f.attributes, GenAIToolCallResult) != nil:
f.tool()
case attribute(f.attributes, GenAIInputMessages) != nil || attribute(f.attributes, GenAIOutputMessages) != nil:
f.messages()
}
f.prependSystemInstructions()
f.fillFinishReasons()
return f.out
}
type formatting struct {
attributes map[string]any
out FormattedMessages
}
func (f *formatting) warn(format string, args ...any) {
f.out.Warnings = append(f.out.Warnings, fmt.Sprintf(format, args...))
}
// messages converts gen_ai.input.messages and gen_ai.output.messages, each by its own shape, and names
// the formatter after the input side, or the output side when the input has none.
func (f *formatting) messages() {
var inputLabel, outputLabel string
if input != nil {
if input := attribute(f.attributes, GenAIInputMessages); input != nil {
var msgs OutputMessages
msgs, inputLabel = f.side(decode(input), RoleUser)
for _, m := range msgs {
f.out.Input = append(f.out.Input, ChatMessage{Role: m.Role, Parts: m.Parts, Name: m.Name})
}
}
if output != nil {
if output := attribute(f.attributes, GenAIOutputMessages); output != nil {
f.out.Output, outputLabel = f.side(decode(output), RoleAssistant)
}
f.out.Formatter = inputLabel
@@ -55,15 +86,45 @@ func FormatMessages(input, output any) FormattedMessages {
} else if outputLabel != "" && outputLabel != inputLabel {
f.warn("output formatted as %s, input as %s", outputLabel, inputLabel)
}
return f.out
}
type formatting struct {
out FormattedMessages
// prependSystemInstructions adds gen_ai.system_instructions, a list of parts, as a leading system message,
// unless the input already starts with a system message of the same text: the OpenInference strands
// processor keeps the attribute and inserts it as message 0 as well. A span without messages, such as
// create_agent, stays without them.
// https://github.com/open-telemetry/semantic-conventions-genai/blob/06ec68e722c45a7218e23ea1bc1339fe4e21ecae/model/gen-ai/gen-ai-system-instructions.json
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-strands-agents/src/openinference/instrumentation/strands_agents/processor.py#L152
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-strands-agents/src/openinference/instrumentation/strands_agents/processor.py#L293
func (f *formatting) prependSystemInstructions() {
raw := attribute(f.attributes, GenAISystemInstructions)
if raw == nil || (len(f.out.Input) == 0 && len(f.out.Output) == 0) {
return
}
parts := systemParts(decode(raw))
if len(parts) == 0 || (len(f.out.Input) > 0 && f.out.Input[0].Role == RoleSystem && textOf(f.out.Input[0].Parts) == textOf(parts)) {
return
}
f.out.Input = append(InputMessages{{Role: RoleSystem, Parts: parts}}, f.out.Input...)
}
func (f *formatting) warn(format string, args ...any) {
f.out.Warnings = append(f.out.Warnings, fmt.Sprintf(format, args...))
// fillFinishReasons applies gen_ai.response.finish_reasons, one entry per output message, over
// whatever the converter found in the payload.
// https://github.com/open-telemetry/semantic-conventions-genai/blob/06ec68e722c45a7218e23ea1bc1339fe4e21ecae/model/gen-ai/registry.yaml#L210
func (f *formatting) fillFinishReasons() {
raw := attribute(f.attributes, GenAIResponseFinishReasons)
if raw == nil || len(f.out.Output) == 0 {
return
}
reasons := stringList(decode(raw))
if len(reasons) != len(f.out.Output) {
f.warn("gen_ai.response.finish_reasons has %d entries for %d output messages", len(reasons), len(f.out.Output))
return
}
for i, reason := range reasons {
if fr := finishReason(reason); fr != nil {
f.out.Output[i].FinishReason = fr
}
}
}
// side converts one attribute value by its shape; role is what a bare string on this side is
@@ -97,6 +158,21 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
return OutputMessages{genericMessage(value)}, FormatterGeneric
}
// attribute walks the document along the dotted key; a nested object comes back whole. Not handled: a
// scalar at a prefix of the path, such as gen_ai.input beside gen_ai.input.messages, which ends the
// walk early.
func attribute(doc map[string]any, key string) any {
var current any = doc
for segment := range strings.SplitSeq(key, ".") {
object, ok := current.(map[string]any)
if !ok {
return nil
}
current = object[segment]
}
return current
}
func firstObject(list []any) (map[string]any, bool) {
if len(list) == 0 {
return nil, false
@@ -155,6 +231,23 @@ func finishReason(value string) *FinishReason {
return &fr
}
// stringList reads a list of strings, or a bare string as a list of one.
func stringList(value any) []string {
switch v := value.(type) {
case []any:
out := make([]string, 0, len(v))
for _, item := range v {
out = append(out, stringOf(item))
}
return out
case string:
if v != "" {
return []string{v}
}
}
return nil
}
func optionalString(value any) *string {
s := stringOf(value)
if s == "" {
@@ -171,8 +264,37 @@ func textPart(content string) TextPart {
return TextPart{Type: PartTypeText, Content: content}
}
// contentPart converts one OpenAI content block, which semconv parts may also carry; a block of
// any other type is kept as sent.
// textOf joins the text parts, for comparing a system message with gen_ai.system_instructions.
func textOf(parts Parts) string {
var texts []string
for _, p := range parts {
if t, ok := p.Value.(TextPart); ok {
texts = append(texts, t.Content)
}
}
return strings.Join(texts, "\n")
}
// systemParts reads a system prompt given outside the history, a list of parts or a bare string.
func systemParts(value any) Parts {
switch v := value.(type) {
case []any:
parts := make(Parts, 0, len(v))
for _, item := range v {
parts = append(parts, contentPart(item))
}
return parts
case string:
if v != "" {
return Parts{part(textPart(v))}
}
}
return nil
}
// contentPart converts one OpenAI content block, which semconv parts may also carry; a text part
// in the semconv spelling has content where OpenAI has text. A block of any other type is kept
// as sent.
func contentPart(item any) Part {
p, ok := item.(map[string]any)
if !ok {
@@ -183,6 +305,9 @@ func contentPart(item any) Part {
}
switch stringOf(p["type"]) {
case "text":
if p["text"] == nil {
return part(textPart(stringOf(p["content"])))
}
return part(textPart(stringOf(p["text"])))
case "refusal":
return part(textPart(stringOf(p["refusal"])))

View File

@@ -170,7 +170,8 @@ func TestFormat_OpenAIChat(t *testing.T) {
]`,
},
{
// openinference capture, input.value of the ChatCompletion span
// ChatCompletionRequestMessageContentPartFile {file: {file_id}}, as OpenInference records the request in input.value:
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L42650
name: "FileByID_FilePart",
input: `{"model": "gpt-4o-mini", "messages": [{"role": "user", "content": [{"type": "text", "text": "Summarise this document in one sentence."}, {"type": "file", "file": {"file_id": "file-6F2ksmvXxt4VdoqmHRw6kL"}}]}], "max_tokens": 60}`,
output: `{"id":"chatcmpl-DktFileCapture0001","object":"chat.completion","model":"gpt-4o-mini-2024-07-18","choices":[{"index":0,"finish_reason":"stop","logprobs":null,"message":{"role":"assistant","content":"The document is a short PDF about dragons.","refusal":null,"annotations":[]}}]}`,
@@ -193,7 +194,8 @@ func TestFormat_OpenAIChat(t *testing.T) {
]`,
},
{
// openinference capture, input.value of the ChatCompletion span
// ChatCompletionRequestMessageContentPartFile {file: {filename, file_data}}, file_data a data URL:
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L42650
name: "FileInline_BlobPart",
input: `{"model": "gpt-4o-mini", "messages": [{"role": "user", "content": [{"type": "text", "text": "Summarise this document in one sentence."}, {"type": "file", "file": {"filename": "draconomicon.pdf", "file_data": "data:application/pdf;base64,JVBERi0xLjQKMSAwIG9iajw8L1R5cGUvQ2F0YWxvZy9QYWdlcyAyIDAgUj4+ZW5kb2JqCjIgMCBvYmo8PC9UeXBlL1BhZ2VzL0tpZHNbXS9Db3VudCAwPj5lbmRvYmoKdHJhaWxlcjw8L1Jvb3QgMSAwIFI+PgolJUVPRgo="}}]}], "max_tokens": 60}`,
formatter: FormatterOpenAIChat,

View File

@@ -6,15 +6,20 @@ import (
)
// Format: the OTel shape itself, [{role, parts}].
// Written by: litellm, langchain, openllmetry, portkey, traceloop LLM spans.
// Decoding is the genai types. Repairs: "text" used for "content" on text parts; OpenAI content
// blocks left inside parts.
// Not yet: "thinking" for "reasoning"; "result" or "output" for "response" on tool call responses.
// Written by: litellm, langchain, openllmetry, portkey, traceloop LLM spans, pydantic-ai.
// Decoding is the genai types. Repairs: "text" used for "content" on text parts; OpenAI content blocks
// left inside parts; tool call arguments as a JSON string and "result" for "response", as pydantic-ai
// writes them:
// https://github.com/pydantic/pydantic-ai/blob/9ac853bb84ffed7ba0edccb026c5b4f3228c1de7/pydantic_ai_slim/pydantic_ai/messages.py#L3130
// https://github.com/pydantic/pydantic-ai/blob/9ac853bb84ffed7ba0edccb026c5b4f3228c1de7/pydantic_ai_slim/pydantic_ai/messages.py#L1724
// The choice finish_reason as the provider spells it, as litellm copies it:
// https://github.com/BerriAI/litellm/blob/9d72bae336ec0d14c41651adcb0d2e1149721f46/litellm/integrations/opentelemetry.py#L2822
// Not yet: "thinking" for "reasoning".
// semconv decodes messages that already follow the schema through the genai types. A part of a
// type the schema does not name is a provider content block, which contentPart converts.
func (f *formatting) semconv(list []any) OutputMessages {
data, err := json.Marshal(semconvTextKey(list))
data, err := json.Marshal(semconvRepairs(list))
if err != nil {
f.warn("messages could not be encoded: %s", err)
return OutputMessages{genericMessage(list)}
@@ -28,6 +33,9 @@ func (f *formatting) semconv(list []any) OutputMessages {
if msgs[i].Parts == nil {
msgs[i].Parts = Parts{}
}
if msgs[i].FinishReason != nil {
msgs[i].FinishReason = finishReason(msgs[i].FinishReason.StringValue())
}
for j, p := range msgs[i].Parts {
if block, ok := p.Value.(GenericPart); ok {
msgs[i].Parts[j] = contentPart(map[string]any(block))
@@ -37,9 +45,9 @@ func (f *formatting) semconv(list []any) OutputMessages {
return msgs
}
// semconvTextKey copies the text and reasoning parts some SDKs write with "text" in place of the
// schema's "content". The input is left untouched, it is also the span's raw attribute.
func semconvTextKey(list []any) []any {
// semconvRepairs copies the parts whose keys SDKs misspell into the schema's spelling. The input
// is left untouched, it is also the span's raw attribute.
func semconvRepairs(list []any) []any {
out := make([]any, 0, len(list))
for _, item := range list {
m, ok := item.(map[string]any)
@@ -50,13 +58,8 @@ func semconvTextKey(list []any) []any {
}
fixedParts := make([]any, 0, len(parts))
for _, p := range parts {
part, ok := p.(map[string]any)
typ := stringOf(part["type"])
if ok && (typ == "text" || typ == "reasoning") && part["content"] == nil && part["text"] != nil {
fixed := maps.Clone(part)
fixed["content"] = fixed["text"]
delete(fixed, "text")
p = fixed
if part, ok := p.(map[string]any); ok {
p = semconvRepairPart(part)
}
fixedParts = append(fixedParts, p)
}
@@ -66,3 +69,24 @@ func semconvTextKey(list []any) []any {
}
return out
}
func semconvRepairPart(part map[string]any) map[string]any {
fixed := maps.Clone(part)
switch stringOf(part["type"]) {
case "text", "reasoning":
if part["content"] == nil && part["text"] != nil {
fixed["content"] = fixed["text"]
delete(fixed, "text")
}
case "tool_call":
if args, ok := part["arguments"].(string); ok {
fixed["arguments"] = decode(args)
}
case "tool_call_response":
if _, ok := part["response"]; !ok && part["result"] != nil {
fixed["response"] = part["result"]
delete(fixed, "result")
}
}
return fixed
}

View File

@@ -161,3 +161,36 @@ func TestFormat_Semconv(t *testing.T) {
},
})
}
func TestFormat_Semconv_Repairs(t *testing.T) {
runFormatCases(t, []formatCase{
{
// pydantic-ai keeps string tool call arguments as sent, writes the tool result under result, and the
// instructions as a gen_ai.system_instructions text part:
// https://github.com/pydantic/pydantic-ai/blob/9ac853bb84ffed7ba0edccb026c5b4f3228c1de7/pydantic_ai_slim/pydantic_ai/messages.py#L3130
// https://github.com/pydantic/pydantic-ai/blob/9ac853bb84ffed7ba0edccb026c5b4f3228c1de7/pydantic_ai_slim/pydantic_ai/messages.py#L1724
// https://github.com/pydantic/pydantic-ai/blob/9ac853bb84ffed7ba0edccb026c5b4f3228c1de7/pydantic_ai_slim/pydantic_ai/models/instrumented.py#L292
name: "PydanticAI_ArgumentsDecoded_ResultBecomesResponse_InstructionsLead",
input: `[{"role":"user","parts":[{"type":"text","content":"Weather in Paris?"}]},{"role":"assistant","parts":[{"type":"tool_call","id":"call_stub1","name":"get_weather","arguments":"{\"city\": \"Paris\"}"}],"finish_reason":"tool_call"},{"role":"user","parts":[{"type":"tool_call_response","id":"call_stub1","name":"get_weather","result":{"city":"Paris","temp_c":18}}]}]`,
output: `[{"role":"assistant","parts":[{"type":"text","content":"Paris is clear at 18C."}],"finish_reason":"stop"}]`,
attributes: map[string]any{GenAISystemInstructions: `[{"type":"text","content":"Use the tool."}]`},
formatter: FormatterSemconv,
wantInput: `[
{"role": "system", "parts": [{"type": "text", "content": "Use the tool."}]},
{"role": "user", "parts": [{"type": "text", "content": "Weather in Paris?"}]},
{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_stub1", "arguments": {"city": "Paris"}}]},
{"role": "user", "parts": [{"type": "tool_call_response", "id": "call_stub1", "response": {"city": "Paris", "temp_c": 18}}]}
]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Paris is clear at 18C."}], "finish_reason": "stop"}]`,
},
{
// litellm copies the choice finish_reason into the semconv message as OpenAI spells it:
// https://github.com/BerriAI/litellm/blob/9d72bae336ec0d14c41651adcb0d2e1149721f46/litellm/integrations/opentelemetry.py#L2822
name: "LiteLLMFinishReasonSpelling_Normalised",
output: `[{"role": "assistant", "parts": [], "finish_reason": "tool_calls"}]`,
formatter: FormatterSemconv,
wantInput: `[]`,
wantOutput: `[{"role": "assistant", "parts": [], "finish_reason": "tool_call"}]`,
},
})
}

View File

@@ -2,6 +2,8 @@ package aiobservabilitytypes
import (
"encoding/json"
"maps"
"strings"
"testing"
"github.com/stretchr/testify/assert"
@@ -12,6 +14,7 @@ type formatCase struct {
name string
input any
output any
attributes map[string]any // flat dotted keys beside the two message attributes
formatter string
warnings []string
wantInput string
@@ -22,7 +25,9 @@ func runFormatCases(t *testing.T, testCases []formatCase) {
t.Helper()
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
formatted := FormatMessages(testCase.input, testCase.output)
attributes := map[string]any{GenAIInputMessages: testCase.input, GenAIOutputMessages: testCase.output}
maps.Copy(attributes, testCase.attributes)
formatted := FormatMessages(document(attributes))
assert.Equal(t, testCase.formatter, formatted.Formatter)
assert.Equal(t, append([]string{}, testCase.warnings...), formatted.Warnings)
@@ -36,6 +41,25 @@ func runFormatCases(t *testing.T, testCases []formatCase) {
}
}
// document nests flat dotted keys the way the JSON column stores them.
func document(flat map[string]any) map[string]any {
doc := map[string]any{}
for key, value := range flat {
segments := strings.Split(key, ".")
current := doc
for _, segment := range segments[:len(segments)-1] {
child, ok := current[segment].(map[string]any)
if !ok {
child = map[string]any{}
current[segment] = child
}
current = child
}
current[segments[len(segments)-1]] = value
}
return doc
}
func TestFormat(t *testing.T) {
runFormatCases(t, []formatCase{
{
@@ -93,9 +117,96 @@ So, the ball costs 5 cents.`,
wantOutput: `[]`,
},
{
name: "NoAttributes_EmptyListsNoFormatter",
name: "NoMessageAttributes_EmptyListsNoFormatter",
attributes: map[string]any{GenAIRequestModel: "gpt-4o", GenAIToolName: "get_weather"},
wantInput: `[]`,
wantOutput: `[]`,
},
})
}
func TestFormat_SystemInstructions(t *testing.T) {
input := `[{"role":"user","parts":[{"type":"text","content":"hi"}]}]`
runFormatCases(t, []formatCase{
{
// the langchain instrumentation moves system messages out of the history into gen_ai.system_instructions,
// a JSON list of parts:
// https://github.com/traceloop/openllmetry/blob/93429cfb9767e6b51db6cb35f6b70f758117669a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/span_utils.py#L304
name: "ListOfParts_LeadsInput",
input: input,
attributes: map[string]any{GenAISystemInstructions: `[{"type": "text", "content": "You are concise."}]`},
formatter: FormatterSemconv,
wantInput: `[
{"role": "system", "parts": [{"type": "text", "content": "You are concise."}]},
{"role": "user", "parts": [{"type": "text", "content": "hi"}]}
]`,
wantOutput: `[]`,
},
{
// the OpenInference strands processor keeps gen_ai.system_instructions on the span and inserts the
// same prompt as message 0
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-strands-agents/src/openinference/instrumentation/strands_agents/processor.py#L152
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-strands-agents/src/openinference/instrumentation/strands_agents/processor.py#L293
name: "SameTextLeadsInput_Skipped",
input: `[{"role":"system","parts":[{"type":"text","content":"You are concise."}]},{"role":"user","parts":[{"type":"text","content":"hi"}]}]`,
attributes: map[string]any{GenAISystemInstructions: `[{"type": "text", "content": "You are concise."}]`},
formatter: FormatterSemconv,
wantInput: `[
{"role": "system", "parts": [{"type": "text", "content": "You are concise."}]},
{"role": "user", "parts": [{"type": "text", "content": "hi"}]}
]`,
wantOutput: `[]`,
},
{
name: "BareString_TextPart",
input: input,
attributes: map[string]any{GenAISystemInstructions: "be brief"},
formatter: FormatterSemconv,
wantInput: `[
{"role": "system", "parts": [{"type": "text", "content": "be brief"}]},
{"role": "user", "parts": [{"type": "text", "content": "hi"}]}
]`,
wantOutput: `[]`,
},
{
// a create_agent span carries instructions alone
name: "NoMessages_Nothing",
attributes: map[string]any{GenAISystemInstructions: "be brief"},
wantInput: `[]`,
wantOutput: `[]`,
},
})
}
func TestFormat_FinishReasons(t *testing.T) {
output := `[{"role":"assistant","parts":[{"type":"text","content":"a"}],"finish_reason":"stop"},{"role":"assistant","parts":[{"type":"text","content":"b"}]}]`
runFormatCases(t, []formatCase{
{
name: "List_OverridesMessages",
output: output,
attributes: map[string]any{GenAIResponseFinishReasons: []any{"tool_calls", "length"}},
formatter: FormatterSemconv,
wantInput: `[]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "a"}], "finish_reason": "tool_call"}, {"role": "assistant", "parts": [{"type": "text", "content": "b"}], "finish_reason": "length"}]`,
},
{
name: "LengthMismatch_LeavesMessages_Warns",
output: output,
attributes: map[string]any{GenAIResponseFinishReasons: []any{"stop"}},
formatter: FormatterSemconv,
warnings: []string{"gen_ai.response.finish_reasons has 1 entries for 2 output messages"},
wantInput: `[]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "a"}], "finish_reason": "stop"}, {"role": "assistant", "parts": [{"type": "text", "content": "b"}]}]`,
},
{
// llm.finish_reason copied by the mapper as a bare string, in the Vercel AI SDK spelling:
// https://github.com/vercel/ai/blob/a5e92fe3c692f8e152fe1eb9ef50a4e734ad0ffb/packages/provider/src/language-model/v2/language-model-v2-finish-reason.ts
name: "BareString_SingleChoice",
output: `{"choices":[{"finish_reason":"stop","message":{"role":"assistant","content":"hi"}}]}`,
attributes: map[string]any{GenAIResponseFinishReasons: "tool-calls"},
formatter: FormatterOpenAIChat,
wantInput: `[]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "hi"}], "finish_reason": "tool_call"}]`,
},
})
}

View File

@@ -0,0 +1,72 @@
package aiobservabilitytypes
// Format: gen_ai.tool.call.arguments and gen_ai.tool.call.result on an execute_tool span, each an
// object or its JSON string:
// https://github.com/open-telemetry/semantic-conventions-genai/blob/06ec68e722c45a7218e23ea1bc1339fe4e21ecae/model/gen-ai/registry.yaml#L545
// Written by: opentelemetry-instrumentation-langchain, traceloop @tool, OpenInference TOOL spans and
// Vercel ai.toolCall.* through the span mapper.
// Envelopes unwrapped: LangChain's on_tool_start and on_tool_end payloads, {input_str, tags, metadata,
// inputs, kwargs} and {output, kwargs}:
// https://github.com/traceloop/openllmetry/blob/93429cfb9767e6b51db6cb35f6b70f758117669a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py#L837
// The traceloop decorator's {args, kwargs}:
// https://github.com/traceloop/openllmetry/blob/93429cfb9767e6b51db6cb35f6b70f758117669a/packages/traceloop-sdk/traceloop/sdk/decorators/base.py#L169
// tool renders the call as an assistant tool_call and the result as a tool tool_call_response, both
// carrying gen_ai.tool.call.id so the pair can be matched.
func (f *formatting) tool() {
id := optionalString(attribute(f.attributes, GenAIToolCallID))
if raw := attribute(f.attributes, GenAIToolCallArguments); raw != nil {
call := ToolCallRequestPart{Type: PartTypeToolCall, Name: stringOf(attribute(f.attributes, GenAIToolName)), ID: id, Arguments: toolArguments(decode(raw))}
f.out.Input = InputMessages{{Role: RoleAssistant, Parts: Parts{part(call)}}}
}
if raw := attribute(f.attributes, GenAIToolCallResult); raw != nil {
result := ToolCallResponsePart{Type: PartTypeToolCallResponse, ID: id, Response: toolResult(decode(raw))}
f.out.Output = OutputMessages{{Role: RoleTool, Parts: Parts{part(result)}}}
}
f.out.Formatter = FormatterTool
}
func toolArguments(value any) any {
m, ok := value.(map[string]any)
if !ok {
return value
}
if _, isLangChain := m["input_str"]; isLangChain && hasKeys(m, "inputs", "kwargs") {
if inputs := m["inputs"]; inputs != nil {
return inputs
}
return decode(m["input_str"])
}
if len(m) == 2 && hasKeys(m, "args", "kwargs") {
args, _ := m["args"].([]any)
kwargs, _ := m["kwargs"].(map[string]any)
switch {
case len(args) == 0:
return m["kwargs"]
case len(kwargs) == 0:
return args
}
}
return value
}
// toolResult unwraps the LangChain envelope; the traceloop decorator encodes a tool returning
// nothing as the literal null.
func toolResult(value any) any {
if m, ok := value.(map[string]any); ok && len(m) == 2 && hasKeys(m, "output", "kwargs") {
return decode(m["output"])
}
if value == "null" {
return nil
}
return value
}
func hasKeys(m map[string]any, keys ...string) bool {
for _, key := range keys {
if _, ok := m[key]; !ok {
return false
}
}
return true
}

View File

@@ -0,0 +1,112 @@
package aiobservabilitytypes
import "testing"
// Captured cases are the tool span attributes as the collector stored them for the SigNoz/scripts
// static-telemetry-generator captures, after the span mapper filled the gen_ai.tool.call.* keys.
func TestFormat_Tool(t *testing.T) {
runFormatCases(t, []formatCase{
{
// gen_ai.tool.call.arguments and gen_ai.tool.call.result written by the langchain instrumentation itself,
// wrapped in its on_tool_start and on_tool_end payloads:
// https://github.com/traceloop/openllmetry/blob/93429cfb9767e6b51db6cb35f6b70f758117669a/packages/opentelemetry-instrumentation-langchain/opentelemetry/instrumentation/langchain/callback_handler.py#L837
name: "LangChainEnvelopes_InputsAndOutputUnwrapped",
attributes: map[string]any{
GenAIToolName: "get_weather",
GenAIToolCallArguments: `{"input_str": "{'city': 'Bengaluru'}", "tags": [], "metadata": {}, "inputs": {"city": "Bengaluru"}, "kwargs": {"color": "green", "name": null, "tool_call_id": null}}`,
GenAIToolCallResult: `{"output": "{\"city\": \"Bengaluru\", \"temp_c\": 18, \"summary\": \"Clear\"}", "kwargs": {"tags": [], "color": "green", "name": "get_weather"}}`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "arguments": {"city": "Bengaluru"}}]}]`,
wantOutput: `[{"role": "tool", "parts": [{"type": "tool_call_response", "response": {"city": "Bengaluru", "temp_c": 18, "summary": "Clear"}}]}]`,
},
{
// traceloop.entity.input and traceloop.entity.output from the @tool decorator, renamed by the mapper;
// the tool returned a JSON string, which json.dumps encoded again:
// https://github.com/traceloop/openllmetry/blob/93429cfb9767e6b51db6cb35f6b70f758117669a/packages/traceloop-sdk/traceloop/sdk/decorators/base.py#L169
name: "TraceloopDecorator_KwargsUnwrapped_ResultDecodedTwice",
attributes: map[string]any{
GenAIToolName: "get_weather",
GenAIToolCallArguments: `{"args": [], "kwargs": {"city": "Bengaluru"}}`,
GenAIToolCallResult: `"{\"city\": \"Bengaluru\", \"temp_c\": 28, \"summary\": \"Partly cloudy\"}"`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "arguments": {"city": "Bengaluru"}}]}]`,
wantOutput: `[{"role": "tool", "parts": [{"type": "tool_call_response", "response": {"city": "Bengaluru", "temp_c": 28, "summary": "Partly cloudy"}}]}]`,
},
{
// traceloop.entity.output of a @tool returning None, json.dumps(None), renamed by the mapper:
// https://github.com/traceloop/openllmetry/blob/93429cfb9767e6b51db6cb35f6b70f758117669a/packages/traceloop-sdk/traceloop/sdk/decorators/base.py#L185
name: "TraceloopDecorator_PositionalArgsOnly_ListKept",
attributes: map[string]any{
GenAIToolName: "lookup",
GenAIToolCallArguments: `{"args": ["OpenAI", "What's the weather in Paris?"], "kwargs": {}}`,
GenAIToolCallResult: `null`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "lookup", "arguments": ["OpenAI", "What's the weather in Paris?"]}]}]`,
wantOutput: `[{"role": "tool", "parts": [{"type": "tool_call_response", "response": null}]}]`,
},
{
name: "TraceloopDecorator_BothArgsAndKwargs_EnvelopeKept",
attributes: map[string]any{
GenAIToolName: "lookup",
GenAIToolCallArguments: `{"args": ["Paris"], "kwargs": {"unit": "c"}}`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "lookup", "arguments": {"args": ["Paris"], "kwargs": {"unit": "c"}}}]}]`,
wantOutput: `[]`,
},
{
// input.value and output.value of an OpenInference TOOL span, renamed by the mapper:
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/traces.md
name: "OpenInferenceToolSpan_PlainJSONAndText",
attributes: map[string]any{
GenAIToolName: "get_weather",
GenAIToolCallArguments: `{"city":"Bangalore"}`,
GenAIToolCallResult: `27C, partly cloudy, light breeze from the south.`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "arguments": {"city": "Bangalore"}}]}]`,
wantOutput: `[{"role": "tool", "parts": [{"type": "tool_call_response", "response": "27C, partly cloudy, light breeze from the south."}]}]`,
},
{
// ai.toolCall.{id,name,args,result} of a Vercel ai.toolCall span, renamed by the mapper:
// https://ai-sdk.dev/docs/ai-sdk-core/telemetry
name: "VercelToolCall_IDOnBothParts",
attributes: map[string]any{
GenAIToolName: "get_current_weather",
GenAIToolCallID: "call_l0ycKcmt1uPCdGWR4SkwoLBv",
GenAIToolCallArguments: `{"city":"Bengaluru","unit":"c"}`,
GenAIToolCallResult: `{"city":"Bengaluru","unit":"c","temperature":27,"condition":"partly cloudy"}`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_current_weather", "id": "call_l0ycKcmt1uPCdGWR4SkwoLBv", "arguments": {"city": "Bengaluru", "unit": "c"}}]}]`,
wantOutput: `[{"role": "tool", "parts": [{"type": "tool_call_response", "id": "call_l0ycKcmt1uPCdGWR4SkwoLBv", "response": {"city": "Bengaluru", "unit": "c", "temperature": 27, "condition": "partly cloudy"}}]}]`,
},
{
// the object form the OTel registry asks for, whole in the JSON column
name: "StructuredArguments_ObjectKept",
attributes: map[string]any{
GenAIToolName: "get_weather",
GenAIToolCallID: "call_1",
GenAIToolCallArguments: map[string]any{"city": "Paris", "days": []any{1.0, 2.0}},
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_1", "arguments": {"city": "Paris", "days": [1, 2]}}]}]`,
wantOutput: `[]`,
},
{
// the mapper groups never fill both; the tool call wins when an SDK does
name: "MessagesBesideToolCall_ToolCallFormatted",
attributes: map[string]any{
GenAIInputMessages: `[{"role": "user", "parts": [{"type": "text", "content": "hi"}]}]`,
GenAIToolName: "lookup",
GenAIToolCallArguments: `{"city": "Paris"}`,
},
formatter: FormatterTool,
wantInput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "lookup", "arguments": {"city": "Paris"}}]}]`,
wantOutput: `[]`,
},
})
}

View File

@@ -4,7 +4,6 @@ import (
"encoding/base64"
"encoding/json"
"maps"
"strings"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/types/aiobservabilitytypes"
@@ -180,7 +179,9 @@ func newThreadSpan(traceID string, storable *StorableSpan) *ThreadSpan {
maps.Copy(resources, storable.ResourcesString)
timeUnixNano := uint64(storable.StartTime.UnixNano())
attributes := threadAttributes(storable)
formatted := aiobservabilitytypes.FormatMessages(rawAttribute(storable, aiobservabilitytypes.GenAIInputMessages), rawAttribute(storable, aiobservabilitytypes.GenAIOutputMessages))
// Spans from before the JSON rollout have no document and are not decoded: the legacy maps
// hold an object split into one key per field.
formatted := aiobservabilitytypes.FormatMessages(storable.AttributesJSON)
return &ThreadSpan{
SpanID: storable.SpanID,
TraceID: traceID,
@@ -214,20 +215,3 @@ func threadAttributes(storable *StorableSpan) map[string]any {
storable.AttributesJSON.FlattenInto("", attributes)
return attributes
}
// rawAttribute reads one attribute for decoding from the JSON document, where an object value is
// still whole. The legacy maps hold objects split into one key per field, so spans from before
// the JSON rollout are not decoded. Not handled: a list of JSON strings, which formats as generic,
// and a scalar at a prefix of the path, such as gen_ai.input beside gen_ai.input.messages, which
// ends the walk early.
func rawAttribute(storable *StorableSpan, key string) any {
var current any = map[string]any(storable.AttributesJSON)
for segment := range strings.SplitSeq(key, ".") {
object, ok := current.(map[string]any)
if !ok {
return nil
}
current = object[segment]
}
return current
}

View File

@@ -73,35 +73,3 @@ func TestThreadAttributes(t *testing.T) {
})
}
}
func TestRawAttribute(t *testing.T) {
arguments := map[string]any{"city": "Paris", "days": []any{1.0, 2.0}}
testCases := []struct {
name string
storable StorableSpan
key string
want any
}{
{
name: "JSONColumn_ObjectReturnedWhole",
storable: StorableSpan{AttributesJSON: telemetrystoretypes.JSONValue{"gen_ai": map[string]any{"tool": map[string]any{"call": map[string]any{"arguments": arguments}}}}},
key: "gen_ai.tool.call.arguments",
want: arguments,
},
{
name: "JSONColumn_MissingPath_Nil",
storable: StorableSpan{AttributesJSON: telemetrystoretypes.JSONValue{"gen_ai": map[string]any{"tool": map[string]any{"name": "get_weather"}}}},
key: "gen_ai.tool.call.arguments",
},
{
name: "LegacyMapsOnly_Nil",
storable: StorableSpan{AttributesString: map[string]string{"gen_ai.output.messages": "sunny", "gen_ai.tool.call.arguments.city": "Paris"}},
key: "gen_ai.output.messages",
},
}
for _, testCase := range testCases {
t.Run(testCase.name, func(t *testing.T) {
assert.Equal(t, testCase.want, rawAttribute(&testCase.storable, testCase.key))
})
}
}

View File

@@ -11,7 +11,7 @@ from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.traces import ATTRIBUTE_JSON_ROLLOUT_TIME, TraceIdGenerator, Traces, TracesKind
def test_thread_returns_message_spans_in_order(
def test_thread_returns_gen_ai_spans_in_order(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
@@ -24,14 +24,22 @@ def test_thread_returns_message_spans_in_order(
root_id, first_llm_id, tool_id, agent_id, second_llm_id, third_llm_id = (TraceIdGenerator.span_id() for _ in range(6))
resources = {"service.name": "tracedetail-thread"}
first_input = json.dumps([{"role": "user", "parts": [{"type": "text", "content": "weather in Bangalore?"}]}])
first_output = json.dumps([{"role": "assistant", "parts": [{"type": "tool_call", "id": "call_1", "name": "get_weather", "arguments": {"city": "Bangalore"}}], "finish_reason": "tool_call"}])
first_output = json.dumps([{"role": "assistant", "parts": [{"type": "tool_call", "id": "call_1", "name": "get_weather", "arguments": {"city": "Bangalore"}}]}])
first_instructions = json.dumps([{"type": "text", "content": "Answer with the tool."}])
second_input = json.dumps([{"role": "tool", "content": "sunny", "tool_call_id": "call_1"}])
insert_traces(
[
Traces(timestamp=now - timedelta(seconds=10), duration=timedelta(seconds=9), trace_id=trace_id, span_id=root_id, name="POST /chat", kind=TracesKind.SPAN_KIND_SERVER, resources=resources, attribute_write_mode="json_only"),
Traces(
timestamp=now - timedelta(seconds=8), trace_id=trace_id, span_id=first_llm_id, parent_span_id=root_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": first_input, "gen_ai.output.messages": first_output}, attribute_write_mode="json_only"
timestamp=now - timedelta(seconds=8),
trace_id=trace_id,
span_id=first_llm_id,
parent_span_id=root_id,
name="chat gpt-4o",
resources=resources,
attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.system_instructions": first_instructions, "gen_ai.input.messages": first_input, "gen_ai.output.messages": first_output, "gen_ai.response.finish_reasons": json.dumps(["tool_calls"])},
attribute_write_mode="json_only",
),
Traces(
timestamp=now - timedelta(seconds=6),
@@ -54,10 +62,14 @@ def test_thread_returns_message_spans_in_order(
assert response.status_code == HTTPStatus.OK, response.text
thread = response.json()["data"]
assert [span["span_id"] for span in thread["spans"]] == [first_llm_id, tool_id, second_llm_id, third_llm_id]
assert [span["span_id"] for span in thread["spans"]] == [first_llm_id, tool_id, agent_id, second_llm_id, third_llm_id]
assert "nextCursor" not in thread
first, tool, input_only, output_only = thread["spans"]
first, tool, agent, input_only, output_only = thread["spans"]
assert agent["attributes"]["gen_ai.agent.name"] == "planner"
assert agent["formatter"] == ""
assert agent["formatted_input"] == []
assert agent["formatted_output"] == []
assert tool["attributes"]["gen_ai.tool.call.arguments"] == json.dumps({"city": "Bangalore"})
assert tool["attributes"]["gen_ai.tool.call.result"] == "sunny"
assert first["time_unix"] == int((now - timedelta(seconds=8)).timestamp() * 1000)
@@ -71,13 +83,12 @@ def test_thread_returns_message_spans_in_order(
tool_call = {"type": "tool_call", "name": "get_weather", "id": "call_1", "arguments": {"city": "Bangalore"}}
assert first["formatter"] == "semconv"
assert first["formatted_input"] == [{"role": "user", "parts": [{"type": "text", "content": "weather in Bangalore?"}]}]
assert first["formatted_input"] == [{"role": "system", "parts": [{"type": "text", "content": "Answer with the tool."}]}, {"role": "user", "parts": [{"type": "text", "content": "weather in Bangalore?"}]}]
assert first["formatted_output"] == [{"role": "assistant", "parts": [tool_call], "finish_reason": "tool_call"}]
assert first["formatter_warnings"] == []
# tool spans carry no messages until their converter lands
assert tool["formatter"] == ""
assert tool["formatted_input"] == []
assert tool["formatted_output"] == []
assert tool["formatter"] == "tool"
assert tool["formatted_input"] == [{"role": "assistant", "parts": [tool_call]}]
assert tool["formatted_output"] == [{"role": "tool", "parts": [{"type": "tool_call_response", "id": "call_1", "response": "sunny"}]}]
assert input_only["formatter"] == "openai.chat"
assert input_only["formatted_input"] == [{"role": "tool", "parts": [{"type": "tool_call_response", "id": "call_1", "response": "sunny"}]}]
assert input_only["formatted_output"] == []
@@ -102,7 +113,7 @@ def test_thread_paginates_with_cursors(
timestamps = [now - timedelta(seconds=6), now - timedelta(seconds=3), now - timedelta(seconds=3)]
insert_traces(
[
Traces(timestamp=timestamp, trace_id=trace_id, span_id=span_id, name="chat gpt-4o", resources={"service.name": "tracedetail-thread-pages"}, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": span_id}])}, attribute_write_mode="json_only")
Traces(timestamp=timestamp, trace_id=trace_id, span_id=span_id, name="chat gpt-4o", resources={"service.name": "tracedetail-thread-pages"}, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": json.dumps([{"role": "user", "content": span_id}])}, attribute_write_mode="json_only")
for span_id, timestamp in zip(span_ids, timestamps, strict=True)
]
)
@@ -164,7 +175,7 @@ def test_thread_paginates_across_buckets(
span_ids = [TraceIdGenerator.span_id() for _ in timestamps]
insert_traces(
[
Traces(timestamp=timestamp, trace_id=trace_id, span_id=span_id, name="chat gpt-4o", resources={"service.name": "tracedetail-thread-buckets"}, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": span_id}])}, attribute_write_mode="json_only")
Traces(timestamp=timestamp, trace_id=trace_id, span_id=span_id, name="chat gpt-4o", resources={"service.name": "tracedetail-thread-buckets"}, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": json.dumps([{"role": "user", "content": span_id}])}, attribute_write_mode="json_only")
for span_id, timestamp in zip(span_ids, timestamps, strict=True)
]
)
@@ -209,16 +220,25 @@ def test_thread_opens_around_span(
resources = {"service.name": "tracedetail-thread-anchor"}
root_id = TraceIdGenerator.span_id()
llm_ids = [TraceIdGenerator.span_id() for _ in range(5)]
agent_id = TraceIdGenerator.span_id()
# agent span, not a thread row, sits between the third and fourth llm spans
http_id = TraceIdGenerator.span_id()
# http span, not a thread row, sits between the third and fourth llm spans
insert_traces(
[
Traces(timestamp=now - timedelta(seconds=20), duration=timedelta(seconds=19), trace_id=trace_id, span_id=root_id, name="POST /chat", kind=TracesKind.SPAN_KIND_SERVER, resources=resources, attribute_write_mode="json_only"),
*(
Traces(timestamp=now - timedelta(seconds=18 - 3 * i), trace_id=trace_id, span_id=span_id, parent_span_id=root_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": span_id}])}, attribute_write_mode="json_only")
Traces(
timestamp=now - timedelta(seconds=18 - 3 * i),
trace_id=trace_id,
span_id=span_id,
parent_span_id=root_id,
name="chat gpt-4o",
resources=resources,
attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": json.dumps([{"role": "user", "content": span_id}])},
attribute_write_mode="json_only",
)
for i, span_id in enumerate(llm_ids)
),
Traces(timestamp=now - timedelta(seconds=11), trace_id=trace_id, span_id=agent_id, parent_span_id=root_id, name="invoke_agent planner", resources=resources, attributes={"gen_ai.agent.name": "planner"}, attribute_write_mode="json_only"),
Traces(timestamp=now - timedelta(seconds=11), trace_id=trace_id, span_id=http_id, parent_span_id=root_id, name="GET /weather", resources=resources, attributes={"http.request.method": "GET"}, attribute_write_mode="json_only"),
]
)
@@ -231,7 +251,7 @@ def test_thread_opens_around_span(
assert response.status_code == HTTPStatus.OK, f"{params}: {response.text}"
return response.json()["data"]
# span with messages: included with its neighbours
# thread span: included with its neighbours
around = get_page({"limit": 3, "spanId": llm_ids[2]})
assert [span["span_id"] for span in around["spans"]] == llm_ids[1:4]
assert around["prevCursor"]
@@ -239,11 +259,11 @@ def test_thread_opens_around_span(
assert [span["span_id"] for span in get_page({"limit": 3, "before": around["prevCursor"]})["spans"]] == llm_ids[:1]
assert [span["span_id"] for span in get_page({"limit": 3, "after": around["nextCursor"]})["spans"]] == llm_ids[4:]
# span without messages: only its neighbours
around_agent = get_page({"limit": 2, "spanId": agent_id})
assert [span["span_id"] for span in around_agent["spans"]] == llm_ids[2:4]
assert around_agent["prevCursor"]
assert around_agent["nextCursor"]
# span outside the thread: only its neighbours
around_http = get_page({"limit": 2, "spanId": http_id})
assert [span["span_id"] for span in around_http["spans"]] == llm_ids[2:4]
assert around_http["prevCursor"]
assert around_http["nextCursor"]
# near the start, the short side gives its room to the other
at_start = get_page({"limit": 3, "spanId": llm_ids[0]})
@@ -287,11 +307,11 @@ def test_thread_reads_spans_across_json_rollout(
legacy_id, json_id = TraceIdGenerator.span_id(), TraceIdGenerator.span_id()
insert_traces(
[
Traces(timestamp=rollout - timedelta(minutes=10), trace_id=before_trace_id, span_id=before_ids[0], name="chat gpt-4o", resources=resources, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": "first"}])}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout - timedelta(minutes=8), trace_id=before_trace_id, span_id=TraceIdGenerator.span_id(), name="invoke_agent planner", resources=resources, attributes={"gen_ai.agent.name": "planner"}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout - timedelta(minutes=5), trace_id=before_trace_id, span_id=before_ids[1], name="chat gpt-4o", resources=resources, attributes={"gen_ai.output.messages": json.dumps([{"role": "assistant", "content": "second"}])}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout - timedelta(minutes=5), trace_id=straddle_trace_id, span_id=legacy_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": "legacy"}])}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout + timedelta(minutes=5), trace_id=straddle_trace_id, span_id=json_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": "json"}])}, attribute_write_mode="json_only"),
Traces(timestamp=rollout - timedelta(minutes=10), trace_id=before_trace_id, span_id=before_ids[0], name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": json.dumps([{"role": "user", "content": "first"}])}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout - timedelta(minutes=8), trace_id=before_trace_id, span_id=TraceIdGenerator.span_id(), name="GET /weather", resources=resources, attributes={"http.request.method": "GET"}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout - timedelta(minutes=5), trace_id=before_trace_id, span_id=before_ids[1], name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.output.messages": json.dumps([{"role": "assistant", "content": "second"}])}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout - timedelta(minutes=5), trace_id=straddle_trace_id, span_id=legacy_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": json.dumps([{"role": "user", "content": "legacy"}])}, attribute_write_mode="legacy_only"),
Traces(timestamp=rollout + timedelta(minutes=5), trace_id=straddle_trace_id, span_id=json_id, name="chat gpt-4o", resources=resources, attributes={"gen_ai.request.model": "gpt-4o", "gen_ai.input.messages": json.dumps([{"role": "user", "content": "json"}])}, attribute_write_mode="json_only"),
]
)
@@ -322,7 +342,7 @@ def test_thread_reads_spans_across_json_rollout(
assert straddle_spans[1]["formatted_input"] == [{"role": "user", "parts": [{"type": "text", "content": "json"}]}]
def test_thread_without_messages_is_empty(
def test_thread_without_gen_ai_spans_is_empty(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
@@ -331,7 +351,14 @@ def test_thread_without_messages_is_empty(
) -> None:
seed_attribute_evolution("traces", ATTRIBUTE_JSON_ROLLOUT_TIME)
trace_id = TraceIdGenerator.trace_id()
insert_traces([Traces(timestamp=datetime.now(tz=UTC) - timedelta(seconds=5), trace_id=trace_id, span_id=TraceIdGenerator.span_id(), name="GET /health", resources={"service.name": "tracedetail-thread-empty"}, attribute_write_mode="json_only")])
resources = {"service.name": "tracedetail-thread-empty"}
# messages without a gate key, as a failed call or an unmapped SDK writes them, are not a row either
insert_traces(
[
Traces(timestamp=datetime.now(tz=UTC) - timedelta(seconds=5), trace_id=trace_id, span_id=TraceIdGenerator.span_id(), name="GET /health", resources=resources, attribute_write_mode="json_only"),
Traces(timestamp=datetime.now(tz=UTC) - timedelta(seconds=4), trace_id=trace_id, span_id=TraceIdGenerator.span_id(), name="chat", resources=resources, attributes={"gen_ai.input.messages": json.dumps([{"role": "user", "content": "hi"}])}, attribute_write_mode="json_only"),
]
)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
response = requests.get(signoz.self.host_configs["8080"].get(f"/api/v1/traces/{trace_id}/thread"), headers={"Authorization": f"Bearer {token}"}, timeout=10)