From 075d41f5b3cda4a19026bd295aaec947f872ec37 Mon Sep 17 00:00:00 2001
From: Blizzard <blizzardzhang@icloud.com>
Date: Fri, 26 Jun 2026 11:10:18 +0800
Subject: [PATCH] =?UTF-8?q?feat(llm):=20=E6=9C=AC=E5=9C=B0=E6=A8=A1?=
 =?UTF-8?q?=E5=9E=8B=E6=8E=A5=E5=85=A5=20vLLM=20/=20Ollama=20+=20reasoning?=
 =?UTF-8?q?=20=E6=80=9D=E8=80=83=E8=BF=87=E7=A8=8B=E5=85=A5=E8=BD=A8?=
 =?UTF-8?q?=E8=BF=B9?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

vLLM 与 Ollama 都暴露 OpenAI 兼容 API（底层 go-openai 请求 {base}/chat/completions），
故统一走 openai 客户端，按 provider 归一化连接参数：
- Ollama/vLLM 的 OpenAI 端点固定在 /v1，BaseURL 漏写自动补全（否则打到 /chat/completions 404）
- 本地后端默认不校验 api_key → 缺省补占位（ollama→"ollama"，vllm→"EMPTY"；openai 客户端要求非空）
- 显式 key 一律尊重（vLLM --api-key 启动）；在线 provider 原样不动（DeepSeek 两种都收）

reasoning_content 适配：ChatStream 增 onReasoning 回调，捕获 reasoning 模型
（本地 Qwen3 思考 / DeepSeek-R1 / QwQ、在线 deepseek-v4-pro）的思考分片。思考阶段分片
Content 为空本就不污染答案；runAgent 把思考累计后 surface 到 exec 轨迹「推理过程」事件。

控制台 ModelManager 加 ollama 选项；llm 包补 normalizeBaseURL / apiKeyOrPlaceholder 单测。
四模块全绿。live：经 admin 配 ollama qwen2.5:0.5b → dispatcher 热切（日志 base 自动 .../v1）
→ POST /v1/chat/completions 200 端到端出答案；切回 deepseek-v4-pro → exec 轨迹现「推理过程：思考26字…」。

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
---
 project_analysis.md                           |  2 +-
 .../src/components/ModelManager.tsx           |  3 +-
 sundynix-dispatcher/internal/eino/graph.go    |  9 ++-
 .../internal/eino/integration_test.go         |  2 +-
 sundynix-dispatcher/internal/eino/model.go    |  4 +-
 .../internal/eino/orchestrator.go             |  2 +-
 sundynix-dispatcher/internal/llm/pool.go      | 67 ++++++++++++++++---
 sundynix-dispatcher/internal/llm/pool_test.go | 45 +++++++++++++
 8 files changed, 118 insertions(+), 16 deletions(-)
 create mode 100644 sundynix-dispatcher/internal/llm/pool_test.go

diff --git a/project_analysis.md b/project_analysis.md
index 8977289..ca68218 100644
--- a/project_analysis.md
+++ b/project_analysis.md
@@ -204,7 +204,7 @@ Harness = 围绕 LLM 的可靠性 / 安全 / 质量治理层。4 个组件均为
 | 优先级 | 项目 | 说明 |
 |:------:|------|------|
 | P0 | 真实负载 + 长稳压测，给容量曲线 | "生产级并发"需数据背书 |
-| P0 | 本地模型(vLLM/Ollama) + 推理模型 reasoning_content 适配 | 对齐生产 Qwen |
+| ~~P0~~ ✅ | ~~本地模型(vLLM/Ollama) + reasoning_content 适配~~：Pool provider 感知（Ollama/vLLM 自动补 /v1 + 占位 key），统一走 OpenAI 兼容路径；ChatStream 加 onReasoning，思考过程 surface 到 exec 轨迹（不污染答案）。控制台加 ollama 选项。live：Ollama qwen2.5:0.5b 端到端出答案；deepseek-v4-pro「推理过程」入轨迹 | 对齐生产 Qwen |
 | P1 | 高可用：网关/调度多副本 + NATS 集群 + 自愈 | 解单点 |
 | P1 | 备份/灾备演练（PG/Milvus/Neo4j） | 数据安全 |
 | ~~P1~~ ✅ | ~~优雅停机 drain~~：三 Go 服务全覆盖（gateway HTTP Shutdown / dispatcher 在途任务跑完 / mcp-go 在途工具回完），SIGTERM 后等在途至 SHUTDOWN_DRAIN_TIMEOUT(默认30s)再退；在途任务 ctx 脱离信号 ctx 不被掐断。剩 exec 轨迹 Redis 回放 | 可靠性 |
