mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
Phase 2 收尾。证明「对外协议由适配器决定、内核只调度」不是纸面主张:
Anthropic 客户端打 /v1/messages,拿到的是 Anthropic 形状。
- internal/gateway/messages.go:入向把 Anthropic 请求解析进统一格式
(content blocks 原样保留——只有适配器知道自己的上游接受哪种块类型;
system 支持字符串与 [{type:text,text}] 两种写法)。
调度/配额/AUTO 链/冷却/审计/插件 hook 全部复用 /v1/chat/completions
的同一条路径(serveUnified),差别只有两端线格式。
- internal/lua/adapters/anthropic.lua:新增 serialize_response /
serialize_open_preamble / serialize_chunk / serialize_close,
把统一格式译回 Anthropic Messages 形状(含 content 分块、
stop_reason 反向映射、input/output_tokens 改名)。
- 流式是 message_start → content_block_delta → message_delta →
message_stop;DoneMarker 为空所以内核**不会**擅自加 [DONE]。
内核出向零 Anthropic 字段名:解析是表驱动的,序列化属于适配器。
测试:
- messages_test.go 5 个端到端用例(真实 core+gateway + 只懂 Anthropic 的
假上游):形状/stop_reason/块顺序/usage 命名、流式事件序列、入向校验、
鉴权、以及「未声明钩子时回落 OpenAI 兜底」。
- 修正两个 serializer 测试的前提断言(anthropic 现在确实声明了钩子,
这正是 Phase 2 的意图),并发测试改为按适配器各自校验形状。
golden 字节级未漂移;go test -tags luajit ./... 全绿。
404 lines
13 KiB
Go
404 lines
13 KiB
Go
package gateway
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"io"
|
|
"llmsproxy/internal/config"
|
|
"net/http"
|
|
"strings"
|
|
"time"
|
|
)
|
|
|
|
// HTTP surface for model discovery.
|
|
//
|
|
// Two endpoints, deliberately separate:
|
|
//
|
|
// POST /api/sources/discover — probe a URL+key, return the catalog.
|
|
// Nothing is written.
|
|
// POST /api/sources/discover/apply — probe, then write the source.
|
|
//
|
|
// The split exists because discovery mutates nothing and is therefore safe to
|
|
// call repeatedly while the operator is still typing a key, whereas apply
|
|
// rewrites credentials and the AUTO chain's inputs. Making them one endpoint
|
|
// with a flag would mean a "just check it" click can silently create a source.
|
|
//
|
|
// Both are admin-only, matching /api/sources: the request body carries an API
|
|
// key, and a model catalog reveals which upstreams an operator uses.
|
|
|
|
// discoverRequest is the body of both endpoints.
|
|
type discoverRequest struct {
|
|
// BaseURL is what the operator pasted. It may or may not include a
|
|
// version segment; discovery works that out rather than making the
|
|
// caller know this gateway's path-composition rules.
|
|
BaseURL string `json:"base_url"`
|
|
APIKey string `json:"api_key"`
|
|
|
|
// Name is required for apply, ignored for discover. It defaults from the
|
|
// base URL's host when omitted, because "https://api.vendor.com/v1" is a
|
|
// better starting name than "new-source".
|
|
Name string `json:"name"`
|
|
|
|
// Adapter is optional. Discovery assumes "openai" because that is what a
|
|
// /v1/models catalog implies; a non-OpenAI upstream (Anthropic, Gemini)
|
|
// has no such endpoint, so discovery will legitimately fail for those and
|
|
// the operator configures them by hand.
|
|
Adapter string `json:"adapter"`
|
|
|
|
// Temperature and MaxConcurrent are carried into the written source.
|
|
Temperature float64 `json:"temperature"`
|
|
MaxConcurrent int `json:"max_concurrent"`
|
|
|
|
// ReplaceModels drops the source's existing model list before applying the
|
|
// discovered one. It defaults to true for a NEW source and is required to
|
|
// be spelled out for an existing one, so a re-discovery cannot silently
|
|
// delete hand-tuned priorities.
|
|
ReplaceModels *bool `json:"replace_models,omitempty"`
|
|
|
|
// Select narrows what gets written. Empty means everything discovered.
|
|
// Ids that are not in the catalog are an error rather than a silent skip:
|
|
// a typo here would produce a source that 404s on every request.
|
|
Include []string `json:"include,omitempty"`
|
|
Exclude []string `json:"exclude,omitempty"`
|
|
}
|
|
|
|
// discoverResponse is the JSON the WebUI binds to.
|
|
type discoverResponse struct {
|
|
OK bool `json:"ok"`
|
|
|
|
// Normalized source fields, ready to be echoed back to /api/sources.
|
|
BaseURL string `json:"base_url"`
|
|
Adapter string `json:"adapter"`
|
|
Models []string `json:"models"`
|
|
ImageModels []string `json:"image_models"`
|
|
|
|
// Endpoint is set only when discovery had to move a version segment.
|
|
Endpoint string `json:"endpoint"`
|
|
// ImageEndpoint is required whenever base_url carries a version segment:
|
|
// the gateway's default image path is the ABSOLUTE "/v1/images/generations"
|
|
// while chat's default is the relative "/chat/completions", so the same
|
|
// base_url that yields a working chat URL yields a 404 image URL.
|
|
ImageEndpoint string `json:"image_endpoint"`
|
|
|
|
// Diagnostics for the operator.
|
|
ModelsPath string `json:"models_path"`
|
|
Tried []string `json:"tried"`
|
|
LatencyMS int64 `json:"latency_ms"`
|
|
Note string `json:"note,omitempty"`
|
|
Warnings []string `json:"warnings,omitempty"`
|
|
|
|
// Set by apply only.
|
|
SourceName string `json:"source_name,omitempty"`
|
|
Applied bool `json:"applied,omitempty"`
|
|
}
|
|
|
|
func (g *Gateway) handleSourceDiscoverAPI(w http.ResponseWriter, r *http.Request) {
|
|
if reqRole(r.Context()) != "admin" {
|
|
writeError(w, http.StatusForbidden, "forbidden", "admin role required")
|
|
return
|
|
}
|
|
apply := strings.HasSuffix(r.URL.Path, "/apply")
|
|
if r.Method != http.MethodPost {
|
|
w.Header().Set("Allow", "POST")
|
|
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed",
|
|
"use POST with {\"base_url\":…,\"api_key\":…}")
|
|
return
|
|
}
|
|
|
|
body, err := io.ReadAll(io.LimitReader(r.Body, 1<<20))
|
|
if err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request", "cannot read body")
|
|
return
|
|
}
|
|
var req discoverRequest
|
|
if err := json.Unmarshal(body, &req); err != nil {
|
|
writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error())
|
|
return
|
|
}
|
|
req.BaseURL = strings.TrimSpace(req.BaseURL)
|
|
if req.BaseURL == "" {
|
|
writeError(w, http.StatusBadRequest, "invalid_request", "base_url required")
|
|
return
|
|
}
|
|
if req.Adapter == "" {
|
|
req.Adapter = "openai"
|
|
}
|
|
|
|
// Bounded so a hung upstream cannot pin the handler; discovery has its own
|
|
// per-request timeout on top.
|
|
ctx, cancel := context.WithTimeout(r.Context(), 60*time.Second)
|
|
defer cancel()
|
|
|
|
res, err := DiscoverModels(ctx, req.BaseURL, req.APIKey)
|
|
if err != nil {
|
|
// A rejected key must not be reported as an upstream outage: the two
|
|
// need opposite actions from the operator, and 502 sends them to the
|
|
// wrong one.
|
|
status := http.StatusBadGateway
|
|
var f *discoveryFailure
|
|
if errors.As(err, &f) {
|
|
status = f.StatusOrDefault()
|
|
}
|
|
writeError(w, status, "discover_failed", err.Error())
|
|
return
|
|
}
|
|
|
|
chat, image, warns, err := selectModels(res.Models, res.ImageModels, req.Include, req.Exclude)
|
|
if err != nil {
|
|
// 400: the operator asked for a model the catalog does not have.
|
|
writeError(w, http.StatusBadRequest, "invalid_request", err.Error())
|
|
return
|
|
}
|
|
|
|
out := &discoverResponse{
|
|
OK: true,
|
|
BaseURL: res.BaseURL,
|
|
Adapter: req.Adapter,
|
|
Models: chat,
|
|
ImageModels: image,
|
|
Endpoint: res.Endpoint,
|
|
ImageEndpoint: res.ImageEndpoint,
|
|
ModelsPath: res.ModelsPath,
|
|
Tried: res.Tried,
|
|
LatencyMS: res.LatencyMS,
|
|
Note: res.Note,
|
|
Warnings: warns,
|
|
}
|
|
if len(chat) == 0 && len(image) == 0 {
|
|
out.Warnings = append(out.Warnings,
|
|
"every discovered model was excluded; applying this would create a source with no models")
|
|
}
|
|
|
|
if !apply {
|
|
writeJSON(w, http.StatusOK, out)
|
|
return
|
|
}
|
|
|
|
name := strings.TrimSpace(req.Name)
|
|
if name == "" {
|
|
name = defaultSourceName(req.BaseURL)
|
|
}
|
|
if name == "" {
|
|
writeError(w, http.StatusBadRequest, "invalid_request",
|
|
"name required: cannot derive one from base_url")
|
|
return
|
|
}
|
|
if strings.ContainsAny(name, "/\\ \t") || name == "." || name == ".." {
|
|
writeError(w, http.StatusBadRequest, "invalid_request",
|
|
"source name must not contain slashes or whitespace")
|
|
return
|
|
}
|
|
|
|
exists := g.sourceExists(name)
|
|
if exists && req.ReplaceModels == nil {
|
|
// Refusing here is deliberate. Re-running discovery on an existing
|
|
// source and having it silently replace a hand-tuned model list —
|
|
// including priorities chosen for the AUTO chain — is the kind of
|
|
// "helpful" behaviour that loses an afternoon.
|
|
writeError(w, http.StatusConflict, "conflict",
|
|
"source "+name+" already exists; set replace_models=true to overwrite its model list, "+
|
|
"or choose a different name")
|
|
return
|
|
}
|
|
|
|
// Build the source directly rather than round-tripping through the
|
|
// /api/sources payload: resolveAPIKey carries the "an empty api_key means
|
|
// keep the stored one" rule, and going through a second decoder here would
|
|
// duplicate (and eventually diverge from) that logic.
|
|
src := config.Source{
|
|
Name: name,
|
|
BaseURL: res.BaseURL,
|
|
APIKey: g.resolveAPIKey(name, req.APIKey, exists),
|
|
Adapter: req.Adapter,
|
|
Endpoint: res.Endpoint,
|
|
ImageEndpoint: res.ImageEndpoint,
|
|
Models: modelsToConfig(chat, image),
|
|
Temperature: req.Temperature,
|
|
}
|
|
if req.MaxConcurrent > 0 {
|
|
src.MaxConcurrent = req.MaxConcurrent
|
|
}
|
|
|
|
// An existing source may carry fields discovery does not know about
|
|
// (proxy_url, headers, api_key_env, timeout). Rebuilding from scratch would
|
|
// drop them, so they are inherited unless the request overrides them.
|
|
if exists {
|
|
old := g.sourceByName(name)
|
|
if old != nil {
|
|
src.ProxyURL = old.ProxyURL
|
|
src.APIKeyEnv = old.APIKeyEnv
|
|
src.Timeout = old.Timeout
|
|
src.QueueTimeout = old.QueueTimeout
|
|
src.Headers = old.Headers
|
|
src.Meta = old.Meta
|
|
src.MaxTokens = old.MaxTokens
|
|
src.RPM = old.RPM
|
|
// image_endpoint is NOT inherited when discovery supplied one:
|
|
// the stored value was correct for the OLD base_url, and the base
|
|
// is exactly what may have changed. Inheriting it would keep a
|
|
// stale value that 404s against the new base.
|
|
if res.ImageEndpoint == "" {
|
|
src.ImageEndpoint = old.ImageEndpoint
|
|
}
|
|
if src.Adapter == "" {
|
|
src.Adapter = old.Adapter
|
|
}
|
|
if req.MaxConcurrent <= 0 {
|
|
src.MaxConcurrent = old.MaxConcurrent
|
|
}
|
|
// replace_models=false keeps the curated list. Discovery still
|
|
// refreshed base_url/endpoint/credential, which is the point of
|
|
// re-running it.
|
|
if req.ReplaceModels != nil && !*req.ReplaceModels {
|
|
src.Models = old.Models
|
|
out.Warnings = append(out.Warnings,
|
|
"model list left unchanged (replace_models=false)")
|
|
}
|
|
}
|
|
}
|
|
|
|
if err := g.core.AddSource(src); err != nil {
|
|
writeError(w, http.StatusBadRequest, "source_error", err.Error())
|
|
return
|
|
}
|
|
|
|
out.SourceName = name
|
|
out.Applied = true
|
|
writeJSON(w, http.StatusOK, out)
|
|
}
|
|
|
|
// selectModels applies the include/exclude filters.
|
|
//
|
|
// An include entry the catalog does not have is an error: silently dropping it
|
|
// produces a source that looks configured and 404s at request time, which is
|
|
// far harder to diagnose than a rejected apply. An exclude entry that is not
|
|
// present is fine — the operator is allowed to be defensive.
|
|
func selectModels(chat, image, include, exclude []string) ([]string, []string, []string, error) {
|
|
have := map[string]bool{}
|
|
for _, m := range chat {
|
|
have[m] = true
|
|
}
|
|
for _, m := range image {
|
|
have[m] = true
|
|
}
|
|
|
|
if len(include) > 0 {
|
|
var missing []string
|
|
for _, want := range include {
|
|
if !have[want] {
|
|
missing = append(missing, want)
|
|
}
|
|
}
|
|
if len(missing) > 0 {
|
|
return nil, nil, nil, &discoverError{
|
|
msg: "these models are not in the upstream catalog: " + strings.Join(missing, ", "),
|
|
}
|
|
}
|
|
keep := map[string]bool{}
|
|
for _, want := range include {
|
|
keep[want] = true
|
|
}
|
|
var c, i []string
|
|
for _, m := range chat {
|
|
if keep[m] {
|
|
c = append(c, m)
|
|
}
|
|
}
|
|
for _, m := range image {
|
|
if keep[m] {
|
|
i = append(i, m)
|
|
}
|
|
}
|
|
chat, image = c, i
|
|
}
|
|
|
|
if len(exclude) > 0 {
|
|
drop := map[string]bool{}
|
|
for _, m := range exclude {
|
|
drop[m] = true
|
|
}
|
|
var c, i []string
|
|
for _, m := range chat {
|
|
if !drop[m] {
|
|
c = append(c, m)
|
|
}
|
|
}
|
|
for _, m := range image {
|
|
if !drop[m] {
|
|
i = append(i, m)
|
|
}
|
|
}
|
|
chat, image = c, i
|
|
}
|
|
|
|
// chat and image must stay non-nil: the JSON contract forbids null here.
|
|
if chat == nil {
|
|
chat = []string{}
|
|
}
|
|
if image == nil {
|
|
image = []string{}
|
|
}
|
|
|
|
var warns []string
|
|
if len(chat) == 0 && len(image) == 0 {
|
|
// Caller can only get here by excluding everything, or from a catalog
|
|
// that was already empty. Either way "silence" would read as success.
|
|
warns = append(warns,
|
|
"no models selected; applying this would create a source that cannot serve any request")
|
|
return chat, image, warns, nil
|
|
}
|
|
if len(image) > 0 && len(chat) == 0 {
|
|
warns = append(warns,
|
|
"every model is an image model; AUTO chat requests have nothing to select")
|
|
}
|
|
if len(chat) > 0 && len(image) > 0 {
|
|
warns = append(warns,
|
|
"image models are only reachable through /v1/images/generations, not AUTO chat")
|
|
}
|
|
return chat, image, warns, nil
|
|
}
|
|
|
|
// discoverError is a client mistake rather than an upstream failure, which is
|
|
// why it gets its own type instead of fmt.Errorf.
|
|
type discoverError struct{ msg string }
|
|
|
|
func (e *discoverError) Error() string { return e.msg }
|
|
|
|
// defaultSourceName derives a source name from a base URL host, stripping the
|
|
// version segment and any "api." prefix. It returns "" for input that is not a
|
|
// URL at all — the caller must not fall back to echoing garbage into a source
|
|
// name, since the name becomes a config key and an API path segment.
|
|
func defaultSourceName(baseURL string) string {
|
|
s := strings.TrimSpace(baseURL)
|
|
if !strings.Contains(s, "://") {
|
|
return ""
|
|
}
|
|
s = s[strings.Index(s, "://")+3:]
|
|
if i := strings.IndexAny(s, "/?#"); i >= 0 {
|
|
s = s[:i]
|
|
}
|
|
if i := strings.LastIndex(s, ":"); i > 0 && !strings.Contains(s[i+1:], "]") {
|
|
s = s[:i] // strip port
|
|
}
|
|
// A host must still look like one after trimming the cosmetic prefixes.
|
|
s = strings.TrimPrefix(s, "api.")
|
|
s = strings.TrimPrefix(s, "www.")
|
|
if s == "" || strings.ContainsAny(s, " \t\"'<>") {
|
|
return ""
|
|
}
|
|
for _, r := range s {
|
|
switch {
|
|
case r >= 'a' && r <= 'z', r >= 'A' && r <= 'Z', r >= '0' && r <= '9':
|
|
case r == '.' || r == '-' || r == '_':
|
|
default:
|
|
return ""
|
|
}
|
|
}
|
|
if !strings.Contains(s, ".") {
|
|
// A bare label is far more likely to be a mistake than a hostname.
|
|
return ""
|
|
}
|
|
return s
|
|
}
|