feat(llm): 本地模型接入 vLLM / Ollama + reasoning 思考过程入轨迹
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>
This commit is contained in:
+1
-1
@@ -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 回放 | 可靠性 |
|
||||
|
||||
@@ -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">
|
||||
|
||||
@@ -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(暴露原因,便于监控告警),
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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)) })
|
||||
}
|
||||
|
||||
@@ -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。
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user