diff --git a/apps/edge/internal/openai/server_test.go b/apps/edge/internal/openai/server_test.go index 445624b..04f2652 100644 --- a/apps/edge/internal/openai/server_test.go +++ b/apps/edge/internal/openai/server_test.go @@ -1,6 +1,7 @@ package openai import ( + "bufio" "context" "encoding/json" "net/http" @@ -1092,6 +1093,61 @@ func TestChatCompletionsStreamSynthesizesToolCallsFromClineTextBlock(t *testing. } } +func TestChatCompletionsStreamWithToolsFlushesTextBeforeComplete(t *testing.T) { + fake := &fakeRunService{events: make(chan *iop.RunEvent)} + srv := NewServer(config.EdgeOpenAIConf{Adapter: "openai_compat"}, fake, nil) + httpSrv := httptest.NewServer(srv.routes()) + defer httpSrv.Close() + + resp, err := http.Post(httpSrv.URL+"/v1/chat/completions", "application/json", strings.NewReader(`{ + "model":"stream-model", + "stream":true, + "messages":[{"role":"user","content":"explain sudo"}], + "tools":[{"type":"function","function":{"name":"run_commands"}}], + "tool_choice":"auto" + }`)) + if err != nil { + t.Fatalf("post: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("status: got %d", resp.StatusCode) + } + + lines := make(chan string, 16) + scanDone := make(chan struct{}) + go func() { + defer close(scanDone) + scanner := bufio.NewScanner(resp.Body) + for scanner.Scan() { + lines <- scanner.Text() + } + }() + + fake.events <- &iop.RunEvent{Type: "delta", Delta: "streaming before complete"} + + deadline := time.After(2 * time.Second) + for { + select { + case line := <-lines: + if strings.Contains(line, `"finish_reason"`) { + t.Fatalf("saw finish before streamed content: %s", line) + } + if strings.Contains(line, `"content":"streaming before complete"`) { + fake.events <- &iop.RunEvent{Type: "complete"} + select { + case <-scanDone: + case <-time.After(2 * time.Second): + t.Fatal("stream did not close after complete") + } + return + } + case <-deadline: + t.Fatal("tools request did not stream content before complete") + } + } +} + func TestChatCompletionsStreamSynthesizesToolCallsFromClineTemplateCall(t *testing.T) { fake := &fakeRunService{events: make(chan *iop.RunEvent, 3)} fake.events <- &iop.RunEvent{Type: "delta", Delta: "먼저 현재 git 상태를 확인하겠습니다.\n\n"} diff --git a/apps/edge/internal/openai/stream.go b/apps/edge/internal/openai/stream.go index 39f6389..84611ae 100644 --- a/apps/edge/internal/openai/stream.go +++ b/apps/edge/internal/openai/stream.go @@ -6,6 +6,7 @@ import ( "net/http" "strings" "time" + "unicode" "go.uber.org/zap" @@ -36,13 +37,18 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re }}, }) - if len(req.Tools) > 0 || (outputPolicy.Strict && outputPolicy.StreamBuffer) { + if outputPolicy.Strict && outputPolicy.StreamBuffer { s.streamBufferedChatCompletion(w, r, handle, id, created, model, flusher, outputPolicy, req.Tools) return } var contentBuilder strings.Builder var reasoningBuilder strings.Builder + var emittedContent strings.Builder + var toolTextFilter *streamToolTextFilter + if len(req.Tools) > 0 { + toolTextFilter = &streamToolTextFilter{} + } defer func() { s.logger.Info("openai chat completion stream closed", zap.String("run_id", handle.Dispatch().RunID), @@ -86,17 +92,16 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re if event.GetDelta() == "" { continue } - contentBuilder.WriteString(event.GetDelta()) - writeSSE(w, flusher, chatCompletionChunk{ - ID: id, - Object: "chat.completion.chunk", - Created: created, - Model: model, - Choices: []chatCompletionChunkChoice{{ - Index: 0, - Delta: chatDelta{Content: event.GetDelta()}, - }}, - }) + delta := event.GetDelta() + contentBuilder.WriteString(delta) + if toolTextFilter != nil { + delta = toolTextFilter.Append(delta) + } + if delta == "" { + continue + } + emittedContent.WriteString(delta) + writeContentDeltaSSE(w, flusher, id, created, model, delta) case "reasoning_delta": if event.GetDelta() == "" { continue @@ -116,10 +121,42 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re }}, }) case "complete": - finishReason := event.GetMetadata()["finish_reason"] + metadata := event.GetMetadata() + finishReason := metadata["finish_reason"] if finishReason == "" { finishReason = "stop" } + nativeToolCalls := toolCallsFromRunMetadata(metadata) + if len(nativeToolCalls) > 0 { + if toolTextFilter != nil { + pending := toolTextFilter.Flush() + if pending != "" { + emittedContent.WriteString(pending) + writeContentDeltaSSE(w, flusher, id, created, model, pending) + } + } + nativeToolCalls = normalizeNativeToolCallsForResponse(nativeToolCalls, req.Tools) + writeToolCallsDeltaSSE(w, flusher, id, created, model, nativeToolCalls) + finishReason = "tool_calls" + } else if toolTextFilter != nil && shouldSynthesizeTextToolCalls(handle.Dispatch(), isTextToolFallback(metadata)) { + cleaned, toolCalls := synthesizeToolCallsFromText(contentBuilder.String(), req.Tools, handle.Dispatch().RunID) + if len(toolCalls) > 0 { + if remaining := unstreamedCleanedText(cleaned, emittedContent.String()); remaining != "" { + emittedContent.WriteString(remaining) + writeContentDeltaSSE(w, flusher, id, created, model, remaining) + } + writeToolCallsDeltaSSE(w, flusher, id, created, model, toolCalls) + finishReason = "tool_calls" + } else if pending := toolTextFilter.Flush(); pending != "" { + emittedContent.WriteString(pending) + writeContentDeltaSSE(w, flusher, id, created, model, pending) + } + } else if toolTextFilter != nil { + if pending := toolTextFilter.Flush(); pending != "" { + emittedContent.WriteString(pending) + writeContentDeltaSSE(w, flusher, id, created, model, pending) + } + } writeSSE(w, flusher, chatCompletionChunk{ ID: id, Object: "chat.completion.chunk", @@ -152,6 +189,116 @@ func (s *Server) streamChatCompletion(w http.ResponseWriter, r *http.Request, re } } +type streamToolTextFilter struct { + pending string +} + +func (f *streamToolTextFilter) Append(delta string) string { + f.pending += delta + if idx := firstStreamToolTextCandidateIndex(f.pending); idx >= 0 { + out := strings.TrimRightFunc(f.pending[:idx], unicode.IsSpace) + f.pending = f.pending[idx:] + return out + } + flushLen := streamToolTextSafeFlushLen(f.pending) + out := strings.TrimRightFunc(f.pending[:flushLen], unicode.IsSpace) + f.pending = f.pending[len(out):] + return out +} + +func (f *streamToolTextFilter) Flush() string { + out := f.pending + f.pending = "" + return out +} + +func firstStreamToolTextCandidateIndex(s string) int { + xml := strings.Index(strings.ToLower(s), "