第一季-源码学习篇-第 2 篇-模型管理与异构接入-《从零吃透企业级 AI 平台:元景万悟学习手记》

一个请求的旅程：万悟如何管理 10 种不同的 AI 模型

📌 本文是《从零吃透企业级 AI 平台：元景万悟源码学习手记》系列第一季·源码学习篇的第 2 篇🎯 读完本文你将：① 理解适配器模式在模型管理中的应用 ② 能接入 3 种以上不同类型的模型 ③ 能自己写一个模型适配器⏱️ 预计阅读时间：30 分钟 | 动手实践：50 分钟💻 前置要求：已完成第 1 篇的部署，万悟平台可正常访问

一、这篇文章要解决什么问题？

想象你是一个公司的 IT 管理员。老板说：

"研发部要用 GPT-4o，市场部要用 DeepSeek，运维部要用本地 Ollama（数据不能出内网），设计部要用元景大模型。你给统一管起来。"

这四个模型的 API 格式完全不同：

OpenAI 用 /v1/chat/completions
Ollama 用 /api/chat
DeepSeek 兼容 OpenAI 但有细微差异
元景有自己的私有协议

如果每接一个模型就写一套全新代码，你的系统会变成一坨"意大利面条"。

万悟的解法是：适配器模式（Adapter Pattern）。 不管底层是什么模型，上层统一走一个接口。就像万能充电头——不管你是 Type-C、Lightning 还是 Micro-USB，插上就能充。

二、核心概念：用大白话讲清楚

2.1 什么是 OpenAI API 兼容协议？

2023 年之后，几乎所有大模型厂商都"抄"了 OpenAI 的 API 格式。这成了事实标准：

// 请求（所有模型都长这样）POST/v1/chat/completions{"model": "qwen2.5:7b","messages": [    {"role": "system", "content": "你是一个助手"},    {"role": "user", "content": "你好"}  ],"stream": true}// 响应（流式，一行一行返回）data: {"choices":[{"delta":{"content":"你"}}]}data: {"choices":[{"delta":{"content":"好"}}]}data: {"choices":[{"delta":{"content":"！"}}]}data: [DONE]

💡 类比：这就像 USB 接口标准。不管你是键盘、鼠标还是U盘，只要符合 USB 规范，插上电脑就能用。OpenAI API 格式就是 AI 模型的"USB 标准"。

2.2 什么是适配器模式？

┌─────────────────────────────────────────────┐│              万悟统一调用接口                          ││         Complete(messages) → response               ││         StreamComplete(messages) → stream           │└────────────────────┬────────────────────────┘                                                  │                   ┌────────────┼────────────────┐                  ▼                            ▼                                     ▼┌──────────┐         ┌──────────┐   ┌──────────────┐│ OpenAI   │                │ Ollama   │            │ 元景/自定义   ││ Adapter  │                │ Adapter  │            │ Adapter      ││                │                │                │            │              ││ 直接转发  │                │ 格式转换  │           │ 协议翻译     ││ (几乎    │                    │ /api/chat│            │ 私有协议 →   ││  透传)   │                    │ → /v1/..│              │ OpenAI 格式  │└──────────┘         └──────────┘   └──────────────┘

核心思想：定义一个 Go interface，所有模型都实现这个 interface。上层代码只认 interface，不关心具体是哪个模型。

2.3 流式输出（SSE）是什么？

你在大模型对话时看到文字"一个一个蹦出来"，这就是 SSE（Server-Sent Events）：

HTTP/1.1 200 OKContent-Type: text/event-stream    ← 关键：不是 application/jsonCache-Control: no-cacheConnection: keep-alivedata: {"choices":[{"delta":{"content":"你"}}]}data: {"choices":[{"delta":{"content":"好"}}]}data: [DONE]

💡 类比：普通 HTTP 像寄快递——打包好一次性送到。SSE 像水龙头——打开后水（数据）持续流出来，直到你关掉（[DONE]）。

三、万悟是怎么实现的？（源码篇）

3.1 定位源码

模型管理相关代码主要在：

internal/├── assistant-service/│   ├── model/              # 模型适配器核心代码│   │   ├── adapter.go      # interface 定义（最重要！）│   │   ├── openai.go       # OpenAI 兼容适配器│   │   ├── ollama.go       # Ollama 适配器│   │   └── yuanjing.go     # 元景适配器│   └── handler/│       └── chat.go         # 对话入口，调用适配器└── operate-service/    └── model/              # 模型 CRUD（增删改查配置）

