Compare commits

...

2 Commits

Author SHA1 Message Date
nityanandagohain
006a272297 fix: address comment 2026-10-30 18:02:48 +05:30
nityanandagohain
de24f430c2 feat(aiobservability): read the messages the Langfuse SDK writes 2026-10-30 17:38:50 +05:30
10 changed files with 172 additions and 17 deletions

View File

@@ -17,7 +17,7 @@ Condition: `model`, `llm.input_messages` (OpenInference sets `llm.model_name` fr
| `gen_ai.response.model`, last source for the request model | vLLM, which writes only the response model |
| `ai.model.id`, `ai.model.provider`, `ai.response.model`, `ai.response.finishReason`, `ai.prompt.messages`, `ai.response.text`, `ai.response.toolCalls`, `ai.usage.promptTokens`, `ai.usage.completionTokens` | Vercel AI SDK v4, Mastra |
| `ai.usage.inputTokens`, `ai.usage.outputTokens`, `ai.usage.cachedInputTokens`, `ai.usage.reasoningTokens` | Vercel AI SDK v5 |
| `langfuse.observation.model.name`, `langfuse.session.id` | Langfuse SDK v3 |
| `langfuse.observation.model.name`, `langfuse.observation.input`, `langfuse.observation.output`, `langfuse.observation.type`, `langfuse.session.id` | Langfuse SDK v3 (Python) and v4 (JS), LiteLLM `langfuse_otel` callback |
| `session.id` | OpenInference, Langfuse, Claude Code |
| `gen_ai.client.session_id` | opentelemetry-hooks runner (Cursor, Copilot, Claude Code hooks) |
| `model`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_creation_tokens` | Claude Code `claude_code.llm_request` spans ([docs](https://code.claude.com/docs/en/monitoring-usage#span-attributes)) |

View File

@@ -1,5 +1,5 @@
{
"version": 4,
"version": 5,
"definition": {
"name": "gen_ai.llm",
"condition": {
@@ -134,6 +134,12 @@
"fieldContext": "attribute",
"config": {
"sources": [
{
"key": "langfuse.observation.type",
"context": "attribute",
"operation": "copy",
"priority": 40
},
{
"key": "openinference.span.kind",
"context": "attribute",
@@ -356,6 +362,12 @@
"fieldContext": "attribute",
"config": {
"sources": [
{
"key": "langfuse.observation.input",
"context": "attribute",
"operation": "move",
"priority": 40
},
{
"key": "gen_ai.prompt",
"context": "attribute",
@@ -382,6 +394,12 @@
"fieldContext": "attribute",
"config": {
"sources": [
{
"key": "langfuse.observation.output",
"context": "attribute",
"operation": "move",
"priority": 40
},
{
"key": "gen_ai.completion",
"context": "attribute",

View File

@@ -121,10 +121,13 @@ func (f *formatting) messages() {
// OpenRouter's answer object or an Anthropic response beside the OpenAI chat request it answers is one
// format, not a mix: an Anthropic request without system and without Anthropic-only blocks reads as an
// OpenAI chat request.
// OpenAI chat request. A side kept as generic already warned, and a side read from OpenInference's
// buckets beside the raw payload of the other is the documented fallback.
func (f *formatting) warnMixedFormats(inputLabel, outputLabel string) {
switch {
case inputLabel == "" || outputLabel == "" || outputLabel == inputLabel:
case inputLabel == FormatterGeneric || outputLabel == FormatterGeneric:
case inputLabel == FormatterOpenInference || outputLabel == FormatterOpenInference:
case inputLabel == FormatterOpenAIChat && (outputLabel == FormatterOpenRouter || outputLabel == FormatterAnthropic):
default:
f.warn("output formatted as %s, input as %s", outputLabel, inputLabel)
@@ -183,11 +186,16 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
return f.semconv(v), FormatterSemconv
case isAnthropicMessages(v):
return f.anthropicMessages(v), FormatterAnthropic
case isResponsesItems(v):
return foldAssistant(f.responsesItems(v)), FormatterOpenAIResponses
case first["role"] != nil:
return f.openAIChatMessages(v), FormatterOpenAIChat
}
}
case map[string]any:
if inner, ok := superJSON(v); ok {
return f.side(decode(inner), role)
}
switch {
case isLangChainRun(v):
return f.langChainRun(v), FormatterLangChain
@@ -214,6 +222,9 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
if messages, ok := v["messages"].([]any); ok {
return f.openAIChatMessages(messages), FormatterOpenAIChat
}
if (v["role"] != nil && hasKeys(v, "content")) || v["tool_calls"] != nil {
return OutputMessages{f.openAIChatSingle(v)}, FormatterOpenAIChat
}
case string:
if v == "" {
return OutputMessages{}, FormatterText
@@ -309,6 +320,23 @@ func foldAssistant(msgs OutputMessages) OutputMessages {
return out
}
// superJSON unwraps {json, meta?}, the envelope superjson, the transformer tRPC apps serialise with,
// puts around a value; a Langfuse generation records it around the request when the app passes the
// serialised form:
// https://github.com/flightcontrolhq/superjson/blob/aaa65e36e8e31da740265a56b27c965ca1c7b754/README.md
func superJSON(m map[string]any) (any, bool) {
inner, ok := m["json"]
if !ok {
return nil, false
}
for key := range m {
if key != "json" && key != "meta" {
return nil, false
}
}
return inner, true
}
// decode parses a JSON-encoded attribute value, twice when an SDK encoded the string again.
// Anything that is not JSON is returned as is.
func decode(value any) any {

View File

@@ -4,8 +4,9 @@ import (
"strings"
)
// Format: OpenAI Chat Completions, a request {messages}, a response {choices}, or a bare message list.
// Written by: bifrost, openrouter requests, the raw OpenAI calls OpenInference records.
// Format: OpenAI Chat Completions, a request {messages}, a response {choices}, a bare message list, or a
// single message.
// Written by: bifrost, openrouter requests, the raw OpenAI calls OpenInference records, the Langfuse SDK.
// Handled: string and block content, tool_calls nested or flattened, tool messages, refusal,
// reasoning_content, finish reasons.
// Not yet: streaming delta chunks, legacy completions choices[].text, audio output.
@@ -88,6 +89,19 @@ func (f *formatting) openAIChatMessage(m map[string]any) OutputMessage {
return msg
}
// openAIChatSingle converts a message recorded outside a list: the first choice's message the Langfuse
// OpenAI wrapper writes as the output, or the additional_kwargs of a LangChain AIMessage, {tool_calls}
// without a role, which the Langfuse LangChain handler writes for a tool-calling turn:
// https://github.com/langfuse/langfuse-python/blob/18baf80450dd5f9c1f85916eb2d1b3c72d2052e1/langfuse/openai.py#L446
// https://github.com/langfuse/langfuse-python/blob/18baf80450dd5f9c1f85916eb2d1b3c72d2052e1/langfuse/langchain/CallbackHandler.py#L1558
func (f *formatting) openAIChatSingle(m map[string]any) OutputMessage {
msg := f.openAIChatMessage(m)
if msg.Role.IsZero() {
msg.Role = RoleAssistant
}
return msg
}
// toolCallPart reads {id, function: {name, arguments}}; a gateway may flatten it to {id, name, arguments|args}.
func toolCallPart(c map[string]any) Part {
call := ToolCallRequestPart{Type: PartTypeToolCall, ID: optionalString(c["id"]), Name: stringOf(c["name"])}

View File

@@ -205,6 +205,30 @@ func TestFormat_OpenAIChat(t *testing.T) {
]`,
wantOutput: `[]`,
},
{
// Langfuse SDK generation: the history as a list, the first choice's message as the output
name: "SingleMessageOutput_AssistantMessage",
input: `[{"role": "system", "content": "You are a support agent on a phone call."}, {"role": "user", "content": "Hello?"}]`,
output: `{"role": "assistant", "content": "Hi, this is a courtesy call about your order. Do you have a minute?"}`,
wantInput: `[{"role": "system", "parts": [{"type": "text", "content": "You are a support agent on a phone call."}]}, {"role": "user", "parts": [{"type": "text", "content": "Hello?"}]}]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Hi, this is a courtesy call about your order. Do you have a minute?"}]}]`,
},
{
// a structured output that happens to carry a role key is not a message
name: "ObjectWithRoleKeyButNoContent_KeptAsGeneric",
output: `{"role": "admin", "permissions": ["read", "write"]}`,
warnings: []string{"message format not recognised, kept as generic"},
wantInput: `[]`,
wantOutput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"permissions\":[\"read\",\"write\"],\"role\":\"admin\"}"}]}]`,
},
{
// Langfuse LangChain handler: the additional_kwargs of the AIMessage, without a role, as the output
// of a tool-calling turn
name: "RolelessToolCallsOutput_AssistantToolCallPart",
output: `{"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"Oslo\"}"}}], "refusal": null}`,
wantInput: `[]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_1", "arguments": {"city": "Oslo"}}]}]`,
},
{
name: "ResponseWithScalarChoice_KeptAsGeneric",
output: `{"choices":["oops",{"finish_reason":"stop","index":1,"message":{"role":"assistant","content":"hi"}}]}`,

View File

@@ -2,13 +2,15 @@ package aiobservabilitytypes
import "strings"
// Format: OpenAI Responses, a request {input, instructions} or a response {output, status} of items:
// Format: OpenAI Responses, a request {input, instructions}, a response {output, status} or a bare list
// of items:
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L47597
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L69691
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L53834
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L57596
// Written by: gateways recording raw Responses bodies, and the raw calls OpenInference records for the
// Responses API and the OpenAI Agents SDK when its flattened messages are absent.
// Written by: gateways recording raw Responses bodies, the raw calls OpenInference records for the
// Responses API and the OpenAI Agents SDK when its flattened messages are absent, and LiteLLM's
// langfuse_otel callback, the function_call items of a tool-calling turn as a bare list.
// Handled: message items with input_text, output_text, refusal, input_image and input_file content;
// function_call and custom_tool_call with their outputs; reasoning summaries; compaction; every other
// *_call item as a server tool call and *_output item as its response; the status and incomplete_details
@@ -32,6 +34,19 @@ func isResponsesRequest(m map[string]any) bool {
return isItem
}
// isResponsesItems looks for an item only the Responses API has, a tool call or its output, a reasoning
// or compaction item, all without a role; a list of messages has none.
func isResponsesItems(list []any) bool {
for _, item := range list {
m, _ := item.(map[string]any)
typ := stringOf(m["type"])
if m["role"] == nil && (typ == "reasoning" || typ == "compaction" || typ == "mcp_approval_response" || strings.HasSuffix(typ, "_call") || strings.HasSuffix(typ, "_output")) {
return true
}
}
return false
}
func isResponsesResponse(m map[string]any) bool {
if _, ok := m["output"].([]any); !ok {
return false

View File

@@ -73,6 +73,23 @@ func TestFormat_OpenAIResponses(t *testing.T) {
wantInput: `[{"role": "system", "parts": [{"type": "text", "content": "Be brief."}]}, {"role": "user", "parts": [{"type": "text", "content": "Say hi"}]}]`,
wantOutput: `[]`,
},
{
// LiteLLM langfuse_otel: the function_call items of a tool-calling turn as a bare list
name: "BareFunctionCallItems_OneAssistantMessage",
output: `[{"id": "chatcmpl-1", "name": "set_stage", "call_id": "call_7", "type": "function_call", "arguments": {"fields": {"stage": "consent"}}}]`,
wantInput: `[]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "set_stage", "id": "call_7", "arguments": {"fields": {"stage": "consent"}}}]}]`,
},
{
name: "BareItemsWithMessage_ToolRound",
input: `[{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Weather in Oslo?"}]}, {"type": "function_call", "call_id": "call_9", "name": "get_weather", "arguments": "{\"city\":\"Oslo\"}"}, {"type": "function_call_output", "call_id": "call_9", "output": "4C, cloudy"}]`,
wantInput: `[
{"role": "user", "parts": [{"type": "text", "content": "Weather in Oslo?"}]},
{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_9", "arguments": {"city": "Oslo"}}]},
{"role": "tool", "parts": [{"type": "tool_call_response", "id": "call_9", "response": "4C, cloudy"}]}
]`,
wantOutput: `[]`,
},
{
// openinference capture, CreateEmbeddings span: an embeddings request also has input and model
name: "EmbeddingsRequest_NotResponses_KeptAsGeneric",

View File

@@ -4,8 +4,9 @@ import "github.com/SigNoz/signoz/pkg/valuer"
// Format: the flattened llm.input_messages.N.message.* and llm.output_messages.N.message.* keys, the
// provider-neutral copy of the conversation the OpenInference instrumentations write on LLM spans. A side
// without them falls back to the raw provider payload the mapper moved onto gen_ai.*.messages; pipecat
// streams the reply into output.value alone.
// without them, or whose buckets hold no message, falls back to the raw provider payload the mapper moved
// onto gen_ai.*.messages; pipecat streams the reply into output.value alone, LiteLLM's OpenInference mode
// leaves the output bucket empty on a tool-calling turn.
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/semantic_conventions.md#L77
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/tool_calling.md#L37
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/multimodal_attributes.md#L35
@@ -28,15 +29,14 @@ func (f *formatting) openInference() bool {
return false
}
inputLabel, outputLabel := FormatterOpenInference, FormatterOpenInference
if len(input) > 0 {
f.out.Input = inputMessages(openInferenceMessages(input))
if msgs := openInferenceMessages(input); len(msgs) > 0 {
f.out.Input = inputMessages(msgs)
} else if raw := attribute(f.attributes, GenAIInputMessages); raw != nil {
var msgs OutputMessages
msgs, inputLabel = f.side(decode(raw), RoleUser)
f.out.Input = inputMessages(msgs)
}
if len(output) > 0 {
f.out.Output = openInferenceMessages(output)
if msgs := openInferenceMessages(output); len(msgs) > 0 {
f.out.Output = msgs
if fr := finishReason(stringOf(attribute(f.attributes, openInferenceFinishReason))); fr != nil {
for i := range f.out.Output {
f.out.Output[i].FinishReason = fr

View File

@@ -160,16 +160,30 @@ func TestFormat_OpenInference(t *testing.T) {
// which the mapper moves onto gen_ai.output.messages
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-pipecat/src/openinference/instrumentation/pipecat/_attributes.py#L430
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-pipecat/src/openinference/instrumentation/pipecat/_observer.py#L628
name: "OutputKeysAbsent_RawOutputFormatted_Warns",
name: "OutputKeysAbsent_RawOutputFormatted",
attributes: map[string]any{
"llm.input_messages.0.message.role": "user",
"llm.input_messages.0.message.content": "hi",
GenAIOutputMessages: "Hello there.",
},
warnings: []string{"bare assistant text, role assumed", "output formatted as text, input as openinference"},
warnings: []string{"bare assistant text, role assumed"},
wantInput: `[{"role": "user", "parts": [{"type": "text", "content": "hi"}]}]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Hello there."}]}]`,
},
{
// LiteLLM's OpenInference mode leaves the output bucket empty on a tool-calling turn; the calls are
// the Langfuse output the mapper moved onto gen_ai.output.messages
name: "EmptyOutputBucket_RawOutputFormatted",
attributes: map[string]any{
"llm.input_messages.0.message.role": "user",
"llm.input_messages.0.message.content": "Yes, go ahead.",
"llm.output_messages.0.message.role": "assistant",
"llm.output_messages.0.message.content": "",
GenAIOutputMessages: `[{"id": "chatcmpl-2", "name": "set_stage", "call_id": "call_8", "type": "function_call", "arguments": {"fields": {"stage": "email"}}}]`,
},
wantInput: `[{"role": "user", "parts": [{"type": "text", "content": "Yes, go ahead."}]}]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "set_stage", "id": "call_8", "arguments": {"fields": {"stage": "email"}}}]}]`,
},
{
// message.role left unset by some instrumentors: tool_call_id, tool_calls or neither decide it,
// as the collector-contrib normalizer does

View File

@@ -122,6 +122,31 @@ So, the ball costs 5 cents.`,
wantInput: `[]`,
wantOutput: `[]`,
},
{
// Langfuse SDK generation: the app passes the request through superjson, the output is the parsed
// structured output
name: "SuperJSONEnvelope_Unwrapped_StructuredOutputGeneric",
input: `{"json": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "Extract the main article of the page as JSON."}, {"role": "user", "content": "# Council approves new bike lanes"}], "temperature": 0}}`,
output: `{"hasArticle": true, "article": {"title": "Council approves new bike lanes"}}`,
warnings: []string{"message format not recognised, kept as generic"},
wantInput: `[{"role": "system", "parts": [{"type": "text", "content": "Extract the main article of the page as JSON."}]}, {"role": "user", "parts": [{"type": "text", "content": "# Council approves new bike lanes"}]}]`,
wantOutput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"article\":{\"title\":\"Council approves new bike lanes\"},\"hasArticle\":true}"}]}]`,
},
{
name: "SuperJSONEnvelopeWithMeta_Unwrapped",
input: `{"json": {"messages": [{"role": "user", "content": "What time is it in Oslo?"}]}, "meta": {"values": {"messages.0.at": ["Date"]}}}`,
output: `{"json": {"choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "It is 14:05 in Oslo."}}]}, "meta": {"values": {}}}`,
wantInput: `[{"role": "user", "parts": [{"type": "text", "content": "What time is it in Oslo?"}]}]`,
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "It is 14:05 in Oslo."}], "finish_reason": "stop"}]`,
},
{
// a json key beside others is a payload of its own, not the envelope
name: "JSONKeyBesideOthers_NotUnwrapped_Generic",
input: `{"json": {"messages": []}, "schema": "v1"}`,
warnings: []string{"message format not recognised, kept as generic"},
wantInput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"json\":{\"messages\":[]},\"schema\":\"v1\"}"}]}]`,
wantOutput: `[]`,
},
{
name: "NoMessageAttributes_EmptyLists",
attributes: map[string]any{GenAIRequestModel: "gpt-4o", GenAIToolName: "get_weather"},