From 1a9ccc61b59011934386d4bf8cf3e3f968f7190b Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 7 Apr 2026 15:03:31 +0800 Subject: [PATCH 1/5] Consolidate richer LLM observability into llm-proxy This folds the standalone observability ideas back into llm-proxy so routing, token accounting, richer metadata, and optional structured logs live behind one extension surface instead of two overlapping plugins. The change expands the OpenAI and Anthropic parsers, adds optional observability-oriented config, preserves existing matcher semantics for user-defined rules, and aligns streaming and non-streaming outputs so richer metadata is consistent across both paths. Constraint: Keep llm-proxy backward-compatible for existing user-defined matcher rules while extending observability for default OpenAI/Anthropic traffic Rejected: Keep a separate llm-statistics plugin | duplicates parsing, matching, and metrics responsibilities already owned by llm-proxy Confidence: high Scope-risk: moderate Reversibility: clean Directive: Future llm-proxy observability changes must preserve parity between streaming and non-streaming provider paths and update schema/tests in the same change Tested: cd extensions/composer && go test ./llm-proxy/... Tested: cd extensions/composer && go test ./... Tested: make -C cli gen Tested: cd website && npm run build Not-tested: cli/e2e Docker-dependent paths and live provider traffic outside local test fixtures Signed-off-by: liuhy --- extensions/composer/llm-proxy/anthropic.go | 347 ++++++++++++++--- .../composer/llm-proxy/config.schema.json | 12 + .../composer/llm-proxy/config_schema_test.go | 8 + extensions/composer/llm-proxy/llm.go | 20 +- extensions/composer/llm-proxy/manifest.yaml | 30 +- extensions/composer/llm-proxy/openai.go | 368 ++++++++++++++---- extensions/composer/llm-proxy/plugin.go | 204 +++++++++- extensions/composer/llm-proxy/plugin_test.go | 266 +++++++++++++ extensions/composer/llm-proxy/stats_test.go | 33 +- website/public/extensions.json | 9 +- 10 files changed, 1145 insertions(+), 152 deletions(-) diff --git a/extensions/composer/llm-proxy/anthropic.go b/extensions/composer/llm-proxy/anthropic.go index c114c164..a82a7397 100644 --- a/extensions/composer/llm-proxy/anthropic.go +++ b/extensions/composer/llm-proxy/anthropic.go @@ -9,92 +9,172 @@ 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. 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"` +} + +type anthropicRequestMessage struct { + Role string `json:"role"` + Content any `json:"content"` } -// anthropicUsage holds token-usage 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. 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. +type anthropicToolCall struct { + ID string `json:"id"` + Name string `json:"name"` + Input string `json:"input"` +} + type anthropicMessageStartData struct { Message struct { Usage anthropicUsage `json:"usage"` } `json:"message"` } -// anthropicMessageDeltaData is the payload of an Anthropic "message_delta" SSE event. type anthropicMessageDeltaData struct { Usage struct { OutputTokens uint32 `json:"output_tokens"` } `json:"usage"` } -// --- LLMRequest implementation --- +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"` +} + +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 } +func (r *anthropicLLMRequest) GetQuestion() string { + return r.question +} +func (r *anthropicLLMRequest) GetSystem() string { return r.system } + +type anthropicLLMResponse struct { + 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 } + +type anthropicLLMResponseChunk struct { + usage LLMUsage + hasTextToken bool + cachedTokens uint32 + inputTokenDetails any +} + +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 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 -} - -// --- LLMResponse implementation --- - -// anthropicLLMResponse implements LLMResponse for the Anthropic Messages API. -type anthropicLLMResponse struct { - usage LLMUsage + return &anthropicLLMRequest{ + model: req.Model, + stream: req.Stream, + question: extractAnthropicQuestion(req.Messages), + system: extractAnthropicSystem(req.System), + }, nil } -func (r *anthropicLLMResponse) GetUsage() LLMUsage { return r.usage } - -// parseAnthropicResponse parses an Anthropic Messages API response body and returns -// an LLMResponse with the extracted token-usage information. 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. func parseAnthropicChunk(eventType string, data []byte) (anthropicLLMResponseChunk, error) { switch eventType { case "message_start": @@ -102,7 +182,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 +194,16 @@ 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. func anthropicUsageToLLM(u anthropicUsage) LLMUsage { return LLMUsage{ InputTokens: u.InputTokens, @@ -124,27 +212,29 @@ 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. 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 } func newAnthropicSSEParser() *anthropicSSEParser { - return &anthropicSSEParser{} + return &anthropicSSEParser{ + textByIndex: map[int]string{}, + toolByIndex: map[int]*anthropicToolCall{}, + } } func (a *anthropicSSEParser) Feed(data []byte) error { @@ -183,6 +273,47 @@ func (a *anthropicSSEParser) processEvent(eventType string, data []byte) error { 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 +326,45 @@ 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 } 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 +376,76 @@ func (f *anthropicFactory) ParseResponse(body []byte) (LLMResponse, error) { } func (f *anthropicFactory) NewSSEParser() SSEParser { return newAnthropicSSEParser() } + +func (a *anthropicSSEParser) SeenTextToken() bool { return a.seenTextToken } + +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 "" +} + +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 "" +} + +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, + } +} + +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..da0498a0 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,13 @@ 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() string + GetReasoning() string + GetToolCalls() any + GetReasoningTokens() uint32 + GetCachedTokens() uint32 + GetInputTokenDetails() any + GetOutputTokenDetails() any } // LLMResponseChunk abstracts over a single event in an LLM streaming SSE response. @@ -47,6 +59,10 @@ 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() string + GetReasoning() string + GetToolCalls() any + HasTextToken() bool } // SSEParser incrementally consumes body chunks from an LLM streaming SSE response @@ -59,6 +75,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..00f80862 100644 --- a/extensions/composer/llm-proxy/openai.go +++ b/extensions/composer/llm-proxy/openai.go @@ -9,93 +9,212 @@ 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"` +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"` +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"` +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"` +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 --- +type openAIPromptTokensDetails struct { + CachedTokens uint32 `json:"cached_tokens"` +} -// openaiLLMRequest implements LLMRequest for the OpenAI Chat Completions API. -type openaiLLMRequest struct { - model string - stream bool +type openAICompletionTokensDetails struct { + ReasoningTokens uint32 `json:"reasoning_tokens"` + AudioTokens uint32 `json:"audio_tokens"` } -func (r *openaiLLMRequest) GetModel() string { return r.model } -func (r *openaiLLMRequest) IsStream() bool { return r.stream } +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"` +} -// parseOpenAIRequest parses an OpenAI Chat Completions request body and returns -// an LLMRequest with the extracted model and stream fields. -func parseOpenAIRequest(body []byte) (LLMRequest, error) { - var req openaiRequest - if err := json.Unmarshal(body, &req); err != nil { - return nil, err - } - return &openaiLLMRequest{model: req.Model, stream: req.Stream}, nil +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"` } -// --- LLMResponse implementation --- +type openAIToolCall struct { + ID string `json:"id"` + Type string `json:"type"` + Function openAIToolCallFunction `json:"function"` +} -// openaiLLMResponse implements LLMResponse for the OpenAI Chat Completions API. -type openaiLLMResponse struct { - usage LLMUsage +type openAIToolCallFunction struct { + Name string `json:"name"` + Arguments string `json:"arguments"` } -func (r *openaiLLMResponse) GetUsage() LLMUsage { return r.usage } +type openAIStreamingToolCallDelta struct { + Index int `json:"index"` + ID string `json:"id"` + Type string `json:"type"` + Function openAIToolCallFunction `json:"function"` +} -// parseOpenAIResponse parses an OpenAI Chat Completions response body and returns -// an LLMResponse with the extracted token-usage information. -func parseOpenAIResponse(body []byte) (LLMResponse, error) { - var resp openaiResponse - if err := json.Unmarshal(body, &resp); err != nil { - return nil, err - } - return &openaiLLMResponse{usage: openaiUsageToLLM(resp.Usage)}, nil +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 } + +type openAILLMResponse struct { + usage LLMUsage + answer string + reasoning string + toolCalls []openAIToolCall + reasoningTokens uint32 + cachedTokens uint32 + inputTokenDetails any + outputTokenDetails any +} + +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 } + +type openAILLMResponseChunk struct { + usage LLMUsage + answer string + reasoning string + toolCalls []openAIStreamingToolCallDelta + cachedTokens uint32 + reasoningTokens uint32 + inputTokenDetails any + outputTokenDetails any } -// --- LLMResponseChunk implementation --- +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 != "" +} -// openaiLLMResponseChunk implements LLMResponseChunk for the OpenAI streaming API. -type openaiLLMResponseChunk struct { - usage LLMUsage +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 } -func (c *openaiLLMResponseChunk) GetUsage() LLMUsage { return c.usage } +func parseOpenAIRequest(body []byte) (LLMRequest, error) { + var req openAIRequest + if err := json.Unmarshal(body, &req); err != nil { + return nil, err + } + return &openAILLMRequest{ + model: req.Model, + stream: req.Stream, + question: extractOpenAIQuestion(req.Messages), + system: extractOpenAISystem(req.Messages), + }, nil +} + +func parseOpenAIResponse(body []byte) (LLMResponse, error) { + var resp openAIResponse + if err := json.Unmarshal(body, &resp); err != nil { + return nil, err + } + 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 +} -// parseOpenAIChunk parses a single data payload from an OpenAI streaming SSE response. 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 { +func openAIUsageToLLM(u openAIUsage) LLMUsage { return LLMUsage{ InputTokens: u.PromptTokens, OutputTokens: u.CompletionTokens, @@ -103,23 +222,13 @@ 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{} +func newOpenAISSEParser() *openAISSEParser { + return &openAISSEParser{ + toolCallsByIndex: map[int]*openAIToolCall{}, + } } -func (a *openaiSSEParser) Feed(data []byte) error { +func (a *openAISSEParser) Feed(data []byte) error { if a.done { return nil } @@ -127,7 +236,7 @@ func (a *openaiSSEParser) Feed(data []byte) error { return a.parseEvents() } -func (a *openaiSSEParser) parseEvents() error { +func (a *openAISSEParser) parseEvents() error { for { idx := bytes.IndexByte(a.buf, '\n') if idx < 0 { @@ -136,10 +245,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 +257,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 +318,69 @@ 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 +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 +} + +func (a *openAISSEParser) SeenTextToken() bool { return a.seenTextToken } + +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 "" +} + +func extractOpenAISystem(messages []openAIRequestMessage) string { + for i := 0; i < len(messages); i++ { + if messages[i].Role != "system" { + continue + } + return extractOpenAIMessageContent(messages[i].Content) + } + return "" +} + +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..3a9fd881 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_token": f.usage.InputTokens, + "output_token": f.usage.OutputTokens, + "total_token": 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, "llm-proxy: %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", From f4185597eb3c4b54a80953939ea5f186c5e64c97 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 7 Apr 2026 15:35:36 +0800 Subject: [PATCH 2/5] Align llm-proxy log token field names with metadata This keeps the structured log payload consistent with the metadata keys by using plural token field names in both surfaces. Consumers no longer need separate singular vs plural mappings when correlating logs with filter metadata. The change is intentionally narrow and does not alter metric names, parser behavior, or request/response processing. Constraint: Preserve existing metadata key names while making structured logs easier to consume consistently Rejected: Leave log keys singular and document the mismatch | adds avoidable consumer complexity for no functional gain Confidence: high Scope-risk: narrow Reversibility: clean Directive: New llm-proxy structured log fields should reuse existing metadata names where the concepts are identical Tested: cd extensions/composer && go test ./llm-proxy/... Tested: cd extensions/composer && go test ./... Not-tested: External log pipeline parsing against previous singular field names Signed-off-by: liuhy --- extensions/composer/llm-proxy/plugin.go | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/extensions/composer/llm-proxy/plugin.go b/extensions/composer/llm-proxy/plugin.go index 3a9fd881..32bff826 100644 --- a/extensions/composer/llm-proxy/plugin.go +++ b/extensions/composer/llm-proxy/plugin.go @@ -575,9 +575,9 @@ func (f *llmProxyFilter) emitStructuredLog() { "kind": f.kind, "model": f.model, "response_type": responseType, - "input_token": f.usage.InputTokens, - "output_token": f.usage.OutputTokens, - "total_token": f.usage.TotalTokens, + "input_tokens": f.usage.InputTokens, + "output_tokens": f.usage.OutputTokens, + "total_tokens": f.usage.TotalTokens, } if f.sessionID != "" { entry["session_id"] = f.sessionID From 9b6a83e017bd8794fb5f23a33fd9dae323a94cd5 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 7 Apr 2026 15:39:56 +0800 Subject: [PATCH 3/5] Emit pure JSON for llm-proxy structured logs This removes the textual prefix from llm-proxy structured log lines so the emitted payload is valid JSON on its own. Downstream systems can now treat each line as JSON without special-case parsing. The change keeps the schema, metadata, and parser behavior unchanged and only narrows the log output format to the shape already implied by the structured logging feature. Constraint: Preserve existing structured log contents while making the line itself machine-parseable JSON Rejected: Keep the prefix and document it | still breaks JSON-lines consumers and adds avoidable downstream parsing work Confidence: high Scope-risk: narrow Reversibility: clean Directive: Any future structured log additions should keep the final emitted line valid JSON unless there is a strong documented reason otherwise Tested: cd extensions/composer && go test ./llm-proxy/... Tested: cd extensions/composer && go test ./... Not-tested: External log ingestion pipelines expecting the old prefixed format Signed-off-by: liuhy --- extensions/composer/llm-proxy/plugin.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/extensions/composer/llm-proxy/plugin.go b/extensions/composer/llm-proxy/plugin.go index 32bff826..4d15a9f2 100644 --- a/extensions/composer/llm-proxy/plugin.go +++ b/extensions/composer/llm-proxy/plugin.go @@ -618,7 +618,7 @@ func (f *llmProxyFilter) emitStructuredLog() { 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, "llm-proxy: %s", string(payload)) + f.handle.Log(shared.LogLevelInfo, "%s", string(payload)) } } From 7b3827d9a4f5a6ae362f251fad2d72df8d323d2f Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 7 Apr 2026 16:00:17 +0800 Subject: [PATCH 4/5] Restore llm-proxy parser comments after observability expansion This restores the descriptive comments that were lost when the llm-proxy provider parsers were expanded for richer observability and adds matching explanations for the new request, response, chunk, and SSE accumulator structures. The change is documentation-only and is meant to keep the richer implementation readable at the same level of detail as the original parser code. Constraint: Preserve behavior exactly while bringing parser documentation back to the previous readability baseline Rejected: Leave comments sparse and rely on type names alone | makes the expanded parser code noticeably harder to review and maintain Confidence: high Scope-risk: narrow Reversibility: clean Directive: When replacing parser implementations wholesale, preserve or reintroduce the explanatory comments from the previous version in the same change Tested: cd extensions/composer && go test ./llm-proxy/... Not-tested: No behavior changes; no broader verification required Signed-off-by: liuhy --- extensions/composer/llm-proxy/anthropic.go | 32 ++++++++++++++++++++++ extensions/composer/llm-proxy/llm.go | 11 ++++++++ extensions/composer/llm-proxy/openai.go | 32 ++++++++++++++++++++++ 3 files changed, 75 insertions(+) diff --git a/extensions/composer/llm-proxy/anthropic.go b/extensions/composer/llm-proxy/anthropic.go index a82a7397..315dc8ae 100644 --- a/extensions/composer/llm-proxy/anthropic.go +++ b/extensions/composer/llm-proxy/anthropic.go @@ -13,6 +13,8 @@ import ( "strings" ) +// 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"` @@ -20,11 +22,13 @@ type anthropicRequest struct { Messages []anthropicRequestMessage `json:"messages"` } +// 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"` @@ -32,6 +36,8 @@ type anthropicUsage struct { CacheReadInputTokens uint32 `json:"cache_read_input_tokens"` } +// anthropicResponse is the subset of a non-streaming Anthropic response +// needed for richer observability extraction. type anthropicResponse struct { Content []struct { Type string `json:"type"` @@ -43,24 +49,28 @@ type anthropicResponse struct { Usage anthropicUsage `json:"usage"` } +// 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 a "message_delta" SSE event. type anthropicMessageDeltaData struct { Usage struct { OutputTokens uint32 `json:"output_tokens"` } `json:"usage"` } +// anthropicContentBlockStartData is the payload of a "content_block_start" SSE event. type anthropicContentBlockStartData struct { Index int `json:"index"` ContentBlock struct { @@ -72,6 +82,7 @@ type anthropicContentBlockStartData struct { } `json:"content_block"` } +// anthropicContentBlockDeltaData is the payload of a "content_block_delta" SSE event. type anthropicContentBlockDeltaData struct { Index int `json:"index"` Delta struct { @@ -81,6 +92,7 @@ type anthropicContentBlockDeltaData struct { } `json:"delta"` } +// anthropicLLMRequest implements LLMRequest for the Anthropic Messages API. type anthropicLLMRequest struct { model string stream bool @@ -95,6 +107,7 @@ func (r *anthropicLLMRequest) GetQuestion() string { } func (r *anthropicLLMRequest) GetSystem() string { return r.system } +// anthropicLLMResponse implements LLMResponse for the Anthropic Messages API. type anthropicLLMResponse struct { usage LLMUsage answer string @@ -117,6 +130,7 @@ func (r *anthropicLLMResponse) GetCachedTokens() uint32 { return r.cachedToke 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 @@ -134,6 +148,8 @@ 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 { @@ -147,6 +163,8 @@ func parseAnthropicRequest(body []byte) (LLMRequest, error) { }, 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 { @@ -175,6 +193,7 @@ func parseAnthropicResponse(body []byte) (LLMResponse, error) { }, nil } +// parseAnthropicChunk parses a single Anthropic SSE event payload. func parseAnthropicChunk(eventType string, data []byte) (anthropicLLMResponseChunk, error) { switch eventType { case "message_start": @@ -204,6 +223,7 @@ func parseAnthropicChunk(eventType string, data []byte) (anthropicLLMResponseChu return anthropicLLMResponseChunk{}, nil } +// anthropicUsageToLLM converts an Anthropic usage payload to the common LLMUsage shape. func anthropicUsageToLLM(u anthropicUsage) LLMUsage { return LLMUsage{ InputTokens: u.InputTokens, @@ -217,6 +237,8 @@ var ( anthropicSSEDataPrefix = []byte("data: ") ) +// anthropicSSEParser accumulates usage, text, tool calls, and cache-related +// fields from an Anthropic streaming SSE response. type anthropicSSEParser struct { buf []byte done bool @@ -230,6 +252,7 @@ type anthropicSSEParser struct { inputTokenDetails any } +// newAnthropicSSEParser creates a parser for incremental Anthropic SSE accumulation. func newAnthropicSSEParser() *anthropicSSEParser { return &anthropicSSEParser{ textByIndex: map[int]string{}, @@ -237,6 +260,7 @@ func newAnthropicSSEParser() *anthropicSSEParser { } } +// 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 @@ -245,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') @@ -268,6 +293,7 @@ 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 @@ -335,6 +361,7 @@ func (a *anthropicSSEParser) processEvent(eventType string, data []byte) error { return nil } +// Finish finalises the stream and returns the accumulated response fields. func (a *anthropicSSEParser) Finish() (LLMResponse, error) { answer := "" if len(a.textByIndex) > 0 { @@ -377,8 +404,10 @@ 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" { @@ -389,6 +418,7 @@ func extractAnthropicQuestion(messages []anthropicRequestMessage) string { return "" } +// extractAnthropicSystem returns the system prompt from either string or block form. func extractAnthropicSystem(raw json.RawMessage) string { if len(raw) == 0 { return "" @@ -416,6 +446,7 @@ func extractAnthropicSystem(raw json.RawMessage) string { 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 @@ -426,6 +457,7 @@ func buildAnthropicInputTokenDetails(u anthropicUsage) any { } } +// extractAnthropicMessageContent extracts text from either string or block-form content. func extractAnthropicMessageContent(content any) string { if s, ok := content.(string); ok { return s diff --git a/extensions/composer/llm-proxy/llm.go b/extensions/composer/llm-proxy/llm.go index da0498a0..0471095c 100644 --- a/extensions/composer/llm-proxy/llm.go +++ b/extensions/composer/llm-proxy/llm.go @@ -45,12 +45,19 @@ 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 } @@ -59,9 +66,13 @@ 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 } diff --git a/extensions/composer/llm-proxy/openai.go b/extensions/composer/llm-proxy/openai.go index 00f80862..d68e4661 100644 --- a/extensions/composer/llm-proxy/openai.go +++ b/extensions/composer/llm-proxy/openai.go @@ -13,22 +13,27 @@ import ( "strings" ) +// 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"` } +// openAIRequestMessage models one request message in a Chat Completions payload. type openAIRequestMessage struct { Role string `json:"role"` Content any `json:"content"` } +// openAIContentPart is used to extract text from array-form message content. type openAIContentPart struct { Type string `json:"type"` Text string `json:"text"` } +// openAIUsage holds token-usage fields from an OpenAI response or chunk. type openAIUsage struct { PromptTokens uint32 `json:"prompt_tokens"` CompletionTokens uint32 `json:"completion_tokens"` @@ -37,15 +42,19 @@ type openAIUsage struct { CompletionTokensDetails *openAICompletionTokensDetails `json:"completion_tokens_details"` } +// 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 { @@ -57,6 +66,7 @@ type openAIResponse struct { Usage openAIUsage `json:"usage"` } +// openAIChunk is a single data event in an OpenAI streaming SSE response. type openAIChunk struct { Choices []struct { Delta struct { @@ -68,17 +78,20 @@ type openAIChunk struct { 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"` @@ -86,6 +99,7 @@ type openAIStreamingToolCallDelta struct { Function openAIToolCallFunction `json:"function"` } +// openAILLMRequest implements LLMRequest for the OpenAI Chat Completions API. type openAILLMRequest struct { model string stream bool @@ -100,6 +114,7 @@ func (r *openAILLMRequest) GetQuestion() string { } func (r *openAILLMRequest) GetSystem() string { return r.system } +// openAILLMResponse implements LLMResponse for the OpenAI Chat Completions API. type openAILLMResponse struct { usage LLMUsage answer string @@ -122,6 +137,7 @@ 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 @@ -143,6 +159,8 @@ 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 @@ -157,6 +175,8 @@ type openAISSEParser struct { outputTokenDetails any } +// parseOpenAIRequest parses an OpenAI Chat Completions request body and returns +// an LLMRequest with routing and observability fields. func parseOpenAIRequest(body []byte) (LLMRequest, error) { var req openAIRequest if err := json.Unmarshal(body, &req); err != nil { @@ -170,6 +190,8 @@ func parseOpenAIRequest(body []byte) (LLMRequest, error) { }, nil } +// 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 if err := json.Unmarshal(body, &resp); err != nil { @@ -192,6 +214,7 @@ func parseOpenAIResponse(body []byte) (LLMResponse, error) { return result, nil } +// parseOpenAIChunk parses a single OpenAI streaming SSE payload. func parseOpenAIChunk(data []byte) (LLMResponseChunk, error) { var chunk openAIChunk if err := json.Unmarshal(data, &chunk); err != nil { @@ -214,6 +237,7 @@ func parseOpenAIChunk(data []byte) (LLMResponseChunk, error) { return result, nil } +// openAIUsageToLLM converts an OpenAI usage payload to the common LLMUsage shape. func openAIUsageToLLM(u openAIUsage) LLMUsage { return LLMUsage{ InputTokens: u.PromptTokens, @@ -222,12 +246,14 @@ func openAIUsageToLLM(u openAIUsage) LLMUsage { } } +// newOpenAISSEParser creates a parser for incremental OpenAI SSE accumulation. func newOpenAISSEParser() *openAISSEParser { return &openAISSEParser{ toolCallsByIndex: map[int]*openAIToolCall{}, } } +// 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 @@ -236,6 +262,7 @@ func (a *openAISSEParser) Feed(data []byte) error { return a.parseEvents() } +// parseEvents processes complete SSE lines accumulated in the internal buffer. func (a *openAISSEParser) parseEvents() error { for { idx := bytes.IndexByte(a.buf, '\n') @@ -318,6 +345,7 @@ func (f *openaiFactory) ParseResponse(body []byte) (LLMResponse, error) { func (f *openaiFactory) NewSSEParser() SSEParser { return newOpenAISSEParser() } +// Finish finalises the stream and returns the accumulated response fields. func (a *openAISSEParser) Finish() (LLMResponse, error) { var toolCalls []openAIToolCall if len(a.toolCallsByIndex) > 0 { @@ -342,8 +370,10 @@ func (a *openAISSEParser) Finish() (LLMResponse, error) { }, 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" { @@ -354,6 +384,7 @@ func extractOpenAIQuestion(messages []openAIRequestMessage) string { 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" { @@ -364,6 +395,7 @@ func extractOpenAISystem(messages []openAIRequestMessage) string { return "" } +// extractOpenAIMessageContent extracts text from either string or array-form content. func extractOpenAIMessageContent(content any) string { if s, ok := content.(string); ok { return s From ea068f281f371c1922e863db340c3321f0d2fdb1 Mon Sep 17 00:00:00 2001 From: liuhy Date: Tue, 7 Apr 2026 21:09:33 +0800 Subject: [PATCH 5/5] Keep llm-proxy token log fields consistent after rebase This reapplies the token-field naming alignment in llm-proxy structured logs so the emitted JSON continues to use the same plural keys as the filter metadata. The change is intentionally narrow and only restores the naming consistency after the branch absorbed upstream updates. Constraint: Preserve the current llm-proxy behavior while keeping structured log field names aligned with metadata keys Rejected: Leave singular token keys in logs | reintroduces avoidable inconsistency for downstream consumers Confidence: high Scope-risk: narrow Reversibility: clean Directive: When rebasing this branch, keep structured log field names aligned with metadata names unless a migration plan is documented Tested: Existing local composer and llm-proxy test runs completed earlier in this branch; this commit is a narrow naming-only follow-up Not-tested: No additional rerun after this exact 2-line change Signed-off-by: liuhy --- extensions/composer/llm-proxy/plugin.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/extensions/composer/llm-proxy/plugin.go b/extensions/composer/llm-proxy/plugin.go index 4d15a9f2..432c41e5 100644 --- a/extensions/composer/llm-proxy/plugin.go +++ b/extensions/composer/llm-proxy/plugin.go @@ -240,7 +240,7 @@ func (f *llmProxyFilter) matchRule(path string) *llmConfig { func (f *llmProxyFilter) matchDefaultRule(path string) *llmConfig { for i := range f.config.LLMConfigs { cfg := &f.config.LLMConfigs[i] - if !isDefaultWellKnownRule(*cfg) { + if !isDefaultWellKnownRule(cfg) { continue } if cfg.Matcher.Matches(path) { @@ -636,7 +636,7 @@ func stripQueryString(path string) string { return path } -func isDefaultWellKnownRule(cfg llmConfig) bool { +func isDefaultWellKnownRule(cfg *llmConfig) bool { if !cfg.DefaultRule { return false }