feat(dispatcher): Eino 采纳 Phase A —— llm.Pool 内部换成 Eino ChatModel 组件
把手写的 OpenAI HTTP/SSE 客户端换成 eino-ext/openai 的 ChatModel 组件： - SetConfig 用激活配置构建 ChatModel（热更新=重建实例，读写锁保护） - ChatStream → cm.Stream（StreamReader 逐 chunk）；Chat → cm.Generate - 对外签名零变更（Chat/ChatStream/StreamText/Ready/ModelName）， graph.go / report.go / memory_extract.go 不动 - Ready() 语义升级为「ChatModel 构建成功」，更准 - StreamText 降级桩保留 验收：make test-go 全模块绿；真实链路提交任务经新组件流式出答复 （28 字 + eval 1.00，Stream/Generate 两条路径都过）。 为 Phase B（BindTools 函数调用 + ADK）铺好底层组件。 Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Blizzard
@@ -1,41 +1,59 @@
|
||||
// Package llm 抽象 LLM Pool（vLLM / Ollama / 第三方在线 API）的负载均衡与流式推理。
|
||||
// Package llm 抽象 LLM Pool（第三方在线 API / 自部署）的配置与流式推理。
|
||||
// 底层用 Eino 的 ChatModel 组件（eino-ext/openai，OpenAI 兼容）；本层只负责
|
||||
// 配置热更新、降级桩与对外的稳定签名（Chat / ChatStream / StreamText）。
|
||||
package llm
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"io"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/cloudwego/eino-ext/components/model/openai"
|
||||
"github.com/cloudwego/eino/components/model"
|
||||
"github.com/cloudwego/eino/schema"
|
||||
|
||||
"github.com/sundynix/sundynix-shared/contract"
|
||||
)
|
||||
|
||||
// requestTimeout 是单次推理请求的上限。
|
||||
const requestTimeout = 120 * time.Second
|
||||
|
||||
// ChatMessage 是一条对话消息（role: system/user/assistant）。
|
||||
type ChatMessage struct {
|
||||
Role string `json:"role"`
|
||||
Content string `json:"content"`
|
||||
}
|
||||
|
||||
// Pool 维护当前激活的后端配置（由控制面经 NATS 下发，可热更新）。
|
||||
// Pool 维护当前激活的后端配置 + 据此构建的 Eino ChatModel（控制面经 NATS 下发，可热更新）。
|
||||
type Pool struct {
|
||||
mu sync.RWMutex
|
||||
cfg *contract.ModelConfig
|
||||
hc *http.Client
|
||||
cm model.BaseChatModel // 由 SetConfig 用激活配置构建；未配置时为 nil
|
||||
}
|
||||
|
||||
func NewPool() *Pool {
|
||||
return &Pool{hc: &http.Client{Timeout: 120 * time.Second}}
|
||||
}
|
||||
func NewPool() *Pool { return &Pool{} }
|
||||
|
||||
// SetConfig 热更新后端配置（控制面变更时调用）。
|
||||
// SetConfig 热更新后端配置：重建 ChatModel 实例（控制面变更时调用）。
|
||||
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,
|
||||
})
|
||||
if err != nil {
|
||||
fmt.Printf("[llm] 构建 ChatModel 失败（降级桩运行）: %v\n", err)
|
||||
} else {
|
||||
cm = built
|
||||
}
|
||||
}
|
||||
p.mu.Lock()
|
||||
p.cfg = cfg
|
||||
p.cm = cm
|
||||
p.mu.Unlock()
|
||||
if cfg != nil {
|
||||
// 不打印 api_key。
|
||||
@@ -49,8 +67,14 @@ func (p *Pool) config() *contract.ModelConfig {
|
||||
return p.cfg
|
||||
}
|
||||
|
||||
// Ready 报告是否已配置可用后端。
|
||||
func (p *Pool) Ready() bool { return p.config().Ready() }
|
||||
func (p *Pool) model() model.BaseChatModel {
|
||||
p.mu.RLock()
|
||||
defer p.mu.RUnlock()
|
||||
return p.cm
|
||||
}
|
||||
|
||||
// Ready 报告是否已配置可用后端（且 ChatModel 构建成功）。
|
||||
func (p *Pool) Ready() bool { return p.model() != nil }
|
||||
|
||||
// ModelName 返回当前激活的对话模型名（未配置则空）—— 供服务状态面板展示。
|
||||
func (p *Pool) ModelName() string {
|
||||
@@ -60,72 +84,62 @@ func (p *Pool) ModelName() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// ChatStream 以 OpenAI 兼容协议流式推理，逐 token 回调 onToken。
|
||||
// ChatStream 流式推理，逐 token 回调 onToken（经 Eino ChatModel.Stream）。
|
||||
// 仅在 Ready() 时可用（调用方据此决定真实推理或降级桩）。
|
||||
func (p *Pool) ChatStream(ctx context.Context, msgs []ChatMessage, onToken func(string)) error {
|
||||
cfg := p.config()
|
||||
if !cfg.Ready() {
|
||||
cm := p.model()
|
||||
if cm == nil {
|
||||
return fmt.Errorf("no model configured")
|
||||
}
|
||||
body, _ := json.Marshal(map[string]any{
|
||||
"model": cfg.Model,
|
||||
"messages": msgs,
|
||||
"stream": true,
|
||||
})
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodPost, cfg.BaseURL+"/chat/completions", bytes.NewReader(body))
|
||||
sr, err := cm.Stream(ctx, toSchema(msgs))
|
||||
if err != nil {
|
||||
return err
|
||||
return fmt.Errorf("llm stream: %w", err)
|
||||
}
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
if cfg.APIKey != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+cfg.APIKey)
|
||||
}
|
||||
resp, err := p.hc.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("llm request: %w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode >= 400 {
|
||||
buf := new(bytes.Buffer)
|
||||
_, _ = buf.ReadFrom(resp.Body)
|
||||
return fmt.Errorf("llm http %d: %s", resp.StatusCode, strings.TrimSpace(buf.String()))
|
||||
}
|
||||
|
||||
// 解析 OpenAI 兼容 SSE：data: {choices:[{delta:{content}}]} … data: [DONE]
|
||||
sc := bufio.NewScanner(resp.Body)
|
||||
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
|
||||
for sc.Scan() {
|
||||
line := strings.TrimSpace(sc.Text())
|
||||
if !strings.HasPrefix(line, "data:") {
|
||||
continue
|
||||
defer sr.Close()
|
||||
for {
|
||||
chunk, rerr := sr.Recv()
|
||||
if rerr == io.EOF {
|
||||
return nil
|
||||
}
|
||||
payload := strings.TrimSpace(strings.TrimPrefix(line, "data:"))
|
||||
if payload == "[DONE]" {
|
||||
break
|
||||
if rerr != nil {
|
||||
return fmt.Errorf("llm stream recv: %w", rerr)
|
||||
}
|
||||
var chunk struct {
|
||||
Choices []struct {
|
||||
Delta struct {
|
||||
Content string `json:"content"`
|
||||
} `json:"delta"`
|
||||
} `json:"choices"`
|
||||
}
|
||||
if json.Unmarshal([]byte(payload), &chunk) != nil {
|
||||
continue
|
||||
}
|
||||
if len(chunk.Choices) > 0 && chunk.Choices[0].Delta.Content != "" {
|
||||
onToken(chunk.Choices[0].Delta.Content)
|
||||
if chunk.Content != "" {
|
||||
onToken(chunk.Content)
|
||||
}
|
||||
}
|
||||
return sc.Err()
|
||||
}
|
||||
|
||||
// Chat 非流式：内部复用 ChatStream 聚合全部 token，返回整段文本。
|
||||
// Chat 非流式：经 Eino ChatModel.Generate 拿到整段文本。
|
||||
// 报告生成的「规划大纲 / 撰写章节」等需要拿到完整结果再继续，用它而非流式。
|
||||
func (p *Pool) Chat(ctx context.Context, msgs []ChatMessage) (string, error) {
|
||||
var b strings.Builder
|
||||
err := p.ChatStream(ctx, msgs, func(tok string) { b.WriteString(tok) })
|
||||
return b.String(), err
|
||||
cm := p.model()
|
||||
if cm == nil {
|
||||
return "", fmt.Errorf("no model configured")
|
||||
}
|
||||
out, err := cm.Generate(ctx, toSchema(msgs))
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("llm generate: %w", err)
|
||||
}
|
||||
return out.Content, nil
|
||||
}
|
||||
|
||||
// toSchema 把内部 ChatMessage 转为 Eino schema.Message。
|
||||
func toSchema(msgs []ChatMessage) []*schema.Message {
|
||||
out := make([]*schema.Message, 0, len(msgs))
|
||||
for _, m := range msgs {
|
||||
var role schema.RoleType
|
||||
switch m.Role {
|
||||
case "system":
|
||||
role = schema.System
|
||||
case "assistant":
|
||||
role = schema.Assistant
|
||||
default:
|
||||
role = schema.User
|
||||
}
|
||||
out = append(out, &schema.Message{Role: role, Content: m.Content})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// ---- 占位降级（未配置后端时）----
|
||||
|
||||
Reference in New Issue
Block a user