diff --git a/extensions/composer/llm-proxy/anthropic.go b/extensions/composer/llm-proxy/anthropic.go index c114c164..315dc8ae 100644 --- a/extensions/composer/llm-proxy/anthropic.go +++ b/extensions/composer/llm-proxy/anthropic.go @@ -9,92 +9,191 @@ import ( "bytes" "encoding/json" "fmt" + "sort" + "strings" ) -// anthropicRequest is the minimal subset of an Anthropic Messages API request body -// needed to extract the model name and streaming flag. +// anthropicRequest is the subset of an Anthropic Messages request body +// needed for routing and richer observability extraction. type anthropicRequest struct { - Model string `json:"model"` - Stream bool `json:"stream"` + Model string `json:"model"` + Stream bool `json:"stream"` + System json.RawMessage `json:"system"` + Messages []anthropicRequestMessage `json:"messages"` } -// anthropicUsage holds token-usage fields from an Anthropic response. +// anthropicRequestMessage models one request message in the Anthropic payload. +type anthropicRequestMessage struct { + Role string `json:"role"` + Content any `json:"content"` +} + +// anthropicUsage holds token-usage and cache-related fields from an Anthropic response. type anthropicUsage struct { - InputTokens uint32 `json:"input_tokens"` - OutputTokens uint32 `json:"output_tokens"` + InputTokens uint32 `json:"input_tokens"` + OutputTokens uint32 `json:"output_tokens"` + CacheCreationInputTokens uint32 `json:"cache_creation_input_tokens"` + CacheReadInputTokens uint32 `json:"cache_read_input_tokens"` } -// anthropicResponse is the minimal subset of an Anthropic Messages API response body. +// anthropicResponse is the subset of a non-streaming Anthropic response +// needed for richer observability extraction. type anthropicResponse struct { + Content []struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Input json.RawMessage `json:"input,omitempty"` + } `json:"content"` Usage anthropicUsage `json:"usage"` } -// anthropicMessageStartData is the payload of an Anthropic "message_start" SSE event. +// anthropicToolCall represents a tool use block in Anthropic responses. +type anthropicToolCall struct { + ID string `json:"id"` + Name string `json:"name"` + Input string `json:"input"` +} + +// anthropicMessageStartData is the payload of a "message_start" SSE event. type anthropicMessageStartData struct { Message struct { Usage anthropicUsage `json:"usage"` } `json:"message"` } -// anthropicMessageDeltaData is the payload of an Anthropic "message_delta" SSE event. +// anthropicMessageDeltaData is the payload of a "message_delta" SSE event. type anthropicMessageDeltaData struct { Usage struct { OutputTokens uint32 `json:"output_tokens"` } `json:"usage"` } -// --- LLMRequest implementation --- +// anthropicContentBlockStartData is the payload of a "content_block_start" SSE event. +type anthropicContentBlockStartData struct { + Index int `json:"index"` + ContentBlock struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Input json.RawMessage `json:"input,omitempty"` + } `json:"content_block"` +} + +// anthropicContentBlockDeltaData is the payload of a "content_block_delta" SSE event. +type anthropicContentBlockDeltaData struct { + Index int `json:"index"` + Delta struct { + Type string `json:"type"` + Text string `json:"text,omitempty"` + PartialJSON string `json:"partial_json,omitempty"` + } `json:"delta"` +} // anthropicLLMRequest implements LLMRequest for the Anthropic Messages API. type anthropicLLMRequest struct { - model string - stream bool + model string + stream bool + question string + system string } func (r *anthropicLLMRequest) GetModel() string { return r.model } func (r *anthropicLLMRequest) IsStream() bool { return r.stream } - -// parseAnthropicRequest parses an Anthropic Messages API request body and returns -// an LLMRequest with the extracted model and stream fields. -func parseAnthropicRequest(body []byte) (LLMRequest, error) { - var req anthropicRequest - if err := json.Unmarshal(body, &req); err != nil { - return nil, err - } - return &anthropicLLMRequest{model: req.Model, stream: req.Stream}, nil +func (r *anthropicLLMRequest) GetQuestion() string { + return r.question } - -// --- LLMResponse implementation --- +func (r *anthropicLLMRequest) GetSystem() string { return r.system } // anthropicLLMResponse implements LLMResponse for the Anthropic Messages API. type anthropicLLMResponse struct { - usage LLMUsage + usage LLMUsage + answer string + reasoning string + toolCalls []anthropicToolCall + reasoningTokens uint32 + cachedTokens uint32 + inputTokenDetails any + outputTokenDetails any } func (r *anthropicLLMResponse) GetUsage() LLMUsage { return r.usage } +func (r *anthropicLLMResponse) GetAnswer() string { return r.answer } +func (r *anthropicLLMResponse) GetReasoning() string { + return r.reasoning +} +func (r *anthropicLLMResponse) GetToolCalls() any { return r.toolCalls } +func (r *anthropicLLMResponse) GetReasoningTokens() uint32 { return r.reasoningTokens } +func (r *anthropicLLMResponse) GetCachedTokens() uint32 { return r.cachedTokens } +func (r *anthropicLLMResponse) GetInputTokenDetails() any { return r.inputTokenDetails } +func (r *anthropicLLMResponse) GetOutputTokenDetails() any { return r.outputTokenDetails } + +// anthropicLLMResponseChunk implements LLMResponseChunk for Anthropic streaming SSE. +type anthropicLLMResponseChunk struct { + usage LLMUsage + hasTextToken bool + cachedTokens uint32 + inputTokenDetails any +} -// parseAnthropicResponse parses an Anthropic Messages API response body and returns -// an LLMResponse with the extracted token-usage information. +func (c *anthropicLLMResponseChunk) GetUsage() LLMUsage { return c.usage } +func (c *anthropicLLMResponseChunk) GetAnswer() string { return "" } +func (c *anthropicLLMResponseChunk) GetReasoning() string { + return "" +} +func (c *anthropicLLMResponseChunk) GetToolCalls() any { return nil } +func (c *anthropicLLMResponseChunk) HasTextToken() bool { + return c.hasTextToken +} + +// parseAnthropicRequest parses an Anthropic Messages request body and returns +// an LLMRequest with routing and observability fields. +func parseAnthropicRequest(body []byte) (LLMRequest, error) { + var req anthropicRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + return &anthropicLLMRequest{ + model: req.Model, + stream: req.Stream, + question: extractAnthropicQuestion(req.Messages), + system: extractAnthropicSystem(req.System), + }, nil +} + +// parseAnthropicResponse parses a non-streaming Anthropic response and extracts +// usage plus richer observability fields. func parseAnthropicResponse(body []byte) (LLMResponse, error) { var resp anthropicResponse if err := json.Unmarshal(body, &resp); err != nil { return nil, err } - return &anthropicLLMResponse{usage: anthropicUsageToLLM(resp.Usage)}, nil -} - -// --- LLMResponseChunk implementation --- - -// anthropicLLMResponseChunk implements LLMResponseChunk for the Anthropic streaming API. -type anthropicLLMResponseChunk struct { - usage LLMUsage + answer := "" + toolCalls := make([]anthropicToolCall, 0) + for _, item := range resp.Content { + if item.Type == "text" { + answer += item.Text + } + if item.Type == "tool_use" { + toolCalls = append(toolCalls, anthropicToolCall{ + ID: item.ID, + Name: item.Name, + Input: string(item.Input), + }) + } + } + return &anthropicLLMResponse{ + usage: anthropicUsageToLLM(resp.Usage), + answer: answer, + toolCalls: toolCalls, + cachedTokens: resp.Usage.CacheReadInputTokens, + inputTokenDetails: buildAnthropicInputTokenDetails(resp.Usage), + }, nil } -func (c *anthropicLLMResponseChunk) GetUsage() LLMUsage { return c.usage } - -// parseAnthropicChunk parses a single Anthropic SSE event and returns an -// LLMResponseChunk containing any usage data carried by that event. -// eventType is the value from the preceding "event:" SSE line. +// parseAnthropicChunk parses a single Anthropic SSE event payload. func parseAnthropicChunk(eventType string, data []byte) (anthropicLLMResponseChunk, error) { switch eventType { case "message_start": @@ -102,7 +201,11 @@ func parseAnthropicChunk(eventType string, data []byte) (anthropicLLMResponseChu if err := json.Unmarshal(data, &msg); err != nil { return anthropicLLMResponseChunk{}, err } - return anthropicLLMResponseChunk{usage: anthropicUsageToLLM(msg.Message.Usage)}, nil + return anthropicLLMResponseChunk{ + usage: anthropicUsageToLLM(msg.Message.Usage), + cachedTokens: msg.Message.Usage.CacheReadInputTokens, + inputTokenDetails: buildAnthropicInputTokenDetails(msg.Message.Usage), + }, nil case "message_delta": var delta anthropicMessageDeltaData @@ -110,12 +213,17 @@ func parseAnthropicChunk(eventType string, data []byte) (anthropicLLMResponseChu return anthropicLLMResponseChunk{}, err } return anthropicLLMResponseChunk{usage: LLMUsage{OutputTokens: delta.Usage.OutputTokens}}, nil + case "content_block_delta": + var delta anthropicContentBlockDeltaData + if err := json.Unmarshal(data, &delta); err != nil { + return anthropicLLMResponseChunk{}, err + } + return anthropicLLMResponseChunk{hasTextToken: delta.Delta.Type == "text_delta" && delta.Delta.Text != ""}, nil } return anthropicLLMResponseChunk{}, nil } -// anthropicUsageToLLM converts an anthropicUsage to an LLMUsage. -// Returns the zero value when u is nil. +// anthropicUsageToLLM converts an Anthropic usage payload to the common LLMUsage shape. func anthropicUsageToLLM(u anthropicUsage) LLMUsage { return LLMUsage{ InputTokens: u.InputTokens, @@ -124,29 +232,35 @@ func anthropicUsageToLLM(u anthropicUsage) LLMUsage { } } -// --- SSE accumulator --- - var ( anthropicSSEEventPrefix = []byte("event: ") anthropicSSEDataPrefix = []byte("data: ") ) -// anthropicSSEParser accumulates usage information from an Anthropic streaming SSE response. -// It consumes body chunks as they arrive and produces an LLMResponse when finished. +// anthropicSSEParser accumulates usage, text, tool calls, and cache-related +// fields from an Anthropic streaming SSE response. type anthropicSSEParser struct { - buf []byte - done bool - inputTokens uint32 - outputTokens uint32 - // currentEvent tracks the most recently seen "event:" value so that the - // following "data:" line can be routed to the correct handler. - currentEvent string + buf []byte + done bool + inputTokens uint32 + outputTokens uint32 + cachedTokens uint32 + currentEvent string + textByIndex map[int]string + toolByIndex map[int]*anthropicToolCall + seenTextToken bool + inputTokenDetails any } +// newAnthropicSSEParser creates a parser for incremental Anthropic SSE accumulation. func newAnthropicSSEParser() *anthropicSSEParser { - return &anthropicSSEParser{} + return &anthropicSSEParser{ + textByIndex: map[int]string{}, + toolByIndex: map[int]*anthropicToolCall{}, + } } +// Feed appends a new response body chunk and parses any complete SSE events. func (a *anthropicSSEParser) Feed(data []byte) error { if a.done { return nil @@ -155,6 +269,7 @@ func (a *anthropicSSEParser) Feed(data []byte) error { return a.parseEvents() } +// parseEvents processes complete SSE lines accumulated in the internal buffer. func (a *anthropicSSEParser) parseEvents() error { for { idx := bytes.IndexByte(a.buf, '\n') @@ -178,11 +293,53 @@ func (a *anthropicSSEParser) parseEvents() error { } } +// processEvent handles a single parsed Anthropic SSE event. func (a *anthropicSSEParser) processEvent(eventType string, data []byte) error { if eventType == "message_stop" { a.done = true return nil } + if eventType == "content_block_start" { + var block anthropicContentBlockStartData + if err := json.Unmarshal(data, &block); err != nil { + return fmt.Errorf("llm-proxy: failed to parse Anthropic SSE event %q: %w", eventType, err) + } + if block.ContentBlock.Type == "text" { + a.textByIndex[block.Index] = block.ContentBlock.Text + if block.ContentBlock.Text != "" { + a.seenTextToken = true + } + return nil + } + if block.ContentBlock.Type == "tool_use" { + a.toolByIndex[block.Index] = &anthropicToolCall{ + ID: block.ContentBlock.ID, + Name: block.ContentBlock.Name, + Input: string(block.ContentBlock.Input), + } + } + return nil + } + if eventType == "content_block_delta" { + var delta anthropicContentBlockDeltaData + if err := json.Unmarshal(data, &delta); err != nil { + return fmt.Errorf("llm-proxy: failed to parse Anthropic SSE event %q: %w", eventType, err) + } + switch delta.Delta.Type { + case "text_delta": + a.seenTextToken = true + a.textByIndex[delta.Index] += delta.Delta.Text + case "input_json_delta": + if tc, ok := a.toolByIndex[delta.Index]; ok { + if tc.Input == "" || tc.Input == "{}" { + tc.Input = delta.Delta.PartialJSON + } else { + tc.Input += delta.Delta.PartialJSON + } + } + } + return nil + } chunk, err := parseAnthropicChunk(eventType, data) if err != nil { return fmt.Errorf("llm-proxy: failed to parse Anthropic SSE event %q: %w", eventType, err) @@ -195,18 +352,46 @@ func (a *anthropicSSEParser) processEvent(eventType string, data []byte) error { a.outputTokens = u.OutputTokens } } + if chunk.cachedTokens > 0 { + a.cachedTokens = chunk.cachedTokens + } + if chunk.inputTokenDetails != nil { + a.inputTokenDetails = chunk.inputTokenDetails + } return nil } +// Finish finalises the stream and returns the accumulated response fields. func (a *anthropicSSEParser) Finish() (LLMResponse, error) { + answer := "" + if len(a.textByIndex) > 0 { + indexes := make([]int, 0, len(a.textByIndex)) + for idx := range a.textByIndex { + indexes = append(indexes, idx) + } + sort.Ints(indexes) + for _, idx := range indexes { + answer += a.textByIndex[idx] + } + } + toolCalls := make([]anthropicToolCall, 0, len(a.toolByIndex)) + if len(a.toolByIndex) > 0 { + indexes := make([]int, 0, len(a.toolByIndex)) + for idx := range a.toolByIndex { + indexes = append(indexes, idx) + } + sort.Ints(indexes) + for _, idx := range indexes { + toolCalls = append(toolCalls, *a.toolByIndex[idx]) + } + } return &anthropicLLMResponse{usage: LLMUsage{ InputTokens: a.inputTokens, OutputTokens: a.outputTokens, TotalTokens: a.inputTokens + a.outputTokens, - }}, nil + }, answer: answer, toolCalls: toolCalls, cachedTokens: a.cachedTokens, inputTokenDetails: a.inputTokenDetails}, nil } -// anthropicFactory implements LLMFactory for the Anthropic Messages API. type anthropicFactory struct{} func (f *anthropicFactory) ParseRequest(body []byte) (LLMRequest, error) { @@ -218,3 +403,81 @@ func (f *anthropicFactory) ParseResponse(body []byte) (LLMResponse, error) { } func (f *anthropicFactory) NewSSEParser() SSEParser { return newAnthropicSSEParser() } + +// SeenTextToken reports whether the stream has emitted a real text token yet. +func (a *anthropicSSEParser) SeenTextToken() bool { return a.seenTextToken } + +// extractAnthropicQuestion returns the last user message content from the request. +func extractAnthropicQuestion(messages []anthropicRequestMessage) string { + for i := len(messages) - 1; i >= 0; i-- { + if messages[i].Role != "user" { + continue + } + return extractAnthropicMessageContent(messages[i].Content) + } + return "" +} + +// extractAnthropicSystem returns the system prompt from either string or block form. +func extractAnthropicSystem(raw json.RawMessage) string { + if len(raw) == 0 { + return "" + } + var s string + if err := json.Unmarshal(raw, &s); err == nil { + return s + } + var blocks []struct { + Type string `json:"type"` + Text string `json:"text"` + } + if err := json.Unmarshal(raw, &blocks); err == nil { + out := "" + for i, b := range blocks { + if b.Type == "text" && b.Text != "" { + if i > 0 && out != "" { + out += "\n" + } + out += b.Text + } + } + return out + } + return "" +} + +// buildAnthropicInputTokenDetails returns cache-related input token detail fields when present. +func buildAnthropicInputTokenDetails(u anthropicUsage) any { + if u.CacheCreationInputTokens == 0 && u.CacheReadInputTokens == 0 { + return nil + } + return map[string]uint32{ + "cache_creation_input_tokens": u.CacheCreationInputTokens, + "cache_read_input_tokens": u.CacheReadInputTokens, + } +} + +// extractAnthropicMessageContent extracts text from either string or block-form content. +func extractAnthropicMessageContent(content any) string { + if s, ok := content.(string); ok { + return s + } + raw, err := json.Marshal(content) + if err != nil { + return "" + } + var blocks []struct { + Type string `json:"type"` + Text string `json:"text"` + } + if err := json.Unmarshal(raw, &blocks); err != nil { + return "" + } + texts := make([]string, 0, len(blocks)) + for _, b := range blocks { + if b.Type == "text" && b.Text != "" { + texts = append(texts, b.Text) + } + } + return strings.Join(texts, "\n") +} diff --git a/extensions/composer/llm-proxy/config.schema.json b/extensions/composer/llm-proxy/config.schema.json index 3a1a3a00..2d1699b4 100644 --- a/extensions/composer/llm-proxy/config.schema.json +++ b/extensions/composer/llm-proxy/config.schema.json @@ -34,6 +34,18 @@ "type": "string", "description": "Request header to set with the extracted model name." }, + "use_default_attributes": { + "type": "boolean", + "description": "Emit the full built-in structured log payload." + }, + "use_default_response_attributes": { + "type": "boolean", + "description": "Emit the lightweight built-in structured log payload." + }, + "session_id_header": { + "type": "string", + "description": "Optional request header used to extract a session or conversation ID." + }, "clear_route_cache": { "type": "boolean", "description": "Clear Envoy route cache after setting metadata, enabling route re-selection. Defaults to false." diff --git a/extensions/composer/llm-proxy/config_schema_test.go b/extensions/composer/llm-proxy/config_schema_test.go index 64f3307b..24e59be4 100644 --- a/extensions/composer/llm-proxy/config_schema_test.go +++ b/extensions/composer/llm-proxy/config_schema_test.go @@ -20,6 +20,9 @@ func TestConfigSchema(t *testing.T) { ], "metadata_namespace": "custom.ns", "llm_model_header": "x-llm-model", + "use_default_attributes": true, + "use_default_response_attributes": false, + "session_id_header": "x-session-id", "clear_route_cache": true }`) }) @@ -33,4 +36,9 @@ func TestConfigSchema(t *testing.T) { ] }`) }) + t.Run("invalid wrong type", func(t *testing.T) { + internaltesting.AssertSchemaInvalid(t, "config.schema.json", `{ + "use_default_attributes": "yes" + }`) + }) } diff --git a/extensions/composer/llm-proxy/llm.go b/extensions/composer/llm-proxy/llm.go index 4c535e11..0471095c 100644 --- a/extensions/composer/llm-proxy/llm.go +++ b/extensions/composer/llm-proxy/llm.go @@ -4,7 +4,8 @@ // the root of the repo. // Package llmproxy implements an HTTP filter that identifies LLM API requests -// and extracts model, stream, and token-usage information into filter metadata. +// and extracts model, stream, token-usage, and richer observability attributes +// into filter metadata. package llmproxy const ( @@ -33,6 +34,10 @@ type LLMRequest interface { GetModel() string // IsStream returns whether the request asks for a streaming (SSE) response. IsStream() bool + // GetQuestion returns the user question extracted from the request if available. + GetQuestion() string + // GetSystem returns the system prompt extracted from the request if available. + GetSystem() string } // LLMResponse abstracts over different LLM API non-streaming response formats. @@ -40,6 +45,20 @@ type LLMResponse interface { // GetUsage returns token-usage information extracted from the response body. // The zero value of LLMUsage indicates that no usage data was present. GetUsage() LLMUsage + // GetAnswer returns the textual assistant answer extracted from the response if available. + GetAnswer() string + // GetReasoning returns provider-specific reasoning content when available. + GetReasoning() string + // GetToolCalls returns any tool call payload emitted by the model. + GetToolCalls() any + // GetReasoningTokens returns the number of reasoning tokens when provided. + GetReasoningTokens() uint32 + // GetCachedTokens returns cached input token counts when provided. + GetCachedTokens() uint32 + // GetInputTokenDetails returns provider-specific prompt/input token detail fields. + GetInputTokenDetails() any + // GetOutputTokenDetails returns provider-specific completion/output token detail fields. + GetOutputTokenDetails() any } // LLMResponseChunk abstracts over a single event in an LLM streaming SSE response. @@ -47,6 +66,14 @@ type LLMResponseChunk interface { // GetUsage returns token-usage information carried by this chunk. // The zero value of LLMUsage indicates that the chunk carries no usage data. GetUsage() LLMUsage + // GetAnswer returns any assistant text carried by this chunk. + GetAnswer() string + // GetReasoning returns any reasoning content carried by this chunk. + GetReasoning() string + // GetToolCalls returns any tool call delta carried by this chunk. + GetToolCalls() any + // HasTextToken reports whether the chunk contains a real text token. + HasTextToken() bool } // SSEParser incrementally consumes body chunks from an LLM streaming SSE response @@ -59,6 +86,8 @@ type SSEParser interface { // Finish finalises parsing and returns the accumulated LLMResponse and any // terminal error encountered while processing the stream. Finish() (LLMResponse, error) + // SeenTextToken reports whether the stream has emitted a real text token yet. + SeenTextToken() bool } // LLMFactory creates the per-API-type parsers for a specific LLM provider. diff --git a/extensions/composer/llm-proxy/manifest.yaml b/extensions/composer/llm-proxy/manifest.yaml index b211e213..d8fbc663 100644 --- a/extensions/composer/llm-proxy/manifest.yaml +++ b/extensions/composer/llm-proxy/manifest.yaml @@ -9,7 +9,7 @@ categories: - AI author: Tetrate featured: true -description: Routes LLM API requests by the module name and monitors token usage and latency via Envoy metadata and metrics. +description: Routes LLM API requests and emits richer observability metadata, metrics, and optional structured logs. longDescription: | An HTTP filter plugin that inspects incoming requests against a set of configured path-matcher rules to identify the LLM provider API in use (OpenAI Chat Completions, @@ -18,9 +18,11 @@ longDescription: | 1. Parses the request body to extract the model name and streaming flag, then writes them to Envoy's dynamic filter metadata. 2. Parses the response body (JSON for non-streaming, SSE for streaming) to extract - token-usage information and writes it to filter metadata. + token-usage information and richer observability fields such as `question`, + `system`, `answer`, `reasoning`, and `tool_calls`. 3. Records Envoy metrics (counters and histograms) for request counts, token usage, time-to-first-token (TTFT), and time-per-output-token (TPOT). + 4. Optionally emits lightweight or full structured logs for downstream observability. Requests whose path does not match any rule are passed through without modification. @@ -38,9 +40,20 @@ longDescription: | | `kind` | string | API kind: `"openai"`, `"anthropic"`, or `"custom"` | | `model` | string | Model name extracted from the request body | | `is_stream` | bool | Whether the request asks for a streaming (SSE) response | + | `response_type` | string | `"stream"` or `"nonstream"` | + | `session_id` | string | Session or conversation ID extracted from `session_id_header`, if configured | + | `question` | string | User question extracted from the request body, when available | + | `system` | string | System prompt extracted from the request body, when available | + | `answer` | string | Assistant response content, when available | + | `reasoning` | string | Provider-specific reasoning content, when available | + | `tool_calls` | array | Tool calls emitted by the model, when available | | `input_tokens` | uint32 | Input / prompt token count from the response | | `output_tokens` | uint32 | Output / completion token count from the response | | `total_tokens` | uint32 | Total token count from the response | + | `reasoning_tokens` | uint32 | Reasoning token count when provided by the upstream API | + | `cached_tokens` | uint32 | Cached input token count when provided by the upstream API | + | `input_token_details` | object | Provider-specific prompt/input token detail fields | + | `output_token_details` | object | Provider-specific completion/output token detail fields | | `request_ttft` | int64 | Time to first token in milliseconds | | `request_tpot` | int64 | Average time per output token in milliseconds | @@ -67,6 +80,9 @@ longDescription: | | `llm_configs[].kind` | string | yes | — | `"openai"`, `"anthropic"`, or `"custom"` | | `metadata_namespace` | string | no | `io.builtonenvoy.llm-proxy` | Filter metadata namespace | | `llm_model_header` | string | no | `""` | If set, the extracted model name is written to this request header | + | `use_default_attributes` | bool | no | `false` | Emit the full built-in structured log payload | + | `use_default_response_attributes` | bool | no | `false` | Emit the lightweight built-in structured log payload | + | `session_id_header` | string | no | `""` | Optional request header used to extract a session or conversation ID | | `clear_route_cache` | bool | no | `false` | Clear the route cache after request parsing so Envoy can re-select the route based on updated metadata | type: go @@ -130,3 +146,13 @@ examples: "llm_model_header": "x-llm-model", "clear_route_cache": true }' + - title: Full observability output + description: | + Emit richer metadata and a structured log entry including prompts, answers, + tool calls, and provider token-detail fields. + code: | + boe run --extension llm-proxy \ + --config '{ + "use_default_attributes": true, + "session_id_header": "x-session-id" + }' diff --git a/extensions/composer/llm-proxy/openai.go b/extensions/composer/llm-proxy/openai.go index af811036..d68e4661 100644 --- a/extensions/composer/llm-proxy/openai.go +++ b/extensions/composer/llm-proxy/openai.go @@ -9,93 +9,236 @@ import ( "bytes" "encoding/json" "fmt" + "sort" + "strings" ) -// openaiRequest is the minimal subset of an OpenAI Chat Completions request body -// needed to extract the model name and streaming flag. -type openaiRequest struct { - Model string `json:"model"` - Stream bool `json:"stream"` +// openAIRequest is the subset of an OpenAI Chat Completions request body +// needed for routing and richer observability extraction. +type openAIRequest struct { + Model string `json:"model"` + Stream bool `json:"stream"` + Messages []openAIRequestMessage `json:"messages"` } -// openaiUsage holds token-usage fields from an OpenAI response or chunk. -type openaiUsage struct { - PromptTokens uint32 `json:"prompt_tokens"` - CompletionTokens uint32 `json:"completion_tokens"` - TotalTokens uint32 `json:"total_tokens"` +// openAIRequestMessage models one request message in a Chat Completions payload. +type openAIRequestMessage struct { + Role string `json:"role"` + Content any `json:"content"` } -// openaiResponse is the minimal subset of an OpenAI Chat Completions response body. -type openaiResponse struct { - Usage openaiUsage `json:"usage"` +// openAIContentPart is used to extract text from array-form message content. +type openAIContentPart struct { + Type string `json:"type"` + Text string `json:"text"` } -// openaiChunk is a single data event in an OpenAI streaming SSE response. -type openaiChunk struct { - Usage openaiUsage `json:"usage"` +// openAIUsage holds token-usage fields from an OpenAI response or chunk. +type openAIUsage struct { + PromptTokens uint32 `json:"prompt_tokens"` + CompletionTokens uint32 `json:"completion_tokens"` + TotalTokens uint32 `json:"total_tokens"` + PromptTokensDetails *openAIPromptTokensDetails `json:"prompt_tokens_details"` + CompletionTokensDetails *openAICompletionTokensDetails `json:"completion_tokens_details"` } -// --- LLMRequest implementation --- +// openAIPromptTokensDetails holds provider-specific prompt token detail fields. +type openAIPromptTokensDetails struct { + CachedTokens uint32 `json:"cached_tokens"` +} + +// openAICompletionTokensDetails holds provider-specific completion token detail fields. +type openAICompletionTokensDetails struct { + ReasoningTokens uint32 `json:"reasoning_tokens"` + AudioTokens uint32 `json:"audio_tokens"` +} + +// openAIResponse is the subset of a non-streaming Chat Completions response +// needed for richer observability extraction. +type openAIResponse struct { + Choices []struct { + Message struct { + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + ToolCalls []openAIToolCall `json:"tool_calls"` + } `json:"message"` + } `json:"choices"` + Usage openAIUsage `json:"usage"` +} + +// openAIChunk is a single data event in an OpenAI streaming SSE response. +type openAIChunk struct { + Choices []struct { + Delta struct { + Content string `json:"content"` + ReasoningContent string `json:"reasoning_content"` + ToolCalls []openAIStreamingToolCallDelta `json:"tool_calls"` + } `json:"delta"` + } `json:"choices"` + Usage openAIUsage `json:"usage"` +} + +// openAIToolCall represents a tool call in a non-streaming response. +type openAIToolCall struct { + ID string `json:"id"` + Type string `json:"type"` + Function openAIToolCallFunction `json:"function"` +} + +// openAIToolCallFunction represents the function call payload of a tool call. +type openAIToolCallFunction struct { + Name string `json:"name"` + Arguments string `json:"arguments"` +} + +// openAIStreamingToolCallDelta represents an incremental tool call update in a stream. +type openAIStreamingToolCallDelta struct { + Index int `json:"index"` + ID string `json:"id"` + Type string `json:"type"` + Function openAIToolCallFunction `json:"function"` +} + +// openAILLMRequest implements LLMRequest for the OpenAI Chat Completions API. +type openAILLMRequest struct { + model string + stream bool + question string + system string +} + +func (r *openAILLMRequest) GetModel() string { return r.model } +func (r *openAILLMRequest) IsStream() bool { return r.stream } +func (r *openAILLMRequest) GetQuestion() string { + return r.question +} +func (r *openAILLMRequest) GetSystem() string { return r.system } -// openaiLLMRequest implements LLMRequest for the OpenAI Chat Completions API. -type openaiLLMRequest struct { - model string - stream bool +// openAILLMResponse implements LLMResponse for the OpenAI Chat Completions API. +type openAILLMResponse struct { + usage LLMUsage + answer string + reasoning string + toolCalls []openAIToolCall + reasoningTokens uint32 + cachedTokens uint32 + inputTokenDetails any + outputTokenDetails any } -func (r *openaiLLMRequest) GetModel() string { return r.model } -func (r *openaiLLMRequest) IsStream() bool { return r.stream } +func (r *openAILLMResponse) GetUsage() LLMUsage { return r.usage } +func (r *openAILLMResponse) GetAnswer() string { return r.answer } +func (r *openAILLMResponse) GetReasoning() string { + return r.reasoning +} +func (r *openAILLMResponse) GetToolCalls() any { return r.toolCalls } +func (r *openAILLMResponse) GetReasoningTokens() uint32 { return r.reasoningTokens } +func (r *openAILLMResponse) GetCachedTokens() uint32 { return r.cachedTokens } +func (r *openAILLMResponse) GetInputTokenDetails() any { return r.inputTokenDetails } +func (r *openAILLMResponse) GetOutputTokenDetails() any { return r.outputTokenDetails } + +// openAILLMResponseChunk implements LLMResponseChunk for the OpenAI streaming API. +type openAILLMResponseChunk struct { + usage LLMUsage + answer string + reasoning string + toolCalls []openAIStreamingToolCallDelta + cachedTokens uint32 + reasoningTokens uint32 + inputTokenDetails any + outputTokenDetails any +} + +func (c *openAILLMResponseChunk) GetUsage() LLMUsage { return c.usage } +func (c *openAILLMResponseChunk) GetAnswer() string { return c.answer } +func (c *openAILLMResponseChunk) GetReasoning() string { + return c.reasoning +} +func (c *openAILLMResponseChunk) GetToolCalls() any { return c.toolCalls } +func (c *openAILLMResponseChunk) HasTextToken() bool { + return c.answer != "" || c.reasoning != "" +} + +// openAISSEParser accumulates usage, text, reasoning, tool calls, and token-detail +// fields from an OpenAI streaming SSE response and produces an LLMResponse when finished. +type openAISSEParser struct { + buf []byte + done bool + usage LLMUsage + answer string + reasoning string + toolCallsByIndex map[int]*openAIToolCall + seenTextToken bool + cachedTokens uint32 + reasoningTokens uint32 + inputTokenDetails any + outputTokenDetails any +} // parseOpenAIRequest parses an OpenAI Chat Completions request body and returns -// an LLMRequest with the extracted model and stream fields. +// an LLMRequest with routing and observability fields. func parseOpenAIRequest(body []byte) (LLMRequest, error) { - var req openaiRequest + var req openAIRequest if err := json.Unmarshal(body, &req); err != nil { return nil, err } - return &openaiLLMRequest{model: req.Model, stream: req.Stream}, nil + return &openAILLMRequest{ + model: req.Model, + stream: req.Stream, + question: extractOpenAIQuestion(req.Messages), + system: extractOpenAISystem(req.Messages), + }, nil } -// --- LLMResponse implementation --- - -// openaiLLMResponse implements LLMResponse for the OpenAI Chat Completions API. -type openaiLLMResponse struct { - usage LLMUsage -} - -func (r *openaiLLMResponse) GetUsage() LLMUsage { return r.usage } - -// parseOpenAIResponse parses an OpenAI Chat Completions response body and returns -// an LLMResponse with the extracted token-usage information. +// parseOpenAIResponse parses a non-streaming OpenAI Chat Completions response +// and extracts token usage plus richer observability fields. func parseOpenAIResponse(body []byte) (LLMResponse, error) { - var resp openaiResponse + var resp openAIResponse if err := json.Unmarshal(body, &resp); err != nil { return nil, err } - return &openaiLLMResponse{usage: openaiUsageToLLM(resp.Usage)}, nil -} - -// --- LLMResponseChunk implementation --- - -// openaiLLMResponseChunk implements LLMResponseChunk for the OpenAI streaming API. -type openaiLLMResponseChunk struct { - usage LLMUsage + result := &openAILLMResponse{usage: openAIUsageToLLM(resp.Usage)} + if len(resp.Choices) > 0 { + result.answer = resp.Choices[0].Message.Content + result.reasoning = resp.Choices[0].Message.ReasoningContent + result.toolCalls = resp.Choices[0].Message.ToolCalls + } + if resp.Usage.PromptTokensDetails != nil { + result.cachedTokens = resp.Usage.PromptTokensDetails.CachedTokens + result.inputTokenDetails = resp.Usage.PromptTokensDetails + } + if resp.Usage.CompletionTokensDetails != nil { + result.reasoningTokens = resp.Usage.CompletionTokensDetails.ReasoningTokens + result.outputTokenDetails = resp.Usage.CompletionTokensDetails + } + return result, nil } -func (c *openaiLLMResponseChunk) GetUsage() LLMUsage { return c.usage } - -// parseOpenAIChunk parses a single data payload from an OpenAI streaming SSE response. +// parseOpenAIChunk parses a single OpenAI streaming SSE payload. func parseOpenAIChunk(data []byte) (LLMResponseChunk, error) { - var chunk openaiChunk + var chunk openAIChunk if err := json.Unmarshal(data, &chunk); err != nil { return nil, err } - return &openaiLLMResponseChunk{usage: openaiUsageToLLM(chunk.Usage)}, nil + result := &openAILLMResponseChunk{usage: openAIUsageToLLM(chunk.Usage)} + if len(chunk.Choices) > 0 { + result.answer = chunk.Choices[0].Delta.Content + result.reasoning = chunk.Choices[0].Delta.ReasoningContent + result.toolCalls = chunk.Choices[0].Delta.ToolCalls + } + if chunk.Usage.PromptTokensDetails != nil { + result.cachedTokens = chunk.Usage.PromptTokensDetails.CachedTokens + result.inputTokenDetails = chunk.Usage.PromptTokensDetails + } + if chunk.Usage.CompletionTokensDetails != nil { + result.reasoningTokens = chunk.Usage.CompletionTokensDetails.ReasoningTokens + result.outputTokenDetails = chunk.Usage.CompletionTokensDetails + } + return result, nil } -// openaiUsageToLLM converts an openaiUsage to an LLMUsage. -// Returns the zero value when u is nil. -func openaiUsageToLLM(u openaiUsage) LLMUsage { +// openAIUsageToLLM converts an OpenAI usage payload to the common LLMUsage shape. +func openAIUsageToLLM(u openAIUsage) LLMUsage { return LLMUsage{ InputTokens: u.PromptTokens, OutputTokens: u.CompletionTokens, @@ -103,23 +246,15 @@ func openaiUsageToLLM(u openaiUsage) LLMUsage { } } -// --- SSE accumulator --- - -var openaiSSEDataPrefix = []byte("data: ") - -// openaiSSEParser accumulates usage information from an OpenAI streaming SSE response. -// It consumes body chunks as they arrive and produces an LLMResponse when finished. -type openaiSSEParser struct { - buf []byte - done bool - usage LLMUsage -} - -func newOpenAISSEParser() *openaiSSEParser { - return &openaiSSEParser{} +// newOpenAISSEParser creates a parser for incremental OpenAI SSE accumulation. +func newOpenAISSEParser() *openAISSEParser { + return &openAISSEParser{ + toolCallsByIndex: map[int]*openAIToolCall{}, + } } -func (a *openaiSSEParser) Feed(data []byte) error { +// Feed appends a new response body chunk and parses any complete SSE events. +func (a *openAISSEParser) Feed(data []byte) error { if a.done { return nil } @@ -127,7 +262,8 @@ func (a *openaiSSEParser) Feed(data []byte) error { return a.parseEvents() } -func (a *openaiSSEParser) parseEvents() error { +// parseEvents processes complete SSE lines accumulated in the internal buffer. +func (a *openAISSEParser) parseEvents() error { for { idx := bytes.IndexByte(a.buf, '\n') if idx < 0 { @@ -136,10 +272,10 @@ func (a *openaiSSEParser) parseEvents() error { line := bytes.TrimSpace(a.buf[:idx]) a.buf = a.buf[idx+1:] - if !bytes.HasPrefix(line, openaiSSEDataPrefix) { + if !bytes.HasPrefix(line, []byte("data: ")) { continue } - payload := bytes.TrimPrefix(line, openaiSSEDataPrefix) + payload := bytes.TrimPrefix(line, []byte("data: ")) if bytes.Equal(payload, []byte("[DONE]")) { a.done = true return nil @@ -148,8 +284,50 @@ func (a *openaiSSEParser) parseEvents() error { if err != nil { return fmt.Errorf("llm-proxy: failed to parse OpenAI streaming chunk: %w", err) } - if u := chunk.GetUsage(); u != (LLMUsage{}) { - a.usage = u + if usage := chunk.GetUsage(); usage != (LLMUsage{}) { + a.usage = usage + } + if chunkOpenAI, ok := chunk.(*openAILLMResponseChunk); ok { + if chunkOpenAI.cachedTokens > 0 { + a.cachedTokens = chunkOpenAI.cachedTokens + } + if chunkOpenAI.reasoningTokens > 0 { + a.reasoningTokens = chunkOpenAI.reasoningTokens + } + if chunkOpenAI.inputTokenDetails != nil { + a.inputTokenDetails = chunkOpenAI.inputTokenDetails + } + if chunkOpenAI.outputTokenDetails != nil { + a.outputTokenDetails = chunkOpenAI.outputTokenDetails + } + } + if chunk.HasTextToken() { + a.seenTextToken = true + } + a.answer += chunk.GetAnswer() + a.reasoning += chunk.GetReasoning() + for _, delta := range chunk.GetToolCalls().([]openAIStreamingToolCallDelta) { + if tc, ok := a.toolCallsByIndex[delta.Index]; ok { + if delta.ID != "" { + tc.ID = delta.ID + } + if delta.Type != "" { + tc.Type = delta.Type + } + if delta.Function.Name != "" { + tc.Function.Name = delta.Function.Name + } + tc.Function.Arguments += delta.Function.Arguments + } else { + a.toolCallsByIndex[delta.Index] = &openAIToolCall{ + ID: delta.ID, + Type: delta.Type, + Function: openAIToolCallFunction{ + Name: delta.Function.Name, + Arguments: delta.Function.Arguments, + }, + } + } } } } @@ -167,6 +345,74 @@ func (f *openaiFactory) ParseResponse(body []byte) (LLMResponse, error) { func (f *openaiFactory) NewSSEParser() SSEParser { return newOpenAISSEParser() } -func (a *openaiSSEParser) Finish() (LLMResponse, error) { - return &openaiLLMResponse{usage: a.usage}, nil +// Finish finalises the stream and returns the accumulated response fields. +func (a *openAISSEParser) Finish() (LLMResponse, error) { + var toolCalls []openAIToolCall + if len(a.toolCallsByIndex) > 0 { + indexes := make([]int, 0, len(a.toolCallsByIndex)) + for idx := range a.toolCallsByIndex { + indexes = append(indexes, idx) + } + sort.Ints(indexes) + for _, idx := range indexes { + toolCalls = append(toolCalls, *a.toolCallsByIndex[idx]) + } + } + return &openAILLMResponse{ + usage: a.usage, + answer: a.answer, + reasoning: a.reasoning, + toolCalls: toolCalls, + cachedTokens: a.cachedTokens, + reasoningTokens: a.reasoningTokens, + inputTokenDetails: a.inputTokenDetails, + outputTokenDetails: a.outputTokenDetails, + }, nil +} + +// SeenTextToken reports whether the stream has emitted a real text token yet. +func (a *openAISSEParser) SeenTextToken() bool { return a.seenTextToken } + +// extractOpenAIQuestion returns the last user message content from the request. +func extractOpenAIQuestion(messages []openAIRequestMessage) string { + for i := len(messages) - 1; i >= 0; i-- { + if messages[i].Role != "user" { + continue + } + return extractOpenAIMessageContent(messages[i].Content) + } + return "" +} + +// extractOpenAISystem returns the first system message content from the request. +func extractOpenAISystem(messages []openAIRequestMessage) string { + for i := 0; i < len(messages); i++ { + if messages[i].Role != "system" { + continue + } + return extractOpenAIMessageContent(messages[i].Content) + } + return "" +} + +// extractOpenAIMessageContent extracts text from either string or array-form content. +func extractOpenAIMessageContent(content any) string { + if s, ok := content.(string); ok { + return s + } + raw, err := json.Marshal(content) + if err != nil { + return "" + } + var parts []openAIContentPart + if err := json.Unmarshal(raw, &parts); err != nil { + return "" + } + texts := make([]string, 0, len(parts)) + for _, p := range parts { + if p.Type == "text" && p.Text != "" { + texts = append(texts, p.Text) + } + } + return strings.Join(texts, "\n") } diff --git a/extensions/composer/llm-proxy/plugin.go b/extensions/composer/llm-proxy/plugin.go index 59073509..432c41e5 100644 --- a/extensions/composer/llm-proxy/plugin.go +++ b/extensions/composer/llm-proxy/plugin.go @@ -16,6 +16,7 @@ package llmproxy import ( "encoding/json" "fmt" + "reflect" "strings" "time" @@ -44,6 +45,9 @@ type llmConfig struct { // Factory is the factory for parsing requests/responses for this rule; set during // filter initialization. Factory LLMFactory `json:"-"` + // DefaultRule marks rules that were auto-injected by ValidateAndParse rather + // than explicitly provided by the user. + DefaultRule bool `json:"-"` } func (c *llmConfig) ValidateAndParse() error { @@ -81,6 +85,12 @@ type llmProxyConfig struct { MetadataNamespace string `json:"metadata_namespace"` // Header key to set the extracted model name if any. LLMModelHeader string `json:"llm_model_header"` + // Emit the full built-in structured log payload when true. + UseDefaultAttributes bool `json:"use_default_attributes"` + // Emit the lightweight structured log payload when true. + UseDefaultResponseAttributes bool `json:"use_default_response_attributes"` + // Optional request header to extract a session or conversation ID from. + SessionIDHeader string `json:"session_id_header"` // ClearRouteCache indicates whether to clear route cache to reselect route // based on the extracted model and metadata. // Only one of ClearRouteCache and ClearClusterCache can be true. @@ -114,14 +124,16 @@ func (c *llmProxyConfig) ValidateAndParse() error { // for the most common case. if !hasOpenAI { c.LLMConfigs = append([]llmConfig{{ - Matcher: pkg.StringMatcher{Suffix: "/v1/chat/completions"}, - Kind: KindOpenAI, + Matcher: pkg.StringMatcher{Suffix: "/v1/chat/completions"}, + Kind: KindOpenAI, + DefaultRule: true, }}, c.LLMConfigs...) } if !hasAnthropic { c.LLMConfigs = append([]llmConfig{{ - Matcher: pkg.StringMatcher{Suffix: "/v1/messages"}, - Kind: KindAnthropic, + Matcher: pkg.StringMatcher{Suffix: "/v1/messages"}, + Kind: KindAnthropic, + DefaultRule: true, }}, c.LLMConfigs...) } @@ -196,9 +208,12 @@ type llmProxyFilter struct { sseParser SSEParser // llmReq holds the parsed LLM request; set after the request body is processed. - llmReq LLMRequest - model string - isStream bool + llmReq LLMRequest + model string + isStream bool + sessionID string + question string + system string // llmResp holds the parsed LLM response; set after the response body is processed. llmResp LLMResponse @@ -222,10 +237,29 @@ func (f *llmProxyFilter) matchRule(path string) *llmConfig { return nil } +func (f *llmProxyFilter) matchDefaultRule(path string) *llmConfig { + for i := range f.config.LLMConfigs { + cfg := &f.config.LLMConfigs[i] + if !isDefaultWellKnownRule(cfg) { + continue + } + if cfg.Matcher.Matches(path) { + return cfg + } + } + return nil +} + func (f *llmProxyFilter) OnRequestHeaders(headers shared.HeaderMap, endOfStream bool) shared.HeadersStatus { pathBuffer, _ := f.handle.GetAttributeString(shared.AttributeIDRequestPath) path := pathBuffer.ToUnsafeString() rule := f.matchRule(path) + if rule == nil { + strippedPath := stripQueryString(path) + if strippedPath != path { + rule = f.matchDefaultRule(strippedPath) + } + } if rule == nil || rule.Factory == nil { // Unknown path: pass through without any processing. f.handle.Log(shared.LogLevelDebug, "llm-proxy: no matching valid rule found for path %q", path) @@ -318,9 +352,7 @@ func (f *llmProxyFilter) OnResponseBody(body shared.BodyBuffer, endOfStream bool return shared.BodyStatusContinue } - // Record the time when the first chunk arrives even for non-streaming responses, - // so that TTFT and TPOT can be computed consistently. - if f.firstChunkAt.IsZero() { + if f.sseParser == nil && f.firstChunkAt.IsZero() { f.firstChunkAt = time.Now() } @@ -332,6 +364,9 @@ func (f *llmProxyFilter) OnResponseBody(body shared.BodyBuffer, endOfStream bool f.onError(fmt.Sprintf("event stream error: %s", err.Error())) return shared.BodyStatusContinue } + if f.firstChunkAt.IsZero() && f.sseParser.SeenTextToken() { + f.firstChunkAt = time.Now() + } } } if endOfStream { @@ -385,6 +420,20 @@ func (f *llmProxyFilter) onRequestSuccess() { f.handle.SetMetadata(ns, "kind", f.kind) f.handle.SetMetadata(ns, "model", f.model) f.handle.SetMetadata(ns, "is_stream", f.isStream) + if f.isStream { + f.handle.SetMetadata(ns, "response_type", "stream") + } else { + f.handle.SetMetadata(ns, "response_type", "nonstream") + } + if f.sessionID != "" { + f.handle.SetMetadata(ns, "session_id", f.sessionID) + } + if f.question != "" { + f.handle.SetMetadata(ns, "question", f.question) + } + if f.system != "" { + f.handle.SetMetadata(ns, "system", f.system) + } f.handle.IncrementCounterValue(f.config.stats.requestTotal, 1, f.kind, f.model) @@ -401,6 +450,27 @@ func (f *llmProxyFilter) onResponseSuccess() { f.handle.SetMetadata(ns, "input_tokens", f.usage.InputTokens) f.handle.SetMetadata(ns, "output_tokens", f.usage.OutputTokens) f.handle.SetMetadata(ns, "total_tokens", f.usage.TotalTokens) + if answer := f.llmResp.GetAnswer(); answer != "" { + f.handle.SetMetadata(ns, "answer", answer) + } + if reasoning := f.llmResp.GetReasoning(); reasoning != "" { + f.handle.SetMetadata(ns, "reasoning", reasoning) + } + if reasoningTokens := f.llmResp.GetReasoningTokens(); reasoningTokens > 0 { + f.handle.SetMetadata(ns, "reasoning_tokens", reasoningTokens) + } + if cachedTokens := f.llmResp.GetCachedTokens(); cachedTokens > 0 { + f.handle.SetMetadata(ns, "cached_tokens", cachedTokens) + } + if toolCalls := f.llmResp.GetToolCalls(); hasToolCalls(toolCalls) { + f.handle.SetMetadata(ns, "tool_calls", toolCalls) + } + if inputDetails := f.llmResp.GetInputTokenDetails(); inputDetails != nil { + f.handle.SetMetadata(ns, "input_token_details", inputDetails) + } + if outputDetails := f.llmResp.GetOutputTokenDetails(); outputDetails != nil { + f.handle.SetMetadata(ns, "output_token_details", outputDetails) + } // Set tokens stats. Token counts are always non-negative; clamp to 0 to satisfy // the static analyser before converting to uint64. @@ -413,6 +483,7 @@ func (f *llmProxyFilter) onResponseSuccess() { // Handle some corner cases to avoid error. if f.requestSentAt.IsZero() || f.firstChunkAt.IsZero() || f.usage.OutputTokens == 0 { + f.emitStructuredLog() return } @@ -426,6 +497,7 @@ func (f *llmProxyFilter) onResponseSuccess() { f.handle.RecordHistogramValue(f.config.stats.requestTTFT, uint64(max(ttfp, 0)), f.kind, f.model) // nolint:gosec f.handle.RecordHistogramValue(f.config.stats.requestTPOT, uint64(max(tpot, 0)), f.kind, f.model) + f.emitStructuredLog() } // parseRequestBody reads the complete request body, parses it via the matched @@ -448,6 +520,9 @@ func (f *llmProxyFilter) parseRequestBody() { } f.isStream = req.IsStream() f.llmReq = req + f.question = req.GetQuestion() + f.system = req.GetSystem() + f.sessionID = f.extractSessionID() // Get the request correctly parsed and metadata set, we can set some metadata or stats now. f.onRequestSuccess() @@ -487,6 +562,115 @@ func (f *llmProxyFilter) finishStreamingResponse() { f.onResponseSuccess() } +func (f *llmProxyFilter) emitStructuredLog() { + if !f.config.UseDefaultResponseAttributes && !f.config.UseDefaultAttributes { + return + } + + responseType := "nonstream" + if f.isStream { + responseType = "stream" + } + entry := map[string]any{ + "kind": f.kind, + "model": f.model, + "response_type": responseType, + "input_tokens": f.usage.InputTokens, + "output_tokens": f.usage.OutputTokens, + "total_tokens": f.usage.TotalTokens, + } + if f.sessionID != "" { + entry["session_id"] = f.sessionID + } + if f.config.UseDefaultAttributes { + if f.question != "" { + entry["question"] = f.question + } + if f.system != "" { + entry["system"] = f.system + } + if answer := f.llmResp.GetAnswer(); answer != "" { + entry["answer"] = answer + } + if reasoning := f.llmResp.GetReasoning(); reasoning != "" { + entry["reasoning"] = reasoning + } + if reasoningTokens := f.llmResp.GetReasoningTokens(); reasoningTokens > 0 { + entry["reasoning_tokens"] = reasoningTokens + } + if cachedTokens := f.llmResp.GetCachedTokens(); cachedTokens > 0 { + entry["cached_tokens"] = cachedTokens + } + if toolCalls := f.llmResp.GetToolCalls(); hasToolCalls(toolCalls) { + entry["tool_calls"] = toolCalls + } + if inputDetails := f.llmResp.GetInputTokenDetails(); inputDetails != nil { + entry["input_token_details"] = inputDetails + } + if outputDetails := f.llmResp.GetOutputTokenDetails(); outputDetails != nil { + entry["output_token_details"] = outputDetails + } + } + if !f.requestSentAt.IsZero() { + entry["llm_service_duration_ms"] = time.Since(f.requestSentAt).Milliseconds() + } + if f.isStream && !f.requestSentAt.IsZero() && !f.firstChunkAt.IsZero() { + entry["llm_first_token_duration_ms"] = f.firstChunkAt.Sub(f.requestSentAt).Milliseconds() + } + if payload, err := json.Marshal(entry); err == nil { + f.handle.Log(shared.LogLevelInfo, "%s", string(payload)) + } +} + +func (f *llmProxyFilter) extractSessionID() string { + if f.config.SessionIDHeader == "" { + return "" + } + return f.handle.RequestHeaders().GetOne(f.config.SessionIDHeader).ToUnsafeString() +} + +func stripQueryString(path string) string { + if idx := strings.IndexByte(path, '?'); idx >= 0 { + return path[:idx] + } + return path +} + +func isDefaultWellKnownRule(cfg *llmConfig) bool { + if !cfg.DefaultRule { + return false + } + if cfg.Matcher.Prefix != "" || cfg.Matcher.Regex != "" { + return false + } + switch cfg.Kind { + case KindOpenAI: + return cfg.Matcher.Suffix == "/v1/chat/completions" + case KindAnthropic: + return cfg.Matcher.Suffix == "/v1/messages" + default: + return false + } +} + +func hasToolCalls(toolCalls any) bool { + if toolCalls == nil { + return false + } + v := reflect.ValueOf(toolCalls) + if !v.IsValid() { + return false + } + switch v.Kind() { + case reflect.Slice, reflect.Array, reflect.Map, reflect.String: + return v.Len() > 0 + case reflect.Interface, reflect.Pointer: + return !v.IsNil() + default: + return true + } +} + // ExtensionName is the name used to refer to this plugin in Envoy configuration. const ExtensionName = "llm-proxy" diff --git a/extensions/composer/llm-proxy/plugin_test.go b/extensions/composer/llm-proxy/plugin_test.go index 52139c14..575eb5b5 100644 --- a/extensions/composer/llm-proxy/plugin_test.go +++ b/extensions/composer/llm-proxy/plugin_test.go @@ -6,7 +6,9 @@ package llmproxy import ( + "encoding/json" "testing" + "time" "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/shared" "github.com/envoyproxy/envoy/source/extensions/dynamic_modules/sdk/go/shared/fake" @@ -222,6 +224,74 @@ func TestOnRequestHeaders_MatchedPath_HasBody(t *testing.T) { require.True(t, filter.matched) } +func TestOnRequestHeaders_DefaultSuffixRule_StripsQueryString(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + mockHandle.EXPECT().GetAttributeString(shared.AttributeIDRequestPath). + Return(pkg.UnsafeBufferFromString("/v1/chat/completions?api-version=2024-10-21"), true) + mockHandle.EXPECT().Log(shared.LogLevelDebug, gomock.Any(), gomock.Any(), gomock.Any()).Times(1) + + cfg := &llmProxyConfig{} + require.NoError(t, cfg.ValidateAndParse()) + + filter := &llmProxyFilter{handle: mockHandle, config: cfg} + headers := fake.NewFakeHeaderMap(map[string][]string{"content-type": {"application/json"}}) + result := filter.OnRequestHeaders(headers, false) + require.Equal(t, shared.HeadersStatusStop, result) + require.True(t, filter.matched) + require.Equal(t, KindOpenAI, filter.kind) +} + +func TestOnRequestHeaders_CustomMatcher_PreservesQueryStringSemantics(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + mockHandle.EXPECT().GetAttributeString(shared.AttributeIDRequestPath). + Return(pkg.UnsafeBufferFromString("/custom/v1/chat?provider=openai"), true) + mockHandle.EXPECT().Log(shared.LogLevelDebug, gomock.Any(), gomock.Any(), gomock.Any()).Times(1) + + cfg := &llmProxyConfig{ + LLMConfigs: []llmConfig{{ + Matcher: pkg.StringMatcher{Suffix: "?provider=openai"}, + Kind: KindCustom, + Factory: &customFactory{}, + }}, + MetadataNamespace: defaultMetadataNamespace, + } + + filter := &llmProxyFilter{handle: mockHandle, config: cfg} + headers := fake.NewFakeHeaderMap(map[string][]string{"content-type": {"application/json"}}) + result := filter.OnRequestHeaders(headers, false) + require.Equal(t, shared.HeadersStatusStop, result) + require.True(t, filter.matched) + require.Equal(t, KindCustom, filter.kind) +} + +func TestOnRequestHeaders_UserProvidedSuffixRule_DoesNotUseQueryFallback(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + mockHandle.EXPECT().GetAttributeString(shared.AttributeIDRequestPath). + Return(pkg.UnsafeBufferFromString("/v1/chat/completions?api-version=2024-10-21"), true) + mockHandle.EXPECT().Log(shared.LogLevelDebug, gomock.Any(), gomock.Any()).Times(1) + + cfg := &llmProxyConfig{ + LLMConfigs: []llmConfig{{ + Matcher: pkg.StringMatcher{Suffix: "/v1/chat/completions"}, + Kind: KindOpenAI, + Factory: &openaiFactory{}, + }}, + MetadataNamespace: defaultMetadataNamespace, + } + + filter := &llmProxyFilter{handle: mockHandle, config: cfg} + headers := fake.NewFakeHeaderMap(map[string][]string{"content-type": {"application/json"}}) + result := filter.OnRequestHeaders(headers, false) + require.Equal(t, shared.HeadersStatusContinue, result) + require.False(t, filter.matched) +} + func TestOnRequestHeaders_NonJSONContentType_Error(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -275,6 +345,7 @@ func TestOnRequestBody_OpenAI_SetsMetadata(t *testing.T) { mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "kind", "openai").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "model", "gpt-4o").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "is_stream", false).Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "response_type", "nonstream").Times(1) mockHandle.EXPECT().IncrementCounterValue(idRequestTotal, uint64(1), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) filter := &llmProxyFilter{ @@ -305,6 +376,7 @@ func TestOnRequestBody_Anthropic_SetsMetadata(t *testing.T) { mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "kind", "anthropic").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "model", "claude-3-5-sonnet-20241022").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "is_stream", true).Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "response_type", "stream").Times(1) mockHandle.EXPECT().IncrementCounterValue(idRequestTotal, uint64(1), "anthropic", "claude-3-5-sonnet-20241022").Return(shared.MetricsSuccess).Times(1) filter := &llmProxyFilter{ @@ -318,6 +390,47 @@ func TestOnRequestBody_Anthropic_SetsMetadata(t *testing.T) { require.Equal(t, shared.BodyStatusContinue, result) } +func TestOnRequestBody_OpenAI_RicherMetadata(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + + body := []byte(`{ + "model":"gpt-4o", + "stream":false, + "messages":[ + {"role":"system","content":"You are concise."}, + {"role":"user","content":"What is 2+2?"} + ] + }`) + mockHandle.EXPECT().BufferedRequestBody().Return(fake.NewFakeBodyBuffer(body)).AnyTimes() + mockHandle.EXPECT().ReceivedRequestBody().Return(nil).AnyTimes() + mockHandle.EXPECT().RequestHeaders().Return(fake.NewFakeHeaderMap(map[string][]string{ + "x-session-id": {"sess-123"}, + })).AnyTimes() + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "kind", "openai").Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "model", "gpt-4o").Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "is_stream", false).Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "response_type", "nonstream").Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "session_id", "sess-123").Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "question", "What is 2+2?").Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "system", "You are concise.").Times(1) + mockHandle.EXPECT().IncrementCounterValue(idRequestTotal, uint64(1), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + + cfg := defaultCfgWithStats(newTestStats(ctrl)) + cfg.SessionIDHeader = "x-session-id" + + filter := &llmProxyFilter{ + handle: mockHandle, + config: cfg, + matched: true, + kind: KindOpenAI, + factory: &openaiFactory{}, + } + result := filter.OnRequestBody(fake.NewFakeBodyBuffer(body), true) + require.Equal(t, shared.BodyStatusContinue, result) +} + func TestOnRequestBody_InvalidJSON_LogsDebug(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -349,6 +462,7 @@ func TestOnRequestTrailers_NotProcessed_ParsesBody(t *testing.T) { mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "kind", "openai").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "model", "gpt-4o").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "is_stream", false).Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "response_type", "nonstream").Times(1) mockHandle.EXPECT().IncrementCounterValue(idRequestTotal, uint64(1), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) filter := &llmProxyFilter{ @@ -510,6 +624,157 @@ func TestOnResponseBody_Anthropic_SetsUsageMetadata(t *testing.T) { require.Equal(t, shared.BodyStatusContinue, result) } +func TestOnResponseBody_OpenAI_RicherMetadataAndLog(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + + body := []byte(`{ + "choices":[ + {"message":{ + "content":"4", + "reasoning_content":"Simple arithmetic", + "tool_calls":[ + {"id":"call_1","type":"function","function":{"name":"get_weather","arguments":"{\"loc\":\"NYC\"}"}} + ] + }} + ], + "usage":{ + "prompt_tokens":100, + "completion_tokens":50, + "total_tokens":150, + "prompt_tokens_details":{"cached_tokens":80}, + "completion_tokens_details":{"reasoning_tokens":25} + } + }`) + mockHandle.EXPECT().BufferedResponseBody().Return(fake.NewFakeBodyBuffer(body)).AnyTimes() + mockHandle.EXPECT().ReceivedResponseBody().Return(nil).AnyTimes() + captured := map[string]any{} + mockHandle.EXPECT().SetMetadata(gomock.Any(), gomock.Any(), gomock.Any()).Do( + func(_ string, key string, value any) { + captured[key] = value + }, + ).AnyTimes() + mockHandle.EXPECT().IncrementCounterValue(idInputTokens, uint64(100), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().IncrementCounterValue(idOutputTokens, uint64(50), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().IncrementCounterValue(idTotalTokens, uint64(150), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().RecordHistogramValue(idTTFT, gomock.Any(), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().RecordHistogramValue(idTPOT, gomock.Any(), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().Log(shared.LogLevelInfo, gomock.Any(), gomock.Any()).Do(func(_ shared.LogLevel, _ string, payload string) { + var entry map[string]any + require.NoError(t, json.Unmarshal([]byte(payload), &entry)) + require.Equal(t, "What is 2+2?", entry["question"]) + require.Equal(t, "You are concise.", entry["system"]) + require.Equal(t, "4", entry["answer"]) + require.Equal(t, "Simple arithmetic", entry["reasoning"]) + require.Equal(t, "sess-123", entry["session_id"]) + }).Times(1) + + cfg := defaultCfgWithStats(newTestStats(ctrl)) + cfg.UseDefaultAttributes = true + cfg.SessionIDHeader = "x-session-id" + + filter := &llmProxyFilter{ + handle: mockHandle, + config: cfg, + matched: true, + factory: &openaiFactory{}, + model: "gpt-4o", + kind: KindOpenAI, + llmReq: &openAILLMRequest{model: "gpt-4o", question: "What is 2+2?", system: "You are concise."}, + question: "What is 2+2?", + system: "You are concise.", + sessionID: "sess-123", + requestSentAt: time.Now().Add(-100 * time.Millisecond), + firstChunkAt: time.Now().Add(-50 * time.Millisecond), + } + result := filter.OnResponseBody(fake.NewFakeBodyBuffer(body), true) + require.Equal(t, shared.BodyStatusContinue, result) + require.Equal(t, "4", captured["answer"]) + require.Equal(t, "Simple arithmetic", captured["reasoning"]) + require.EqualValues(t, 25, captured["reasoning_tokens"]) + require.EqualValues(t, 80, captured["cached_tokens"]) + toolCalls, ok := captured["tool_calls"].([]openAIToolCall) + require.True(t, ok) + require.Len(t, toolCalls, 1) +} + +func TestOnResponseBody_SSE_OpenAI_RicherDetails(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + + captured := map[string]any{} + mockHandle.EXPECT().SetMetadata(gomock.Any(), gomock.Any(), gomock.Any()).Do( + func(_ string, key string, value any) { + captured[key] = value + }, + ).AnyTimes() + mockHandle.EXPECT().IncrementCounterValue(idInputTokens, uint64(100), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().IncrementCounterValue(idOutputTokens, uint64(50), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().IncrementCounterValue(idTotalTokens, uint64(150), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().RecordHistogramValue(idTTFT, gomock.Any(), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().RecordHistogramValue(idTPOT, gomock.Any(), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + + acc := newOpenAISSEParser() + filter := &llmProxyFilter{ + handle: mockHandle, + config: defaultCfgWithStats(newTestStats(ctrl)), + matched: true, + factory: &openaiFactory{}, + sseParser: acc, + model: "gpt-4o", + kind: KindOpenAI, + requestSentAt: time.Now().Add(-100 * time.Millisecond), + } + + chunk := fake.NewFakeBodyBuffer([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}],\"usage\":{\"prompt_tokens\":100,\"completion_tokens\":50,\"total_tokens\":150,\"prompt_tokens_details\":{\"cached_tokens\":80},\"completion_tokens_details\":{\"reasoning_tokens\":25}}}\n")) + done := fake.NewFakeBodyBuffer([]byte("data: [DONE]\n")) + + require.Equal(t, shared.BodyStatusContinue, filter.OnResponseBody(chunk, false)) + require.Equal(t, shared.BodyStatusContinue, filter.OnResponseBody(done, true)) + require.EqualValues(t, 25, captured["reasoning_tokens"]) + require.EqualValues(t, 80, captured["cached_tokens"]) +} + +func TestOnResponseBody_SSE_Anthropic_RicherDetails(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + mockHandle := mocks.NewMockHttpFilterHandle(ctrl) + + captured := map[string]any{} + mockHandle.EXPECT().SetMetadata(gomock.Any(), gomock.Any(), gomock.Any()).Do( + func(_ string, key string, value any) { + captured[key] = value + }, + ).AnyTimes() + mockHandle.EXPECT().IncrementCounterValue(idInputTokens, uint64(9), "anthropic", "claude-sonnet-4-20250514").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().IncrementCounterValue(idOutputTokens, uint64(5), "anthropic", "claude-sonnet-4-20250514").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().IncrementCounterValue(idTotalTokens, uint64(14), "anthropic", "claude-sonnet-4-20250514").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().RecordHistogramValue(idTTFT, gomock.Any(), "anthropic", "claude-sonnet-4-20250514").Return(shared.MetricsSuccess).Times(1) + mockHandle.EXPECT().RecordHistogramValue(idTPOT, gomock.Any(), "anthropic", "claude-sonnet-4-20250514").Return(shared.MetricsSuccess).Times(1) + + acc := newAnthropicSSEParser() + filter := &llmProxyFilter{ + handle: mockHandle, + config: defaultCfgWithStats(newTestStats(ctrl)), + matched: true, + factory: &anthropicFactory{}, + sseParser: acc, + model: "claude-sonnet-4-20250514", + kind: KindAnthropic, + requestSentAt: time.Now().Add(-100 * time.Millisecond), + } + + chunk1 := fake.NewFakeBodyBuffer([]byte("event: message_start\ndata: {\"message\":{\"usage\":{\"input_tokens\":9,\"cache_creation_input_tokens\":20,\"cache_read_input_tokens\":30}}}\n\n")) + chunk2 := fake.NewFakeBodyBuffer([]byte("event: content_block_start\ndata: {\"type\":\"content_block_start\",\"index\":0,\"content_block\":{\"type\":\"text\",\"text\":\"hello\"}}\n\nevent: message_delta\ndata: {\"usage\":{\"output_tokens\":5}}\n\nevent: message_stop\ndata: {\"type\":\"message_stop\"}\n\n")) + + require.Equal(t, shared.BodyStatusContinue, filter.OnResponseBody(chunk1, false)) + require.Equal(t, shared.BodyStatusContinue, filter.OnResponseBody(chunk2, true)) + require.EqualValues(t, 30, captured["cached_tokens"]) + require.NotNil(t, captured["input_token_details"]) +} + // TestOnResponseBody_NoUsageInResponse verifies that onResponseSuccess always sets // all three token metadata keys (even as zero values) when no usage is present. func TestOnResponseBody_NoUsageInResponse(t *testing.T) { @@ -674,6 +939,7 @@ func TestCustomAPIType_RequestParsedLikeOpenAI(t *testing.T) { mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "kind", "custom").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "model", "my-model").Times(1) mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "is_stream", false).Times(1) + mockHandle.EXPECT().SetMetadata(defaultMetadataNamespace, "response_type", "nonstream").Times(1) mockHandle.EXPECT().BufferedRequestBody().Return(nil).AnyTimes() mockHandle.EXPECT().ReceivedRequestBody().Return( fake.NewFakeBodyBuffer([]byte(`{"model":"my-model","messages":[]}`)), diff --git a/extensions/composer/llm-proxy/stats_test.go b/extensions/composer/llm-proxy/stats_test.go index 63bc1043..f484f557 100644 --- a/extensions/composer/llm-proxy/stats_test.go +++ b/extensions/composer/llm-proxy/stats_test.go @@ -138,6 +138,35 @@ func TestStats_NonStreamingResponse_RecordsTokenCounters(t *testing.T) { filter.OnResponseBody(fake.NewFakeBodyBuffer(body), true) } +func TestStats_NonStreamingResponse_RecordsTTFT_TPOT_WhenRequestSentAtSet(t *testing.T) { + ctrl := gomock.NewController(t) + defer ctrl.Finish() + s := newTestStats(ctrl) + + body := []byte(`{"choices":[],"usage":{"prompt_tokens":10,"completion_tokens":20,"total_tokens":30}}`) + handle := mocks.NewMockHttpFilterHandle(ctrl) + handle.EXPECT().BufferedResponseBody().Return(fake.NewFakeBodyBuffer(body)).AnyTimes() + handle.EXPECT().ReceivedResponseBody().Return(nil).AnyTimes() + handle.EXPECT().SetMetadata(gomock.Any(), gomock.Any(), gomock.Any()).AnyTimes() + handle.EXPECT().IncrementCounterValue(idInputTokens, uint64(10), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + handle.EXPECT().IncrementCounterValue(idOutputTokens, uint64(20), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + handle.EXPECT().IncrementCounterValue(idTotalTokens, uint64(30), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + handle.EXPECT().RecordHistogramValue(idTTFT, gomock.Any(), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + handle.EXPECT().RecordHistogramValue(idTPOT, gomock.Any(), "openai", "gpt-4o").Return(shared.MetricsSuccess).Times(1) + + filter := &llmProxyFilter{ + handle: handle, + config: defaultCfgWithStats(s), + matched: true, + kind: KindOpenAI, + factory: &openaiFactory{}, + model: "gpt-4o", + requestSentAt: time.Now().Add(-100 * time.Millisecond), + } + filter.OnResponseBody(fake.NewFakeBodyBuffer(body), true) + require.False(t, filter.firstChunkAt.IsZero(), "firstChunkAt must be set for non-streaming responses") +} + func TestStats_StreamingResponse_RecordsTTFT_TPOT_Tokens(t *testing.T) { ctrl := gomock.NewController(t) defer ctrl.Finish() @@ -161,11 +190,11 @@ func TestStats_StreamingResponse_RecordsTTFT_TPOT_Tokens(t *testing.T) { requestSentAt: time.Now().Add(-100 * time.Millisecond), // simulate sent 100 ms ago } - chunk := fake.NewFakeBodyBuffer([]byte("data: {\"choices\":[],\"usage\":{\"prompt_tokens\":8,\"completion_tokens\":4,\"total_tokens\":12}}\n")) + chunk := fake.NewFakeBodyBuffer([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"hello\"}}],\"usage\":{\"prompt_tokens\":8,\"completion_tokens\":4,\"total_tokens\":12}}\n")) done := fake.NewFakeBodyBuffer([]byte("data: [DONE]\n")) require.Equal(t, shared.BodyStatusContinue, filter.OnResponseBody(chunk, false)) - require.False(t, filter.firstChunkAt.IsZero(), "firstChunkAt must be set on first SSE body chunk") + require.False(t, filter.firstChunkAt.IsZero(), "firstChunkAt must be set on first text token chunk") require.Equal(t, shared.BodyStatusContinue, filter.OnResponseBody(done, true)) } diff --git a/website/public/extensions.json b/website/public/extensions.json index 017d9303..bc78bb79 100644 --- a/website/public/extensions.json +++ b/website/public/extensions.json @@ -533,8 +533,8 @@ ], "author": "Tetrate", "featured": true, - "description": "Routes LLM API requests by the module name and monitors token usage and latency via Envoy metadata and metrics.", - "longDescription": "An HTTP filter plugin that inspects incoming requests against a set of configured\npath-matcher rules to identify the LLM provider API in use (OpenAI Chat Completions,\nAnthropic Messages, or a custom OpenAI-compatible API). Once a rule matches, the filter:\n\n1. Parses the request body to extract the model name and streaming flag, then\n writes them to Envoy's dynamic filter metadata.\n2. Parses the response body (JSON for non-streaming, SSE for streaming) to extract\n token-usage information and writes it to filter metadata.\n3. Records Envoy metrics (counters and histograms) for request counts, token usage,\n time-to-first-token (TTFT), and time-per-output-token (TPOT).\n\nRequests whose path does not match any rule are passed through without modification.\n\nIf no rule is explicitly configured for OpenAI or Anthropic, the filter automatically\nadds default suffix-matcher rules for `/v1/chat/completions` (OpenAI) and\n`/v1/messages` (Anthropic), so it works out of the box with no configuration.\n\n## Metadata Keys\n\nAll keys are written under the configured `metadata_namespace`\n(default: `io.builtonenvoy.llm-proxy`).\n\n| Key | Type | Description |\n|-----|------|-------------|\n| `kind` | string | API kind: `\"openai\"`, `\"anthropic\"`, or `\"custom\"` |\n| `model` | string | Model name extracted from the request body |\n| `is_stream` | bool | Whether the request asks for a streaming (SSE) response |\n| `input_tokens` | uint32 | Input / prompt token count from the response |\n| `output_tokens` | uint32 | Output / completion token count from the response |\n| `total_tokens` | uint32 | Total token count from the response |\n| `request_ttft` | int64 | Time to first token in milliseconds |\n| `request_tpot` | int64 | Average time per output token in milliseconds |\n\n## Metrics\n\nAll metrics are tagged with `kind` and `model` labels.\n\n| Metric | Type | Description |\n|--------|------|-------------|\n| `llm_proxy_request_total` | counter | Successfully parsed LLM requests |\n| `llm_proxy_request_error` | counter | Requests that failed to parse |\n| `llm_proxy_input_tokens` | counter | Accumulated input token counts |\n| `llm_proxy_output_tokens` | counter | Accumulated output token counts |\n| `llm_proxy_total_tokens` | counter | Accumulated total token counts |\n| `llm_proxy_request_ttft` | histogram | Time to first token in milliseconds |\n| `llm_proxy_request_tpot` | histogram | Average time per output token in milliseconds |\n\n## Configuration Reference\n\n| Field | Type | Required | Default | Description |\n|-------|------|----------|---------|-------------|\n| `llm_configs` | array | no | auto | Ordered list of path-matcher rules; first match wins |\n| `llm_configs[].matcher` | object | yes | — | Path matcher: set exactly one of `prefix`, `suffix`, or `regex` |\n| `llm_configs[].kind` | string | yes | — | `\"openai\"`, `\"anthropic\"`, or `\"custom\"` |\n| `metadata_namespace` | string | no | `io.builtonenvoy.llm-proxy` | Filter metadata namespace |\n| `llm_model_header` | string | no | `\"\"` | If set, the extracted model name is written to this request header |\n| `clear_route_cache` | bool | no | `false` | Clear the route cache after request parsing so Envoy can re-select the route based on updated metadata |\n", + "description": "Routes LLM API requests and emits richer observability metadata, metrics, and optional structured logs.", + "longDescription": "An HTTP filter plugin that inspects incoming requests against a set of configured\npath-matcher rules to identify the LLM provider API in use (OpenAI Chat Completions,\nAnthropic Messages, or a custom OpenAI-compatible API). Once a rule matches, the filter:\n\n1. Parses the request body to extract the model name and streaming flag, then\n writes them to Envoy's dynamic filter metadata.\n2. Parses the response body (JSON for non-streaming, SSE for streaming) to extract\n token-usage information and richer observability fields such as `question`,\n `system`, `answer`, `reasoning`, and `tool_calls`.\n3. Records Envoy metrics (counters and histograms) for request counts, token usage,\n time-to-first-token (TTFT), and time-per-output-token (TPOT).\n4. Optionally emits lightweight or full structured logs for downstream observability.\n\nRequests whose path does not match any rule are passed through without modification.\n\nIf no rule is explicitly configured for OpenAI or Anthropic, the filter automatically\nadds default suffix-matcher rules for `/v1/chat/completions` (OpenAI) and\n`/v1/messages` (Anthropic), so it works out of the box with no configuration.\n\n## Metadata Keys\n\nAll keys are written under the configured `metadata_namespace`\n(default: `io.builtonenvoy.llm-proxy`).\n\n| Key | Type | Description |\n|-----|------|-------------|\n| `kind` | string | API kind: `\"openai\"`, `\"anthropic\"`, or `\"custom\"` |\n| `model` | string | Model name extracted from the request body |\n| `is_stream` | bool | Whether the request asks for a streaming (SSE) response |\n| `response_type` | string | `\"stream\"` or `\"nonstream\"` |\n| `session_id` | string | Session or conversation ID extracted from `session_id_header`, if configured |\n| `question` | string | User question extracted from the request body, when available |\n| `system` | string | System prompt extracted from the request body, when available |\n| `answer` | string | Assistant response content, when available |\n| `reasoning` | string | Provider-specific reasoning content, when available |\n| `tool_calls` | array | Tool calls emitted by the model, when available |\n| `input_tokens` | uint32 | Input / prompt token count from the response |\n| `output_tokens` | uint32 | Output / completion token count from the response |\n| `total_tokens` | uint32 | Total token count from the response |\n| `reasoning_tokens` | uint32 | Reasoning token count when provided by the upstream API |\n| `cached_tokens` | uint32 | Cached input token count when provided by the upstream API |\n| `input_token_details` | object | Provider-specific prompt/input token detail fields |\n| `output_token_details` | object | Provider-specific completion/output token detail fields |\n| `request_ttft` | int64 | Time to first token in milliseconds |\n| `request_tpot` | int64 | Average time per output token in milliseconds |\n\n## Metrics\n\nAll metrics are tagged with `kind` and `model` labels.\n\n| Metric | Type | Description |\n|--------|------|-------------|\n| `llm_proxy_request_total` | counter | Successfully parsed LLM requests |\n| `llm_proxy_request_error` | counter | Requests that failed to parse |\n| `llm_proxy_input_tokens` | counter | Accumulated input token counts |\n| `llm_proxy_output_tokens` | counter | Accumulated output token counts |\n| `llm_proxy_total_tokens` | counter | Accumulated total token counts |\n| `llm_proxy_request_ttft` | histogram | Time to first token in milliseconds |\n| `llm_proxy_request_tpot` | histogram | Average time per output token in milliseconds |\n\n## Configuration Reference\n\n| Field | Type | Required | Default | Description |\n|-------|------|----------|---------|-------------|\n| `llm_configs` | array | no | auto | Ordered list of path-matcher rules; first match wins |\n| `llm_configs[].matcher` | object | yes | — | Path matcher: set exactly one of `prefix`, `suffix`, or `regex` |\n| `llm_configs[].kind` | string | yes | — | `\"openai\"`, `\"anthropic\"`, or `\"custom\"` |\n| `metadata_namespace` | string | no | `io.builtonenvoy.llm-proxy` | Filter metadata namespace |\n| `llm_model_header` | string | no | `\"\"` | If set, the extracted model name is written to this request header |\n| `use_default_attributes` | bool | no | `false` | Emit the full built-in structured log payload |\n| `use_default_response_attributes` | bool | no | `false` | Emit the lightweight built-in structured log payload |\n| `session_id_header` | string | no | `\"\"` | Optional request header used to extract a session or conversation ID |\n| `clear_route_cache` | bool | no | `false` | Clear the route cache after request parsing so Envoy can re-select the route based on updated metadata |\n", "type": "go", "tags": [ "go", @@ -567,6 +567,11 @@ "title": "Route to different clusters based on model name", "description": "Use `llm_model_header` to inject the extracted model name as a request header,\nthen configure an Envoy route to select a cluster based on that header.\nEnable `clear_route_cache` so Envoy re-evaluates the route after the header is set.\n", "code": "boe run --extension llm-proxy \\\n --config '{\n \"llm_model_header\": \"x-llm-model\",\n \"clear_route_cache\": true\n }'\n" + }, + { + "title": "Full observability output", + "description": "Emit richer metadata and a structured log entry including prompts, answers,\ntool calls, and provider token-detail fields.\n", + "code": "boe run --extension llm-proxy \\\n --config '{\n \"use_default_attributes\": true,\n \"session_id_header\": \"x-session-id\"\n }'\n" } ], "minEnvoyVersion": "1.37.1",