diff --git a/sundynix-admin/src/components/ModelManager.tsx b/sundynix-admin/src/components/ModelManager.tsx
index 268019e..dfd9b6c 100644
--- a/sundynix-admin/src/components/ModelManager.tsx
+++ b/sundynix-admin/src/components/ModelManager.tsx
@@ -132,8 +132,9 @@ export function ModelManager({
               value={form.provider}
               onChange={(e) => set("provider", e.target.value)}
             >
-              <option value="openai-compatible">openai-compatible</option>
+              <option value="openai-compatible">openai-compatible（在线 API）</option>
               <option value="vllm">vllm（自部署）</option>
+              <option value="ollama">ollama（本地）</option>
             </select>
           </label>
           <label className="text-xs text-gray-500">
diff --git a/sundynix-dispatcher/internal/eino/graph.go b/sundynix-dispatcher/internal/eino/graph.go
index 1997431..585c546 100644
--- a/sundynix-dispatcher/internal/eino/graph.go
+++ b/sundynix-dispatcher/internal/eino/graph.go
@@ -301,12 +301,19 @@ func (o *Orchestrator) runAgent(ctx context.Context, taskID string, b *board, sy
 		n++
 	}
 	send := func(s string) { emit(red.Push(s)) }
+	// reasoning 模型（本地 Qwen3 思考 / DeepSeek-R1 / QwQ 等）的「思考过程」：不进答案、
+	// 累计后 surface 到观测轨迹（让用户看到模型在想什么，又不污染最终回答）。
+	var reasoning strings.Builder
+	onReasoning := func(s string) { reasoning.WriteString(s) }
 	var err error
 	if o.pool.Ready() {
-		err = o.pool.ChatStream(ctx, toChatMessages(msgs), send)
+		err = o.pool.ChatStream(ctx, toChatMessages(msgs), send, onReasoning)
 	} else {
 		err = o.pool.StreamText(ctx, replyFor(msgs), func(tok []byte) { send(string(tok)) })
 	}
+	if rc := reasoning.String(); rc != "" {
+		tr.info(node, "model", "推理过程", fmt.Sprintf("思考 %d 字：%s", len([]rune(rc)), truncate(rc, 200)))
+	}
 	if err != nil {
 		tr.emit(node, "model", "error", "模型流式推理", err.Error(), time.Since(t0).Milliseconds())
 		// 未产出任何 token 即失败 → 标记致命错，让任务判 failed（暴露原因，便于监控告警），
diff --git a/sundynix-dispatcher/internal/eino/integration_test.go b/sundynix-dispatcher/internal/eino/integration_test.go
index d1fd975..47ace69 100644
--- a/sundynix-dispatcher/internal/eino/integration_test.go
+++ b/sundynix-dispatcher/internal/eino/integration_test.go
@@ -24,7 +24,7 @@ type fakeLLM struct {
 }
 
 func (f *fakeLLM) Ready() bool { return f.ready }
-func (f *fakeLLM) ChatStream(_ context.Context, msgs []llm.ChatMessage, onToken func(string)) error {
+func (f *fakeLLM) ChatStream(_ context.Context, msgs []llm.ChatMessage, onToken func(string), _ func(string)) error {
 	if f.stream != nil {
 		onToken(f.stream(msgs))
 	}
diff --git a/sundynix-dispatcher/internal/eino/model.go b/sundynix-dispatcher/internal/eino/model.go
index b24759d..5513867 100644
--- a/sundynix-dispatcher/internal/eino/model.go
+++ b/sundynix-dispatcher/internal/eino/model.go
@@ -24,7 +24,7 @@ func (pm *poolModel) Generate(ctx context.Context, input []*schema.Message, _ ..
 	var sb strings.Builder
 	var err error
 	if pm.pool.Ready() {
-		err = pm.pool.ChatStream(ctx, toChatMessages(input), func(tok string) { sb.WriteString(tok) })
+		err = pm.pool.ChatStream(ctx, toChatMessages(input), func(tok string) { sb.WriteString(tok) }, nil)
 	} else {
 		err = pm.pool.StreamText(ctx, replyFor(input), func(tok []byte) { sb.Write(tok) })
 	}
@@ -44,7 +44,7 @@ func (pm *poolModel) Stream(ctx context.Context, input []*schema.Message, _ ...m
 		send := func(s string) { sw.Send(schema.AssistantMessage(s, nil), nil) }
 		var err error
 		if ready {
-			err = pm.pool.ChatStream(ctx, toChatMessages(input), send)
+			err = pm.pool.ChatStream(ctx, toChatMessages(input), send, nil)
 		} else {
 			err = pm.pool.StreamText(ctx, replyFor(input), func(tok []byte) { send(string(tok)) })
 		}
diff --git a/sundynix-dispatcher/internal/eino/orchestrator.go b/sundynix-dispatcher/internal/eino/orchestrator.go
index aa64d14..76015f9 100644
--- a/sundynix-dispatcher/internal/eino/orchestrator.go
+++ b/sundynix-dispatcher/internal/eino/orchestrator.go
@@ -68,7 +68,7 @@ var errRejected = errors.New("approval rejected")
 // LLM 是编排所需的语言模型能力（生产由 *llm.Pool 实现）。抽成接口便于测试注入假模型。
 type LLM interface {
 	Ready() bool
-	ChatStream(ctx context.Context, msgs []llm.ChatMessage, onToken func(string)) error
+	ChatStream(ctx context.Context, msgs []llm.ChatMessage, onToken func(string), onReasoning func(string)) error
 	StreamText(ctx context.Context, text string, onToken func([]byte)) error
 	Chat(ctx context.Context, msgs []llm.ChatMessage) (string, error)
 	// ToolCallingModel 返回支持函数调用的模型（ReAct agent 用）；不支持则返回 nil。
diff --git a/sundynix-dispatcher/internal/llm/pool.go b/sundynix-dispatcher/internal/llm/pool.go
index fa1842e..c659a84 100644
--- a/sundynix-dispatcher/internal/llm/pool.go
+++ b/sundynix-dispatcher/internal/llm/pool.go
@@ -7,6 +7,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"strings"
 	"sync"
 	"time"
 
@@ -42,12 +43,7 @@ func NewPool() *Pool { return &Pool{} }
 func (p *Pool) SetConfig(cfg *contract.ModelConfig) {
 	var cm model.BaseChatModel
 	if cfg != nil && cfg.Ready() {
-		built, err := openai.NewChatModel(context.Background(), &openai.ChatModelConfig{
-			APIKey:  cfg.APIKey,
-			BaseURL: cfg.BaseURL,
-			Model:   cfg.Model,
-			Timeout: requestTimeout,
-		})
+		built, err := buildChatModel(cfg)
 		if err != nil {
 			fmt.Printf("[llm] 构建 ChatModel 失败（降级桩运行）: %v\n", err)
 		} else {
@@ -60,10 +56,58 @@ func (p *Pool) SetConfig(cfg *contract.ModelConfig) {
 	p.mu.Unlock()
 	if cfg != nil {
 		// 不打印 api_key。
-		fmt.Printf("[llm] model config set: provider=%s base=%s model=%s\n", cfg.Provider, cfg.BaseURL, cfg.Model)
+		fmt.Printf("[llm] model config set: provider=%s base=%s model=%s\n", cfg.Provider, normalizeBaseURL(cfg), cfg.Model)
 	}
 }
 
+// buildChatModel 据 provider 归一化连接参数后构建 OpenAI 兼容 ChatModel。
+// vLLM 与 Ollama 都暴露 OpenAI 兼容 API（底层 go-openai 请求 {base}/chat/completions），
+// 故统一走 openai 客户端，仅差在 BaseURL 是否带 /v1 与是否需要占位 key：
+//   - Ollama：端点固定为 {host}/v1，且不校验 api_key → 缺省补 /v1 + 占位 key "ollama"。
+//   - vLLM：openai server 在 /v1，默认不校验 key（除非 --api-key 启动）→ 补 /v1 + 占位 "EMPTY"。
+//   - openai-compatible（DeepSeek/OpenAI 等在线）：BaseURL 原样（DeepSeek 两种都收，OpenAI 默认带 /v1）。
+func buildChatModel(cfg *contract.ModelConfig) (model.BaseChatModel, error) {
+	return openai.NewChatModel(context.Background(), &openai.ChatModelConfig{
+		APIKey:  apiKeyOrPlaceholder(cfg),
+		BaseURL: normalizeBaseURL(cfg),
+		Model:   cfg.Model,
+		Timeout: requestTimeout,
+	})
+}
+
+// 本地后端 provider 标识（与控制台下拉、contract.ModelConfig.Provider 对齐）。
+const (
+	providerOllama = "ollama"
+	providerVLLM   = "vllm"
+)
+
+// normalizeBaseURL 为本地后端补全 /v1 端点（Ollama/vLLM 的 OpenAI 兼容 API 均在 /v1，
+// 漏写会打到 /chat/completions 而 404）；在线 provider 原样返回。
+func normalizeBaseURL(cfg *contract.ModelConfig) string {
+	base := strings.TrimRight(cfg.BaseURL, "/")
+	switch strings.ToLower(cfg.Provider) {
+	case providerOllama, providerVLLM:
+		if base != "" && !strings.HasSuffix(base, "/v1") {
+			base += "/v1"
+		}
+	}
+	return base
+}
+
+// apiKeyOrPlaceholder 为不校验 key 的本地后端补占位符（openai 客户端要求 api_key 非空）。
+func apiKeyOrPlaceholder(cfg *contract.ModelConfig) string {
+	if cfg.APIKey != "" {
+		return cfg.APIKey
+	}
+	switch strings.ToLower(cfg.Provider) {
+	case providerOllama:
+		return "ollama"
+	case providerVLLM:
+		return "EMPTY"
+	}
+	return cfg.APIKey
+}
+
 func (p *Pool) config() *contract.ModelConfig {
 	p.mu.RLock()
 	defer p.mu.RUnlock()
@@ -99,8 +143,10 @@ func (p *Pool) ModelName() string {
 }
 
 // ChatStream 流式推理，逐 token 回调 onToken（经 Eino ChatModel.Stream）。
-// 仅在 Ready() 时可用（调用方据此决定真实推理或降级桩）。
-func (p *Pool) ChatStream(ctx context.Context, msgs []ChatMessage, onToken func(string)) error {
+// onReasoning（可空）接收 reasoning 模型（DeepSeek-R1 / Qwen3 思考 / QwQ 等）的「思考过程」分片：
+// 思考阶段的分片只有 ReasoningContent、Content 为空，本就不会污染答案；onReasoning 让调用方
+// 可把思考流单独 surface 到观测轨迹。仅在 Ready() 时可用（调用方据此决定真实推理或降级桩）。
+func (p *Pool) ChatStream(ctx context.Context, msgs []ChatMessage, onToken func(string), onReasoning func(string)) error {
 	cm := p.model()
 	if cm == nil {
 		return fmt.Errorf("no model configured")
@@ -125,6 +171,9 @@ func (p *Pool) ChatStream(ctx context.Context, msgs []ChatMessage, onToken func(
 		if rerr != nil {
 			return fmt.Errorf("llm stream recv: %w", rerr)
 		}
+		if chunk.ReasoningContent != "" && onReasoning != nil {
+			onReasoning(chunk.ReasoningContent)
+		}
 		if chunk.Content != "" {
 			onToken(chunk.Content)
 		}
diff --git a/sundynix-dispatcher/internal/llm/pool_test.go b/sundynix-dispatcher/internal/llm/pool_test.go
new file mode 100644
index 0000000..6b551f8
--- /dev/null
+++ b/sundynix-dispatcher/internal/llm/pool_test.go
@@ -0,0 +1,45 @@
+package llm
+
+import (
+	"testing"
+
+	"github.com/sundynix/sundynix-shared/contract"
+)
+
+func TestNormalizeBaseURL(t *testing.T) {
+	cases := []struct {
+		provider, in, want string
+	}{
+		{"ollama", "http://localhost:11434", "http://localhost:11434/v1"},    // 补 /v1
+		{"ollama", "http://localhost:11434/", "http://localhost:11434/v1"},   // 去尾斜杠再补
+		{"ollama", "http://localhost:11434/v1", "http://localhost:11434/v1"}, // 已有不重复补
+		{"vllm", "http://gpu-node:8000", "http://gpu-node:8000/v1"},
+		{"vllm", "http://gpu-node:8000/v1", "http://gpu-node:8000/v1"},
+		{"openai-compatible", "https://api.deepseek.com", "https://api.deepseek.com"}, // 在线不动
+		{"", "https://api.deepseek.com", "https://api.deepseek.com"},
+	}
+	for _, c := range cases {
+		got := normalizeBaseURL(&contract.ModelConfig{Provider: c.provider, BaseURL: c.in})
+		if got != c.want {
+			t.Errorf("normalizeBaseURL(%s,%q)=%q want %q", c.provider, c.in, got, c.want)
+		}
+	}
+}
+
+func TestAPIKeyOrPlaceholder(t *testing.T) {
+	// 本地后端缺 key → 占位（openai 客户端要求非空）。
+	if got := apiKeyOrPlaceholder(&contract.ModelConfig{Provider: "ollama"}); got != "ollama" {
+		t.Errorf("ollama 缺 key 应占位 'ollama', got %q", got)
+	}
+	if got := apiKeyOrPlaceholder(&contract.ModelConfig{Provider: "vllm"}); got != "EMPTY" {
+		t.Errorf("vllm 缺 key 应占位 'EMPTY', got %q", got)
+	}
+	// 显式 key 一律尊重（vLLM 带 --api-key 启动的情况）。
+	if got := apiKeyOrPlaceholder(&contract.ModelConfig{Provider: "vllm", APIKey: "sk-real"}); got != "sk-real" {
+		t.Errorf("显式 key 应原样, got %q", got)
+	}
+	// 在线 provider 缺 key 不占位（保持空，由 Ready/调用方处理）。
+	if got := apiKeyOrPlaceholder(&contract.ModelConfig{Provider: "openai-compatible"}); got != "" {
+		t.Errorf("在线 provider 缺 key 不应占位, got %q", got)
+	}
+}