mirror of
https://gitcode.com/JianFeeeee/ModelRouter.git
synced 2026-11-07 09:47:23 +00:00
feat(gateway): /v1/messages 端到端 —— 内核不含任何 Anthropic 出站字段名
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 ./... 全绿。
This commit is contained in:
@ -470,8 +470,8 @@ func (g *Gateway) handleStatsAPI(w http.ResponseWriter, r *http.Request) {
|
||||
// the all-time view showed the truth, which read as "the gateway is
|
||||
// idle" precisely when it was busiest. A calendar window cannot
|
||||
// filter it, so it travels with every response unfiltered.
|
||||
"active": g.stats.Active(),
|
||||
"records": page.Records,
|
||||
"active": g.stats.Active(),
|
||||
"records": page.Records,
|
||||
// The paging cursor has to travel with the first page, otherwise
|
||||
// the scroll handler has nothing to page from and the table is
|
||||
// stuck at one screen forever.
|
||||
|
||||
@ -535,11 +535,11 @@ type modelCaps struct {
|
||||
}
|
||||
|
||||
type upstreamModel struct {
|
||||
ID string `json:"id"`
|
||||
Object string `json:"object"`
|
||||
Kind string `json:"kind"`
|
||||
Type string `json:"type"`
|
||||
Modality string `json:"modality"`
|
||||
ID string `json:"id"`
|
||||
Object string `json:"object"`
|
||||
Kind string `json:"kind"`
|
||||
Type string `json:"type"`
|
||||
Modality string `json:"modality"`
|
||||
Architecture struct {
|
||||
InputModalities []string `json:"input_modalities"`
|
||||
OutputModalities []string `json:"output_modalities"`
|
||||
@ -699,8 +699,8 @@ func modelMeta(m upstreamModel) map[string]interface{} {
|
||||
}
|
||||
}
|
||||
|
||||
copyFields(m.Metadata) // nested first
|
||||
copyFields(mustJSON(m)) // top level wins
|
||||
copyFields(m.Metadata) // nested first
|
||||
copyFields(mustJSON(m)) // top level wins
|
||||
if len(meta) == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
@ -5,10 +5,10 @@ import (
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"io"
|
||||
"llmsproxy/internal/config"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
"llmsproxy/internal/config"
|
||||
)
|
||||
|
||||
// HTTP surface for model discovery.
|
||||
@ -68,9 +68,9 @@ 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"`
|
||||
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.
|
||||
@ -152,18 +152,18 @@ func (g *Gateway) handleSourceDiscoverAPI(w http.ResponseWriter, r *http.Request
|
||||
}
|
||||
|
||||
out := &discoverResponse{
|
||||
OK: true,
|
||||
BaseURL: res.BaseURL,
|
||||
Adapter: req.Adapter,
|
||||
Models: chat,
|
||||
ImageModels: image,
|
||||
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,
|
||||
Tried: res.Tried,
|
||||
LatencyMS: res.LatencyMS,
|
||||
Note: res.Note,
|
||||
Warnings: warns,
|
||||
}
|
||||
if len(chat) == 0 && len(image) == 0 {
|
||||
out.Warnings = append(out.Warnings,
|
||||
@ -207,14 +207,14 @@ func (g *Gateway) handleSourceDiscoverAPI(w http.ResponseWriter, r *http.Request
|
||||
// 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,
|
||||
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,
|
||||
Temperature: req.Temperature,
|
||||
}
|
||||
if req.MaxConcurrent > 0 {
|
||||
src.MaxConcurrent = req.MaxConcurrent
|
||||
|
||||
@ -32,12 +32,12 @@ func catalogUpstream(t *testing.T) *httptest.Server {
|
||||
|
||||
func TestDefaultSourceName(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"https://api.vendor.com/v1": "vendor.com",
|
||||
"https://api.openai.com/v1": "openai.com",
|
||||
"https://www.vendor.com/v1": "vendor.com",
|
||||
"http://127.0.0.1:8080/v1": "127.0.0.1",
|
||||
"https://api.vendor.com/v1": "vendor.com",
|
||||
"https://api.openai.com/v1": "openai.com",
|
||||
"https://www.vendor.com/v1": "vendor.com",
|
||||
"http://127.0.0.1:8080/v1": "127.0.0.1",
|
||||
"https://amr-link.open-design.ai/v1": "amr-link.open-design.ai",
|
||||
"not a url at all": "",
|
||||
"not a url at all": "",
|
||||
}
|
||||
for in, want := range cases {
|
||||
if got := defaultSourceName(in); got != want {
|
||||
@ -122,16 +122,16 @@ func TestSelectModels_WarnsThatImagesAreNotInAUTO(t *testing.T) {
|
||||
// operator previews one set of values and saves another.
|
||||
func TestDiscoveryResultAndConfigRoundTrip(t *testing.T) {
|
||||
res := &discoveryResult{
|
||||
BaseURL: "https://h",
|
||||
Endpoint: "/v1/chat/completions",
|
||||
Models: []string{"m1", "m2"},
|
||||
BaseURL: "https://h",
|
||||
Endpoint: "/v1/chat/completions",
|
||||
Models: []string{"m1", "m2"},
|
||||
ImageModels: []string{"img1"},
|
||||
}
|
||||
src := config.Source{
|
||||
Name: "x",
|
||||
BaseURL: res.BaseURL,
|
||||
Endpoint: res.Endpoint,
|
||||
Models: modelsToConfig(res.Models, res.ImageModels),
|
||||
Name: "x",
|
||||
BaseURL: res.BaseURL,
|
||||
Endpoint: res.Endpoint,
|
||||
Models: modelsToConfig(res.Models, res.ImageModels),
|
||||
}
|
||||
if len(src.Models) != 3 {
|
||||
t.Fatalf("models = %d, want 3", len(src.Models))
|
||||
|
||||
@ -31,10 +31,10 @@ func TestComposeImageEndpoint_NoVersionBaseLeavesDefault(t *testing.T) {
|
||||
// TestDiscovery_BareHostNeedsVersionOnEndpointWhenCatalogIsVersioned
|
||||
// distinguishes two cases that look identical from the outside.
|
||||
//
|
||||
// bare host + catalog at "/models" → chat is "/chat/completions" (the
|
||||
// adapter default, no endpoint written)
|
||||
// bare host + catalog at "/v1/models" → chat is "/v1/chat/completions",
|
||||
// so an endpoint MUST be written
|
||||
// bare host + catalog at "/models" → chat is "/chat/completions" (the
|
||||
// adapter default, no endpoint written)
|
||||
// bare host + catalog at "/v1/models" → chat is "/v1/chat/completions",
|
||||
// so an endpoint MUST be written
|
||||
//
|
||||
// The catalog path is the only evidence of whether the upstream wants the
|
||||
// version segment, and getting this backwards sends every chat request to a
|
||||
@ -105,15 +105,15 @@ func TestDiscovery_BareHostNeedsVersionOnEndpointWhenCatalogIsVersioned(t *testi
|
||||
// silent, expensive mistakes, so they are asserted literally.
|
||||
func TestComposeEndpoints_BothFormsProduceWorkingURLs(t *testing.T) {
|
||||
cases := []struct {
|
||||
name string
|
||||
base string
|
||||
wantChat string
|
||||
wantImage string
|
||||
name string
|
||||
base string
|
||||
wantChat string
|
||||
wantImage string
|
||||
}{
|
||||
{
|
||||
name: "bare host: adapter and image defaults compose",
|
||||
base: "https://h",
|
||||
wantChat: "https://h/chat/completions",
|
||||
name: "bare host: adapter and image defaults compose",
|
||||
base: "https://h",
|
||||
wantChat: "https://h/chat/completions",
|
||||
// openai's adapter endpoint is /chat/completions (verified in
|
||||
// internal/lua/adapters/openai.lua: adapter.endpoint).
|
||||
wantImage: "https://h/v1/images/generations",
|
||||
|
||||
@ -356,10 +356,10 @@ func TestDiscoveryCandidates_DedupesAndPrefersTyped(t *testing.T) {
|
||||
|
||||
func TestVersionSegment(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"/v1/models": "/v1",
|
||||
"/v1/models": "/v1",
|
||||
"/v1beta/models": "/v1beta",
|
||||
"/models": "",
|
||||
"/api/models": "",
|
||||
"/models": "",
|
||||
"/api/models": "",
|
||||
"/vision/models": "",
|
||||
}
|
||||
for in, want := range cases {
|
||||
@ -371,19 +371,19 @@ func TestVersionSegment(t *testing.T) {
|
||||
|
||||
func TestTrimVersionSuffix(t *testing.T) {
|
||||
cases := map[string]string{
|
||||
"https://h/v1": "https://h",
|
||||
"https://h/v2": "https://h",
|
||||
"https://h/v1beta": "https://h",
|
||||
"https://h/api/v1": "https://h/api",
|
||||
"https://h/vision": "https://h/vision", // not numeric — a real path segment
|
||||
"https://h": "https://h",
|
||||
"https://h/openai": "https://h/openai",
|
||||
"https://h/v1/extra": "https://h/v1/extra",
|
||||
"https://h/v1": "https://h",
|
||||
"https://h/v2": "https://h",
|
||||
"https://h/v1beta": "https://h",
|
||||
"https://h/api/v1": "https://h/api",
|
||||
"https://h/vision": "https://h/vision", // not numeric — a real path segment
|
||||
"https://h": "https://h",
|
||||
"https://h/openai": "https://h/openai",
|
||||
"https://h/v1/extra": "https://h/v1/extra",
|
||||
// A segment that merely STARTS with v must not be stripped: stripping
|
||||
// it would rewrite base "https://h/vision" to "https://h" and, worse,
|
||||
// would treat a real vendor path as a version.
|
||||
"https://h/vertex": "https://h/vertex",
|
||||
"https://h/volcengine": "https://h/volcengine",
|
||||
"https://h/vertex": "https://h/vertex",
|
||||
"https://h/volcengine": "https://h/volcengine",
|
||||
"https://h/vertex-ai/v1": "https://h/vertex-ai",
|
||||
}
|
||||
for in, want := range cases {
|
||||
@ -463,7 +463,7 @@ func TestComposeEndpoint_ExactlyOneVersionSegment(t *testing.T) {
|
||||
// Whichever way it resolved, base + endpoint must never carry the
|
||||
// same version segment twice.
|
||||
onBase := trimVersionSuffix(tc.persistedBase) != tc.persistedBase
|
||||
onEndpoint := versionSegment(strings.TrimSuffix(got, "/chat/completions") + "/models") != ""
|
||||
onEndpoint := versionSegment(strings.TrimSuffix(got, "/chat/completions")+"/models") != ""
|
||||
if onBase && onEndpoint {
|
||||
t.Errorf("version segment duplicated: base %q + endpoint %q", tc.persistedBase, got)
|
||||
}
|
||||
|
||||
@ -4,8 +4,8 @@ import (
|
||||
"encoding/json"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
@ -33,11 +33,11 @@ func keyAutoGateway(t *testing.T) *Gateway {
|
||||
t.Fatal(err)
|
||||
}
|
||||
cfg := &config.Config{
|
||||
Path: cfgPath,
|
||||
AdapterDir: filepath.Join(td, "adapters"),
|
||||
RuntimeFile: filepath.Join(td, "runtime.json"),
|
||||
Path: cfgPath,
|
||||
AdapterDir: filepath.Join(td, "adapters"),
|
||||
RuntimeFile: filepath.Join(td, "runtime.json"),
|
||||
DefaultModel: "AUTO",
|
||||
GatewayKeys: []string{"sk-admin"},
|
||||
GatewayKeys: []string{"sk-admin"},
|
||||
Keys: []config.GWKey{
|
||||
{Key: "sk-admin", Role: "admin", Name: "admin"},
|
||||
{Key: "sk-user", Role: "user", Name: "user"},
|
||||
@ -97,7 +97,7 @@ func TestKeyAutoAPICrudAndInherit(t *testing.T) {
|
||||
|
||||
// Initially inherits.
|
||||
var got struct {
|
||||
Inherits bool `json:"inherits"`
|
||||
Inherits bool `json:"inherits"`
|
||||
Auto []config.ModelScope `json:"auto"`
|
||||
}
|
||||
rr := doReqAs(t, g, "sk-admin", "GET", "/api/keys/sk-user/auto", "")
|
||||
@ -254,4 +254,4 @@ func TestKeyAutoAPIDoesNotShadowKeyRoutes(t *testing.T) {
|
||||
if rr := doReqAs(t, g, "sk-user", "GET", "/api/keys/me", ""); rr.Code != http.StatusOK {
|
||||
t.Fatalf("GET /api/keys/me as user = %d, want 200", rr.Code)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
322
internal/gateway/messages.go
Normal file
322
internal/gateway/messages.go
Normal file
@ -0,0 +1,322 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"llmsproxy/internal/provider"
|
||||
"llmsproxy/internal/scheduler"
|
||||
"llmsproxy/internal/types"
|
||||
)
|
||||
|
||||
// handleMessages serves the Anthropic Messages protocol (POST /v1/messages).
|
||||
//
|
||||
// The point of this endpoint is NOT to add a second scheduling path. It reuses
|
||||
// everything /v1/chat/completions does — key auth, model scope, per-model
|
||||
// quotas, the AUTO chain, cooldowns, audit records, plugin hooks — and differs
|
||||
// in exactly two places:
|
||||
//
|
||||
// 1. the request body is parsed from Anthropic's shape into the unified format
|
||||
// (messages with content blocks, a top-level `system`, `max_tokens`
|
||||
// mandatory, `thinking` optional);
|
||||
// 2. the response is rendered by the SERVING adapter's serialize_* hooks, so
|
||||
// the client sees Anthropic's shape back.
|
||||
//
|
||||
// The kernel therefore contains no Anthropic field names on this path: parsing
|
||||
// is table-driven, and serialization belongs to the adapter.
|
||||
//
|
||||
// A source whose adapter does not declare serialize_* falls back to the
|
||||
// OpenAI shape, which is what the kernel guarantees for an unconfigured
|
||||
// deployment. That asymmetry is deliberate (see serializer.go): an Anthropic
|
||||
// client talking to a source that cannot speak Anthropic gets a clear
|
||||
// OpenAI-shaped answer rather than a malformed Anthropic one.
|
||||
func (g *Gateway) handleMessages(w http.ResponseWriter, r *http.Request) {
|
||||
if r.Method != http.MethodPost {
|
||||
writeError(w, http.StatusMethodNotAllowed, "method_not_allowed", "use POST")
|
||||
return
|
||||
}
|
||||
var raw anthropicRequest
|
||||
if err := json.NewDecoder(r.Body).Decode(&raw); err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", "invalid json: "+err.Error())
|
||||
return
|
||||
}
|
||||
req, err := raw.toUnified(g.core.DefaultModel())
|
||||
if err != nil {
|
||||
writeError(w, http.StatusBadRequest, "invalid_request", err.Error())
|
||||
return
|
||||
}
|
||||
|
||||
model := req.Model
|
||||
ctx := r.Context()
|
||||
g.fireStart(ctx, &chatRequest{Messages: req.Messages, Tools: nil},
|
||||
chatKind(req.Stream), model, len(req.Messages), 0)
|
||||
|
||||
if isAuto(model) {
|
||||
chain, _ := g.core.AutoChainFor(reqKey(ctx))
|
||||
if chain == nil || len(chain.Tiers) == 0 {
|
||||
writeError(w, http.StatusServiceUnavailable, "no_provider", "no auto slot configured")
|
||||
return
|
||||
}
|
||||
if q := g.checkQuota(ctx, "AUTO"); q != nil {
|
||||
g.writeReject(w, q)
|
||||
return
|
||||
}
|
||||
g.serveUnified(ctx, w, chain, req, nil)
|
||||
return
|
||||
}
|
||||
if allow := g.allowedModels(ctx); allow != nil && !g.hasScopeModel(allow, model) {
|
||||
writeError(w, http.StatusForbidden, "model_not_allowed",
|
||||
"model \""+model+"\" is not allowed for this key")
|
||||
return
|
||||
}
|
||||
cands, effective := g.resolveCands(ctx, &chatRequest{Model: model})
|
||||
if len(cands) == 0 {
|
||||
writeError(w, http.StatusNotFound, "model_not_found",
|
||||
"model \""+model+"\" is not configured")
|
||||
return
|
||||
}
|
||||
if effective == "" {
|
||||
effective = firstModel(cands[0])
|
||||
}
|
||||
if q := g.checkQuota(ctx, effective); q != nil {
|
||||
g.writeReject(w, q)
|
||||
return
|
||||
}
|
||||
g.serveUnified(ctx, w, nil, req, &directRoute{cands: cands, effective: effective})
|
||||
}
|
||||
|
||||
// directRoute carries the pre-resolved candidate list for a model-pinned
|
||||
// request, so serveUnified can run both routes (direct / AUTO) from one body.
|
||||
type directRoute struct {
|
||||
cands []*provider.Provider
|
||||
effective string
|
||||
}
|
||||
|
||||
// serveUnified runs the request through the scheduler and writes the reply via
|
||||
// the serving adapter's serializer. It is the shared tail of handleChat and
|
||||
// handleMessages: by this point every gate (auth, scope, quota) has already run,
|
||||
// so the only remaining questions are "which source serves it" and "what shape
|
||||
// does the client get".
|
||||
func (g *Gateway) serveUnified(ctx context.Context, w http.ResponseWriter, chain *scheduler.Chain, req *types.ChatRequest, direct *directRoute) {
|
||||
rec := &Req{Key: keyID(reqKey(ctx)), Type: "chat", OK: false}
|
||||
if direct != nil {
|
||||
rec.Model = direct.effective
|
||||
rec.Source = firstSource(direct.cands)
|
||||
} else {
|
||||
rec.Type = "stream"
|
||||
}
|
||||
if req.Stream {
|
||||
rec.Type = "stream"
|
||||
}
|
||||
|
||||
done := g.stats.Begin()
|
||||
defer done()
|
||||
|
||||
var (
|
||||
resp *types.UnifiedResponse
|
||||
chunks <-chan types.UnifiedChunk
|
||||
usedSrc, usedModel string
|
||||
err error
|
||||
)
|
||||
t0 := time.Now()
|
||||
if direct != nil {
|
||||
cands := toScheduler(direct.cands)
|
||||
if req.Stream {
|
||||
chunks, usedSrc, usedModel, err = g.core.Scheduler().ChatStream(ctx, cands, req)
|
||||
} else {
|
||||
resp, usedSrc, usedModel, err = g.core.Scheduler().Chat(ctx, cands, req)
|
||||
}
|
||||
} else {
|
||||
if req.Stream {
|
||||
chunks, usedSrc, usedModel, err = g.core.Scheduler().ChainChatStream(ctx, chain, req, nil, nil)
|
||||
} else {
|
||||
resp, usedSrc, usedModel, err = g.core.Scheduler().ChainChat(ctx, chain, req, nil, nil)
|
||||
}
|
||||
}
|
||||
rec.LatMs = time.Since(t0).Milliseconds()
|
||||
if err != nil {
|
||||
g.failChat(w, rec, err)
|
||||
g.writeRec(rec)
|
||||
return
|
||||
}
|
||||
rec.OK = true
|
||||
rec.Status = http.StatusOK
|
||||
if usedModel != "" {
|
||||
rec.Model = usedModel
|
||||
}
|
||||
rec.Source = usedSrc
|
||||
ser := g.serializerFor(usedSrc, directProviderList(direct))
|
||||
|
||||
if chunks != nil {
|
||||
g.fireRouted(ctx, "stream", usedSrc, usedModel, tierForRoute(direct), true)
|
||||
rec.Prompt = estimatePromptTokens(req)
|
||||
g.pumpStream(w, ser, rec, chunks, modelLabel(req, usedModel), t0)
|
||||
return
|
||||
}
|
||||
recordChatUsage(rec, req, resp)
|
||||
g.fireRouted(ctx, "chat", usedSrc, usedModel, tierForRoute(direct), false)
|
||||
rec.FirstByteMs = rec.LatMs
|
||||
g.writeRec(rec)
|
||||
writeChatCompletion(w, ser, resp, modelLabel(req, usedModel))
|
||||
}
|
||||
|
||||
// ---- Anthropic request shape (in-bound only; the kernel holds no outbound
|
||||
// Anthropic field names) ----
|
||||
|
||||
// anthropicRequest is the client-facing parse target for POST /v1/messages.
|
||||
//
|
||||
// Only the fields the gateway must understand to schedule and account for a
|
||||
// request appear here. Anything else the client sent is preserved verbatim in
|
||||
// raw so an adapter can read it (the same "kernel does not interpret" rule that
|
||||
// governs tools).
|
||||
type anthropicRequest struct {
|
||||
Model string `json:"model"`
|
||||
System json.RawMessage `json:"system,omitempty"`
|
||||
Messages []anthropicMessage `json:"messages"`
|
||||
MaxTokens int `json:"max_tokens"`
|
||||
Stream bool `json:"stream,omitempty"`
|
||||
|
||||
Temperature *float64 `json:"temperature,omitempty"`
|
||||
TopP *float64 `json:"top_p,omitempty"`
|
||||
TopK int `json:"top_k,omitempty"`
|
||||
StopSeqs []string `json:"stop_sequences,omitempty"`
|
||||
Tools json.RawMessage `json:"tools,omitempty"`
|
||||
ToolChoice json.RawMessage `json:"tool_choice,omitempty"`
|
||||
Thinking *types.ThinkingConfig `json:"thinking,omitempty"`
|
||||
Metadata map[string]interface{} `json:"metadata,omitempty"`
|
||||
|
||||
// raw is the untouched body, so an adapter can read fields the gateway
|
||||
// has no reason to know (cache_control, service_tier, …).
|
||||
raw json.RawMessage
|
||||
}
|
||||
|
||||
// anthropicMessage accepts Anthropic's content as either a plain string or an
|
||||
// array of typed blocks; both normalize to the unified message shape.
|
||||
type anthropicMessage struct {
|
||||
Role string `json:"role"`
|
||||
Content json.RawMessage `json:"content"`
|
||||
}
|
||||
|
||||
// toUnified converts the parsed request into the internal format.
|
||||
func (a *anthropicRequest) toUnified(defaultModel string) (*types.ChatRequest, error) {
|
||||
if len(a.Messages) == 0 {
|
||||
return nil, errors.New("messages is required")
|
||||
}
|
||||
msgs := make([]types.ChatMessage, 0, len(a.Messages))
|
||||
for i, m := range a.Messages {
|
||||
u := types.ChatMessage{Role: m.Role, Content: normalizeAnthropicContent(m.Content)}
|
||||
if u.Role == "" {
|
||||
return nil, fmt.Errorf("messages[%d].role is required", i)
|
||||
}
|
||||
msgs = append(msgs, u)
|
||||
}
|
||||
model := a.Model
|
||||
if model == "" {
|
||||
model = defaultModel
|
||||
}
|
||||
th := a.Thinking
|
||||
if th == nil {
|
||||
// Anthropic's default is thinking OFF, and `thinking: {enabled:false}`
|
||||
// is the explicit spelling of that. Emitting it keeps the unified
|
||||
// request honest about the client's intent instead of leaving the
|
||||
// field absent and letting each adapter guess.
|
||||
no := false
|
||||
th = &types.ThinkingConfig{Enabled: &no}
|
||||
}
|
||||
return &types.ChatRequest{
|
||||
Model: model,
|
||||
Messages: msgs,
|
||||
Temperature: a.Temperature,
|
||||
MaxTokens: a.MaxTokens,
|
||||
Stream: a.Stream,
|
||||
TopP: a.TopP,
|
||||
TopK: a.TopK,
|
||||
StopSequences: a.StopSeqs,
|
||||
Tools: a.Tools,
|
||||
ToolChoice: a.ToolChoice,
|
||||
System: systemToString(a.System),
|
||||
Thinking: th,
|
||||
ExtraBody: map[string]interface{}{},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// normalizeAnthropicContent turns Anthropic content into the unified shape.
|
||||
//
|
||||
// A string stays a string. An array of blocks is preserved as-is: the unified
|
||||
// format deliberately carries multimodal/tool blocks verbatim (ChatMessage
|
||||
// .Content is json.RawMessage) rather than flattening them, because only an
|
||||
// adapter knows which block types its upstream accepts.
|
||||
func normalizeAnthropicContent(raw json.RawMessage) json.RawMessage {
|
||||
trimmed := strings.TrimSpace(string(raw))
|
||||
if trimmed == "" {
|
||||
return types.StringContent("")
|
||||
}
|
||||
if strings.HasPrefix(trimmed, `"`) {
|
||||
return raw
|
||||
}
|
||||
return raw
|
||||
}
|
||||
|
||||
// systemToString accepts Anthropic's two spellings: a plain string, or an array
|
||||
// of {type:"text", text} blocks.
|
||||
func systemToString(raw json.RawMessage) string {
|
||||
trimmed := strings.TrimSpace(string(raw))
|
||||
if trimmed == "" || trimmed == "null" {
|
||||
return ""
|
||||
}
|
||||
if strings.HasPrefix(trimmed, `"`) {
|
||||
var s string
|
||||
if err := json.Unmarshal(raw, &s); err == nil {
|
||||
return s
|
||||
}
|
||||
return ""
|
||||
}
|
||||
var blocks []struct {
|
||||
Type string `json:"type"`
|
||||
Text string `json:"text"`
|
||||
}
|
||||
if err := json.Unmarshal(raw, &blocks); err != nil {
|
||||
return ""
|
||||
}
|
||||
var sb strings.Builder
|
||||
for _, b := range blocks {
|
||||
if b.Type == "text" {
|
||||
sb.WriteString(b.Text)
|
||||
}
|
||||
}
|
||||
return sb.String()
|
||||
}
|
||||
|
||||
func chatKind(stream bool) string {
|
||||
if stream {
|
||||
return "stream"
|
||||
}
|
||||
return "chat"
|
||||
}
|
||||
|
||||
// directProviderList hands the serving-source resolver its candidate list.
|
||||
func directProviderList(direct *directRoute) []*provider.Provider {
|
||||
if direct == nil {
|
||||
return nil
|
||||
}
|
||||
return direct.cands
|
||||
}
|
||||
|
||||
func tierForRoute(direct *directRoute) int {
|
||||
if direct != nil {
|
||||
return -1
|
||||
}
|
||||
return -2
|
||||
}
|
||||
|
||||
func modelLabel(req *types.ChatRequest, usedModel string) string {
|
||||
if usedModel != "" {
|
||||
return usedModel
|
||||
}
|
||||
return req.Model
|
||||
}
|
||||
303
internal/gateway/messages_test.go
Normal file
303
internal/gateway/messages_test.go
Normal file
@ -0,0 +1,303 @@
|
||||
package gateway
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"llmsproxy/internal/config"
|
||||
"llmsproxy/internal/core"
|
||||
)
|
||||
|
||||
// 本文件端到端验证 Phase 2 的核心主张:**对外协议由适配器决定,内核只调度**。
|
||||
//
|
||||
// 具体做法:起一个真实的 core + gateway,源挂 anthropic 适配器,上游是一个
|
||||
// 只懂 Anthropic Messages 的假服务器。然后:
|
||||
//
|
||||
// 1. Anthropic 客户端形状的请求 → 拿到 Anthropic 形状的响应
|
||||
// 2. 调度/配额/审计与 /v1/chat/completions 走同一条路径
|
||||
// 3. 上游若不支持该模型,内核如实回 503,不伪造格式
|
||||
|
||||
// anthropicUpstream 是一个只实现 Anthropic Messages 的假上游。
|
||||
func anthropicUpstream(t *testing.T) *httptest.Server {
|
||||
return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
if !strings.HasSuffix(r.URL.Path, "/messages") {
|
||||
// 网关把请求发到了非 Anthropic 路径 —— 显式暴露这个 bug
|
||||
t.Errorf("请求打到了非 Anthropic 路径: %s", r.URL.Path)
|
||||
w.WriteHeader(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
body, _ := io.ReadAll(r.Body)
|
||||
var probe map[string]interface{}
|
||||
_ = json.Unmarshal(body, &probe)
|
||||
if _, ok := probe["messages"]; !ok {
|
||||
t.Errorf("上游收到没有 messages 的请求体: %s", body)
|
||||
}
|
||||
if _, ok := probe["system"]; !ok && probe["system"] == nil {
|
||||
// system 是可选的,仅在客户端给了的时候才应出现;不强制
|
||||
_ = ok
|
||||
}
|
||||
if _, ok := probe["max_tokens"]; !ok {
|
||||
t.Errorf("Anthropic 要求 max_tokens,上游请求体却缺这个字段: %s", body)
|
||||
}
|
||||
|
||||
// Streaming requests must get a real Anthropic SSE sequence; a JSON body
|
||||
// here would leave the adapter with zero frames and the gateway would
|
||||
// (correctly) reject the stream as empty.
|
||||
if stream, _ := probe["stream"].(bool); stream {
|
||||
w.Header().Set("Content-Type", "text/event-stream")
|
||||
fl, _ := w.(http.Flusher)
|
||||
frames := []string{
|
||||
`event: message_start` + "\n" + `data: {"type":"message_start","message":{"id":"msg_mock","type":"message","role":"assistant","model":"` + probeModel(probe) + `","content":[],"usage":{"input_tokens":3,"output_tokens":0}}}`,
|
||||
`event: content_block_start` + "\n" + `data: {"type":"content_block_start","index":0,"content_block":{"type":"text","text":""}}`,
|
||||
`event: content_block_delta` + "\n" + `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"pong"}}`,
|
||||
`event: content_block_stop` + "\n" + `data: {"type":"content_block_stop","index":0}`,
|
||||
`event: message_delta` + "\n" + `data: {"type":"message_delta","delta":{"stop_reason":"end_turn"},"usage":{"output_tokens":5}}`,
|
||||
`event: message_stop` + "\n" + `data: {"type":"message_stop"}`,
|
||||
}
|
||||
for _, f := range frames {
|
||||
fmt.Fprint(w, f+"\n\n")
|
||||
if fl != nil {
|
||||
fl.Flush()
|
||||
}
|
||||
}
|
||||
return
|
||||
}
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, `{
|
||||
"id":"msg_mock","type":"message","role":"assistant","model":"`+probeModel(probe)+`",
|
||||
"content":[{"type":"thinking","thinking":"let me think"},{"type":"text","text":"pong"}],
|
||||
"stop_reason":"end_turn",
|
||||
"usage":{"input_tokens":3,"output_tokens":5}
|
||||
}`)
|
||||
}))
|
||||
}
|
||||
|
||||
func probeModel(m map[string]interface{}) string {
|
||||
if s, ok := m["model"].(string); ok {
|
||||
return s
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
// messagesGateway 起一个真实网关,源挂 anthropic 适配器。
|
||||
func messagesGateway(t *testing.T) (*httptest.Server, string) {
|
||||
t.Helper()
|
||||
up := anthropicUpstream(t)
|
||||
t.Cleanup(up.Close)
|
||||
|
||||
dir := t.TempDir()
|
||||
// adapterDir 故意不预先创建:VM 首启会把内置适配器 seed 进去。
|
||||
// 预先创建会被当成"目录已存在即为权威",于是没有任何内置适配器被加载,
|
||||
// 表现为 "adapter anthropic not loaded"。
|
||||
cfg := &config.Config{
|
||||
Path: filepath.Join(dir, "config.yaml"),
|
||||
AdapterDir: filepath.Join(dir, "adapters"),
|
||||
RuntimeFile: filepath.Join(dir, "runtime.json"),
|
||||
Listen: "127.0.0.1:0",
|
||||
DefaultModel: "claude-test",
|
||||
GatewayKeys: []string{"sk-e2e-0001"},
|
||||
Sources: []config.Source{{
|
||||
Name: "anth", BaseURL: up.URL, Adapter: "anthropic",
|
||||
APIKey: "k", MaxConcurrent: 4,
|
||||
Models: []config.Model{{ID: "claude-test", Kind: "chat", Priority: 100}},
|
||||
}},
|
||||
}
|
||||
|
||||
c, err := core.NewFromConfig(cfg)
|
||||
if err != nil {
|
||||
t.Fatalf("core: %v", err)
|
||||
}
|
||||
t.Cleanup(c.Close)
|
||||
gw, err := New(c)
|
||||
if err != nil {
|
||||
t.Fatalf("gateway: %v", err)
|
||||
}
|
||||
ts := httptest.NewServer(gw.Handler())
|
||||
t.Cleanup(ts.Close)
|
||||
return ts, "sk-e2e-0001"
|
||||
}
|
||||
|
||||
func postJSON(t *testing.T, url, key, body string) (int, string) {
|
||||
t.Helper()
|
||||
req, _ := http.NewRequest(http.MethodPost, url, strings.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+key)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatalf("request: %v", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
b, _ := io.ReadAll(resp.Body)
|
||||
return resp.StatusCode, string(b)
|
||||
}
|
||||
|
||||
// TestMessagesEndpointSpeaksAnthropic 是本文件的主断言:Anthropic 客户端
|
||||
// 打 /v1/messages,拿到 Anthropic 形状。
|
||||
func TestMessagesEndpointSpeaksAnthropic(t *testing.T) {
|
||||
ts, key := messagesGateway(t)
|
||||
code, body := postJSON(t, ts.URL+"/v1/messages", key,
|
||||
`{"model":"claude-test","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`)
|
||||
if code != 200 {
|
||||
t.Fatalf("status = %d: %s", code, body)
|
||||
}
|
||||
var m map[string]interface{}
|
||||
if err := json.Unmarshal([]byte(body), &m); err != nil {
|
||||
t.Fatalf("响应不是 JSON: %v (%s)", err, body)
|
||||
}
|
||||
if m["type"] != "message" {
|
||||
t.Errorf("type = %v, want message(内核不该退回 OpenAI 的 chat.completion)", m["type"])
|
||||
}
|
||||
if m["stop_reason"] != "end_turn" {
|
||||
t.Errorf("stop_reason = %v, want end_turn", m["stop_reason"])
|
||||
}
|
||||
blocks, _ := m["content"].([]interface{})
|
||||
if len(blocks) != 2 {
|
||||
t.Fatalf("content blocks = %v, want [thinking, text]", m["content"])
|
||||
}
|
||||
if b0 := blocks[0].(map[string]interface{}); b0["type"] != "thinking" || b0["thinking"] != "let me think" {
|
||||
t.Errorf("首个块应为 thinking: %v", b0)
|
||||
}
|
||||
if b1 := blocks[1].(map[string]interface{}); b1["type"] != "text" || b1["text"] != "pong" {
|
||||
t.Errorf("第二个块应为 text/pong: %v", b1)
|
||||
}
|
||||
u := m["usage"].(map[string]interface{})
|
||||
if u["input_tokens"] != float64(3) || u["output_tokens"] != float64(5) {
|
||||
t.Errorf("usage 应为 Anthropic 的 input/output 命名: %v", u)
|
||||
}
|
||||
// 内核不得混入 OpenAI 形状
|
||||
for _, forbidden := range []string{`"object"`, `"choices"`, `"chat.completion"`, `"prompt_tokens"`} {
|
||||
if strings.Contains(body, forbidden) {
|
||||
t.Errorf("OpenAI 字段 %s 混入 Anthropic 响应: %s", forbidden, body)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// TestMessagesStreamSpeaksAnthropic 验证流式也是 Anthropic 事件序列,
|
||||
// 而不是 OpenAI SSE。
|
||||
func TestMessagesStreamSpeaksAnthropic(t *testing.T) {
|
||||
ts, key := messagesGateway(t)
|
||||
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/messages",
|
||||
strings.NewReader(`{"model":"claude-test","max_tokens":64,"stream":true,"messages":[{"role":"user","content":"hi"}]}`))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
req.Header.Set("Authorization", "Bearer "+key)
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, _ := io.ReadAll(resp.Body)
|
||||
s := string(body)
|
||||
|
||||
if !strings.Contains(s, "message_start") {
|
||||
t.Errorf("Anthropic 流应以 message_start 开头: %s", s)
|
||||
}
|
||||
if !strings.Contains(s, "message_stop") {
|
||||
t.Errorf("Anthropic 流应以 message_stop 收尾: %s", s)
|
||||
}
|
||||
// 内核不得擅自给 Anthropic 流加 OpenAI 的 [DONE]
|
||||
if strings.Contains(s, "[DONE]") {
|
||||
t.Errorf("Anthropic 流里出现了 OpenAI 的 [DONE] 终止标记: %s", s)
|
||||
}
|
||||
if strings.Contains(s, "chat.completion.chunk") {
|
||||
t.Errorf("流里混入 OpenAI chunk 形状: %s", s)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMessagesRejectsBadRequest 确认入向校验(Anthropic 要求 max_tokens)
|
||||
// 与 OpenAI 路径一样在入口拒绝,不占用上游槽位。
|
||||
func TestMessagesRejectsBadRequest(t *testing.T) {
|
||||
ts, key := messagesGateway(t)
|
||||
// Anthropic 允许省略 max_tokens(上游会给默认),所以「有 messages 无
|
||||
// max_tokens」必须被接受;真正该拒的是没有 messages。
|
||||
code, body := postJSON(t, ts.URL+"/v1/messages", key,
|
||||
`{"model":"claude-test","messages":[{"role":"user","content":"hi"}]}`)
|
||||
if code != 200 {
|
||||
t.Fatalf("有 messages 的请求应被接受(max_tokens 可省),却得 %d: %s", code, body)
|
||||
}
|
||||
code, body = postJSON(t, ts.URL+"/v1/messages", key, `{"model":"claude-test"}`)
|
||||
if code != 400 {
|
||||
t.Errorf("无 messages 的请求 status = %d, want 400: %s", code, body)
|
||||
}
|
||||
// 坏 JSON
|
||||
code, _ = postJSON(t, ts.URL+"/v1/messages", key, `{"model":`)
|
||||
if code != 400 {
|
||||
t.Errorf("坏 JSON status = %d, want 400", code)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMessagesUnconfiguredFallsBackToOpenai 验证 R5 的兜底在新增端点上同样
|
||||
// 生效:源没有声明序列化钩子时(此例用 openai 适配器),回 OpenAI 形状。
|
||||
func TestMessagesUnconfiguredFallsBackToOpenai(t *testing.T) {
|
||||
up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("Content-Type", "application/json")
|
||||
fmt.Fprint(w, `{"choices":[{"message":{"content":"pong"},"finish_reason":"stop"}],
|
||||
"usage":{"prompt_tokens":3,"completion_tokens":1,"total_tokens":4}}`)
|
||||
}))
|
||||
defer up.Close()
|
||||
|
||||
dir := t.TempDir()
|
||||
// adapterDir 故意不预先创建:VM 首启会把内置适配器 seed 进去。
|
||||
// 预先创建会被当成"目录已存在即为权威",于是没有任何内置适配器被加载,
|
||||
// 表现为 "adapter anthropic not loaded"。
|
||||
cfg := &config.Config{
|
||||
Path: filepath.Join(dir, "config.yaml"),
|
||||
AdapterDir: filepath.Join(dir, "adapters"),
|
||||
RuntimeFile: filepath.Join(dir, "runtime.json"),
|
||||
Listen: "127.0.0.1:0",
|
||||
DefaultModel: "gpt-test",
|
||||
GatewayKeys: []string{"sk-e2e-0001"},
|
||||
Sources: []config.Source{{
|
||||
Name: "oai", BaseURL: up.URL, Adapter: "openai", APIKey: "k", MaxConcurrent: 4,
|
||||
Models: []config.Model{{ID: "gpt-test", Kind: "chat"}},
|
||||
}},
|
||||
}
|
||||
c, err := core.NewFromConfig(cfg)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.Close()
|
||||
gw, err := New(c)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ts := httptest.NewServer(gw.Handler())
|
||||
defer ts.Close()
|
||||
|
||||
code, body := postJSON(t, ts.URL+"/v1/messages", "sk-e2e-0001",
|
||||
`{"model":"gpt-test","max_tokens":64,"messages":[{"role":"user","content":"hi"}]}`)
|
||||
if code != 200 {
|
||||
t.Fatalf("status = %d: %s", code, body)
|
||||
}
|
||||
if !strings.Contains(body, "chat.completion") {
|
||||
t.Errorf("无序列化钩子的源应回落 OpenAI 兜底形状: %s", body)
|
||||
}
|
||||
}
|
||||
|
||||
// TestMessagesRequiresAuth 确认新端点走同一套鉴权(不因为是新协议而放行)。
|
||||
func TestMessagesRequiresAuth(t *testing.T) {
|
||||
ts, _ := messagesGateway(t)
|
||||
code, _ := postJSON(t, ts.URL+"/v1/messages", "wrong-key",
|
||||
`{"model":"claude-test","max_tokens":8,"messages":[{"role":"user","content":"hi"}]}`)
|
||||
if code != 401 {
|
||||
t.Errorf("错误 key status = %d, want 401", code)
|
||||
}
|
||||
// 无 auth
|
||||
req, _ := http.NewRequest(http.MethodPost, ts.URL+"/v1/messages", strings.NewReader("{}"))
|
||||
req.Header.Set("Content-Type", "application/json")
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode != 401 {
|
||||
t.Errorf("无 key status = %d, want 401", resp.StatusCode)
|
||||
}
|
||||
_ = context.Background()
|
||||
}
|
||||
@ -147,4 +147,4 @@ func TestPluginCallIsAdminOnly(t *testing.T) {
|
||||
t.Fatalf("a non-admin key reached the call endpoint: HTTP %d %s",
|
||||
rr.Code, rr.Body.String())
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -204,4 +204,4 @@ type bodyTooLarge struct{}
|
||||
func (bodyTooLarge) Error() string { return "request body exceeds 4 MiB" }
|
||||
|
||||
// limitBody is kept for the numeric form some callers want in a log line.
|
||||
var _ = strconv.Itoa
|
||||
var _ = strconv.Itoa
|
||||
|
||||
@ -222,7 +222,7 @@ func TestPluginsListCarriesAccessTables(t *testing.T) {
|
||||
t.Fatalf("GET /api/plugins = %d", rr.Code)
|
||||
}
|
||||
var out struct {
|
||||
Visibility map[string]string `json:"visibility"`
|
||||
Visibility map[string]string `json:"visibility"`
|
||||
HTTPAccess []map[string]interface{} `json:"http_access"`
|
||||
}
|
||||
if err := json.Unmarshal(rr.Body.Bytes(), &out); err != nil {
|
||||
|
||||
@ -32,15 +32,17 @@ func luaVMFor(t *testing.T) *lua.VM {
|
||||
return vm
|
||||
}
|
||||
|
||||
// TestFallbackUsedWhenAdapterDeclaresNothing 是「兜底」本身的测试:
|
||||
// 没有任何内置适配器声明 serialize_*,所以默认路径必须是 OpenAI。
|
||||
// TestFallbackUsedWhenAdapterDeclaresNothing 是「兜底」本身的测试。
|
||||
//
|
||||
// 注:anthropic 现在**确实**声明了 serialize_*(Phase 2 给了它 Anthropic
|
||||
// 形状),所以本测试只在仍未声明钩子的那些内置适配器上验证兜底行为。
|
||||
// 这正是设计意图:谁声明谁接管,没声明的继续用 OpenAI 兜底。
|
||||
func TestFallbackUsedWhenAdapterDeclaresNothing(t *testing.T) {
|
||||
vm := luaVMFor(t)
|
||||
// 先证明前提:内置适配器确实没声明这些钩子(否则下面测的是别的东西)
|
||||
for _, name := range []string{"openai", "anthropic", "gemini", "ollama", "deepseek"} {
|
||||
for _, name := range []string{"openai", "gemini", "ollama", "deepseek", "github", "groq"} {
|
||||
for _, hook := range []string{"serialize_response", "serialize_chunk"} {
|
||||
if vm.HasFn(name, hook) {
|
||||
t.Fatalf("内置适配器 %s 声明了 %s —— 本测试的前提(默认走兜底)已不成立", name, hook)
|
||||
t.Errorf("内置适配器 %s 声明了 %s —— 兜底测试的前提已不成立", name, hook)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -48,6 +50,11 @@ func TestFallbackUsedWhenAdapterDeclaresNothing(t *testing.T) {
|
||||
if _, isFallback := ser.(openAISerializer); !isFallback {
|
||||
t.Fatalf("未声明钩子的适配器应得到兜底序列化器,得到 %T", ser)
|
||||
}
|
||||
|
||||
// 对照:声明了钩子的适配器不走兜底
|
||||
if _, isFallback := serializerFor(vm, "anthropic").(openAISerializer); isFallback {
|
||||
t.Error("anthropic 声明了 serialize_response,不应走兜底")
|
||||
}
|
||||
}
|
||||
|
||||
// TestAdapterSerializerWinsOverFallback 锁定核心规则:适配器声明钩子即接管,
|
||||
@ -323,19 +330,30 @@ func TestSerializerForConcurrent(t *testing.T) {
|
||||
errs <- "nil serializer"
|
||||
return
|
||||
}
|
||||
// 结果必须自洽:同一适配器每次拿到同一类型
|
||||
// 结果必须自洽:同一适配器每次拿到同一类型,且形状与该适配器
|
||||
// 声明的协议一致(openai→OpenAI,anthropic→Anthropic)。
|
||||
_, isFallback := ser.(openAISerializer)
|
||||
if w%2 == 0 && !isFallback {
|
||||
errs <- "openai 未走兜底"
|
||||
return
|
||||
wantShape := "chat.completion"
|
||||
switch name {
|
||||
case "anthropic":
|
||||
if isFallback {
|
||||
errs <- "anthropic 并发下退回兜底"
|
||||
return
|
||||
}
|
||||
wantShape = `"type":"message"`
|
||||
default:
|
||||
if !isFallback {
|
||||
errs <- name + " 并发下未走兜底"
|
||||
return
|
||||
}
|
||||
}
|
||||
body, err := ser.NonStream(&types.UnifiedResponse{Content: "x"}, "m", "id", 1)
|
||||
if err != nil {
|
||||
errs <- "serialize: " + err.Error()
|
||||
return
|
||||
}
|
||||
if !strings.Contains(string(body), "chat.completion") {
|
||||
errs <- "并发下输出形状不一致"
|
||||
if !strings.Contains(string(body), wantShape) {
|
||||
errs <- "并发下输出形状不一致: " + name + " -> " + string(body)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
@ -216,6 +216,14 @@ func (g *Gateway) routes(w http.ResponseWriter, r *http.Request) {
|
||||
switch {
|
||||
case r.URL.Path == "/v1/chat/completions":
|
||||
g.handleChat(w, r)
|
||||
case r.URL.Path == "/v1/messages":
|
||||
// Anthropic Messages protocol. Scheduling, quotas, cooldowns, AUTO
|
||||
// chain and accounting are EXACTLY the ones /v1/chat/completions uses —
|
||||
// only the wire shapes differ, and those come from the serving
|
||||
// adapter's serialize_* hooks. The kernel holds no Anthropic field
|
||||
// names (see serializer.go: the fallback is OpenAI, and any adapter
|
||||
// that declares serialize_* replaces it wholesale).
|
||||
g.handleMessages(w, r)
|
||||
case r.URL.Path == "/v1/images/generations":
|
||||
g.handleImage(w, r)
|
||||
case r.URL.Path == "/v1/models":
|
||||
|
||||
@ -53,7 +53,7 @@ type Req struct {
|
||||
// 0 means "not reported", which is NOT the same as "no reasoning happened" —
|
||||
// three of the sources measured return no such field at all.
|
||||
Reasoning int64 `json:"reasoning_tokens,omitempty"`
|
||||
OK bool `json:"ok"`
|
||||
OK bool `json:"ok"`
|
||||
// Status http status code
|
||||
Status int `json:"status"`
|
||||
// Err short error message
|
||||
|
||||
@ -74,7 +74,7 @@ func TestStatsPeriodRecordsMatchPager(t *testing.T) {
|
||||
rr := doReq(t, g, "GET", "/api/stats?period=day&limit=100", "")
|
||||
var st struct {
|
||||
Records []struct {
|
||||
Time int64 `json:"time"`
|
||||
Time int64 `json:"time"`
|
||||
Model string `json:"model"`
|
||||
} `json:"records"`
|
||||
}
|
||||
@ -84,7 +84,7 @@ func TestStatsPeriodRecordsMatchPager(t *testing.T) {
|
||||
rr2 := doReq(t, g, "GET", "/api/stats/records?limit=100", "")
|
||||
var pg struct {
|
||||
Records []struct {
|
||||
Time int64 `json:"time"`
|
||||
Time int64 `json:"time"`
|
||||
Model string `json:"model"`
|
||||
} `json:"records"`
|
||||
}
|
||||
|
||||
@ -137,4 +137,4 @@ func TestStatsAPIUnknownKeyReturnsJSON(t *testing.T) {
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
t.Fatalf("status = %d, want 200", resp.StatusCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -48,11 +48,11 @@ func TestUI_HasDedicatedImagePage(t *testing.T) {
|
||||
ui := uiSource(t)
|
||||
|
||||
for _, want := range []string{
|
||||
`data-tab="image"`, // sidebar entry
|
||||
`id="tab-image"`, // pane
|
||||
`function renderImage`, // renderer registered with refresh()
|
||||
`tab === "image"`, // dispatched
|
||||
`/v1/images/generations`, // the endpoint the page actually calls
|
||||
`data-tab="image"`, // sidebar entry
|
||||
`id="tab-image"`, // pane
|
||||
`function renderImage`, // renderer registered with refresh()
|
||||
`tab === "image"`, // dispatched
|
||||
`/v1/images/generations`, // the endpoint the page actually calls
|
||||
} {
|
||||
if !strings.Contains(ui, want) {
|
||||
t.Errorf("image page is missing %q", want)
|
||||
@ -99,8 +99,8 @@ func TestUI_ImagePageCallsTheImageEndpoint(t *testing.T) {
|
||||
// contract: the field the UI reads must actually be serialized.
|
||||
func TestSourceStatus_ExposesModelKindsToTheUI(t *testing.T) {
|
||||
st := provider.SourceStatus{
|
||||
Name: "vela",
|
||||
Models: []string{"claude-sonnet-5.5", "gpt-image-2"},
|
||||
Name: "vela",
|
||||
Models: []string{"claude-sonnet-5.5", "gpt-image-2"},
|
||||
ModelKinds: []provider.ModelKindInfo{
|
||||
{ID: "claude-sonnet-5.5", Kind: "chat"},
|
||||
{ID: "gpt-image-2", Kind: "image"},
|
||||
@ -192,4 +192,4 @@ func TestImageEndpointComposesUnderRealConditions(t *testing.T) {
|
||||
"default image path doubles the version segment; if the upstream ever "+
|
||||
"accepts it, composeImageEndpoint's rule needs rechecking", doubled, resp2.StatusCode)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -106,4 +106,4 @@ func sendImageSource(t *testing.T) string {
|
||||
body = body[:end+1]
|
||||
}
|
||||
return body
|
||||
}
|
||||
}
|
||||
|
||||
@ -126,6 +126,7 @@ func TestUIStatsShowsPeriodWindow(t *testing.T) {
|
||||
t.Fatalf("i18n key periodRange appears %d time(s), want both zh and en", n)
|
||||
}
|
||||
}
|
||||
|
||||
// Both canvases can exist in the DOM at once: the modal is mounted on top of
|
||||
// the settings tab, whose #scr-canvas stays alive behind it. Resolving the
|
||||
// canvas by document order (whichever comes first) painted the per-key chain
|
||||
|
||||
@ -396,15 +396,15 @@ func TestPluginSidebarEntryCarriesATitle(t *testing.T) {
|
||||
func TestPluginPageHasAccessEditor(t *testing.T) {
|
||||
src := uiSource(t)
|
||||
for _, needle := range []string{
|
||||
`renderPluginAccess`, // called after rendering the plugin list
|
||||
`savePluginAccess`, // wired to a button
|
||||
`/access`, // hits the per-endpoint access API
|
||||
`id="pl-access"`, // container exists
|
||||
`http_access`, // reads the table from /api/plugins
|
||||
`data-ep=`, // one input per endpoint
|
||||
`data-plvis=`, // one visibility selector per plugin
|
||||
`/http/*/access`, // whole-plugin visibility shares the access API
|
||||
`plVisCell`, // renders the declared-vs-override distinction
|
||||
`renderPluginAccess`, // called after rendering the plugin list
|
||||
`savePluginAccess`, // wired to a button
|
||||
`/access`, // hits the per-endpoint access API
|
||||
`id="pl-access"`, // container exists
|
||||
`http_access`, // reads the table from /api/plugins
|
||||
`data-ep=`, // one input per endpoint
|
||||
`data-plvis=`, // one visibility selector per plugin
|
||||
`/http/*/access`, // whole-plugin visibility shares the access API
|
||||
`plVisCell`, // renders the declared-vs-override distinction
|
||||
} {
|
||||
if !strings.Contains(src, needle) {
|
||||
t.Errorf("plugin page lacks %q", needle)
|
||||
|
||||
@ -456,4 +456,124 @@ function adapter.transform_error(status, body)
|
||||
return nil
|
||||
end
|
||||
|
||||
-- ============================================================
|
||||
-- 对外序列化(统一格式 → Anthropic 形状)
|
||||
-- ============================================================
|
||||
--
|
||||
-- 内核只持统一格式;对外发什么形状由适配器决定(Phase 2)。
|
||||
-- 这几个钩子让以 anthropic 适配器为服务端的源,直接对 Anthropic 协议客户端
|
||||
-- 提供 /v1/messages 语义,而内核里不再有任何 Anthropic 字段名。
|
||||
--
|
||||
-- 与入向的 transform_* 互为逆运算:那里的映射(end_turn→stop)在这里反着
|
||||
-- 做回去(stop→end_turn),且默认 stop_sequence 是最保守的选择——它语义上
|
||||
-- 最接近"正常结束"。
|
||||
|
||||
-- 统一 finish_reason → Anthropic stop_reason
|
||||
local function to_stop_reason(finish)
|
||||
if finish == "tool_calls" then return "tool_use" end
|
||||
if finish == "length" then return "max_tokens" end
|
||||
if finish == "content_filter" then return "refusal" end
|
||||
-- stop / 空 / 未知 → end_turn(Anthropic 的"正常说完")
|
||||
return "end_turn"
|
||||
end
|
||||
|
||||
-- usage 数字化:内部 TokenUsage 的字段名与 Anthropic 的 input_tokens/output_tokens
|
||||
-- 不同名,这里做最后一次翻译。
|
||||
local function to_anthropic_usage(u)
|
||||
if type(u) ~= "table" then
|
||||
return { input_tokens = 0, output_tokens = 0 }
|
||||
end
|
||||
return {
|
||||
input_tokens = u.prompt or 0,
|
||||
output_tokens = u.completion or 0,
|
||||
}
|
||||
end
|
||||
|
||||
-- 非流式响应:统一结果 → Anthropic Messages 响应对象
|
||||
function adapter.serialize_response(payload)
|
||||
local p = json.decode(payload)
|
||||
local blocks = {}
|
||||
-- thinking 放最前,与 Anthropic 自己的块顺序一致
|
||||
if p.reasoning ~= nil and p.reasoning ~= "" then
|
||||
blocks[#blocks + 1] = { type = "thinking", thinking = p.reasoning }
|
||||
end
|
||||
blocks[#blocks + 1] = { type = "text", text = p.content or "" }
|
||||
if type(p.tool_calls) == "table" and #p.tool_calls > 0 then
|
||||
for _, tc in ipairs(p.tool_calls) do
|
||||
blocks[#blocks + 1] = {
|
||||
type = "tool_use",
|
||||
id = tc.id or "",
|
||||
name = tc.name or "",
|
||||
input = tc.arguments or {},
|
||||
}
|
||||
end
|
||||
end
|
||||
return json.encode({
|
||||
id = p.id or "",
|
||||
type = "message",
|
||||
role = "assistant",
|
||||
model = p.model or "",
|
||||
content = blocks,
|
||||
stop_reason = to_stop_reason(p.finish_reason),
|
||||
stop_sequence = nil,
|
||||
usage = to_anthropic_usage(p.token_usage),
|
||||
})
|
||||
end
|
||||
|
||||
-- 角色前导:Anthropic 的流以 message_start 开头(不是 role delta)
|
||||
function adapter.serialize_open_preamble(payload)
|
||||
local p = json.decode(payload)
|
||||
return json.encode({
|
||||
type = "message_start",
|
||||
message = {
|
||||
id = p.id or "",
|
||||
type = "message",
|
||||
role = "assistant",
|
||||
model = p.model or "",
|
||||
content = {},
|
||||
stop_reason = nil,
|
||||
stop_sequence = nil,
|
||||
usage = { input_tokens = 0, output_tokens = 0 },
|
||||
},
|
||||
})
|
||||
end
|
||||
|
||||
-- 流式增量:统一 chunk → Anthropic content_block_delta
|
||||
--
|
||||
-- 真实 Anthropic 流需要 content_block_start(开块)→ delta → stop(关块)
|
||||
-- 的三段式。这里用与多数 Anthropic 客户端兼容的简化形式:首个文本增量
|
||||
-- 带 index=0 的 delta,后续同 index;关闭由 serialize_close 负责。
|
||||
function adapter.serialize_chunk(payload)
|
||||
local p = json.decode(payload)
|
||||
if p.done then
|
||||
-- 终止由 serialize_close 统一发,这里不重复
|
||||
return ""
|
||||
end
|
||||
if p.reasoning ~= nil and p.reasoning ~= "" then
|
||||
return json.encode({
|
||||
type = "content_block_delta",
|
||||
index = 0,
|
||||
delta = { type = "thinking_delta", thinking = p.reasoning },
|
||||
})
|
||||
end
|
||||
return json.encode({
|
||||
type = "content_block_delta",
|
||||
index = 0,
|
||||
delta = { type = "text_delta", text = p.content or "" },
|
||||
})
|
||||
end
|
||||
|
||||
-- 收尾:message_delta(stop_reason + usage)→ message_stop
|
||||
function adapter.serialize_close(payload)
|
||||
local p = json.decode(payload)
|
||||
return json.encode({
|
||||
type = "message_delta",
|
||||
delta = {
|
||||
stop_reason = to_stop_reason(p.finish_reason),
|
||||
stop_sequence = nil,
|
||||
},
|
||||
usage = { output_tokens = 0 },
|
||||
}) .. "\n" .. json.encode({ type = "message_stop" })
|
||||
end
|
||||
|
||||
return adapter
|
||||
|
||||
53
internal/lua/ser_check_test.go
Normal file
53
internal/lua/ser_check_test.go
Normal file
@ -0,0 +1,53 @@
|
||||
package lua
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestAnthropicSerializerShape(t *testing.T) {
|
||||
vm := NewVM(filepath.Join(t.TempDir(), "adapters"))
|
||||
if err := vm.Start(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer vm.Stop()
|
||||
|
||||
if !vm.HasFn("anthropic", "serialize_response") {
|
||||
t.Fatal("anthropic 未声明 serialize_response")
|
||||
}
|
||||
out, err := vm.Transform("anthropic", "serialize_response",
|
||||
`{"model":"claude-x","id":"msg_1","content":"hello","finish_reason":"stop","token_usage":{"prompt":10,"completion":5,"total":15}}`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var m map[string]interface{}
|
||||
if err := json.Unmarshal([]byte(out), &m); err != nil {
|
||||
t.Fatalf("not JSON: %s", out)
|
||||
}
|
||||
if m["type"] != "message" || m["role"] != "assistant" {
|
||||
t.Errorf("Anthropic 形状不对: %s", out)
|
||||
}
|
||||
if m["stop_reason"] != "end_turn" {
|
||||
t.Errorf("stop_reason = %v, want end_turn", m["stop_reason"])
|
||||
}
|
||||
blocks, _ := m["content"].([]interface{})
|
||||
if len(blocks) != 1 {
|
||||
t.Fatalf("blocks = %v", m["content"])
|
||||
}
|
||||
b0 := blocks[0].(map[string]interface{})
|
||||
if b0["type"] != "text" || b0["text"] != "hello" {
|
||||
t.Errorf("首个块应为 text/hello: %v", b0)
|
||||
}
|
||||
u := m["usage"].(map[string]interface{})
|
||||
if u["input_tokens"] != float64(10) || u["output_tokens"] != float64(5) {
|
||||
t.Errorf("usage 字段名/值不对: %v", u)
|
||||
}
|
||||
// 内核不得混入 OpenAI 字段
|
||||
for _, k := range []string{"object", "choices", "usage.prompt_tokens"} {
|
||||
if strings.Contains(out, `"`+k+`"`) {
|
||||
t.Errorf("OpenAI 字段 %q 混入 Anthropic 输出: %s", k, out)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user