3.2 核心 Interface 定义

// 文件：internal/assistant-service/model/adapter.go// ModelAdapter 是所有模型适配器的统一接口// 不管底层是 OpenAI、Ollama 还是元景，都必须实现这两个方法typeModelAdapterinterface {// Complete 非流式调用：发请求，等完整回答返回// 参数：上下文、消息列表、模型参数（temperature等）// 返回：完整的回答文本Complete(ctxcontext.Context, messages []Message, opts...Option) (*Response, error)// StreamComplete 流式调用：发请求，通过 channel 逐 token 返回// 参数：同上 + 一个 channel 用于推送增量内容// 返回：error channel（用于传递中途错误）StreamComplete(ctxcontext.Context, messages []Message, chchan<-StreamChunk, opts...Option) error}// Message 对话消息（和 OpenAI 格式一致）typeMessagestruct {Rolestring`json:"role"`// "system" / "user" / "assistant"Contentstring`json:"content"`// 消息内容}// StreamChunk 流式返回的一个片段typeStreamChunkstruct {Deltastring// 本次新增的文字（如 "你"）Donebool// 是否是最后一个片段Errerror// 如果有错误，放在这里}

📌 关键洞察：这个 interface 只有 2 个方法。上层代码（Agent 推理、工作流节点）只调用 adapter.Complete() 或 adapter.StreamComplete()，完全不知道底层是什么模型。这就是"面向接口编程"的威力。

3.3 OpenAI 适配器实现

// 文件：internal/assistant-service/model/openai.go// OpenAIAdapter 实现了 ModelAdapter 接口// 适用于所有 OpenAI API 兼容的模型（GPT-4o、DeepSeek、硅基流动等）typeOpenAIAdapterstruct {baseURLstring// 如 "https://api.openai.com" 或 "https://api.deepseek.com"apiKeystring// API 密钥client*http.Client}// NewOpenAIAdapter 构造函数funcNewOpenAIAdapter(baseURL, apiKeystring) *OpenAIAdapter {return&OpenAIAdapter{baseURL: baseURL,apiKey:  apiKey,client:  &http.Client{Timeout: 120*time.Second},    }}// Complete 非流式调用func (a*OpenAIAdapter) Complete(ctxcontext.Context, messages []Message, opts...Option) (*Response, error) {// 1. 构造请求体（和 OpenAI 官方格式完全一致）reqBody :=map[string]interface{}{"model":    getModelFromOpts(opts),"messages": messages,"stream":   false, // 非流式    }// 2. 发送 HTTP POST 到 /v1/chat/completionsresp, err :=a.doRequest(ctx, "/v1/chat/completions", reqBody)iferr!=nil {returnnil, fmt.Errorf("openai request failed: %w", err)    }// 3. 解析响应，提取 choices[0].message.contentreturnparseOpenAIResponse(resp)}// StreamComplete 流式调用func (a*OpenAIAdapter) StreamComplete(ctxcontext.Context, messages []Message, chchan<-StreamChunk, opts...Option) error {// 1. 构造请求体，stream 设为 truereqBody :=map[string]interface{}{"model":    getModelFromOpts(opts),"messages": messages,"stream":   true, // 👈 关键：开启流式    }// 2. 发送请求，拿到响应流resp, err :=a.doRequestStream(ctx, "/v1/chat/completions", reqBody)iferr!=nil {returnerr    }// 3. 逐行读取 SSE 数据scanner :=bufio.NewScanner(resp.Body)forscanner.Scan() {line :=scanner.Text()// SSE 格式：每行以 "data: " 开头if!strings.HasPrefix(line, "data: ") {continue        }data :=strings.TrimPrefix(line, "data: ")// 结束标志ifdata=="[DONE]" {ch<-StreamChunk{Done: true}returnnil        }// 解析 JSON，提取 delta.contentvarchunkOpenAIStreamResponsejson.Unmarshal([]byte(data), &chunk)iflen(chunk.Choices) >0 {ch<-StreamChunk{Delta: chunk.Choices[0].Delta.Content}        }    }returnscanner.Err()}

3.4 Ollama 适配器（格式转换的核心）

