Skip to content

Commit 349c7f6

Browse files
committed
feat: support account capability routing
1 parent 105914c commit 349c7f6

18 files changed

Lines changed: 485 additions & 6 deletions

backend/internal/plugin/host_service.go

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -725,7 +725,7 @@ func (h *HostService) forward(ctx context.Context, req hostForwardRequest) (map[
725725
}
726726

727727
for attempt := 0; attempt < maxHostForwardAttempts; attempt++ {
728-
acc, err := h.scheduler.SelectAccount(ctx, route.Platform, model, 0, route.GroupID, "", hardExclude...)
728+
acc, err := h.scheduler.SelectAccountWithRequirements(ctx, route.Platform, model, 0, route.GroupID, "", hostAccountRequirements(req), hardExclude...)
729729
if err != nil {
730730
if cerr := hostContextError(err); cerr != nil {
731731
return nil, cerr
@@ -751,6 +751,7 @@ func (h *HostService) forward(ctx context.Context, req hostForwardRequest) (map[
751751
}
752752

753753
headers := hostForwardHeaders(req, route)
754+
applyAccountCapabilityHeaders(headers, accFull)
754755
fwdReq := &sdk.ForwardRequest{
755756
Account: hostSDKAccount(accFull),
756757
Body: hostForwardBody(req.Body),
@@ -871,7 +872,7 @@ func (h *HostService) forwardStream(ctx context.Context, req hostForwardRequest,
871872
}
872873

873874
for attempt := 0; attempt < maxHostForwardAttempts; attempt++ {
874-
acc, err := h.scheduler.SelectAccount(ctx, route.Platform, model, 0, route.GroupID, "", hardExclude...)
875+
acc, err := h.scheduler.SelectAccountWithRequirements(ctx, route.Platform, model, 0, route.GroupID, "", hostAccountRequirements(req), hardExclude...)
875876
if err != nil {
876877
if cerr := hostContextError(err); cerr != nil {
877878
return cerr
@@ -897,10 +898,12 @@ func (h *HostService) forwardStream(ctx context.Context, req hostForwardRequest,
897898
}
898899

899900
fw := &failoverStreamWriter{target: sw}
901+
headers := hostForwardHeaders(req, route)
902+
applyAccountCapabilityHeaders(headers, accFull)
900903
fwdReq := &sdk.ForwardRequest{
901904
Account: hostSDKAccount(accFull),
902905
Body: hostForwardBody(req.Body),
903-
Headers: hostForwardHeaders(req, route),
906+
Headers: headers,
904907
Model: model,
905908
Stream: true,
906909
Writer: fw,
@@ -1466,6 +1469,10 @@ func hostForwardRequirements(req hostForwardRequest) routing.Requirements {
14661469
return routing.Requirements{NeedsImage: requestNeedsImage(req.Path, req.Model, hostForwardBody(req.Body))}
14671470
}
14681471

1472+
func hostAccountRequirements(req hostForwardRequest) scheduler.AccountRequirements {
1473+
return accountRequirementsForRequest(req.Path, req.Model, hostForwardBody(req.Body))
1474+
}
1475+
14691476
func hostForwardReasoningEffort(req hostForwardRequest) string {
14701477
return parseBody(hostForwardBody(req.Body), protoHeadersToHTTPHost(req.Headers).Get("Content-Type")).ReasoningEffort
14711478
}

backend/internal/plugin/quota.go

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -95,13 +95,14 @@ func (f *Forwarder) acquireClientQuota(c *gin.Context, state *forwardState) func
9595
func (f *Forwarder) pickAccount(c *gin.Context, state *forwardState, excludeIDs ...int) error {
9696
var lastErr error
9797
for _, model := range state.schedulingModelCandidates() {
98-
account, err := f.scheduler.SelectAccount(
98+
account, err := f.scheduler.SelectAccountWithRequirements(
9999
c.Request.Context(),
100100
state.requestedPlatform,
101101
model,
102102
state.keyInfo.UserID,
103103
state.keyInfo.GroupID,
104104
state.sessionID,
105+
state.accountReq,
105106
excludeIDs...,
106107
)
107108
if err == nil {

backend/internal/plugin/request.go

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@ import (
1818

1919
"github.com/DouDOU-start/airgate-core/ent"
2020
"github.com/DouDOU-start/airgate-core/internal/auth"
21+
"github.com/DouDOU-start/airgate-core/internal/scheduler"
2122
"github.com/DouDOU-start/airgate-core/internal/server/middleware"
2223
sdk "github.com/DouDOU-start/airgate-sdk/sdkgo"
2324
)
@@ -67,6 +68,7 @@ func (f *Forwarder) parseRequest(c *gin.Context) (*forwardState, bool) {
6768
realtime: parsed.Stream,
6869
sessionID: parsed.SessionID,
6970
reasoningEffort: parsed.ReasoningEffort,
71+
accountReq: accountRequirementsForRequest(path, parsed.Model, body),
7072
requestedPlatform: requestedPlatform,
7173
keyInfo: keyInfo,
7274
plugin: inst,
@@ -183,6 +185,25 @@ func requestHasImageWorkload(path, model string, body []byte) bool {
183185
return isImageAPIPath(path) || isImageModel(model) || hasImageGenerationTool(body)
184186
}
185187

188+
func accountRequirementsForRequest(path, model string, body []byte) scheduler.AccountRequirements {
189+
if isImageAPIPath(path) || isImageModel(model) {
190+
return scheduler.AccountRequirements{
191+
Workload: scheduler.WorkloadImage,
192+
ImageProtocols: []scheduler.ImageProtocol{
193+
scheduler.ImageProtocolImagesAPI,
194+
scheduler.ImageProtocolResponsesTool,
195+
},
196+
}
197+
}
198+
if hasForcedImageGenerationTool(body) {
199+
return scheduler.AccountRequirements{
200+
Workload: scheduler.WorkloadImage,
201+
ImageProtocols: []scheduler.ImageProtocol{scheduler.ImageProtocolResponsesTool},
202+
}
203+
}
204+
return scheduler.AccountRequirements{Workload: scheduler.WorkloadChat}
205+
}
206+
186207
func isImageAPIPath(path string) bool {
187208
if path == "" {
188209
return false
@@ -359,6 +380,7 @@ func buildPluginRequest(c *gin.Context, state *forwardState) *sdk.ForwardRequest
359380
if qs := c.Request.URL.RawQuery; qs != "" {
360381
headers.Set("X-Forwarded-Query", qs)
361382
}
383+
applyAccountCapabilityHeaders(headers, state.account)
362384

363385
req := &sdk.ForwardRequest{
364386
Account: buildSDKAccount(state.account),
@@ -373,6 +395,15 @@ func buildPluginRequest(c *gin.Context, state *forwardState) *sdk.ForwardRequest
373395
return req
374396
}
375397

398+
func applyAccountCapabilityHeaders(headers http.Header, acc *ent.Account) {
399+
if headers == nil || acc == nil {
400+
return
401+
}
402+
if protocols := scheduler.AccountImageProtocols(acc); len(protocols) > 0 {
403+
headers.Set("X-Airgate-Account-Image-Protocols", strings.Join(protocols, ","))
404+
}
405+
}
406+
376407
// buildHeaders 克隆请求头并附加 X-Airgate-* 系列(分组级 service_tier / 强制 instructions / 插件开关)。
377408
func buildHeaders(source http.Header, keyInfo *auth.APIKeyInfo) http.Header {
378409
headers := source.Clone()

backend/internal/plugin/request_test.go

Lines changed: 67 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,10 @@
11
package plugin
22

3-
import "testing"
3+
import (
4+
"testing"
5+
6+
"github.com/DouDOU-start/airgate-core/internal/scheduler"
7+
)
48

59
func TestRequestNeedsImage(t *testing.T) {
610
t.Parallel()
@@ -85,3 +89,65 @@ func TestRequestNeedsImage(t *testing.T) {
8589
})
8690
}
8791
}
92+
93+
func TestAccountRequirementsForRequest(t *testing.T) {
94+
t.Parallel()
95+
96+
tests := []struct {
97+
name string
98+
path string
99+
model string
100+
body []byte
101+
wantWorkload scheduler.Workload
102+
wantProtocols []scheduler.ImageProtocol
103+
}{
104+
{
105+
name: "chat request",
106+
path: "/v1/chat/completions",
107+
model: "gpt-4o",
108+
wantWorkload: scheduler.WorkloadChat,
109+
},
110+
{
111+
name: "images api accepts either image protocol",
112+
path: "/v1/images/generations",
113+
model: "gpt-image-2",
114+
wantWorkload: scheduler.WorkloadImage,
115+
wantProtocols: []scheduler.ImageProtocol{scheduler.ImageProtocolImagesAPI, scheduler.ImageProtocolResponsesTool},
116+
},
117+
{
118+
name: "forced image tool requires responses tool",
119+
path: "/v1/responses",
120+
model: "gpt-5.4",
121+
body: []byte(`{"model":"gpt-5.4","tools":[{"type":"image_generation"}],"tool_choice":"image_generation"}`),
122+
wantWorkload: scheduler.WorkloadImage,
123+
wantProtocols: []scheduler.ImageProtocol{scheduler.ImageProtocolResponsesTool},
124+
},
125+
}
126+
127+
for _, tt := range tests {
128+
tt := tt
129+
t.Run(tt.name, func(t *testing.T) {
130+
t.Parallel()
131+
132+
got := accountRequirementsForRequest(tt.path, tt.model, tt.body)
133+
if got.Workload != tt.wantWorkload {
134+
t.Fatalf("Workload = %q, want %q", got.Workload, tt.wantWorkload)
135+
}
136+
if !sameImageProtocols(got.ImageProtocols, tt.wantProtocols) {
137+
t.Fatalf("ImageProtocols = %v, want %v", got.ImageProtocols, tt.wantProtocols)
138+
}
139+
})
140+
}
141+
}
142+
143+
func sameImageProtocols(a, b []scheduler.ImageProtocol) bool {
144+
if len(a) != len(b) {
145+
return false
146+
}
147+
for i := range a {
148+
if a[i] != b[i] {
149+
return false
150+
}
151+
}
152+
return true
153+
}

backend/internal/plugin/types.go

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -6,6 +6,7 @@ import (
66
"github.com/DouDOU-start/airgate-core/ent"
77
"github.com/DouDOU-start/airgate-core/internal/auth"
88
"github.com/DouDOU-start/airgate-core/internal/routing"
9+
"github.com/DouDOU-start/airgate-core/internal/scheduler"
910
sdk "github.com/DouDOU-start/airgate-sdk/sdkgo"
1011
)
1112

@@ -29,6 +30,7 @@ type forwardState struct {
2930

3031
// 推理强度档位快照。
3132
reasoningEffort string
33+
accountReq scheduler.AccountRequirements
3234

3335
requestedPlatform string
3436
selectedRoute routing.Candidate
Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
package scheduler
2+
3+
import (
4+
"strings"
5+
6+
"github.com/DouDOU-start/airgate-core/ent"
7+
)
8+
9+
type Workload string
10+
11+
const (
12+
WorkloadChat Workload = "chat"
13+
WorkloadImage Workload = "image"
14+
)
15+
16+
type ImageProtocol string
17+
18+
const (
19+
ImageProtocolImagesAPI ImageProtocol = "images_api"
20+
ImageProtocolResponsesTool ImageProtocol = "responses_tool"
21+
)
22+
23+
type AccountRequirements struct {
24+
Workload Workload
25+
ImageProtocols []ImageProtocol
26+
}
27+
28+
func filterAccountsByRequirements(candidates []*ent.Account, req AccountRequirements) []*ent.Account {
29+
if req.Workload == "" && len(req.ImageProtocols) == 0 {
30+
return candidates
31+
}
32+
filtered := make([]*ent.Account, 0, len(candidates))
33+
for _, acc := range candidates {
34+
if accountMatchesRequirements(acc, req) {
35+
filtered = append(filtered, acc)
36+
}
37+
}
38+
return filtered
39+
}
40+
41+
func accountMatchesRequirements(acc *ent.Account, req AccountRequirements) bool {
42+
if acc == nil {
43+
return false
44+
}
45+
if req.Workload != "" && !accountAllowsWorkload(acc, req.Workload) {
46+
return false
47+
}
48+
if len(req.ImageProtocols) > 0 && !accountAllowsAnyImageProtocol(acc, req.ImageProtocols) {
49+
return false
50+
}
51+
return true
52+
}
53+
54+
func accountAllowsWorkload(acc *ent.Account, workload Workload) bool {
55+
allowed := extraStringSet(acc.Extra, "allowed_workloads")
56+
if len(allowed) == 0 {
57+
return true
58+
}
59+
_, ok := allowed[string(workload)]
60+
return ok
61+
}
62+
63+
func accountAllowsAnyImageProtocol(acc *ent.Account, required []ImageProtocol) bool {
64+
allowed := accountImageProtocolSet(acc)
65+
for _, protocol := range required {
66+
if _, ok := allowed[string(protocol)]; ok {
67+
return true
68+
}
69+
}
70+
return false
71+
}
72+
73+
func accountImageProtocolSet(acc *ent.Account) map[string]struct{} {
74+
allowed := extraStringSet(acc.Extra, "image_protocols")
75+
if len(allowed) > 0 {
76+
return allowed
77+
}
78+
if acc != nil {
79+
if strings.TrimSpace(acc.Credentials["access_token"]) != "" {
80+
return map[string]struct{}{string(ImageProtocolResponsesTool): {}}
81+
}
82+
if strings.TrimSpace(acc.Credentials["api_key"]) != "" {
83+
return map[string]struct{}{string(ImageProtocolImagesAPI): {}}
84+
}
85+
}
86+
return map[string]struct{}{
87+
string(ImageProtocolImagesAPI): {},
88+
string(ImageProtocolResponsesTool): {},
89+
}
90+
}
91+
92+
func AccountImageProtocols(acc *ent.Account) []string {
93+
set := accountImageProtocolSet(acc)
94+
if len(set) == 0 {
95+
return nil
96+
}
97+
protocols := make([]string, 0, len(set))
98+
for _, protocol := range []string{string(ImageProtocolImagesAPI), string(ImageProtocolResponsesTool)} {
99+
if _, ok := set[protocol]; ok {
100+
protocols = append(protocols, protocol)
101+
delete(set, protocol)
102+
}
103+
}
104+
for protocol := range set {
105+
protocols = append(protocols, protocol)
106+
}
107+
return protocols
108+
}
109+
110+
func extraStringSet(extra map[string]interface{}, key string) map[string]struct{} {
111+
if len(extra) == 0 {
112+
return nil
113+
}
114+
raw, ok := extra[key]
115+
if !ok || raw == nil {
116+
return nil
117+
}
118+
values := make(map[string]struct{})
119+
add := func(value string) {
120+
for _, part := range strings.Split(value, ",") {
121+
normalized := strings.ToLower(strings.TrimSpace(part))
122+
if normalized != "" {
123+
values[normalized] = struct{}{}
124+
}
125+
}
126+
}
127+
switch v := raw.(type) {
128+
case string:
129+
add(v)
130+
case []string:
131+
for _, item := range v {
132+
add(item)
133+
}
134+
case []interface{}:
135+
for _, item := range v {
136+
if s, ok := item.(string); ok {
137+
add(s)
138+
}
139+
}
140+
}
141+
if len(values) == 0 {
142+
return nil
143+
}
144+
return values
145+
}

0 commit comments

Comments
 (0)