From 9b7829744257e3a7802c0acd121f026bb8ebfebf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=85=AC=E6=98=8E?= <83812544+Ed1s0nZ@users.noreply.github.com> Date: Sat, 15 Aug 2026 01:46:12 +0800 Subject: [PATCH] Add files via upload --- internal/einoobserve/attach.go | 12 ++++-- internal/einoobserve/attach_test.go | 23 +++++++++++ internal/openai/reasoning_payload_test.go | 38 +++++++++++++++++++ .../openai/reasoning_tool_choice_compat.go | 7 ++-- 4 files changed, 73 insertions(+), 7 deletions(-) diff --git a/internal/einoobserve/attach.go b/internal/einoobserve/attach.go index 62c5e4bd..9846a2f7 100644 --- a/internal/einoobserve/attach.go +++ b/internal/einoobserve/attach.go @@ -38,6 +38,7 @@ type Params struct { ConversationID string OrchMode string OrchestratorName string + RunID string } // AttachAgentRunCallbacks returns ctx wrapped with callbacks.InitCallbacks when enabled. @@ -53,7 +54,10 @@ func AttachAgentRunCallbacks(ctx context.Context, cfg *config.MultiAgentEinoCall if mode == "off" { return ctx } - runID := uuid.New().String() + runID := strings.TrimSpace(p.RunID) + if runID == "" { + runID = uuid.New().String() + } if p.Progress != nil && cfg.ShouldEmitEinoTraceSSE(mode) { p.Progress("eino_trace_run", "Eino callbacks session", map[string]interface{}{ "runId": runID, @@ -206,7 +210,7 @@ func (h *runHandler) onStart(ctx context.Context, info *callbacks.RunInfo, input "spanId": spanID, "parentSpanId": parentID, "conversationId": strings.TrimSpace(h.params.ConversationID), - "orchestration": strings.TrimSpace(h.params.OrchMode), + "orchestration": strings.TrimSpace(h.params.OrchMode), "component": string(ri.Component), "name": ri.Name, "type": ri.Type, @@ -255,7 +259,7 @@ func (h *runHandler) onEnd(ctx context.Context, info *callbacks.RunInfo, output "runId": h.runID, "spanId": spanID, "conversationId": strings.TrimSpace(h.params.ConversationID), - "orchestration": strings.TrimSpace(h.params.OrchMode), + "orchestration": strings.TrimSpace(h.params.OrchMode), "component": string(ri.Component), "name": ri.Name, "type": ri.Type, @@ -301,7 +305,7 @@ func (h *runHandler) onError(ctx context.Context, info *callbacks.RunInfo, err e "runId": h.runID, "spanId": spanID, "conversationId": strings.TrimSpace(h.params.ConversationID), - "orchestration": strings.TrimSpace(h.params.OrchMode), + "orchestration": strings.TrimSpace(h.params.OrchMode), "component": string(ri.Component), "name": ri.Name, "type": ri.Type, diff --git a/internal/einoobserve/attach_test.go b/internal/einoobserve/attach_test.go index f4e2d80b..d12290a2 100644 --- a/internal/einoobserve/attach_test.go +++ b/internal/einoobserve/attach_test.go @@ -16,6 +16,29 @@ func TestAttachAgentRunCallbacks_Disabled(t *testing.T) { } } +func TestAttachAgentRunCallbacksUsesProvidedRunID(t *testing.T) { + emit := true + var gotRunID string + ctx := context.Background() + cfg := &config.MultiAgentEinoCallbacksConfig{Enabled: true, Mode: "sse", SseTraceToClient: &emit} + + AttachAgentRunCallbacks(ctx, cfg, Params{ + RunID: "run-shared", + Progress: func(eventType, _ string, data interface{}) { + if eventType != "eino_trace_run" { + return + } + if m, ok := data.(map[string]interface{}); ok { + gotRunID, _ = m["runId"].(string) + } + }, + }) + + if gotRunID != "run-shared" { + t.Fatalf("runId = %q, want run-shared", gotRunID) + } +} + func TestTruncateRunes(t *testing.T) { if got := truncateRunes("abc", 10); got != "abc" { t.Fatalf("got %q", got) diff --git a/internal/openai/reasoning_payload_test.go b/internal/openai/reasoning_payload_test.go index 4bede21e..9f7dccfe 100644 --- a/internal/openai/reasoning_payload_test.go +++ b/internal/openai/reasoning_payload_test.go @@ -205,6 +205,44 @@ func TestReasoningToolChoiceCompatRoundTripperDeepSeek(t *testing.T) { } } +func TestReasoningToolChoiceCompatRoundTripperDeepSeekEndpointWinsOverProfile(t *testing.T) { + var gotBody string + rt := &reasoningToolChoiceCompatRoundTripper{ + cfg: &config.OpenAIConfig{ + BaseURL: "https://api.deepseek.com/v1", + Model: "deepseek-v4-flash", + Reasoning: config.OpenAIReasoningConfig{ + Profile: "openai_compat", + }, + }, + base: roundTripperFunc(func(req *http.Request) (*http.Response, error) { + b, _ := io.ReadAll(req.Body) + gotBody = string(b) + return &http.Response{ + StatusCode: 200, + Body: io.NopCloser(strings.NewReader(`{"choices":[{"message":{"content":"ok"}}]}`)), + Header: http.Header{"Content-Type": []string{"application/json"}}, + }, nil + }), + } + req, err := http.NewRequest(http.MethodPost, "https://api.deepseek.com/v1/chat/completions", strings.NewReader( + `{"model":"deepseek-v4-flash","tool_choice":"required","tools":[],"messages":[]}`, + )) + if err != nil { + t.Fatal(err) + } + _, err = rt.RoundTrip(req) + if err != nil { + t.Fatal(err) + } + if strings.Contains(gotBody, "tool_choice") { + t.Fatalf("expected DeepSeek tool_choice stripped despite openai_compat profile, got %s", gotBody) + } + if !strings.Contains(gotBody, "tools") { + t.Fatalf("expected tools preserved for DeepSeek, got %s", gotBody) + } +} + type roundTripperFunc func(*http.Request) (*http.Response, error) func (f roundTripperFunc) RoundTrip(req *http.Request) (*http.Response, error) { diff --git a/internal/openai/reasoning_tool_choice_compat.go b/internal/openai/reasoning_tool_choice_compat.go index 26841590..fd703222 100644 --- a/internal/openai/reasoning_tool_choice_compat.go +++ b/internal/openai/reasoning_tool_choice_compat.go @@ -55,6 +55,9 @@ func isDeepSeekToolChoiceCompatProfile(cfg *config.OpenAIConfig) bool { if cfg == nil { return false } + if cfg.IsDeepSeekEndpointOrModel() { + return true + } profile := strings.ToLower(strings.TrimSpace(cfg.Reasoning.ProfileEffective())) if profile == "deepseek" || profile == "deepseek_compat" { return true @@ -62,7 +65,5 @@ func isDeepSeekToolChoiceCompatProfile(cfg *config.OpenAIConfig) bool { if profile != "" && profile != "auto" { return false } - baseURL := strings.ToLower(cfg.BaseURL) - model := strings.ToLower(cfg.Model) - return strings.Contains(baseURL, "deepseek") || strings.Contains(model, "deepseek") + return false }