// 文件：internal/assistant-service/model/ollama.go// OllamaAdapter 适用于本地 Ollama 服务// Ollama 原生 API 格式和 OpenAI 不同，需要做转换typeOllamaAdapterstruct {baseURLstring// 如 "http://localhost:11434"client*http.Client}func (a*OllamaAdapter) Complete(ctxcontext.Context, messages []Message, opts...Option) (*Response, error) {// Ollama 原生格式：POST /api/chat// Body: {"model":"qwen2.5:7b", "messages":[...], "stream":false}//// 注意：Ollama 也支持 /v1/chat/completions（OpenAI 兼容模式）// 万悟这里直接走兼容模式，减少转换逻辑reqBody :=map[string]interface{}{"model":    getModelFromOpts(opts),"messages": messages,"stream":   false,    }// 走 Ollama 的 OpenAI 兼容端点resp, err :=a.doRequest(ctx, "/v1/chat/completions", reqBody)iferr!=nil {returnnil, err    }// 响应格式和 OpenAI 一致，直接复用解析逻辑returnparseOpenAIResponse(resp)}

💡 设计亮点：Ollama 从 v0.1.24 开始支持 /v1/chat/completions 兼容端点。万悟直接利用这一点，让 Ollama 适配器几乎和 OpenAI 适配器一样简单。如果 Ollama 不支持兼容模式，你就需要在这里做格式转换（把 OpenAI 格式翻译成 Ollama 原生格式）。

3.5 适配器工厂：根据配置选择适配器

// 文件：internal/assistant-service/model/factory.go// CreateAdapter 根据模型配置创建对应的适配器// 这是"工厂模式"：根据 type 字段决定实例化哪个 adapterfuncCreateAdapter(cfg*ModelConfig) (ModelAdapter, error) {switchcfg.Type {case"openai":// OpenAI / DeepSeek / 硅基流动 / 任何 OpenAI 兼容服务returnNewOpenAIAdapter(cfg.BaseURL, cfg.APIKey), nilcase"ollama":// 本地 OllamareturnNewOllamaAdapter(cfg.BaseURL), nilcase"yuanjing":// 联通元景大模型（私有协议）returnNewYuanjingAdapter(cfg.BaseURL, cfg.APIKey), nildefault:returnnil, fmt.Errorf("unsupported model type: %s", cfg.Type)    }}

3.6 完整调用链

用户发消息 "你好"  → assistant-service/handler/chat.go    → 从数据库加载 Agent 配置（关联的模型 ID）    → 从数据库加载模型配置（type="ollama", baseURL="http://..."）    → factory.CreateAdapter(cfg) → 返回 OllamaAdapter 实例    → adapter.StreamComplete(messages, ch)      → HTTP POST http://host.docker.internal:11434/v1/chat/completions      → 逐行读取 SSE → 写入 channel    → handler 从 channel 读取 → 写入 HTTP Response（SSE 格式）  → 用户浏览器逐字显示 "你好！有什么可以帮你的？"

四、动手跑通（实践篇）

4.1 接入 3 种模型

模型 A：Ollama 本地模型

# 确保 Ollama 已运行ollama list# 预期看到 qwen2.5:7b# 在万悟平台：模型管理 → 添加# 类型：Ollama# 地址：http://host.docker.internal:11434# 模型：qwen2.5:7b

模型 B：DeepSeek 在线模型（OpenAI 兼容）

# 在万悟平台：模型管理 → 添加# 类型：OpenAI API Compatible# 地址：https://api.deepseek.com# API Key：sk-xxxxx（从 DeepSeek 官网获取）# 模型：deepseek-chat

模型 C：硅基流动（OpenAI 兼容）

# 类型：OpenAI API Compatible# 地址：https://api.siliconflow.cn/v1# API Key：sk-xxxxx# 模型：Qwen/Qwen2.5-7B-Instruct

4.2 用 curl 验证适配器

# 通过万悟的统一 API 调用（不直接调模型）# 这验证了"适配器层"是否正常工作curl-N http://localhost:8081/api/v1/chat/completions \-H"Authorization: Bearer " \-H"Content-Type: application/json" \-d'{"model": "qwen2.5:7b","messages": [{"role": "user", "content": "用一句话解释什么是适配器模式"}],"stream": true  }'

✅ 预期结果：看到逐行返回的 SSE 数据：

data: {"choices":[{"delta":{"content":"适配器"}}]}data: {"choices":[{"delta":{"content":"模式是"}}]}data: {"choices":[{"delta":{"content":"一种结构型"}}]}...data: [DONE]

