fix(infra/llm): anthropic stream goroutine respects ctx cancel on send

This commit is contained in:
2026-05-04 09:35:19 +08:00
parent a1ba157d29
commit 027b5d5eba
+23 -6
View File
@@ -77,6 +77,15 @@ func (c *AnthropicClient) Stream(ctx context.Context, modelID string, req Reques
stream := c.sdk.Messages.NewStreaming(ctx, params)
out := make(chan StreamEvent, 16)
// send 在 ctx 取消时立即放弃发送,防止下游 reader 已退出导致 goroutine 永久阻塞。
send := func(ev StreamEvent) bool {
select {
case <-ctx.Done():
return false
case out <- ev:
return true
}
}
go func() {
defer close(out)
var inputTokens int
@@ -93,25 +102,33 @@ func (c *AnthropicClient) Stream(ctx context.Context, modelID string, req Reques
switch ev.Delta.Type {
case "text_delta":
if text := ev.Delta.AsTextDelta().Text; text != "" {
out <- StreamEvent{Type: EventTextDelta, Text: text}
if !send(StreamEvent{Type: EventTextDelta, Text: text}) {
return
}
}
case "thinking_delta":
if thinking := ev.Delta.AsThinkingDelta().Thinking; thinking != "" {
out <- StreamEvent{Type: EventThinkingDelta, Text: thinking}
if !send(StreamEvent{Type: EventThinkingDelta, Text: thinking}) {
return
}
}
}
case "message_delta":
ev := event.AsMessageDelta()
out <- StreamEvent{Type: EventUsage, Usage: &Usage{
if !send(StreamEvent{Type: EventUsage, Usage: &Usage{
PromptTokens: inputTokens,
CompletionTokens: int(ev.Usage.OutputTokens),
}}
}}) {
return
}
case "message_stop":
out <- StreamEvent{Type: EventDone, Reason: "end_turn"}
if !send(StreamEvent{Type: EventDone, Reason: "end_turn"}) {
return
}
}
}
if err := stream.Err(); err != nil {
out <- StreamEvent{Type: EventError, Err: fmt.Errorf("anthropic stream: %w", err)}
send(StreamEvent{Type: EventError, Err: fmt.Errorf("anthropic stream: %w", err)})
}
}()
return out, nil