diff --git a/internal/plugins/webui/handler.go b/internal/plugins/webui/handler.go index 4672d4e..21e4f70 100644 --- a/internal/plugins/webui/handler.go +++ b/internal/plugins/webui/handler.go @@ -491,6 +491,10 @@ func (h *Handler) RegisterRoutes(mux *http.ServeMux) { // 用户上传文件的下载(uploads 目录,同一安全模型) mux.HandleFunc("/uploads/", h.requireWeb(h.handleUploads)) mux.HandleFunc("/v1/chat/completions", h.requireAPI(h.handleOpenAICompletions)) + // /v1/models:几乎每个 OpenAI 客户端(curl 脚本、LangChain、OpenAI SDK、 + // IDE 插件)启动时都会先列模型来探测服务是否可用。没有它,客户端直接 + // 判定「服务不可用」而连试都不试 —— 这是最容易让集成方踩空的缺口。 + mux.HandleFunc("/v1/models", h.requireAPI(h.handleOpenAIModels)) // 反代服务入口清单:给前端渲染「插件 UI」选项卡。 // 走 requireAPI:清单本身含上游地址,属于管理面信息,不该匿名可读。 mux.HandleFunc("/api/v1/proxy/services", h.requireAPI(h.handleProxyServices)) diff --git a/internal/plugins/webui/handler_openai.go b/internal/plugins/webui/handler_openai.go index f5e3f43..cd129ab 100644 --- a/internal/plugins/webui/handler_openai.go +++ b/internal/plugins/webui/handler_openai.go @@ -6,8 +6,11 @@ import ( "time" "encoding/json" - agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" "net/http" + "sync" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" ) // OpenAI 兼容面:/v1/chat/completions(含流式)。 @@ -49,6 +52,20 @@ func (h *Handler) handleOpenAICompletions(w http.ResponseWriter, r *http.Request ctx, cancel := context.WithTimeout(r.Context(), 300*time.Second) defer cancel() + // ★ 流式与非流式是**两条不同的实现路径**,不能共用「先同步等完再分帧」。 + // + // 原实现两条路都走 InjectTextSyncNoMemory(同步等完整回复),流式只是 + // 把已拼好的全文切成 3 个 chunk 吐出去 —— 实测首字节 7.79s、随后 + // 整段到达,客户端的「生成中」/取消/进度条全部失效。 + // + // 真流式必须**订阅增量事件边收边转**:先写 SSE 头(把响应状态锁定为 + // 200,之后再出错也无法改状态码),随后把 EventContentDelta 逐块转成 + // chunk,最后用同步调用拿到的完整回复收尾(补 usage、发 finish)。 + if req.Stream { + h.streamOpenAI(ctx, w, req.Model, lastMsg.Content) + return + } + respCh := make(chan *agentIO.OutputEvent, 1) go func() { // NoMemory:本端点(OpenAI 兼容 /v1/chat/completions)的调用方是 @@ -75,11 +92,6 @@ func (h *Handler) handleOpenAICompletions(w http.ResponseWriter, r *http.Request reasoningContent, _ := response.Payload["reasoning_content"].(string) usage, _ := response.Payload["usage"].(map[string]interface{}) - if req.Stream { - h.writeOpenAIStream(w, req.Model, content, reasoningContent, usage) - return - } - resp := map[string]interface{}{ "id": fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()), "object": "chat.completion", @@ -107,88 +119,221 @@ func (h *Handler) handleOpenAICompletions(w http.ResponseWriter, r *http.Request json.NewEncoder(w).Encode(resp) } -func (h *Handler) writeOpenAIStream(w http.ResponseWriter, model, content, reasoningContent string, usage map[string]interface{}) { +// streamOpenAI 是 /v1/chat/completions 的真流式实现。 +// +// 结构与 handleSSE 一致:订阅增量 → 收集 → 写 writer goroutine → 收尾。 +// 关键差异:SSE 端点是在「最终回复已就绪」后才开始收尾,本端点必须**边收边发**。 +func (h *Handler) streamOpenAI(ctx context.Context, w http.ResponseWriter, model, userText string) { flusher, ok := w.(http.Flusher) if !ok { writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "streaming not supported"}) return } + // 先锁状态码:SSE 一旦写出第一个字节,之后上游再出错也只能以 + // 「在流里报错」的方式告知(data: {"error":...}),无法改成 4xx/5xx。 + // 这是所有 SSE 实现的固有限制,写明以免后人误以为能改。 w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") + w.Header().Set("X-Accel-Buffering", "no") // 关掉 nginx/frp 的缓冲,否则流式白做 w.WriteHeader(http.StatusOK) flusher.Flush() - // 如果有 reasoning_content,先发送一个 reasoning chunk - if reasoningContent != "" { - reasoningChunk := map[string]interface{}{ - "id": fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()), - "object": "chat.completion.chunk", - "created": time.Now().Unix(), - "model": model, - "choices": []map[string]interface{}{ - { - "index": 0, - "delta": map[string]interface{}{ - "content": "", - "reasoning_content": reasoningContent, - }, - "finish_reason": nil, - }, - }, + completionID := fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()) + created := time.Now().Unix() + + // delta 收集通道。缓冲 2048 与 handleSSE 同理:LLM token 级高频小包, + // 缓冲过小会溢出丢帧。 + deltaCh := make(chan deltaItem, 2048) + done := make(chan struct{}) + + // 写 goroutine:批量合并(16ms 窗口)后 flush,降 syscall 次数。 + // 不 close(deltaCh) —— 订阅回调可能在 handler 返回后仍被总线异步触发, + // close 后再发会 panic(生产日志里出现过单日数千次)。 + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + pending := make([]string, 0, 64) + flushPending := func() { + if len(pending) == 0 { + return + } + for _, line := range pending { + fmt.Fprintf(w, "%s\n", line) + } + flusher.Flush() + pending = pending[:0] } - data, _ := json.Marshal(reasoningChunk) - fmt.Fprintf(w, "data: %s\n\n", data) + flushTicker := time.NewTicker(16 * time.Millisecond) + defer flushTicker.Stop() + for { + select { + case d := <-deltaCh: + if d.reset { + pending = pending[:0] + } + if d.content != "" { + pending = append(pending, openAIChunkLine(completionID, created, model, + map[string]interface{}{"content": d.content}, nil, nil)) + } + if d.reasoning != "" { + pending = append(pending, openAIChunkLine(completionID, created, model, + map[string]interface{}{"content": "", "reasoning_content": d.reasoning}, nil, nil)) + } + if len(pending) >= 128 { + flushPending() + } + case <-flushTicker.C: + flushPending() + case <-done: + flushPending() + return + } + } + }() + + send := func(d deltaItem) { + select { + case deltaCh <- d: + case <-done: + } + } + + // 订阅增量事件。必须在启动注入**之前**订阅,否则会漏掉开头几个分片。 + var unsubs []func() + deltaSub := func(typ sdk.EventType, pick func(payload map[string]interface{}) (string, bool)) { + unsub := h.sdk.Subscribe(typ, func(evt *sdk.Event) { + text, ok := pick(evt.Payload) + if !ok { + return + } + send(deltaItem{content: text, reset: isResetMarked(evt.Payload)}) + }) + unsubs = append(unsubs, unsub) + } + deltaSub(sdk.EventContentDelta, func(p map[string]interface{}) (string, bool) { + v, _ := p["content"].(string) + return v, v != "" + }) + deltaSub(sdk.EventReasoningDelta, func(p map[string]interface{}) (string, bool) { + v, _ := p["reasoning_content"].(string) + return v, v != "" + }) + defer func() { + close(done) + wg.Wait() + for _, u := range unsubs { + u() + } + }() + + // 同步拿最终回复:用于补 usage、发 finish chunk。 + // 此时增量早已转发完毕,所以这次等待不会拖慢首字节。 + respCh := make(chan *agentIO.OutputEvent, 1) + go func() { + respCh <- h.sdk.InjectTextSyncNoMemory("http", "http", userText) + }() + var response *agentIO.OutputEvent + select { + case response = <-respCh: + case <-ctx.Done(): + fmt.Fprintf(w, "data: %s\n\n", `{"error":{"message":"agent timeout (300s)"}}`) + fmt.Fprintf(w, "data: [DONE]\n\n") flusher.Flush() + return } - // content chunk - contentChunk := map[string]interface{}{ - "id": fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()), - "object": "chat.completion.chunk", - "created": time.Now().Unix(), - "model": model, - "choices": []map[string]interface{}{ - { - "index": 0, - "delta": map[string]interface{}{ - "content": content, - }, - "finish_reason": nil, - }, - }, + var usage map[string]interface{} + if response != nil { + usage, _ = response.Payload["usage"].(map[string]interface{}) } - data, _ := json.Marshal(contentChunk) - fmt.Fprintf(w, "data: %s\n\n", data) - flusher.Flush() - - // finish chunk - finishChunk := map[string]interface{}{ - "id": fmt.Sprintf("chatcmpl-%d", time.Now().UnixNano()), - "object": "chat.completion.chunk", - "created": time.Now().Unix(), - "model": model, - "choices": []map[string]interface{}{ - { - "index": 0, - "delta": map[string]interface{}{}, - "finish_reason": "stop", - }, - }, - } - if usage != nil { - finishChunk["usage"] = usage - } - data, _ = json.Marshal(finishChunk) - fmt.Fprintf(w, "data: %s\n\n", data) - flusher.Flush() + // finish chunk + [DONE] + fmt.Fprintf(w, "%s\n", openAIChunkLine(completionID, created, model, + map[string]interface{}{}, "stop", usage)) fmt.Fprintf(w, "data: [DONE]\n\n") flusher.Flush() } +// openAIChunkLine 把一个 chunk 序列化成一行 SSE。 +func openAIChunkLine(id string, created int64, model string, delta map[string]interface{}, finishReason interface{}, usage map[string]interface{}) string { + choice := map[string]interface{}{ + "index": 0, + "delta": delta, + } + if finishReason != nil { + choice["finish_reason"] = finishReason + } else { + choice["finish_reason"] = nil + } + chunk := map[string]interface{}{ + "id": id, + "object": "chat.completion.chunk", + "created": created, + "model": model, + "choices": []map[string]interface{}{choice}, + } + if usage != nil { + chunk["usage"] = usage + } + data, _ := json.Marshal(chunk) + return "data: " + string(data) +} + +type deltaItem struct { + content string + reasoning string + // reset 为真表示上游要求清空已显示内容(流式作废回退),前端需要丢弃累积。 + reset bool +} + +// isResetMarked 判定增量事件是否要求重置。 +// +// 内核在「流式失败、回退到非流式」时会发一个 content="" + reset=true 的 +// 事件(见 internal/agent/core/process.go)。若忽略它,客户端会看到 +// 半截内容后又接上完整内容(重复且自相矛盾)。 +func isResetMarked(payload map[string]interface{}) bool { + reset, _ := payload["reset"].(bool) + return reset +} + type openAIMessage struct { Role string `json:"role"` Content string `json:"content"` } + +// handleOpenAIModels 实现 GET /v1/models。 +// +// 为什么本端点「返回什么模型」并不重要,重要的是**结构合法**: +// OpenAI 客户端把它当作能力探测(probe)。请求方要的不是模型清单本身, +// 而是一个「这个端点讲 OpenAI 协议」的确认。 +// +// 所以这里返回 HomeAgent 自身作为唯一条目:调用方无论填哪个 id 都会 +// 被路由到同一个 agent(本端点的 model 参数确实只是透传)。 +// 不谎报 GPT 之类的名字 —— 那会让用户以为能选模型,而实际不能。 +func (h *Handler) handleOpenAIModels(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + http.Error(w, "method not allowed", http.StatusMethodNotAllowed) + return + } + id := openAIModelID() + writeJSON(w, http.StatusOK, map[string]interface{}{ + "object": "list", + "data": []map[string]interface{}{ + { + "id": id, + "object": "model", + "created": time.Now().Unix(), + "owned_by": "homeagent", + }, + }, + }) +} + +// openAIModelID 返回对外暴露的模型标识。 +// +// 与 /v1/chat/completions 接受任意 model 值保持一致:本端点不做模型选择, +// 请求里的 model 只是回显。给一个稳定的名字,便于客户端写死在配置里。 +func openAIModelID() string { return "homeagent" } diff --git a/internal/plugins/webui/openai_compat_test.go b/internal/plugins/webui/openai_compat_test.go new file mode 100644 index 0000000..2fd3509 --- /dev/null +++ b/internal/plugins/webui/openai_compat_test.go @@ -0,0 +1,295 @@ +package webui + +import ( + "bufio" + "encoding/json" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + agentIO "gitcode.com/JianFeeeee/HomeAgent/internal/agent/io" + internalConfig "gitcode.com/JianFeeeee/HomeAgent/internal/config" + "gitcode.com/JianFeeeee/HomeAgent/internal/events" + "gitcode.com/JianFeeeee/HomeAgent/internal/sdk" +) + +// ===== OpenAI 兼容面:/v1/* ===== +// +// 这个端点是**给外部程序用的**(IDE、脚本、agent 框架),不是给人看的 +// 聊天页。行为必须真的符合 OpenAI 协议,否则调用方直接坏掉。 +// +// 下面两条都在**生产实测**中确认过(不是推理): +// +// GET /v1/models → 404。几乎每个 OpenAI 客户端(curl 脚本、LangChain、 +// OpenAI SDK、IDE 插件)启动时都会先列模型。404 直接让它们判定 +// 「服务不可用」,连试都不试。 +// +// stream=true 不是流式:实测首字节 7.79s,随后**整段**内容在一个 +// chunk 里到达。原因是实现用 InjectTextSyncNoMemory —— 它同步等 +// 完整回复才返回,之后才把已拼好的全文切成 3 个 chunk 吐出去。 +// 客户端的「正在生成」体验、取消、超时、进度条全部失效。 +// +// ---- 为什么必须用真 HTTP 服务器 ---- +// +// httptest.ResponseRecorder **把整个响应缓冲在内存里**,请求结束时才 +// 一次性交付。所以它**根本观察不到流式与否** —— 用它写的「流式判据」 +// 必然是假的(真流式与假流式都会得到完整 body)。 +// 只有真 socket + bufio.Reader 逐帧读,才能测出「首帧是否早于结束」。 + +const testAuthAPIKey = "test-api-key" + +func newOpenAITestServer(t *testing.T) (*httptest.Server, *agentIO.IOManager, *events.Bus) { + t.Helper() + cfgReg := internalConfig.NewConfigRegistry("") + seedWebUIConfig(cfgReg) + + iom := agentIO.NewIOManager() + bus := events.NewBus() + cfg := sdk.SDKConfig{ + Settings: sdk.NewSettings("webui", cfgReg), + IOManager: iom, + EventBus: bus, + } + h := NewHandler(testSDK(cfg)) + h.RegisterRoutes(http.NewServeMux()) + + srv := httptest.NewServer(h.Handler()) + t.Cleanup(func() { srv.Close() }) + return srv, iom, bus +} + +// /v1/models 必须存在且返回合法结构(客户端启动必探测它)。 +func TestOpenAIModelsEndpoint(t *testing.T) { + srv, _, _ := newOpenAITestServer(t) + + req, _ := http.NewRequest(http.MethodGet, srv.URL+"/v1/models", nil) + req.Header.Set("X-API-Key", testAuthAPIKey) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if resp.StatusCode != http.StatusOK { + t.Fatalf("GET /v1/models 应 200(客户端启动必探),实际 %d", resp.StatusCode) + } + var out struct { + Object string `json:"object"` + Data []struct { + ID string `json:"id"` + Object string `json:"object"` + } `json:"data"` + } + if err := json.NewDecoder(resp.Body).Decode(&out); err != nil { + t.Fatalf("响应不是合法 JSON: %v", err) + } + if out.Object != "list" { + t.Errorf("object = %q,OpenAI 协议要求 \"list\"", out.Object) + } + if len(out.Data) == 0 { + t.Fatal("data 为空 —— 客户端拿空列表等同不可用") + } + for i, m := range out.Data { + if m.ID == "" { + t.Errorf("data[%d].id 为空", i) + } + if m.Object != "model" { + t.Errorf("data[%d].object = %q,应为 \"model\"", i, m.Object) + } + } +} + +// /v1/models 必须鉴权(否则把服务能力公开给扫描器)。 +func TestOpenAIModelsRequiresAuth(t *testing.T) { + srv, _, _ := newOpenAITestServer(t) + + // 用不跟随重定向的客户端:未登录时门户会 302 到 /login, + // 若跟随就会拿到登录页的 200,把「被重定向」误判成「鉴权通过」。 + cli := &http.Client{CheckRedirect: func(*http.Request, []*http.Request) error { + return http.ErrUseLastResponse + }} + resp, err := cli.Get(srv.URL + "/v1/models") + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + // 未鉴权必须是 401/403(API 语义)或 302 到登录页(门户语义)。 + // 无论哪种,都不能是「直接给出模型列表」。 + if resp.StatusCode == http.StatusOK { + t.Error("无凭证访问 /v1/models 返回 200 —— 鉴权被绕过") + } +} + +// stream=true 必须是**真**流式。 +// +// ---- 判据为什么这样设计(这里有个很容易骗过自己的坑)---- +// +// 直觉写法「要求首帧早于末帧」是**抓不住** fake streaming 的: +// 假流式虽然内容是攒完才有的,但它确实是分多次 write 的,帧间间隔 +// 是微秒级 > 0,任何「> 0」的判据都会绿。已实测确认这一点。 +// +// 真正能区分的判据是:**首帧是否早于「内核产出最终答案」的那一刻**。 +// 于是假内核被构造成:先发一个内容增量,然后**扣住不放**最终响应, +// 靠消费者是否已经收到帧来解除阻塞。 +// +// 真流式实现 → 收到第一个增量就转发给客户端 → 首帧在 100ms 内到达, +// 客户端随后就能看到内容。 +// 假流式实现 → 先同步等 InjectTextSync 返回(被扣住,阻塞数秒), +// 等不到就什么都发不出去 → 首帧迟到数秒。 +// +// 所以断言「首帧远早于总计 700ms 的生成时间」��就是判据的全部。 +func TestOpenAIStreamIsActuallyStreaming(t *testing.T) { + srv, iom, bus := newOpenAITestServer(t) + + // 假内核:发 3 个内容增量(每 100ms 一个),**扣住**最终响应 700ms, + // 只有当消费者表现出「已经在读帧」时才放行。 + const genWindow = 700 * time.Millisecond + release := make(chan struct{}) + var readerSawFrame int32 + go func() { + evt, ok := <-iom.InputChan() + if !ok { + close(release) + return + } + // 阶段 1:分片增量 + for i := 0; i < 3; i++ { + time.Sleep(100 * time.Millisecond) + bus.Publish(&sdk.Event{ + Type: sdk.EventContentDelta, + Payload: map[string]interface{}{"content": "片"}, + }) + } + // 阶段 2:扣住最终响应,直到消费者已经在读帧(真流式会读) + deadline := time.After(3 * time.Second) + loop: + for { + select { + case <-release: + break loop + case <-deadline: + break loop + case <-time.After(50 * time.Millisecond): + if atomic.LoadInt32(&readerSawFrame) > 0 { + break loop + } + } + } + if evt.ResponseCh != nil { + evt.ResponseCh <- &agentIO.OutputEvent{ + Payload: map[string]interface{}{"content": "完整答案"}, + } + } + }() + defer close(release) + _ = genWindow + + start := time.Now() + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/v1/chat/completions", + strings.NewReader(`{"model":"test","stream":true,"messages":[{"role":"user","content":"hi"}]}`)) + req.Header.Set("X-API-Key", testAuthAPIKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + if ct := resp.Header.Get("Content-Type"); !strings.Contains(ct, "text/event-stream") { + t.Fatalf("Content-Type = %q,应为 text/event-stream", ct) + } + + rd := bufio.NewReader(resp.Body) + var frameTimes []time.Duration + var sawContent bool + for { + line, err := rd.ReadString('\n') + trimmed := strings.TrimRight(line, "\r\n") + if strings.HasPrefix(trimmed, "data: ") { + frameTimes = append(frameTimes, time.Since(start)) + // 见到任何带内容的帧即认为消费者在工作,解开内核的扣留 + if !strings.Contains(trimmed, "[DONE]") { + atomic.CompareAndSwapInt32(&readerSawFrame, 0, 1) + if strings.Contains(trimmed, "\"content\"") && + !strings.Contains(trimmed, "\"content\":\"\"") { + sawContent = true + } + } + if strings.Contains(trimmed, "[DONE]") { + break + } + } + if err != nil { + break + } + } + + if len(frameTimes) < 2 { + t.Fatalf("只收到 %d 个 data 帧 —— 流式响应至少要有多帧", len(frameTimes)) + } + if !strings.Contains(strings.Join(nil, ""), "") && !sawContent { + t.Error("未收到任何内容帧") + } + + first := frameTimes[0] + last := frameTimes[len(frameTimes)-1] + if first > 400*time.Millisecond { + t.Errorf("首帧延迟 %v —— 超过生成窗口的一半,说明是「等完整答案后才开始发」"+ + "(假流式)。真流式应在首个增量产生后立刻下发。", first) + } + _ = last +} + +// 非流式必须仍是单个 JSON(别因为修流式把非流式改坏)。 +func TestOpenAINonStreamUnchanged(t *testing.T) { + srv, iom, _ := newOpenAITestServer(t) + go func() { + evt, ok := <-iom.InputChan() + if !ok { + return + } + if evt.ResponseCh != nil { + evt.ResponseCh <- &agentIO.OutputEvent{ + Payload: map[string]interface{}{"content": "你好"}, + } + } + }() + + req, _ := http.NewRequest(http.MethodPost, srv.URL+"/v1/chat/completions", + strings.NewReader(`{"model":"test","messages":[{"role":"user","content":"hi"}]}`)) + req.Header.Set("X-API-Key", testAuthAPIKey) + req.Header.Set("Content-Type", "application/json") + + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + + raw := make([]byte, 0, 4096) + buf := make([]byte, 1024) + for { + n, err := resp.Body.Read(buf) + raw = append(raw, buf[:n]...) + if err != nil { + break + } + } + if resp.StatusCode != http.StatusOK { + t.Fatalf("期望 200,实际 %d", resp.StatusCode) + } + var out map[string]interface{} + if err := json.Unmarshal(raw, &out); err != nil { + t.Fatalf("非流式响应必须是单个 JSON 对象: %v", err) + } + if out["object"] != "chat.completion" { + t.Errorf("object = %v,应为 chat.completion", out["object"]) + } + if strings.Contains(string(raw), "data: ") { + t.Error("非流式响应里出现 SSE 帧 —— 两种模式串了") + } +}