From c7b2cd8ae02031373f194a7ff4fd7df638220aec Mon Sep 17 00:00:00 2001 From: AskaEth Date: Tue, 8 Sep 2026 23:52:47 +0800 Subject: [PATCH] =?UTF-8?q?fix(model):=20=E8=A7=A3=E6=9E=90=20stop=5Freaso?= =?UTF-8?q?n=20+=20max=5Ftokens=20=E6=88=AA=E6=96=AD=E9=87=8D=E8=AF=95/?= =?UTF-8?q?=E9=9D=9E=E7=A9=BA=E5=85=9C=E5=BA=95=20=E2=80=94=20=E4=B8=8D?= =?UTF-8?q?=E5=86=8D=E9=9D=99=E9=BB=98=E7=A9=BA=E5=9B=9E=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - model.go: Response.StopReason; anthropic wire加stop_reason; openai finish_reason映射(两协议统一,原值透传) - truncation.go(新,159行): effectiveMaxTokens/retryMaxTokens(×2封顶32768)/isTruncatedEmpty(空正文+无tool_use+stop_reason∈{max_tokens,length})/withEmptyBodyHint/callWithTruncationRetry/callStreamWithTruncationRetry/dropThinking/emitTextSegment - anthropic.go Chat/ChatStream 走重试+兜底; anthropic_encode.go(新)拆编码(anthropic.go 529→357合规); anthropic_stream message_delta落StopReason - agent/loop: 把模型层StopReason带到llm.call span的model_stop_reason属性(修掉trace硬编码stop的误导); loop.Result.StopReason语义未动 - 触发仅当截断空正文; tool_use空正文永不触发(native_toolcall_test仍过); 至多1次重试; 兜底回'（思考过长，未能生成回复）'绝不用thinking - 新增7测试(非流式/流式重试+兜底+tool_use不触发+封顶+openai映射), build/vet/test绿+3平台交叉编译 --- internal/runtime/agent/step.go | 9 +- internal/runtime/loop/step.go | 16 +- internal/runtime/loop/stream.go | 4 +- internal/runtime/model/anthropic.go | 242 +++----------- internal/runtime/model/anthropic_encode.go | 178 ++++++++++ internal/runtime/model/anthropic_stream.go | 6 + internal/runtime/model/model.go | 5 + internal/runtime/model/openai.go | 4 + internal/runtime/model/truncation.go | 159 +++++++++ internal/runtime/model/truncation_test.go | 369 +++++++++++++++++++++ 10 files changed, 792 insertions(+), 200 deletions(-) create mode 100644 internal/runtime/model/anthropic_encode.go create mode 100644 internal/runtime/model/truncation.go create mode 100644 internal/runtime/model/truncation_test.go diff --git a/internal/runtime/agent/step.go b/internal/runtime/agent/step.go index fcc13e0..5e32955 100644 --- a/internal/runtime/agent/step.go +++ b/internal/runtime/agent/step.go @@ -46,6 +46,11 @@ type StepResult struct { ToolCalls []ToolCall // StopReason classifies how the step ended (see the StopReason* consts). StopReason string + // ModelStopReason is the provider's raw stop/finish reason for the model + // call (e.g. Anthropic max_tokens / end_turn / tool_use), so a trace can + // distinguish a token-budget truncation from a normal end. Empty when the + // step never reached the model (guardrail denial). + ModelStopReason string } // RunStep executes one step of the agent loop. It assembles a model.Request from @@ -101,9 +106,9 @@ func (a *Agent) RunStep(ctx context.Context, sessionID string, messages []model. } if calls, ok := ExtractToolCalls(resp); ok { - return &StepResult{Final: false, ToolCalls: calls, StopReason: StopReasonToolCalls}, nil + return &StepResult{Final: false, ToolCalls: calls, StopReason: StopReasonToolCalls, ModelStopReason: resp.StopReason}, nil } - return &StepResult{Final: true, Text: resp.Content, StopReason: StopReasonStop}, nil + return &StepResult{Final: true, Text: resp.Content, StopReason: StopReasonStop, ModelStopReason: resp.StopReason}, nil } // ExtractToolCalls decodes tool calls from a model.Response. It prefers the diff --git a/internal/runtime/loop/step.go b/internal/runtime/loop/step.go index 1e265b3..95dce2b 100644 --- a/internal/runtime/loop/step.go +++ b/internal/runtime/loop/step.go @@ -17,6 +17,10 @@ type stepOutcome struct { text string toolCalls []agent.ToolCall stopReason string + // modelStopReason is the provider's raw stop/finish reason for the model + // call (Anthropic max_tokens / end_turn / tool_use), recorded on the llm_call + // span so a truncation is visible even when the step itself ended normally. + modelStopReason string } // step runs a single model call. When an OnStream callback is configured it @@ -45,6 +49,9 @@ func (l *Loop) step(ctx context.Context, sessionID string, messages []model.Mess } l.spanProp(llmSpan, "stop_reason", out.stopReason) + if out.modelStopReason != "" { + l.spanProp(llmSpan, "model_stop_reason", out.modelStopReason) + } l.spanProp(llmSpan, "tool_calls", len(out.toolCalls)) l.finishSpan(llmSpan, trace.SpanStatusOK) return out, nil @@ -58,9 +65,10 @@ func (l *Loop) agentStep(ctx context.Context, sessionID string, messages []model return nil, fmt.Errorf("loop: agent step: %w", err) } return &stepOutcome{ - final: sr.Final, - text: sr.Text, - toolCalls: sr.ToolCalls, - stopReason: sr.StopReason, + final: sr.Final, + text: sr.Text, + toolCalls: sr.ToolCalls, + stopReason: sr.StopReason, + modelStopReason: sr.ModelStopReason, }, nil } diff --git a/internal/runtime/loop/stream.go b/internal/runtime/loop/stream.go index 99ed5b6..fe6a2a6 100644 --- a/internal/runtime/loop/stream.go +++ b/internal/runtime/loop/stream.go @@ -53,7 +53,7 @@ func (l *Loop) streamStep(ctx context.Context, messages []model.Message) (*stepO } if calls, ok := decodeToolCalls(resp); ok { - return &stepOutcome{final: false, toolCalls: calls, stopReason: agent.StopReasonToolCalls}, nil + return &stepOutcome{final: false, toolCalls: calls, stopReason: agent.StopReasonToolCalls, modelStopReason: resp.StopReason}, nil } - return &stepOutcome{final: true, text: resp.Content, stopReason: agent.StopReasonStop}, nil + return &stepOutcome{final: true, text: resp.Content, stopReason: agent.StopReasonStop, modelStopReason: resp.StopReason}, nil } diff --git a/internal/runtime/model/anthropic.go b/internal/runtime/model/anthropic.go index 5730ad1..bb38cfb 100644 --- a/internal/runtime/model/anthropic.go +++ b/internal/runtime/model/anthropic.go @@ -55,10 +55,11 @@ type anthropicTool struct { // anthropicResponse is the JSON body returned by /v1/messages. type anthropicResponse struct { - ID string `json:"id"` - Model string `json:"model"` - Content []anthropicBlock `json:"content"` - Usage *anthropicUsage `json:"usage,omitempty"` + ID string `json:"id"` + Model string `json:"model"` + Content []anthropicBlock `json:"content"` + StopReason string `json:"stop_reason,omitempty"` + Usage *anthropicUsage `json:"usage,omitempty"` } // anthropicBlock is a single content block in a response. For tool_use blocks the @@ -132,7 +133,11 @@ func NewAnthropicChat(cfg config.LLMConfig, opts ...AnthropicOption) *AnthropicC // Name returns the adapter identifier. func (a *AnthropicChatAdapter) Name() string { return a.name } -// Chat performs a non-streaming completion. +// Chat performs a non-streaming completion. A response cut off by the token +// budget with no visible output (stop_reason max_tokens and neither text nor a +// tool_use block) is retried once with a doubled budget and, if that still +// yields nothing, replaced by a non-empty placeholder so the caller never sees a +// silent empty reply. func (a *AnthropicChatAdapter) Chat(ctx context.Context, req Request, opts Options) (Response, error) { start := time.Now() if a.baseURL == "" { @@ -145,6 +150,22 @@ func (a *AnthropicChatAdapter) Chat(ctx context.Context, req Request, opts Optio return Response{}, err } req.Model = model + out, err := callWithTruncationRetry(req.MaxTokens, anthropicDefaultMaxTokens, + func(limit int) (Response, error) { + attempt := req + attempt.MaxTokens = limit + return a.chatOnce(ctx, attempt) + }) + if err != nil { + return Response{}, err + } + a.record(req, start, out.TotalTokens) + return out, nil +} + +// chatOnce performs a single non-streaming /v1/messages call and normalizes the +// wire body into a Response. +func (a *AnthropicChatAdapter) chatOnce(ctx context.Context, req Request) (Response, error) { body, err := a.buildRequest(req, false) if err != nil { return Response{}, err @@ -165,16 +186,16 @@ func (a *AnthropicChatAdapter) Chat(ctx context.Context, req Request, opts Optio if err := json.NewDecoder(resp.Body).Decode(&wire); err != nil { return Response{}, fmt.Errorf("model: %w: decode response: %v", ErrUpstream, err) } - out := responseFromAnthropicWire(&wire) - a.record(req, start, out.TotalTokens) - return out, nil + return responseFromAnthropicWire(&wire), nil } // ChatStream streams structured events via onEvent and returns the aggregated // response with usage when the provider reports it. Text blocks are delivered as // text_start/text_delta/text_end events; tool_use blocks are delivered as // tool_call_start/tool_call_delta/tool_call_end so the frontend can render each -// segment independently. +// segment independently. A budget-truncated empty stream is retried once with a +// doubled budget (thinking from the retry is suppressed to avoid a duplicate +// block) and, if still empty, closed with a non-empty placeholder text segment. func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) { start := time.Now() if a.baseURL == "" { @@ -185,6 +206,23 @@ func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onEv return Response{}, err } req.Model = model + out, err := callStreamWithTruncationRetry(req.MaxTokens, anthropicDefaultMaxTokens, + func(limit int, ev func(StreamEvent)) (Response, error) { + attempt := req + attempt.MaxTokens = limit + return a.streamOnce(ctx, attempt, ev) + }, onEvent) + if err != nil { + return Response{}, err + } + out.Model = model + a.record(req, start, out.TotalTokens) + return out, nil +} + +// streamOnce performs a single streaming /v1/messages call and aggregates its +// SSE events into a Response. +func (a *AnthropicChatAdapter) streamOnce(ctx context.Context, req Request, onEvent func(StreamEvent)) (Response, error) { body, err := a.buildRequest(req, true) if err != nil { return Response{}, err @@ -201,13 +239,7 @@ func (a *AnthropicChatAdapter) ChatStream(ctx context.Context, req Request, onEv if resp.StatusCode != http.StatusOK { return Response{}, a.upstreamError(resp) } - out, err := scanAnthropicSSE(resp.Body, onEvent) - if err != nil { - return Response{}, err - } - out.Model = model - a.record(req, start, out.TotalTokens) - return out, nil + return scanAnthropicSSE(resp.Body, onEvent) } // resolveModel returns the effective model name. @@ -231,10 +263,7 @@ func (a *AnthropicChatAdapter) buildRequest(req Request, stream bool) ([]byte, e if err != nil { return nil, err } - tokenLimit := req.MaxTokens - if tokenLimit == 0 { - tokenLimit = anthropicDefaultMaxTokens - } + tokenLimit := effectiveMaxTokens(req.MaxTokens, anthropicDefaultMaxTokens) wire := anthropicRequest{ Model: req.Model, System: system, @@ -287,178 +316,6 @@ func (a *AnthropicChatAdapter) record(req Request, start time.Time, tokens int) }) } -// buildAnthropicMessages splits a normalized message list into a top-level system -// string and the remaining messages in Anthropic's content-block form. All -// RoleSystem messages are concatenated into a single system prompt (Anthropic -// carries it at the request root, not in messages); messages with no content -// parts are dropped because Anthropic rejects empty content. -func buildAnthropicMessages(msgs []Message) (string, []anthropicMessage, error) { - var sys strings.Builder - out := make([]anthropicMessage, 0, len(msgs)) - for _, m := range msgs { - if m.Role == RoleSystem { - if t := messageText(m); t != "" { - if sys.Len() > 0 { - sys.WriteString("\n") - } - sys.WriteString(t) - } - continue - } - blocks, err := encodeAnthropicMessage(m) - if err != nil { - return "", nil, err - } - if len(blocks) == 0 { - continue - } - out = append(out, anthropicMessage{Role: anthropicRole(m.Role), Content: blocks}) - } - return sys.String(), out, nil -} - -// messageText concatenates the text-bearing parts of a message (system prompts -// and tool-result bodies). Tool-result parts carry their body in Text, so a -// RoleTool turn's content serializes correctly for OpenAI's tool message. -func messageText(m Message) string { - var b strings.Builder - for _, part := range m.Content { - if part.Type == ContentTypeText || part.Type == ContentTypeToolResult { - b.WriteString(part.Text) - } - } - return b.String() -} - -// anthropicRole maps a normalized Role to the Anthropic role vocabulary. system -// messages are extracted to the top-level system field and never reach here. -func anthropicRole(r Role) string { - switch r { - case RoleAssistant: - return "assistant" - default: - // user, tool and any future roles all land on the user turn; tool_result - // blocks (M2) belong to the user role in Anthropic. - return "user" - } -} - -// encodeAnthropicMessage turns a normalized message into Anthropic content -// blocks. A RoleTool turn is ALWAYS rendered as native tool_result blocks (never -// a user plain-text turn), because Anthropic requires a tool result to arrive as -// a tool_result block carrying the tool_use_id and the is_error flag. -func encodeAnthropicMessage(m Message) ([]any, error) { - if m.Role == RoleTool { - return encodeAnthropicToolResult(m) - } - return encodeAnthropicContent(m.Content) -} - -// encodeAnthropicToolResult renders a RoleTool turn as tool_result blocks. Each -// ContentTypeToolResult part becomes one block; legacy text parts are wrapped as -// tool_result blocks too so the turn never degrades into a plain user message. -// The tool_use_id falls back to the message-level ToolCallID when a part does not -// carry its own. -func encodeAnthropicToolResult(m Message) ([]any, error) { - blocks := make([]any, 0, len(m.Content)) - for _, p := range m.Content { - switch p.Type { - case ContentTypeToolResult, ContentTypeText: - id := p.ToolCallID - if id == "" { - id = m.ToolCallID - } - blocks = append(blocks, AnthropicToolResult(id, p.Text, p.IsError)) - } - } - if len(blocks) == 0 { - blocks = append(blocks, AnthropicToolResult(m.ToolCallID, "", false)) - } - return blocks, nil -} - -// encodeAnthropicContent turns normalized content parts into Anthropic content -// blocks. Text / image / tool_use (assistant) parts are handled here; tool_result -// parts are rendered at the message level by encodeAnthropicToolResult. -func encodeAnthropicContent(parts []Content) ([]any, error) { - blocks := make([]any, 0, len(parts)) - for _, p := range parts { - switch p.Type { - case ContentTypeText: - blocks = append(blocks, map[string]any{"type": "text", "text": p.Text}) - case ContentTypeImage: - mediaType, data, err := parseAnthropicImageData(p.ImageURL) - if err != nil { - return nil, err - } - blocks = append(blocks, map[string]any{ - "type": "image", - "source": map[string]any{ - "type": "base64", - "media_type": mediaType, - "data": data, - }, - }) - case ContentTypeToolCall: - args := p.ToolArgs - if args == nil { - args = map[string]any{} - } - blocks = append(blocks, map[string]any{ - "type": "tool_use", - "id": p.ToolCallID, - "name": p.ToolCallName, - "input": args, - }) - case ContentTypeToolResult: - blocks = append(blocks, AnthropicToolResult(p.ToolCallID, p.Text, p.IsError)) - default: - return nil, fmt.Errorf("model: %w: unsupported content type %q", ErrBadRequest, p.Type) - } - } - return blocks, nil -} - -// parseAnthropicImageData extracts the media type and base64 payload from a data: -// URI. Anthropic requires a base64 image source and cannot reference a remote -// URL, so any other URI form is rejected as a bad request. -func parseAnthropicImageData(uri string) (string, string, error) { - if !strings.HasPrefix(uri, anthropicDataPrefix) { - return "", "", fmt.Errorf("model: %w: anthropic image requires a data: URI, got %q", ErrBadRequest, uri) - } - rest := strings.TrimPrefix(uri, anthropicDataPrefix) - comma := strings.Index(rest, ",") - if comma < 0 { - return "", "", fmt.Errorf("model: %w: malformed data: URI %q", ErrBadRequest, uri) - } - meta := rest[:comma] - data := rest[comma+1:] - mediaType := meta - if idx := strings.Index(meta, ";"); idx >= 0 { - mediaType = meta[:idx] - } - if mediaType == "" { - return "", "", fmt.Errorf("model: %w: data: URI missing media type in %q", ErrBadRequest, uri) - } - return mediaType, data, nil -} - -// buildAnthropicTools converts ToolSpec schemas into Anthropic tools. -func buildAnthropicTools(tools []ToolSpec) []anthropicTool { - if len(tools) == 0 { - return nil - } - out := make([]anthropicTool, 0, len(tools)) - for _, t := range tools { - out = append(out, anthropicTool{ - Name: t.Name, - Description: t.Description, - InputSchema: t.Parameters, - }) - } - return out -} - // responseFromAnthropicWire converts a non-streaming /v1/messages response into a // normalized Response. Each text block contributes to the assistant content; each // tool_use block becomes a ToolCall whose Args is taken directly from the decoded @@ -478,6 +335,7 @@ func responseFromAnthropicWire(wire *anthropicResponse) Response { out.Content = b.String() out.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: out.Content}}} out.ToolCalls = toolCalls + out.StopReason = wire.StopReason if u := wire.Usage; u != nil { out.PromptTokens = u.InputTokens out.CompletionTokens = u.OutputTokens diff --git a/internal/runtime/model/anthropic_encode.go b/internal/runtime/model/anthropic_encode.go new file mode 100644 index 0000000..a997451 --- /dev/null +++ b/internal/runtime/model/anthropic_encode.go @@ -0,0 +1,178 @@ +package model + +import ( + "fmt" + "strings" +) + +// buildAnthropicMessages splits a normalized message list into a top-level system +// string and the remaining messages in Anthropic's content-block form. All +// RoleSystem messages are concatenated into a single system prompt (Anthropic +// carries it at the request root, not in messages); messages with no content +// parts are dropped because Anthropic rejects empty content. +func buildAnthropicMessages(msgs []Message) (string, []anthropicMessage, error) { + var sys strings.Builder + out := make([]anthropicMessage, 0, len(msgs)) + for _, m := range msgs { + if m.Role == RoleSystem { + if t := messageText(m); t != "" { + if sys.Len() > 0 { + sys.WriteString("\n") + } + sys.WriteString(t) + } + continue + } + blocks, err := encodeAnthropicMessage(m) + if err != nil { + return "", nil, err + } + if len(blocks) == 0 { + continue + } + out = append(out, anthropicMessage{Role: anthropicRole(m.Role), Content: blocks}) + } + return sys.String(), out, nil +} + +// messageText concatenates the text-bearing parts of a message (system prompts +// and tool-result bodies). Tool-result parts carry their body in Text, so a +// RoleTool turn's content serializes correctly for OpenAI's tool message. +func messageText(m Message) string { + var b strings.Builder + for _, part := range m.Content { + if part.Type == ContentTypeText || part.Type == ContentTypeToolResult { + b.WriteString(part.Text) + } + } + return b.String() +} + +// anthropicRole maps a normalized Role to the Anthropic role vocabulary. system +// messages are extracted to the top-level system field and never reach here. +func anthropicRole(r Role) string { + switch r { + case RoleAssistant: + return "assistant" + default: + // user, tool and any future roles all land on the user turn; tool_result + // blocks (M2) belong to the user role in Anthropic. + return "user" + } +} + +// encodeAnthropicMessage turns a normalized message into Anthropic content +// blocks. A RoleTool turn is ALWAYS rendered as native tool_result blocks (never +// a user plain-text turn), because Anthropic requires a tool result to arrive as +// a tool_result block carrying the tool_use_id and the is_error flag. +func encodeAnthropicMessage(m Message) ([]any, error) { + if m.Role == RoleTool { + return encodeAnthropicToolResult(m) + } + return encodeAnthropicContent(m.Content) +} + +// encodeAnthropicToolResult renders a RoleTool turn as tool_result blocks. Each +// ContentTypeToolResult part becomes one block; legacy text parts are wrapped as +// tool_result blocks too so the turn never degrades into a plain user message. +// The tool_use_id falls back to the message-level ToolCallID when a part does not +// carry its own. +func encodeAnthropicToolResult(m Message) ([]any, error) { + blocks := make([]any, 0, len(m.Content)) + for _, p := range m.Content { + switch p.Type { + case ContentTypeToolResult, ContentTypeText: + id := p.ToolCallID + if id == "" { + id = m.ToolCallID + } + blocks = append(blocks, AnthropicToolResult(id, p.Text, p.IsError)) + } + } + if len(blocks) == 0 { + blocks = append(blocks, AnthropicToolResult(m.ToolCallID, "", false)) + } + return blocks, nil +} + +// encodeAnthropicContent turns normalized content parts into Anthropic content +// blocks. Text / image / tool_use (assistant) parts are handled here; tool_result +// parts are rendered at the message level by encodeAnthropicToolResult. +func encodeAnthropicContent(parts []Content) ([]any, error) { + blocks := make([]any, 0, len(parts)) + for _, p := range parts { + switch p.Type { + case ContentTypeText: + blocks = append(blocks, map[string]any{"type": "text", "text": p.Text}) + case ContentTypeImage: + mediaType, data, err := parseAnthropicImageData(p.ImageURL) + if err != nil { + return nil, err + } + blocks = append(blocks, map[string]any{ + "type": "image", + "source": map[string]any{ + "type": "base64", + "media_type": mediaType, + "data": data, + }, + }) + case ContentTypeToolCall: + args := p.ToolArgs + if args == nil { + args = map[string]any{} + } + blocks = append(blocks, map[string]any{ + "type": "tool_use", + "id": p.ToolCallID, + "name": p.ToolCallName, + "input": args, + }) + case ContentTypeToolResult: + blocks = append(blocks, AnthropicToolResult(p.ToolCallID, p.Text, p.IsError)) + default: + return nil, fmt.Errorf("model: %w: unsupported content type %q", ErrBadRequest, p.Type) + } + } + return blocks, nil +} + +// parseAnthropicImageData extracts the media type and base64 payload from a data: +// URI. Anthropic requires a base64 image source and cannot reference a remote +// URL, so any other URI form is rejected as a bad request. +func parseAnthropicImageData(uri string) (string, string, error) { + if !strings.HasPrefix(uri, anthropicDataPrefix) { + return "", "", fmt.Errorf("model: %w: anthropic image requires a data: URI, got %q", ErrBadRequest, uri) + } + rest := strings.TrimPrefix(uri, anthropicDataPrefix) + comma := strings.Index(rest, ",") + if comma < 0 { + return "", "", fmt.Errorf("model: %w: malformed data: URI %q", ErrBadRequest, uri) + } + meta := rest[:comma] + data := rest[comma+1:] + mediaType := meta + if idx := strings.Index(meta, ";"); idx >= 0 { + mediaType = meta[:idx] + } + if mediaType == "" { + return "", "", fmt.Errorf("model: %w: data: URI missing media type in %q", ErrBadRequest, uri) + } + return mediaType, data, nil +} + +// buildAnthropicTools converts ToolSpec schemas into Anthropic tools. +func buildAnthropicTools(tools []ToolSpec) []anthropicTool { + if len(tools) == 0 { + return nil + } + out := make([]anthropicTool, 0, len(tools)) + for _, t := range tools { + out = append(out, anthropicTool{ + Name: t.Name, + Description: t.Description, + InputSchema: t.Parameters, + }) + } + return out +} diff --git a/internal/runtime/model/anthropic_stream.go b/internal/runtime/model/anthropic_stream.go index 749d4ed..0aa8ae2 100644 --- a/internal/runtime/model/anthropic_stream.go +++ b/internal/runtime/model/anthropic_stream.go @@ -28,11 +28,14 @@ type anthropicStartBlock struct { } // anthropicDelta is the incremental payload inside a content_block_delta event. +// The message_delta event reuses the same field name for its terminal +// stop_reason (end_turn / max_tokens / tool_use), so it is decoded here too. type anthropicDelta struct { Type string `json:"type"` // text_delta | input_json_delta | thinking_delta Text string `json:"text,omitempty"` PartialJSON string `json:"partial_json,omitempty"` Thinking string `json:"thinking,omitempty"` + StopReason string `json:"stop_reason,omitempty"` } // anthropicStreamError is the payload of an error event. @@ -201,6 +204,9 @@ func applyAnthropicEvent( } } case "message_delta": + if evt.Delta != nil { + agg.StopReason = evt.Delta.StopReason + } if evt.Usage != nil { agg.CompletionTokens = evt.Usage.OutputTokens } diff --git a/internal/runtime/model/model.go b/internal/runtime/model/model.go index 3fdf182..def94a5 100644 --- a/internal/runtime/model/model.go +++ b/internal/runtime/model/model.go @@ -182,6 +182,11 @@ type Response struct { // ToolCalls carries any tool_use blocks the assistant emitted. Empty when the // response is pure text. ToolCalls []ToolCall + // StopReason is the provider's raw stop/finish reason (Anthropic stop_reason: + // end_turn / max_tokens / tool_use; OpenAI finish_reason: stop / length / + // tool_calls), so callers can distinguish a token-budget truncation from a + // normal end. Empty when the provider reports none. + StopReason string // PromptTokens, CompletionTokens and TotalTokens report provider usage when // available; zero when the provider does not return it. PromptTokens int diff --git a/internal/runtime/model/openai.go b/internal/runtime/model/openai.go index d955a12..4c12ae2 100644 --- a/internal/runtime/model/openai.go +++ b/internal/runtime/model/openai.go @@ -488,6 +488,7 @@ func responseFromWire(wire *openAIChatResponse) Response { text := wire.Choices[0].Message.Content out.Content = text out.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: text}}} + out.StopReason = wire.Choices[0].FinishReason for _, tc := range wire.Choices[0].Message.ToolCalls { args := map[string]any{} if tc.Function.Arguments != "" { @@ -546,6 +547,9 @@ func scanSSE(r io.Reader, onEvent func(StreamEvent)) (Response, error) { agg.CompletionTokens = u.CompletionTokens agg.TotalTokens = u.TotalTokens } + if len(chunk.Choices) > 0 && chunk.Choices[0].FinishReason != "" { + agg.StopReason = chunk.Choices[0].FinishReason + } if len(chunk.Choices) == 0 || chunk.Choices[0].Delta == nil { continue } diff --git a/internal/runtime/model/truncation.go b/internal/runtime/model/truncation.go new file mode 100644 index 0000000..3ba2f42 --- /dev/null +++ b/internal/runtime/model/truncation.go @@ -0,0 +1,159 @@ +package model + +// Provider stop/finish reasons that mean the token budget cut the generation +// off: Anthropic reports stop_reason "max_tokens", OpenAI-compatible providers +// report finish_reason "length". Both are normalized into Response.StopReason +// verbatim so a trace keeps the provider's own vocabulary. +const ( + stopReasonMaxTokens = "max_tokens" + stopReasonLength = "length" +) + +// Truncation-recovery bounds. At most one retry with a doubled budget is +// allowed; the cap keeps a pathological budget from growing without bound. The +// hint is the non-empty placeholder returned when even the retry produced no +// visible text, so a truncated turn never surfaces as a silent empty reply. +// It never carries thinking: reasoning is not the answer. +const ( + truncationRetryCap = 32768 + truncationEmptyHint = "（思考过长，未能生成回复）" +) + +// effectiveMaxTokens resolves the budget an attempt actually sends: the explicit +// Request.MaxTokens, else the adapter default. def may be zero when the protocol +// has no adapter default, in which case the provider default applies. +func effectiveMaxTokens(n, def int) int { + if n > 0 { + return n + } + return def +} + +// retryMaxTokens doubles the effective budget for the one-shot truncation retry, +// capped at truncationRetryCap. It returns the effective budget unchanged when +// there is nothing to double (both n and def are zero, i.e. the provider default +// applies), letting the caller skip a retry that would send an identical request. +func retryMaxTokens(n, def int) int { + eff := effectiveMaxTokens(n, def) + if eff <= 0 { + return 0 + } + if bumped := eff * 2; bumped < truncationRetryCap { + return bumped + } + return truncationRetryCap +} + +// isTruncatedStopReason reports whether a stop reason means the token budget cut +// the generation off. +func isTruncatedStopReason(reason string) bool { + return reason == stopReasonMaxTokens || reason == stopReasonLength +} + +// isEmptyBody reports whether a response carries neither visible text nor tool +// calls. A tool-only round is NOT empty: its empty body is normal and belongs to +// the tool loop. +func isEmptyBody(r Response) bool { + return r.Content == "" && len(r.ToolCalls) == 0 +} + +// isTruncatedEmpty reports whether an attempt was cut off by the token budget +// before producing any visible output (no text, no tool call). Only this shape +// triggers the truncation retry/fallback; a normal tool_use round, whose body is +// legitimately empty, does not. +func isTruncatedEmpty(r Response) bool { + return isEmptyBody(r) && isTruncatedStopReason(r.StopReason) +} + +// withEmptyBodyHint fills a non-empty placeholder body when a response has +// neither text nor tool calls, so a truncated turn never surfaces as a silent +// empty reply. A response that already carries text or tool calls is returned +// unchanged. +func withEmptyBodyHint(r Response) Response { + if !isEmptyBody(r) { + return r + } + r.Content = truncationEmptyHint + r.Message = Message{Role: RoleAssistant, Content: []Content{{Type: ContentTypeText, Text: r.Content}}} + return r +} + +// callWithTruncationRetry runs a non-streaming attempt, retrying once with a +// doubled token budget when the first attempt was cut off with no visible +// output, then returning a non-empty placeholder when the retry still produced +// nothing. attempt receives the budget to send. An upstream error on the first +// attempt is returned unchanged (no retry); a failed retry falls back to the +// placeholder rather than surfacing an empty reply. +func callWithTruncationRetry(maxTokens, def int, attempt func(limit int) (Response, error)) (Response, error) { + out, err := attempt(maxTokens) + if err != nil { + return Response{}, err + } + if !isTruncatedEmpty(out) { + return out, nil + } + if bumped := retryMaxTokens(maxTokens, def); bumped > effectiveMaxTokens(maxTokens, def) { + if second, retryErr := attempt(bumped); retryErr == nil { + out = second + } + } + return withEmptyBodyHint(out), nil +} + +// callStreamWithTruncationRetry is the streaming counterpart of +// callWithTruncationRetry. attempt receives the budget and the event callback +// for that attempt. The retry drops thinking events (the first attempt already +// streamed its reasoning, so re-streaming it would duplicate the block) while +// forwarding text/tool events live; when even the retry yields no visible +// output, a non-empty placeholder is emitted as a text segment and returned as +// the response body, so the stream never ends on a silent truncation. +func callStreamWithTruncationRetry( + maxTokens, def int, + attempt func(limit int, onEvent func(StreamEvent)) (Response, error), + onEvent func(StreamEvent), +) (Response, error) { + out, err := attempt(maxTokens, onEvent) + if err != nil { + return Response{}, err + } + if !isTruncatedEmpty(out) { + return out, nil + } + if bumped := retryMaxTokens(maxTokens, def); bumped > effectiveMaxTokens(maxTokens, def) { + if second, retryErr := attempt(bumped, dropThinking(onEvent)); retryErr == nil { + out = second + } + } + if isEmptyBody(out) { + out = withEmptyBodyHint(out) + emitTextSegment(onEvent, out.Content) + } + return out, nil +} + +// dropThinking wraps onEvent so a retry attempt does not re-emit thinking +// segments. Text and tool_call events pass through unchanged. +func dropThinking(onEvent func(StreamEvent)) func(StreamEvent) { + if onEvent == nil { + return nil + } + return func(ev StreamEvent) { + switch ev.Kind { + case StreamKindThinkingStart, StreamKindThinkingDelta, StreamKindThinkingEnd: + return + default: + onEvent(ev) + } + } +} + +// emitTextSegment emits a single-delta text segment so a synthesized placeholder +// reaches streaming consumers that render only message.* events. +func emitTextSegment(onEvent func(StreamEvent), text string) { + if onEvent == nil || text == "" { + return + } + onEvent(StreamEvent{Kind: StreamKindTextStart}) + onEvent(StreamEvent{Kind: StreamKindTextDelta, Delta: text}) + onEvent(StreamEvent{Kind: StreamKindTextEnd}) +} diff --git a/internal/runtime/model/truncation_test.go b/internal/runtime/model/truncation_test.go new file mode 100644 index 0000000..3c6c9e4 --- /dev/null +++ b/internal/runtime/model/truncation_test.go @@ -0,0 +1,369 @@ +package model + +import ( + "context" + "encoding/json" + "io" + "net/http" + "net/http/httptest" + "reflect" + "strings" + "sync" + "testing" + + "git.yeij.top/AskaEth/Cyrene_v2/internal/config" +) + +// anthropicSeqServer answers successive /v1/messages calls with a fixed script +// of bodies (the last body repeats) and records every decoded request body, so a +// test can assert both the retry count and the retried max_tokens. +type anthropicSeqServer struct { + *httptest.Server + mu sync.Mutex + bodies []map[string]any +} + +func newAnthropicSeqServer(t *testing.T, contentType string, bodies ...string) *anthropicSeqServer { + t.Helper() + s := &anthropicSeqServer{} + idx := 0 + s.Server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + s.mu.Lock() + s.bodies = append(s.bodies, body) + n := idx + idx++ + s.mu.Unlock() + if n >= len(bodies) { + n = len(bodies) - 1 + } + w.Header().Set("Content-Type", contentType) + w.WriteHeader(http.StatusOK) + _, _ = io.WriteString(w, bodies[n]) + })) + t.Cleanup(s.Server.Close) + return s +} + +func (s *anthropicSeqServer) calls() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.bodies) +} + +func (s *anthropicSeqServer) body(i int) map[string]any { + s.mu.Lock() + defer s.mu.Unlock() + if i >= len(s.bodies) { + return nil + } + return s.bodies[i] +} + +// anthropicThinkingOnlyResponse is a thinking-only, budget-truncated reply: the +// reasoning consumed the whole max_tokens budget, so no text block was emitted. +const anthropicThinkingOnlyResponse = `{ + "id":"msg_think","model":"claude-test", + "content":[{"type":"thinking","thinking":"SECRET-REASONING long long long"}], + "stop_reason":"max_tokens", + "usage":{"input_tokens":5,"output_tokens":1024} +}` + +const anthropicEndTurnTextResponse = `{ + "id":"msg_text","model":"claude-test", + "content":[{"type":"text","text":"答案"}], + "stop_reason":"end_turn", + "usage":{"input_tokens":5,"output_tokens":10} +}` + +const anthropicToolUseResponse = `{ + "id":"msg_tool","model":"claude-test", + "content":[{"type":"tool_use","id":"toolu_1","name":"datetime","input":{"tz":"Asia/Shanghai"}}], + "stop_reason":"tool_use", + "usage":{"input_tokens":5,"output_tokens":8} +}` + +// anthropicTruncatedThinkingSSE is the streaming form of a thinking-only +// truncation: a thinking block, then message_delta stop_reason max_tokens. +const anthropicTruncatedThinkingSSE = `data: {"type":"message_start","message":{"model":"claude-test","usage":{"input_tokens":5}}} + +data: {"type":"content_block_start","index":0,"content_block":{"type":"thinking","thinking":""}} + +data: {"type":"content_block_delta","index":0,"delta":{"type":"thinking_delta","thinking":"SECRET-REASONING"}} + +data: {"type":"content_block_stop","index":0} + +data: {"type":"message_delta","delta":{"stop_reason":"max_tokens"},"usage":{"output_tokens":1024}} + +data: {"type":"message_stop"} + +` + +const anthropicEndTurnTextSSE = `data: {"type":"message_start","message":{"model":"claude-test","usage":{"input_tokens":5}}} + +data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}} + +data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"好"}} + +data: {"type":"content_block_stop","index":0} + +data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":3}} + +data: {"type":"message_stop"} + +` + +// TestAnthropicChat_TruncatedThinking_RetriesWithDoubledBudget proves the +// primary recovery: a max_tokens truncation with only a thinking block retries +// exactly once with the token budget doubled and surfaces the retry's text. +func TestAnthropicChat_TruncatedThinking_RetriesWithDoubledBudget(t *testing.T) { + t.Parallel() + srv := newAnthropicSeqServer(t, "application/json", anthropicThinkingOnlyResponse, anthropicEndTurnTextResponse) + ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) + + out, err := ad.Chat(context.Background(), Request{ + Messages: []Message{TextMessage(RoleUser, "hi")}, + MaxTokens: 1024, + }, Options{}) + if err != nil { + t.Fatalf("Chat error: %v", err) + } + if srv.calls() != 2 { + t.Fatalf("requests = %d, want 2 (one retry)", srv.calls()) + } + if got := srv.body(0)["max_tokens"]; got != float64(1024) { + t.Errorf("first request max_tokens = %v, want 1024", got) + } + if got := srv.body(1)["max_tokens"]; got != float64(2048) { + t.Errorf("retry request max_tokens = %v, want 2048 (doubled)", got) + } + if out.Content != "答案" { + t.Errorf("Content = %q, want the retry's text", out.Content) + } + if out.StopReason != "end_turn" { + t.Errorf("StopReason = %q, want end_turn", out.StopReason) + } + if strings.Contains(out.Content, "SECRET-REASONING") { + t.Errorf("Content leaked thinking: %q", out.Content) + } +} + +// TestAnthropicChat_TruncatedThinking_RetryStillEmpty_FallsBackToHint proves the +// safety net: when the retry is truncated too, the caller gets a non-empty +// placeholder instead of a silent empty reply, and the thinking never leaks. +func TestAnthropicChat_TruncatedThinking_RetryStillEmpty_FallsBackToHint(t *testing.T) { + t.Parallel() + srv := newAnthropicSeqServer(t, "application/json", anthropicThinkingOnlyResponse, anthropicThinkingOnlyResponse) + ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) + + out, err := ad.Chat(context.Background(), Request{ + Messages: []Message{TextMessage(RoleUser, "hi")}, + MaxTokens: 8192, + }, Options{}) + if err != nil { + t.Fatalf("Chat error: %v", err) + } + if srv.calls() != 2 { + t.Fatalf("requests = %d, want 2 (retry once, never more)", srv.calls()) + } + if got := srv.body(1)["max_tokens"]; got != float64(16384) { + t.Errorf("retry request max_tokens = %v, want 16384", got) + } + if out.Content != truncationEmptyHint { + t.Errorf("Content = %q, want the non-empty fallback hint", out.Content) + } + if out.Content == "" { + t.Error("Content is empty: a truncated turn must never surface as a silent reply") + } + if strings.Contains(out.Content, "SECRET-REASONING") { + t.Errorf("Content leaked thinking: %q", out.Content) + } + if out.Message.Role != RoleAssistant || len(out.Message.Content) != 1 || out.Message.Content[0].Text != truncationEmptyHint { + t.Errorf("Message = %+v, want a single assistant hint part", out.Message) + } +} + +// TestAnthropicChat_NormalStop_NoRetry proves a normal end_turn answer and a +// native tool_use round (whose body is legitimately empty) are left untouched: +// no retry, no fallback hint. +func TestAnthropicChat_NormalStop_NoRetry(t *testing.T) { + t.Parallel() + t.Run("end_turn", func(t *testing.T) { + t.Parallel() + srv := newAnthropicSeqServer(t, "application/json", anthropicEndTurnTextResponse) + ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) + out, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}, MaxTokens: 1024}, Options{}) + if err != nil { + t.Fatalf("Chat error: %v", err) + } + if srv.calls() != 1 { + t.Errorf("requests = %d, want 1 (no retry on end_turn)", srv.calls()) + } + if out.Content != "答案" { + t.Errorf("Content = %q, want 答案", out.Content) + } + }) + t.Run("tool_use", func(t *testing.T) { + t.Parallel() + srv := newAnthropicSeqServer(t, "application/json", anthropicToolUseResponse) + ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) + out, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}, MaxTokens: 1024}, Options{}) + if err != nil { + t.Fatalf("Chat error: %v", err) + } + if srv.calls() != 1 { + t.Errorf("requests = %d, want 1 (a tool_use round is not a truncation)", srv.calls()) + } + if len(out.ToolCalls) != 1 { + t.Fatalf("ToolCalls = %+v, want 1", out.ToolCalls) + } + if out.Content != "" { + t.Errorf("Content = %q, want empty (tool-only round)", out.Content) + } + }) +} + +// TestAnthropicChatStream_TruncatedThinking_RetriesWithDoubledBudget proves the +// streaming retry: the retry's thinking is suppressed (no duplicate block) while +// its text streams through, and the doubled budget is sent. +func TestAnthropicChatStream_TruncatedThinking_RetriesWithDoubledBudget(t *testing.T) { + t.Parallel() + srv := newAnthropicSeqServer(t, "text/event-stream", anthropicTruncatedThinkingSSE, anthropicEndTurnTextSSE) + ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) + + var events []StreamEvent + out, err := ad.ChatStream(context.Background(), Request{ + Messages: []Message{TextMessage(RoleUser, "hi")}, + MaxTokens: 1024, + }, func(ev StreamEvent) { events = append(events, ev) }) + if err != nil { + t.Fatalf("ChatStream error: %v", err) + } + if srv.calls() != 2 { + t.Fatalf("requests = %d, want 2 (one retry)", srv.calls()) + } + if got := srv.body(1)["max_tokens"]; got != float64(2048) { + t.Errorf("retry request max_tokens = %v, want 2048", got) + } + if out.Content != "好" { + t.Errorf("Content = %q, want the retry's text", out.Content) + } + if out.StopReason != "end_turn" { + t.Errorf("StopReason = %q, want end_turn", out.StopReason) + } + want := []StreamEvent{ + {Kind: StreamKindThinkingStart, Index: 0}, + {Kind: StreamKindThinkingDelta, Index: 0, Delta: "SECRET-REASONING"}, + {Kind: StreamKindThinkingEnd, Index: 0}, + {Kind: StreamKindTextStart, Index: 0}, + {Kind: StreamKindTextDelta, Index: 0, Delta: "好"}, + {Kind: StreamKindTextEnd, Index: 0}, + } + if !reflect.DeepEqual(events, want) { + t.Errorf("events = %+v, want %+v (retry thinking suppressed, retry text streamed)", events, want) + } +} + +// TestAnthropicChatStream_TruncatedThinking_RetryStillEmpty_EmitsHint proves the +// streaming safety net: when the retry is truncated too, a non-empty placeholder +// is emitted as a text segment (and returned as the body) so the frontend is not +// left with a silent truncated turn. +func TestAnthropicChatStream_TruncatedThinking_RetryStillEmpty_EmitsHint(t *testing.T) { + t.Parallel() + srv := newAnthropicSeqServer(t, "text/event-stream", anthropicTruncatedThinkingSSE, anthropicTruncatedThinkingSSE) + ad := NewAnthropicChat(config.LLMConfig{AnthropicBaseURL: srv.URL, AnthropicModel: "claude-test"}) + + var events []StreamEvent + out, err := ad.ChatStream(context.Background(), Request{ + Messages: []Message{TextMessage(RoleUser, "hi")}, + MaxTokens: 1024, + }, func(ev StreamEvent) { events = append(events, ev) }) + if err != nil { + t.Fatalf("ChatStream error: %v", err) + } + if srv.calls() != 2 { + t.Fatalf("requests = %d, want 2 (retry once, never more)", srv.calls()) + } + if out.Content != truncationEmptyHint { + t.Errorf("Content = %q, want the non-empty fallback hint", out.Content) + } + want := []StreamEvent{ + {Kind: StreamKindThinkingStart, Index: 0}, + {Kind: StreamKindThinkingDelta, Index: 0, Delta: "SECRET-REASONING"}, + {Kind: StreamKindThinkingEnd, Index: 0}, + {Kind: StreamKindTextStart, Index: 0}, + {Kind: StreamKindTextDelta, Index: 0, Delta: truncationEmptyHint}, + {Kind: StreamKindTextEnd, Index: 0}, + } + if !reflect.DeepEqual(events, want) { + t.Errorf("events = %+v, want %+v (hint streamed, thinking not re-emitted)", events, want) + } +} + +// TestTruncationBudget covers the budget arithmetic that bounds the retry. +func TestTruncationBudget(t *testing.T) { + t.Parallel() + cases := []struct { + name string + requested, def int + wantEffective int + wantRetry int + }{ + {"explicit doubles", 1024, anthropicDefaultMaxTokens, 1024, 2048}, + {"default doubles", 0, anthropicDefaultMaxTokens, anthropicDefaultMaxTokens, 16384}, + {"capped", 20000, anthropicDefaultMaxTokens, 20000, truncationRetryCap}, + {"nothing to double", 0, 0, 0, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + if got := effectiveMaxTokens(tc.requested, tc.def); got != tc.wantEffective { + t.Errorf("effectiveMaxTokens(%d, %d) = %d, want %d", tc.requested, tc.def, got, tc.wantEffective) + } + if got := retryMaxTokens(tc.requested, tc.def); got != tc.wantRetry { + t.Errorf("retryMaxTokens(%d, %d) = %d, want %d", tc.requested, tc.def, got, tc.wantRetry) + } + }) + } +} + +// TestOpenAIChat_StopReasonMapping asserts the OpenAI adapter fills the unified +// Response.StopReason from finish_reason on both the non-streaming and streaming +// paths, so a "length" truncation is identifiable across protocols. +func TestOpenAIChat_StopReasonMapping(t *testing.T) { + t.Parallel() + t.Run("non_streaming", func(t *testing.T) { + t.Parallel() + body := `{"id":"c1","model":"gpt-test","choices":[{"index":0,"message":{"role":"assistant","content":"cut"},"finish_reason":"length"}],"usage":{"prompt_tokens":5,"completion_tokens":64,"total_tokens":69}}` + srv := newCompletionServer(t, http.StatusOK, "application/json", body, nil) + ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "gpt-test"}) + out, err := ad.Chat(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, Options{}) + if err != nil { + t.Fatalf("Chat error: %v", err) + } + if out.StopReason != "length" { + t.Errorf("StopReason = %q, want length", out.StopReason) + } + }) + t.Run("streaming", func(t *testing.T) { + t.Parallel() + sse := strings.Join([]string{ + `data: {"choices":[{"index":0,"delta":{"content":"Hi"}}]}`, + ``, + `data: {"choices":[{"index":0,"delta":{},"finish_reason":"stop"}]}`, + ``, + `data: [DONE]`, + ``, + }, "\n") + srv := newCompletionServer(t, http.StatusOK, "text/event-stream", sse, nil) + ad := NewOpenAIChat(config.LLMConfig{BaseURL: srv.URL, Model: "gpt-test"}) + out, err := ad.ChatStream(context.Background(), Request{Messages: []Message{TextMessage(RoleUser, "hi")}}, nil) + if err != nil { + t.Fatalf("ChatStream error: %v", err) + } + if out.StopReason != "stop" { + t.Errorf("StopReason = %q, want stop", out.StopReason) + } + }) +}