Add llm-gateway module: routing, failover below the tool-calling advisor, per-provider circuit breakers, dollar caps and token limits, tenant-keyed cache; real OpenAI and Anthropic models against a local fake
Co-Authored-By: Claude Sonnet 5.5 <[email protected]> Claude-Session: https://claude.ai/code/session_01JXVi2GMQ7bR5EmbUFdDj7N
This commit is contained in:
@@ -0,0 +1,17 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public class AllProvidersUnavailableException extends RuntimeException {
|
||||
|
||||
private final List<String> trail;
|
||||
|
||||
public AllProvidersUnavailableException(List<String> trail, Throwable cause) {
|
||||
super("No provider could answer: " + trail, cause);
|
||||
this.trail = List.copyOf(trail);
|
||||
}
|
||||
|
||||
public List<String> trail() {
|
||||
return trail;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,8 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
public class BudgetExceededException extends RuntimeException {
|
||||
|
||||
public BudgetExceededException(String message) {
|
||||
super(message);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,90 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.ToLongFunction;
|
||||
|
||||
/**
|
||||
* Dollar caps per tenant, in microdollars. The point of the class is the order of operations:
|
||||
* {@link #reserve} takes the money BEFORE the provider is called, and {@link Reservation#settle}
|
||||
* corrects it afterwards. Checking the balance and recording the spend as two separate steps lets
|
||||
* concurrent requests all pass the check and all spend (see {@code BudgetTest}).
|
||||
*/
|
||||
public final class Budgets {
|
||||
|
||||
/** Money held for one call. Settle with the real cost, or release if the call failed. */
|
||||
public final class Reservation {
|
||||
|
||||
private final Account account;
|
||||
|
||||
private final long held;
|
||||
|
||||
private boolean done;
|
||||
|
||||
private Reservation(Account account, long held) {
|
||||
this.account = account;
|
||||
this.held = held;
|
||||
}
|
||||
|
||||
public synchronized void settle(long actualMicros) {
|
||||
if (!done) {
|
||||
done = true;
|
||||
account.settle(held, actualMicros);
|
||||
}
|
||||
}
|
||||
|
||||
public synchronized void release() {
|
||||
if (!done) {
|
||||
done = true;
|
||||
account.settle(held, 0);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static final class Account {
|
||||
|
||||
private long spent;
|
||||
|
||||
private long held;
|
||||
|
||||
synchronized boolean reserve(long amount, long cap) {
|
||||
if (spent + held + amount > cap) {
|
||||
return false;
|
||||
}
|
||||
held += amount;
|
||||
return true;
|
||||
}
|
||||
|
||||
synchronized void settle(long heldAmount, long actual) {
|
||||
held -= heldAmount;
|
||||
spent += actual;
|
||||
}
|
||||
|
||||
synchronized long spent() {
|
||||
return spent;
|
||||
}
|
||||
}
|
||||
|
||||
private final Map<String, Account> accounts = new ConcurrentHashMap<>();
|
||||
|
||||
private final ToLongFunction<String> capMicros;
|
||||
|
||||
public Budgets(ToLongFunction<String> capMicros) {
|
||||
this.capMicros = capMicros;
|
||||
}
|
||||
|
||||
public Reservation reserve(String tenant, long estimateMicros) {
|
||||
Account a = accounts.computeIfAbsent(tenant, t -> new Account());
|
||||
long cap = capMicros.applyAsLong(tenant);
|
||||
if (!a.reserve(estimateMicros, cap)) {
|
||||
throw new BudgetExceededException("Budget for " + tenant + " would be exceeded: spent "
|
||||
+ a.spent() + " of " + cap + " microdollars, this call may cost up to " + estimateMicros);
|
||||
}
|
||||
return new Reservation(a, estimateMicros);
|
||||
}
|
||||
|
||||
public long spent(String tenant) {
|
||||
Account a = accounts.get(tenant);
|
||||
return a == null ? 0 : a.spent();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,85 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* What the model-level code needs to know about the request it is serving. The tool-calling loop
|
||||
* runs ABOVE the {@code ChatModel}, so tenant and feature cannot travel as method arguments; they
|
||||
* travel in a {@link ScopedValue}, which is bound for exactly the duration of one gateway call.
|
||||
*/
|
||||
public record CallContext(String tenant, String feature, Meter meter) {
|
||||
|
||||
public static final ScopedValue<CallContext> CURRENT = ScopedValue.newInstance();
|
||||
|
||||
public static CallContext anonymous() {
|
||||
return new CallContext("anonymous", "none", new Meter());
|
||||
}
|
||||
|
||||
/** Totals across every model call made for one gateway request, tool rounds and failovers included. */
|
||||
public static final class Meter {
|
||||
|
||||
private final List<String> trail = new ArrayList<>();
|
||||
|
||||
private long promptTokens;
|
||||
|
||||
private long completionTokens;
|
||||
|
||||
private long costMicros;
|
||||
|
||||
private int modelCalls;
|
||||
|
||||
private String provider = "";
|
||||
|
||||
private String model = "";
|
||||
|
||||
private boolean usageMissing;
|
||||
|
||||
public synchronized void trail(String entry) {
|
||||
trail.add(entry);
|
||||
}
|
||||
|
||||
public synchronized void record(Target t, long prompt, long completion, long micros, boolean missing) {
|
||||
promptTokens += prompt;
|
||||
completionTokens += completion;
|
||||
costMicros += micros;
|
||||
modelCalls++;
|
||||
provider = t.name();
|
||||
model = t.model();
|
||||
usageMissing |= missing;
|
||||
}
|
||||
|
||||
public synchronized List<String> trail() {
|
||||
return List.copyOf(trail);
|
||||
}
|
||||
|
||||
public synchronized long promptTokens() {
|
||||
return promptTokens;
|
||||
}
|
||||
|
||||
public synchronized long completionTokens() {
|
||||
return completionTokens;
|
||||
}
|
||||
|
||||
public synchronized long costMicros() {
|
||||
return costMicros;
|
||||
}
|
||||
|
||||
public synchronized int modelCalls() {
|
||||
return modelCalls;
|
||||
}
|
||||
|
||||
/** The provider that answered last. */
|
||||
public synchronized String provider() {
|
||||
return provider;
|
||||
}
|
||||
|
||||
public synchronized String model() {
|
||||
return model;
|
||||
}
|
||||
|
||||
public synchronized boolean usageMissing() {
|
||||
return usageMissing;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,138 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreaker;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
|
||||
import org.springframework.ai.chat.messages.Message;
|
||||
import org.springframework.ai.chat.metadata.Usage;
|
||||
import org.springframework.ai.chat.model.ChatModel;
|
||||
import org.springframework.ai.chat.model.ChatResponse;
|
||||
import org.springframework.ai.chat.prompt.ChatOptions;
|
||||
import org.springframework.ai.chat.prompt.Prompt;
|
||||
import org.springframework.ai.model.tool.ToolCallingChatOptions;
|
||||
|
||||
/**
|
||||
* The gateway's {@link ChatModel}: tries each target in order, one circuit breaker per target,
|
||||
* one budget reservation per attempt, and meters every successful call.
|
||||
*
|
||||
* <p>It sits BELOW the tool-calling advisor on purpose. When a provider fails in the middle of a
|
||||
* tool loop, this class re-sends the conversation (which already holds the tool results) to the
|
||||
* next target. A retry placed above the advisor would start the whole loop again and run the
|
||||
* tools a second time (see {@code ToolLoopPlacementTest}).
|
||||
*/
|
||||
public final class Failover implements ChatModel {
|
||||
|
||||
private final List<Target> targets;
|
||||
|
||||
private final CircuitBreakerRegistry breakers;
|
||||
|
||||
private final Budgets budgets;
|
||||
|
||||
public Failover(List<Target> targets, CircuitBreakerRegistry breakers, Budgets budgets) {
|
||||
this.targets = List.copyOf(targets);
|
||||
this.breakers = breakers;
|
||||
this.budgets = budgets;
|
||||
}
|
||||
|
||||
/**
|
||||
* Must be a {@link ToolCallingChatOptions}: the tool-calling advisor looks at the request's
|
||||
* options and does nothing at all when they are a plain {@code ChatOptions}.
|
||||
*/
|
||||
@Override
|
||||
public ChatOptions getOptions() {
|
||||
return ToolCallingChatOptions.builder().build();
|
||||
}
|
||||
|
||||
@Override
|
||||
public ChatResponse call(Prompt prompt) {
|
||||
CallContext ctx = CallContext.CURRENT.isBound() ? CallContext.CURRENT.get() : CallContext.anonymous();
|
||||
RuntimeException last = null;
|
||||
for (Target t : targets) {
|
||||
CircuitBreaker breaker = breakers.circuitBreaker(t.name());
|
||||
if (!breaker.tryAcquirePermission()) {
|
||||
ctx.meter().trail(t.name() + ": circuit open, not called");
|
||||
continue;
|
||||
}
|
||||
long estimate = t.price().micros(estimatePromptTokens(prompt), maxTokens(prompt, t));
|
||||
Budgets.Reservation reservation;
|
||||
try {
|
||||
reservation = budgets.reserve(ctx.tenant(), estimate);
|
||||
}
|
||||
catch (BudgetExceededException e) {
|
||||
breaker.releasePermission();
|
||||
throw e;
|
||||
}
|
||||
long started = System.nanoTime();
|
||||
try {
|
||||
ChatResponse response = t.chat().call(forTarget(prompt, t));
|
||||
breaker.onSuccess(System.nanoTime() - started, TimeUnit.NANOSECONDS);
|
||||
Usage u = response.getMetadata() == null ? null : response.getMetadata().getUsage();
|
||||
boolean missing = u == null || u.getTotalTokens() == null || u.getTotalTokens() == 0;
|
||||
long in = missing ? estimatePromptTokens(prompt) : u.getPromptTokens();
|
||||
long out = missing ? 0 : u.getCompletionTokens();
|
||||
long actual = t.price().micros(in, out);
|
||||
reservation.settle(actual);
|
||||
ctx.meter().record(t, in, out, actual, missing);
|
||||
ctx.meter().trail(t.name() + ": ok");
|
||||
return response;
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
reservation.release();
|
||||
if (Failures.isTransient(e)) {
|
||||
breaker.onError(System.nanoTime() - started, TimeUnit.NANOSECONDS, e);
|
||||
ctx.meter().trail(t.name() + ": failed with status " + Failures.status(e));
|
||||
last = e;
|
||||
continue;
|
||||
}
|
||||
breaker.releasePermission();
|
||||
ctx.meter().trail(t.name() + ": rejected with status " + Failures.status(e) + ", not retried");
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
throw new AllProvidersUnavailableException(ctx.meter().trail(), last);
|
||||
}
|
||||
|
||||
/**
|
||||
* The options for ONE target: that provider's own defaults (so the model id is its own, never
|
||||
* another vendor's), overlaid with the portable settings and the tool callbacks the caller set.
|
||||
*/
|
||||
static Prompt forTarget(Prompt prompt, Target t) {
|
||||
ChatOptions in = prompt.getOptions();
|
||||
ToolCallingChatOptions.Builder<?> b = (ToolCallingChatOptions.Builder<?>) t.chat().getOptions().mutate();
|
||||
b.model(t.model());
|
||||
if (in != null) {
|
||||
if (in.getMaxTokens() != null) {
|
||||
b.maxTokens(in.getMaxTokens());
|
||||
}
|
||||
if (in.getTemperature() != null) {
|
||||
b.temperature(in.getTemperature());
|
||||
}
|
||||
if (in.getTopP() != null) {
|
||||
b.topP(in.getTopP());
|
||||
}
|
||||
if (in.getStopSequences() != null) {
|
||||
b.stopSequences(in.getStopSequences());
|
||||
}
|
||||
if (in instanceof ToolCallingChatOptions tools) {
|
||||
b.toolCallbacks(tools.getToolCallbacks());
|
||||
b.toolContext(tools.getToolContext());
|
||||
}
|
||||
}
|
||||
return prompt.mutate().chatOptions(b.build()).build();
|
||||
}
|
||||
|
||||
static long estimatePromptTokens(Prompt prompt) {
|
||||
long chars = 0;
|
||||
for (Message m : prompt.getInstructions()) {
|
||||
chars += m.getText() == null ? 0 : m.getText().length();
|
||||
}
|
||||
return (chars + 3) / 4;
|
||||
}
|
||||
|
||||
private static long maxTokens(Prompt prompt, Target t) {
|
||||
ChatOptions in = prompt.getOptions();
|
||||
return in != null && in.getMaxTokens() != null ? in.getMaxTokens() : 1024;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,48 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.io.IOException;
|
||||
|
||||
import com.anthropic.errors.AnthropicIoException;
|
||||
import com.anthropic.errors.AnthropicServiceException;
|
||||
import com.openai.errors.OpenAIIoException;
|
||||
import com.openai.errors.OpenAIServiceException;
|
||||
|
||||
/**
|
||||
* Decides whether a failed model call is worth sending to another provider. Same idea as the
|
||||
* {@code providers} module: the vendor SDKs throw unrelated exception families, so walk the cause
|
||||
* chain for an HTTP status.
|
||||
*/
|
||||
public final class Failures {
|
||||
|
||||
private Failures() {
|
||||
}
|
||||
|
||||
public static int status(Throwable t) {
|
||||
for (Throwable c = t; c != null; c = c.getCause()) {
|
||||
if (c instanceof ProviderFailure e) {
|
||||
return e.status();
|
||||
}
|
||||
if (c instanceof OpenAIServiceException e) {
|
||||
return e.statusCode();
|
||||
}
|
||||
if (c instanceof AnthropicServiceException e) {
|
||||
return e.statusCode();
|
||||
}
|
||||
}
|
||||
return -1;
|
||||
}
|
||||
|
||||
/** True for overload, rate limit, 5xx and network failures. A 400 or 401 is the caller's problem. */
|
||||
public static boolean isTransient(Throwable t) {
|
||||
int status = status(t);
|
||||
if (status != -1) {
|
||||
return status == 408 || status == 429 || status >= 500;
|
||||
}
|
||||
for (Throwable c = t; c != null; c = c.getCause()) {
|
||||
if (c instanceof OpenAIIoException || c instanceof AnthropicIoException || c instanceof IOException) {
|
||||
return true;
|
||||
}
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,98 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.ai.chat.client.ChatClient;
|
||||
import org.springframework.ai.chat.client.advisor.ToolCallingAdvisor;
|
||||
import org.springframework.ai.chat.model.ChatResponse;
|
||||
import org.springframework.ai.support.ToolCallbacks;
|
||||
import org.springframework.ai.tool.ToolCallback;
|
||||
|
||||
/**
|
||||
* The one method every calling service uses: rate limit, cache, route, record. Each request gets
|
||||
* its own {@link Failover} over the targets for its hint, wrapped in a {@code ChatClient} whose
|
||||
* tool loop is the standard {@link ToolCallingAdvisor}.
|
||||
*/
|
||||
public final class Gateway {
|
||||
|
||||
private final Routes routes;
|
||||
|
||||
private final java.util.function.Function<List<Target>, Failover> failovers;
|
||||
|
||||
private final TokenLimiter limiter;
|
||||
|
||||
private final SemanticCache cache;
|
||||
|
||||
private final Map<String, ToolCallback[]> toolbox;
|
||||
|
||||
public Gateway(Routes routes, java.util.function.Function<List<Target>, Failover> failovers,
|
||||
TokenLimiter limiter, SemanticCache cache, Map<String, Object> toolObjects) {
|
||||
this.routes = routes;
|
||||
this.failovers = failovers;
|
||||
this.limiter = limiter;
|
||||
this.cache = cache;
|
||||
this.toolbox = new java.util.LinkedHashMap<>();
|
||||
toolObjects.forEach((name, obj) -> toolbox.put(name, ToolCallbacks.from(obj)));
|
||||
}
|
||||
|
||||
public GatewayResponse complete(String tenant, GatewayRequest req) {
|
||||
List<Target> targets = routes.get(req.modelHint());
|
||||
List<ToolCallback> callbacks = resolveTools(req.tools());
|
||||
|
||||
long estimate = (req.systemPrompt().length() + req.userMessage().length() + 3) / 4 + req.maxTokens();
|
||||
limiter.take(tenant, estimate);
|
||||
CallContext.Meter meter = new CallContext.Meter();
|
||||
try {
|
||||
var cached = cache == null ? java.util.Optional.<String>empty()
|
||||
: cache.lookup(tenant, req.featureTag(), req.userMessage());
|
||||
if (cached.isPresent()) {
|
||||
limiter.settle(tenant, estimate, 0);
|
||||
return new GatewayResponse(cached.get(), "cache", "cache", 0, 0, 0, 0, true, List.of("cache: hit"));
|
||||
}
|
||||
ChatClient client = ChatClient.builder(failovers.apply(targets))
|
||||
.defaultAdvisors(ToolCallingAdvisor.builder().build())
|
||||
.build();
|
||||
CallContext ctx = new CallContext(tenant, req.featureTag(), meter);
|
||||
ChatResponse response = ScopedValue.where(CallContext.CURRENT, ctx).call(() -> {
|
||||
var spec = client.prompt().user(req.userMessage())
|
||||
.options(org.springframework.ai.model.tool.ToolCallingChatOptions.builder()
|
||||
.maxTokens(req.maxTokens()));
|
||||
if (!req.systemPrompt().isBlank()) {
|
||||
spec = spec.system(req.systemPrompt());
|
||||
}
|
||||
if (!callbacks.isEmpty()) {
|
||||
spec = spec.toolCallbacks(callbacks);
|
||||
}
|
||||
return spec.call().chatResponse();
|
||||
});
|
||||
String content = response.getResult().getOutput().getText();
|
||||
limiter.settle(tenant, estimate, meter.promptTokens() + meter.completionTokens());
|
||||
if (cache != null) {
|
||||
cache.put(tenant, req.featureTag(), req.userMessage(), content);
|
||||
}
|
||||
return new GatewayResponse(content, meter.provider(), meter.model(), meter.promptTokens(),
|
||||
meter.completionTokens(), meter.costMicros(), meter.modelCalls(), false, meter.trail());
|
||||
}
|
||||
catch (RuntimeException e) {
|
||||
limiter.settle(tenant, estimate, meter.promptTokens() + meter.completionTokens());
|
||||
throw e;
|
||||
}
|
||||
catch (Exception e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private List<ToolCallback> resolveTools(List<String> names) {
|
||||
java.util.ArrayList<ToolCallback> out = new java.util.ArrayList<>();
|
||||
for (String n : names) {
|
||||
ToolCallback[] cb = toolbox.get(n);
|
||||
if (cb == null) {
|
||||
throw new IllegalArgumentException("Tool '" + n + "' is not on the gateway allow-list "
|
||||
+ toolbox.keySet());
|
||||
}
|
||||
out.addAll(List.of(cb));
|
||||
}
|
||||
return out;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,12 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
@SpringBootApplication
|
||||
public class GatewayApplication {
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(GatewayApplication.class, args);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,82 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.LinkedHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import io.github.bucket4j.TimeMeter;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerConfig;
|
||||
import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry;
|
||||
import org.springframework.ai.anthropic.AnthropicChatModel;
|
||||
import org.springframework.ai.anthropic.AnthropicChatOptions;
|
||||
import org.springframework.ai.chat.model.ChatModel;
|
||||
import org.springframework.ai.embedding.EmbeddingModel;
|
||||
import org.springframework.ai.openai.OpenAiChatModel;
|
||||
import org.springframework.ai.openai.OpenAiChatOptions;
|
||||
import org.springframework.beans.factory.ObjectProvider;
|
||||
import org.springframework.boot.context.properties.EnableConfigurationProperties;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
|
||||
@Configuration
|
||||
@EnableConfigurationProperties(GatewayProperties.class)
|
||||
public class GatewayConfig {
|
||||
|
||||
@Bean
|
||||
Routes routes(GatewayProperties p) {
|
||||
Map<String, Target> byName = new LinkedHashMap<>();
|
||||
for (GatewayProperties.Provider c : p.providers()) {
|
||||
byName.put(c.name(), new Target(c.name(), c.model(), chatModel(c), Price.of(c.inputPrice(), c.outputPrice())));
|
||||
}
|
||||
Map<String, List<Target>> table = new LinkedHashMap<>();
|
||||
p.routes().forEach((hint, names) -> table.put(hint, names.stream().map(byName::get).toList()));
|
||||
return new Routes(table);
|
||||
}
|
||||
|
||||
/** No retries inside the SDK: the gateway's own failover is the retry policy. */
|
||||
public static ChatModel chatModel(GatewayProperties.Provider c) {
|
||||
return switch (c.type()) {
|
||||
case "openai" -> OpenAiChatModel.builder().options(OpenAiChatOptions.builder()
|
||||
.baseUrl(c.baseUrl()).apiKey(c.apiKey()).model(c.model()).maxRetries(0).build()).build();
|
||||
case "anthropic" -> AnthropicChatModel.builder().options(AnthropicChatOptions.builder()
|
||||
.baseUrl(c.baseUrl()).apiKey(c.apiKey()).model(c.model()).maxTokens(1024).maxRetries(0).build())
|
||||
.build();
|
||||
default -> throw new IllegalArgumentException("Unknown provider type " + c.type());
|
||||
};
|
||||
}
|
||||
|
||||
@Bean
|
||||
CircuitBreakerRegistry breakers(GatewayProperties p) {
|
||||
return registry(p.breaker());
|
||||
}
|
||||
|
||||
/** One breaker per provider name, all built from the same settings. */
|
||||
public static CircuitBreakerRegistry registry(GatewayProperties.Breaker b) {
|
||||
return CircuitBreakerRegistry.of(CircuitBreakerConfig.custom()
|
||||
.slidingWindowSize(b.slidingWindowSize())
|
||||
.minimumNumberOfCalls(b.minimumCalls())
|
||||
.failureRateThreshold(b.failureRateThreshold())
|
||||
.waitDurationInOpenState(Duration.ofMillis(b.waitOpenMillis()))
|
||||
.permittedNumberOfCallsInHalfOpenState(1)
|
||||
.build());
|
||||
}
|
||||
|
||||
@Bean
|
||||
Budgets budgets(GatewayProperties p) {
|
||||
return new Budgets(t -> p.tenant(t).budgetMicros());
|
||||
}
|
||||
|
||||
@Bean
|
||||
TokenLimiter tokenLimiter(GatewayProperties p) {
|
||||
return new TokenLimiter(t -> p.tenant(t).tokensPerHour(), TimeMeter.SYSTEM_MILLISECONDS);
|
||||
}
|
||||
|
||||
@Bean
|
||||
Gateway gateway(Routes routes, CircuitBreakerRegistry breakers, Budgets budgets, TokenLimiter limiter,
|
||||
ObjectProvider<EmbeddingModel> embeddings) {
|
||||
EmbeddingModel em = embeddings.getIfAvailable();
|
||||
return new Gateway(routes, targets -> new Failover(targets, breakers, budgets), limiter,
|
||||
em == null ? null : new SemanticCache(em, 0.92), Map.of("refund_order", new OrderTools()));
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,45 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.web.bind.annotation.ExceptionHandler;
|
||||
import org.springframework.web.bind.annotation.PostMapping;
|
||||
import org.springframework.web.bind.annotation.RequestBody;
|
||||
import org.springframework.web.bind.annotation.RequestHeader;
|
||||
import org.springframework.web.bind.annotation.RequestMapping;
|
||||
import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
/**
|
||||
* Deploy inside the private network only. The tenant comes from a header set by the mesh or the
|
||||
* API gateway upstream; the request body has no tenant field to spoof.
|
||||
*/
|
||||
@RestController
|
||||
@RequestMapping("/v1/gateway")
|
||||
public class GatewayController {
|
||||
|
||||
private final Gateway gateway;
|
||||
|
||||
GatewayController(Gateway gateway) {
|
||||
this.gateway = gateway;
|
||||
}
|
||||
|
||||
@PostMapping("/complete")
|
||||
GatewayResponse complete(@RequestHeader("X-Tenant-Id") String tenant, @RequestBody GatewayRequest request) {
|
||||
return gateway.complete(tenant, request);
|
||||
}
|
||||
|
||||
@ExceptionHandler(BudgetExceededException.class)
|
||||
ResponseEntity<String> overBudget(BudgetExceededException e) {
|
||||
return ResponseEntity.status(HttpStatus.TOO_MANY_REQUESTS).body(e.getMessage());
|
||||
}
|
||||
|
||||
@ExceptionHandler(AllProvidersUnavailableException.class)
|
||||
ResponseEntity<String> down(AllProvidersUnavailableException e) {
|
||||
return ResponseEntity.status(HttpStatus.SERVICE_UNAVAILABLE).header("Retry-After", "30").body(e.getMessage());
|
||||
}
|
||||
|
||||
@ExceptionHandler({UnknownHintException.class, IllegalArgumentException.class})
|
||||
ResponseEntity<String> badRequest(RuntimeException e) {
|
||||
return ResponseEntity.badRequest().body(e.getMessage());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
import org.springframework.boot.context.properties.ConfigurationProperties;
|
||||
|
||||
@ConfigurationProperties("gateway")
|
||||
public record GatewayProperties(List<Provider> providers, Map<String, List<String>> routes,
|
||||
Map<String, Tenant> tenants, Tenant defaults, Breaker breaker) {
|
||||
|
||||
/** {@code type} is {@code openai} or {@code anthropic}; prices are dollars per million tokens. */
|
||||
public record Provider(String name, String type, String baseUrl, String apiKey, String model,
|
||||
String inputPrice, String outputPrice) {
|
||||
}
|
||||
|
||||
public record Tenant(long budgetMicros, long tokensPerHour) {
|
||||
}
|
||||
|
||||
public record Breaker(int slidingWindowSize, int minimumCalls, int failureRateThreshold, long waitOpenMillis) {
|
||||
}
|
||||
|
||||
public Tenant tenant(String id) {
|
||||
return tenants != null && tenants.containsKey(id) ? tenants.get(id) : defaults;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,26 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* What a calling service sends. The tenant is NOT here: it comes from a header set upstream.
|
||||
* {@code tools} are names from the gateway's own allow-list, never code from the caller.
|
||||
*/
|
||||
public record GatewayRequest(String featureTag, String systemPrompt, String userMessage, String modelHint,
|
||||
Integer maxTokens, List<String> tools) {
|
||||
|
||||
public GatewayRequest {
|
||||
if (modelHint == null) {
|
||||
modelHint = "smart";
|
||||
}
|
||||
if (maxTokens == null || maxTokens <= 0) {
|
||||
maxTokens = 1024;
|
||||
}
|
||||
if (tools == null) {
|
||||
tools = List.of();
|
||||
}
|
||||
if (systemPrompt == null) {
|
||||
systemPrompt = "";
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
public record GatewayResponse(String content, String provider, String model, long promptTokens,
|
||||
long completionTokens, long costMicros, int modelCalls, boolean servedFromCache, List<String> trail) {
|
||||
}
|
||||
@@ -0,0 +1,17 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import org.springframework.ai.tool.annotation.Tool;
|
||||
import org.springframework.ai.tool.annotation.ToolParam;
|
||||
|
||||
/** A tool with a side effect that can be counted: every execution is one more refund issued. */
|
||||
public class OrderTools {
|
||||
|
||||
public final AtomicInteger executions = new AtomicInteger();
|
||||
|
||||
@Tool(name = "refund_order", description = "Refunds an order and returns the refund id")
|
||||
public String refund(@ToolParam(description = "The order id") String orderId) {
|
||||
return "refund-" + orderId + "-" + executions.incrementAndGet();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.math.BigDecimal;
|
||||
import java.math.RoundingMode;
|
||||
|
||||
/**
|
||||
* Dollars per million tokens. One dollar per million tokens is exactly one microdollar per token,
|
||||
* so the arithmetic stays in whole microdollars and never touches a double.
|
||||
*/
|
||||
public record Price(BigDecimal inputPerMillion, BigDecimal outputPerMillion) {
|
||||
|
||||
public static Price of(String input, String output) {
|
||||
return new Price(new BigDecimal(input), new BigDecimal(output));
|
||||
}
|
||||
|
||||
/** Cost in microdollars, rounded up so a cap is never undercounted. */
|
||||
public long micros(long inputTokens, long outputTokens) {
|
||||
return inputPerMillion.multiply(BigDecimal.valueOf(inputTokens))
|
||||
.add(outputPerMillion.multiply(BigDecimal.valueOf(outputTokens)))
|
||||
.setScale(0, RoundingMode.CEILING).longValueExact();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,16 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
/** A provider call that failed with an HTTP status. Stand-in for the vendor SDK exceptions in tests. */
|
||||
public class ProviderFailure extends RuntimeException {
|
||||
|
||||
private final int status;
|
||||
|
||||
public ProviderFailure(int status, String message) {
|
||||
super(message);
|
||||
this.status = status;
|
||||
}
|
||||
|
||||
public int status() {
|
||||
return status;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/** Maps a model hint to an ordered list of targets. An unknown hint is an error, not a default. */
|
||||
public final class Routes {
|
||||
|
||||
private final Map<String, List<Target>> table;
|
||||
|
||||
public Routes(Map<String, List<Target>> table) {
|
||||
this.table = Map.copyOf(table);
|
||||
}
|
||||
|
||||
public List<Target> get(String hint) {
|
||||
List<Target> t = table.get(hint);
|
||||
if (t == null) {
|
||||
throw new UnknownHintException(hint, table.keySet());
|
||||
}
|
||||
return t;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,59 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Optional;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.concurrent.CopyOnWriteArrayList;
|
||||
|
||||
import org.springframework.ai.embedding.EmbeddingModel;
|
||||
|
||||
/**
|
||||
* Serves a stored answer when a new question embeds close enough to an old one. The key includes
|
||||
* the tenant: a cache shared across tenants answers tenant B with tenant A's data.
|
||||
*/
|
||||
public final class SemanticCache {
|
||||
|
||||
private record Entry(float[] vector, String answer) {
|
||||
}
|
||||
|
||||
private final EmbeddingModel embeddings;
|
||||
|
||||
private final double threshold;
|
||||
|
||||
private final Map<String, List<Entry>> byKey = new ConcurrentHashMap<>();
|
||||
|
||||
public SemanticCache(EmbeddingModel embeddings, double threshold) {
|
||||
this.embeddings = embeddings;
|
||||
this.threshold = threshold;
|
||||
}
|
||||
|
||||
public Optional<String> lookup(String tenant, String feature, String question) {
|
||||
float[] q = embeddings.embed(question);
|
||||
Entry best = null;
|
||||
double bestScore = -1;
|
||||
for (Entry e : byKey.getOrDefault(tenant + "|" + feature, List.of())) {
|
||||
double s = cosine(q, e.vector());
|
||||
if (s >= threshold && s > bestScore) {
|
||||
best = e;
|
||||
bestScore = s;
|
||||
}
|
||||
}
|
||||
return Optional.ofNullable(best).map(Entry::answer);
|
||||
}
|
||||
|
||||
public void put(String tenant, String feature, String question, String answer) {
|
||||
byKey.computeIfAbsent(tenant + "|" + feature, k -> new CopyOnWriteArrayList<>())
|
||||
.add(new Entry(embeddings.embed(question), answer));
|
||||
}
|
||||
|
||||
static double cosine(float[] a, float[] b) {
|
||||
double dot = 0, na = 0, nb = 0;
|
||||
for (int i = 0; i < a.length; i++) {
|
||||
dot += a[i] * b[i];
|
||||
na += a[i] * a[i];
|
||||
nb += b[i] * b[i];
|
||||
}
|
||||
return dot / (Math.sqrt(na) * Math.sqrt(nb) + 1e-10);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,7 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import org.springframework.ai.chat.model.ChatModel;
|
||||
|
||||
/** One provider the gateway can send a call to: a name for logs and breakers, a model, a price. */
|
||||
public record Target(String name, String model, ChatModel chat, Price price) {
|
||||
}
|
||||
@@ -0,0 +1,60 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ConcurrentHashMap;
|
||||
import java.util.function.ToLongFunction;
|
||||
|
||||
import io.github.bucket4j.Bandwidth;
|
||||
import io.github.bucket4j.Bucket;
|
||||
import io.github.bucket4j.TimeMeter;
|
||||
|
||||
/**
|
||||
* Tokens-per-hour per tenant (Bucket4j, in memory). The unit is LLM tokens, not requests, because
|
||||
* one summarisation can cost fifty lookups. Pre-consume an estimate, then settle with the real
|
||||
* total: refund the surplus, or charge the shortfall. The old post refunded against prompt tokens
|
||||
* only and never charged the completion, so every answer was free of the limit.
|
||||
*/
|
||||
public final class TokenLimiter {
|
||||
|
||||
private final Map<String, Bucket> buckets = new ConcurrentHashMap<>();
|
||||
|
||||
private final ToLongFunction<String> perHour;
|
||||
|
||||
private final TimeMeter clock;
|
||||
|
||||
public TokenLimiter(ToLongFunction<String> perHour, TimeMeter clock) {
|
||||
this.perHour = perHour;
|
||||
this.clock = clock;
|
||||
}
|
||||
|
||||
private Bucket bucket(String tenant) {
|
||||
return buckets.computeIfAbsent(tenant, t -> {
|
||||
long cap = perHour.applyAsLong(t);
|
||||
return Bucket.builder().withCustomTimePrecision(clock)
|
||||
.addLimit(Bandwidth.builder().capacity(cap).refillGreedy(cap, Duration.ofHours(1)).build())
|
||||
.build();
|
||||
});
|
||||
}
|
||||
|
||||
public void take(String tenant, long estimate) {
|
||||
if (!bucket(tenant).tryConsume(estimate)) {
|
||||
throw new BudgetExceededException("Token rate limit reached for " + tenant);
|
||||
}
|
||||
}
|
||||
|
||||
/** Correct the estimate once the real token count is known. */
|
||||
public void settle(String tenant, long estimate, long actual) {
|
||||
Bucket b = bucket(tenant);
|
||||
if (actual < estimate) {
|
||||
b.addTokens(estimate - actual);
|
||||
}
|
||||
else if (actual > estimate) {
|
||||
b.consumeIgnoringRateLimits(actual - estimate);
|
||||
}
|
||||
}
|
||||
|
||||
public long available(String tenant) {
|
||||
return bucket(tenant).getAvailableTokens();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,11 @@
|
||||
package com.ankurm.gateway;
|
||||
|
||||
import java.util.Set;
|
||||
import java.util.TreeSet;
|
||||
|
||||
public class UnknownHintException extends RuntimeException {
|
||||
|
||||
public UnknownHintException(String hint, Set<String> known) {
|
||||
super("Unknown model hint '" + hint + "'. Known hints: " + new TreeSet<>(known));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user