Hello Ship — API to production

public record ShippedRequestContext(
    String requestId, String tenantId, String feature,
    String model, String promptVersion, boolean stream) {}

public void logRequestStart(ShippedRequestContext ctx, int inputTokensEst) {
  structuredLog.info("llm_request_start", Map.of(
      "request_id", ctx.requestId(),
      "tenant_id", ctx.tenantId(),
      "model", ctx.model(),
      "prompt_version", ctx.promptVersion(),
      "input_tokens_est", inputTokensEst));
}

public <T> T callWithRetry(Supplier<T> fn, int maxAttempts) {
  int attempt = 0;
  while (true) {
    try {
      return fn.get();
    } catch (RateLimitException | ServiceUnavailableException e) {
      if (++attempt >= maxAttempts) throw e;
      long delayMs = (long) (500 * Math.pow(2, attempt - 1) + Math.random() * 250);
      try { Thread.sleep(delayMs); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw e; }
    }
  }
}

// Spring AI ChatClient — stream helper
public Flux<String> streamOpenAi(List<Message> messages) {
  return chatClient.prompt(new Prompt(messages))
      .options(OpenAiChatOptions.builder().model("gpt-4o-mini").maxTokens(1024).build())
      .stream()
      .map(r -> r.getResult().getOutput().getText())
      .filter(t -> t != null && !t.isBlank());
}

import anthropic

def stream_anthropic(messages: list[dict]):
    client = anthropic.Anthropic()
    with client.messages.stream(
        model="claude-3-5-sonnet-20241022",
        max_tokens=1024,
        messages=messages,
    ) as stream:
        for text in stream.text_stream:
            yield text

// AnthropicMessageChatModel streaming via Spring AI
public Flux<String> streamAnthropic(List<Message> messages) {
  return anthropicChatModel.stream(new Prompt(messages))
      .map(r -> r.getResult().getOutput().getText());
}

import boto3, json

def stream_bedrock_converse(model_id: str, messages: list[dict]):
    client = boto3.client("bedrock-runtime")
    resp = client.converse_stream(modelId=model_id, messages=messages)
    for event in resp["stream"]:
        if "contentBlockDelta" in event:
            delta = event["contentBlockDelta"]["delta"].get("text", "")
            if delta:
                yield delta

// BedrockRuntimeAsyncClient — ConverseStreamResponseHandler
bedrockAsync.converseStream(request, new ConverseStreamResponseHandler() {
  public void onEvent(ConverseStreamOutput event) {
    event.contentBlockDelta().ifPresent(d -> sink.next(d.delta().text()));
  }
});

@PostMapping(value = "/chat/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux<ServerSentEvent<String>> chatStream(@RequestBody ChatRequest body) {
  return streamOpenAi(List.of(new UserMessage(body.message())))
      .map(token -> ServerSentEvent.builder(token).build())
      .concatWith(Flux.just(ServerSentEvent.builder("{\"type\":\"done\"}").event("done").build()))
      .onErrorResume(e -> Flux.just(
          ServerSentEvent.builder("{\"message\":\"stream_failed\"}").event("error").build()));
}

// Flux.merge with interval heartbeat
Flux<ServerSentEvent<String>> tokens = streamOpenAi(messages).map(this::dataEvent);
Flux<ServerSentEvent<String>> heartbeats = Flux.interval(Duration.ofSeconds(15))
    .map(i -> ServerSentEvent.builder("").comment("heartbeat").build());
return Flux.merge(tokens, heartbeats).takeUntilOther(tokens.ignoreElements());

public record ChatResult(String text, String routeUsed) {}

public ChatResult chatWithFallback(List<Message> messages) {
  List<Route> chain = List.of(
      new Route("openai", "gpt-4o-mini"),
      new Route("anthropic", "claude-3-5-haiku-20241022"));
  List<String> errors = new ArrayList<>();
  for (Route route : chain) {
    try {
      return new ChatResult(router.invoke(route, messages), route.label());
    } catch (Exception e) {
      errors.add(route.label() + ": " + e.getMessage());
    }
  }
  throw new ServiceUnavailableException("all fallbacks exhausted: " + errors);
}

@RestController
@RequestMapping("/")
public class ChatController {
  @GetMapping("/health")
  public Map<String, String> health() { return Map.of("status", "ok"); }

  @GetMapping("/ready")
  public Map<String, String> ready() {
    if (System.getenv("OPENAI_API_KEY") == null) throw new ServiceUnavailableException("missing key");
    return Map.of("status", "ready");
  }

  @PostMapping("/chat")
  public ChatResponse chat(@RequestBody ChatRequest req) {
    String requestId = "req_" + UUID.randomUUID().toString().substring(0, 12);
    log.info("chat_start request_id={} tenant_id={}", requestId, req.tenantId());
    var result = chatService.chatWithFallback(req.message());
    log.info("chat_done request_id={} route={}", requestId, result.routeUsed());
    return new ChatResponse(result.text(), requestId, result.routeUsed());
  }
}

# Same FastAPI shell — swap chat_with_fallback primary to Anthropic in config
MODEL_PRIMARY = os.environ.get("PRIMARY_MODEL", "claude-3-5-haiku-20241022")
PROVIDER = os.environ.get("LLM_PROVIDER", "anthropic")

// application.yml — spring.ai.model.chat=anthropic
// ChatService injects AnthropicChatModel when LLM_PROVIDER=anthropic

# ready() check — verify AWS creds + bedrock-runtime reachability
import boto3
def bedrock_ready() -> bool:
    sts = boto3.client("sts")
    sts.get_caller_identity()
    return True

// Bedrock readiness — BedrockRuntimeClient.invokeModel with minimal prompt in /ready (cache 30s)

// logback-spring.xml — LogstashEncoder or custom JSON layout
// MDC.put("request_id", requestId) in filter; clear in finally

public void postDeploySmoke(String baseUrl) {
  rest.get().uri(baseUrl + "/ready").retrieve().toBodilessEntity();
  var resp = rest.post().uri(baseUrl + "/chat")
      .body(new ChatRequest("ping", "smoke"))
      .retrieve().body(ChatResponse.class);
  assert resp != null && resp.answer() != null && !resp.answer().isBlank();
}