4.3 对比实验

在万悟平台创建一个智能体，分别切换 3 种模型，问同一个问题：

"请解释量子计算的基本原理，100字以内"

记录：

响应速度（首 token 延迟）
回答质量
流式输出的流畅度

五、自己造一个 Mini 版（Deep Dive）

🎯 目标：用 Go 写一个 80 行的模型适配器框架

// mini_adapter.go// 依赖：go mod init mini && go get github.com/go-resty/resty/v2packagemainimport ("bufio""context""encoding/json""fmt""net/http""strings")// ========== 1. 定义统一接口 ==========typeMessagestruct {Rolestring`json:"role"`Contentstring`json:"content"`}// ModelAdapter 统一接口（和万悟的设计一致）typeModelAdapterinterface {StreamChat(ctxcontext.Context, messages []Message) (<-chanstring, error)}// ========== 2. OpenAI 适配器 ==========typeOpenAIAdapterstruct {BaseURLstringAPIKeystringModelstring}func (a*OpenAIAdapter) StreamChat(ctxcontext.Context, messages []Message) (<-chanstring, error) {ch :=make(chanstring)gofunc() {deferclose(ch)// 构造请求body :=fmt.Sprintf(`{"model":"%s","messages":%s,"stream":true}`,a.Model, mustJSON(messages))req, _ :=http.NewRequestWithContext(ctx, "POST",a.BaseURL+"/v1/chat/completions",strings.NewReader(body))req.Header.Set("Authorization", "Bearer "+a.APIKey)req.Header.Set("Content-Type", "application/json")resp, err :=http.DefaultClient.Do(req)iferr!=nil {return        }deferresp.Body.Close()// 逐行读取 SSEscanner :=bufio.NewScanner(resp.Body)forscanner.Scan() {line :=scanner.Text()if!strings.HasPrefix(line, "data: ") {continue            }data :=strings.TrimPrefix(line, "data: ")ifdata=="[DONE]" {return            }// 解析 delta.contentvarparsedstruct {Choices []struct {Deltastruct {Contentstring`json:"content"`                    } `json:"delta"`                } `json:"choices"`            }json.Unmarshal([]byte(data), &parsed)iflen(parsed.Choices) >0&&parsed.Choices[0].Delta.Content!="" {ch<-parsed.Choices[0].Delta.Content            }        }    }()returnch, nil}// ========== 3. 工厂函数 ==========funcCreateAdapter(modelType, baseURL, apiKey, modelstring) ModelAdapter {switchmodelType {case"openai":return&OpenAIAdapter{BaseURL: baseURL, APIKey: apiKey, Model: model}// case "ollama": return &OllamaAdapter{...}  ← 你可以自己扩展default:panic("unsupported type")    }}// ========== 4. 测试 ==========funcmain() {adapter :=CreateAdapter("openai","https://api.deepseek.com",  // 或 http://localhost:11434"sk-your-key","deepseek-chat",    )ch, _ :=adapter.StreamChat(context.Background(), []Message{        {Role: "user", Content: "你好，一句话介绍自己"},    })fortoken :=rangech {fmt.Print(token) // 逐字打印    }fmt.Println()}funcmustJSON(vinterface{}) string {b, _ :=json.Marshal(v)returnstring(b)}

运行：

go run mini_adapter.go# 预期输出：你好！我是DeepSeek，一个由深度求索公司开发的AI助手。

🎉 跑通了吗？你刚刚实现的就是万悟模型管理层的核心骨架。万悟在此基础上加了：错误重试、超时控制、Token 计数、多模型负载均衡、模型健康检查……但骨架就是你上面看到的这 80 行代码。

六、总结 & 延伸阅读

本文要点回顾

✅ 万悟用适配器模式统一管理异构模型，上层只认 interface
✅ 所有模型统一走 OpenAI API 兼容协议（/v1/chat/completions）
✅ 流式输出基于 SSE（text/event-stream），逐 token 推送
✅ 工厂函数根据配置 type 字段实例化对应适配器
✅ 你自己实现了一个 80 行的 mini 适配器框架

参考资源

元景万悟 GitHub：https://github.com/UnicomAI/wanwu
源码位置：internal/assistant-service/model/

📮 点赞 / 收藏 / 关注，不错过后续更新🔔 下一篇：《手把手搞懂 RAG：从 PDF 上传到精准回答的全链路拆解》