327 lines
12 KiB
Go
327 lines
12 KiB
Go
package openai
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"net/http"
|
|
"strings"
|
|
"unicode/utf8"
|
|
|
|
edgeservice "iop/apps/edge/internal/service"
|
|
"iop/packages/go/config"
|
|
)
|
|
|
|
type anthropicClientError struct {
|
|
errorType string
|
|
message string
|
|
}
|
|
|
|
func (e *anthropicClientError) Error() string { return e.message }
|
|
|
|
func newAnthropicClientError(errorType string, err error) error {
|
|
if err == nil {
|
|
return nil
|
|
}
|
|
return &anthropicClientError{errorType: errorType, message: err.Error()}
|
|
}
|
|
|
|
func (s *Server) handleAnthropicMessages(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeAnthropicError(w, http.StatusMethodNotAllowed, "invalid_request_error", "method not allowed")
|
|
return
|
|
}
|
|
defer r.Body.Close()
|
|
if err := validateAnthropicHeaders(r, false); err != nil {
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
body, err := readOpenAIIngressBody(w, r, s.maxIngressSnapshotBytes())
|
|
if err != nil {
|
|
writeAnthropicIngressError(w, err)
|
|
return
|
|
}
|
|
envelope, err := decodeAnthropicEnvelope(body)
|
|
if err != nil {
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
dispatch, ok := s.resolveRouteDispatch(envelope.Model)
|
|
if !ok || !dispatch.ProviderPool {
|
|
writeAnthropicError(w, http.StatusBadRequest, "not_supported_error", "model does not resolve to a protocol profile")
|
|
return
|
|
}
|
|
|
|
needsTools := anthropicRequestNeedsTools(body)
|
|
poolReq := s.anthropicPoolRequest(r, dispatch, envelope, body, config.OperationMessages, needsTools)
|
|
result, err := s.service.SubmitProviderPool(r.Context(), poolReq)
|
|
if err != nil {
|
|
s.writeAnthropicDispatchError(w, err)
|
|
return
|
|
}
|
|
if result == nil || result.Tunnel == nil || result.Path != edgeservice.ProviderPoolPathTunnel {
|
|
writeAnthropicError(w, http.StatusBadGateway, "api_error", "selected provider did not return a tunnel")
|
|
return
|
|
}
|
|
defer result.Tunnel.Close()
|
|
|
|
switch result.DispatchInfo.ProfileDriver {
|
|
case string(config.ProtocolDriverAnthropicMessages):
|
|
s.writeAnthropicNativeTunnelResponse(w, r, result.Tunnel)
|
|
case string(config.ProtocolDriverOpenAIChat):
|
|
s.writeAnthropicChatBridgeResponse(w, r, result.Tunnel, envelope)
|
|
default:
|
|
writeAnthropicError(w, http.StatusBadGateway, "api_error", "selected provider returned an unsupported protocol driver")
|
|
}
|
|
}
|
|
|
|
func (s *Server) handleAnthropicCountTokens(w http.ResponseWriter, r *http.Request) {
|
|
if r.Method != http.MethodPost {
|
|
writeAnthropicError(w, http.StatusMethodNotAllowed, "invalid_request_error", "method not allowed")
|
|
return
|
|
}
|
|
defer r.Body.Close()
|
|
if err := validateAnthropicHeaders(r, false); err != nil {
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
body, err := readOpenAIIngressBody(w, r, s.maxIngressSnapshotBytes())
|
|
if err != nil {
|
|
writeAnthropicIngressError(w, err)
|
|
return
|
|
}
|
|
envelope, err := decodeAnthropicEnvelope(body)
|
|
if err != nil {
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", err.Error())
|
|
return
|
|
}
|
|
dispatch, ok := s.resolveRouteDispatch(envelope.Model)
|
|
if !ok || !dispatch.ProviderPool {
|
|
writeAnthropicError(w, http.StatusBadRequest, "not_supported_error", "model does not resolve to a protocol profile")
|
|
return
|
|
}
|
|
|
|
if entry := s.findProviderPoolEntry(envelope.Model); entry != nil && entry.TokenCounter != nil {
|
|
req, decodeErr := decodeAnthropicMessageRequest(body, false)
|
|
if decodeErr != nil {
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", decodeErr.Error())
|
|
return
|
|
}
|
|
count, countErr := countAnthropicInputTokens(req, *entry.TokenCounter)
|
|
if countErr != nil {
|
|
writeAnthropicError(w, http.StatusBadRequest, "not_supported_error", countErr.Error())
|
|
return
|
|
}
|
|
writeJSON(w, http.StatusOK, anthropicCountTokensResponse{InputTokens: count})
|
|
return
|
|
}
|
|
|
|
poolReq := s.anthropicPoolRequest(r, dispatch, envelope, body, config.OperationCountTokens, false)
|
|
result, err := s.service.SubmitProviderPool(r.Context(), poolReq)
|
|
if err != nil {
|
|
s.writeAnthropicDispatchError(w, err)
|
|
return
|
|
}
|
|
if result == nil || result.Tunnel == nil || result.Path != edgeservice.ProviderPoolPathTunnel ||
|
|
result.DispatchInfo.ProfileDriver != string(config.ProtocolDriverAnthropicMessages) {
|
|
writeAnthropicError(w, http.StatusBadGateway, "api_error", "selected provider did not return a native count-tokens tunnel")
|
|
return
|
|
}
|
|
defer result.Tunnel.Close()
|
|
s.writeAnthropicNativeTunnelResponse(w, r, result.Tunnel)
|
|
}
|
|
|
|
func (s *Server) anthropicPoolRequest(
|
|
r *http.Request,
|
|
dispatch routeDispatch,
|
|
envelope anthropicRequestEnvelope,
|
|
body []byte,
|
|
operation config.ProtocolOperation,
|
|
needsTools bool,
|
|
) edgeservice.ProviderPoolDispatchRequest {
|
|
metadata := principalMetadata(r.Context())
|
|
if metadata == nil {
|
|
metadata = make(map[string]string)
|
|
}
|
|
metadata["anthropic_model"] = envelope.Model
|
|
metadata["anthropic_stream"] = fmt.Sprintf("%t", envelope.Stream)
|
|
estimate := estimateInputTokensBytes(body, metadata, nil, nil)
|
|
contextClass := classifyContext(estimate, s.longContextThreshold())
|
|
|
|
poolReq := edgeservice.ProviderPoolDispatchRequest{
|
|
Run: edgeservice.SubmitRunRequest{
|
|
NodeRef: dispatch.NodeRef, ModelGroupKey: strings.TrimSpace(envelope.Model),
|
|
ProviderID: dispatch.ProviderID, UsageAttribution: dispatch.UsageAttribution,
|
|
SessionID: dispatch.SessionID, TimeoutSec: dispatch.TimeoutSec,
|
|
MaxQueue: dispatch.MaxQueue, QueueTimeoutMS: dispatch.QueueTimeoutMS,
|
|
Metadata: metadata, EstimatedInputTokens: estimate, ContextClass: contextClass,
|
|
ProviderPool: true,
|
|
},
|
|
Tunnel: edgeservice.SubmitProviderTunnelRequest{
|
|
ModelGroupKey: strings.TrimSpace(envelope.Model), ProviderID: dispatch.ProviderID,
|
|
UsageAttribution: dispatch.UsageAttribution, SessionID: dispatch.SessionID,
|
|
Method: http.MethodPost, Path: r.URL.Path, Stream: envelope.Stream,
|
|
TimeoutSec: dispatch.TimeoutSec, MaxQueue: dispatch.MaxQueue,
|
|
QueueTimeoutMS: dispatch.QueueTimeoutMS, Metadata: metadata,
|
|
EstimatedInputTokens: estimate, ContextClass: contextClass, ProviderPool: true,
|
|
},
|
|
}
|
|
poolReq.AcceptCandidate = anthropicCandidatePredicate(operation, envelope.Stream, needsTools)
|
|
poolReq.PrepareProtocolTunnel = func(tunnelReq edgeservice.SubmitProviderTunnelRequest, selected edgeservice.ProviderPoolCandidate) (edgeservice.SubmitProviderTunnelRequest, error) {
|
|
if selected.ProtocolProfile == nil {
|
|
return tunnelReq, newAnthropicClientError("not_supported_error", fmt.Errorf("selected provider has no concrete protocol profile"))
|
|
}
|
|
profile := selected.ProtocolProfile.Clone()
|
|
headers, err := s.anthropicUpstreamHeaders(r, profile, profile.Driver == config.ProtocolDriverAnthropicMessages)
|
|
if err != nil {
|
|
return tunnelReq, newAnthropicClientError("invalid_request_error", err)
|
|
}
|
|
tunnelReq.Headers = headers
|
|
switch profile.Driver {
|
|
case config.ProtocolDriverAnthropicMessages:
|
|
tunnelReq.Operation = string(operation)
|
|
tunnelReq.BuildBody = func(target string) ([]byte, error) {
|
|
return rewriteResponsesModel(body, target)
|
|
}
|
|
case config.ProtocolDriverOpenAIChat:
|
|
if operation != config.OperationMessages {
|
|
return tunnelReq, newAnthropicClientError("not_supported_error", fmt.Errorf("selected Chat profile has no native count-tokens operation"))
|
|
}
|
|
if err := validateAnthropicHeaders(r, true); err != nil {
|
|
return tunnelReq, newAnthropicClientError("invalid_request_error", err)
|
|
}
|
|
bridged, _, err := prepareAnthropicChatBridge(body, selected.ActualModel, profile)
|
|
if err != nil {
|
|
return tunnelReq, newAnthropicClientError("invalid_request_error", err)
|
|
}
|
|
tunnelReq.Operation = string(config.OperationChatCompletions)
|
|
tunnelReq.Body = bridged
|
|
tunnelReq.BuildBody = nil
|
|
default:
|
|
return tunnelReq, newAnthropicClientError("not_supported_error", fmt.Errorf("unsupported protocol driver %q", profile.Driver))
|
|
}
|
|
return tunnelReq, nil
|
|
}
|
|
return poolReq
|
|
}
|
|
|
|
func anthropicCandidatePredicate(operation config.ProtocolOperation, stream, needsTools bool) edgeservice.ProviderPoolCandidatePredicate {
|
|
return func(candidate edgeservice.ProviderPoolCandidate) bool {
|
|
profile := candidate.ProtocolProfile
|
|
if profile == nil || candidate.ExecutionPath != string(edgeservice.ProviderPoolPathTunnel) {
|
|
return false
|
|
}
|
|
if stream && !profile.HasCapability("streaming") {
|
|
return false
|
|
}
|
|
if needsTools && !profile.HasCapability("tool_calling") {
|
|
return false
|
|
}
|
|
switch operation {
|
|
case config.OperationCountTokens:
|
|
return profile.Driver == config.ProtocolDriverAnthropicMessages &&
|
|
profile.HasCapability("count_tokens") && profileHasOperation(*profile, config.OperationCountTokens)
|
|
case config.OperationMessages:
|
|
if profile.Driver == config.ProtocolDriverAnthropicMessages {
|
|
return profile.HasCapability("messages") && profileHasOperation(*profile, config.OperationMessages)
|
|
}
|
|
return profile.Driver == config.ProtocolDriverOpenAIChat &&
|
|
profile.HasCapability("chat") && profileHasOperation(*profile, config.OperationChatCompletions)
|
|
default:
|
|
return false
|
|
}
|
|
}
|
|
}
|
|
|
|
func profileHasOperation(profile config.ConcreteProtocolProfile, operation config.ProtocolOperation) bool {
|
|
_, ok := profile.Operations[string(operation)]
|
|
return ok
|
|
}
|
|
|
|
func (s *Server) anthropicUpstreamHeaders(r *http.Request, profile config.ConcreteProtocolProfile, native bool) (map[string]string, error) {
|
|
headers := map[string]string{"Content-Type": "application/json"}
|
|
if accept := strings.TrimSpace(r.Header.Get("Accept")); accept != "" {
|
|
headers["Accept"] = accept
|
|
}
|
|
if native {
|
|
headers[anthropicVersionHeader] = strings.TrimSpace(r.Header.Get(anthropicVersionHeader))
|
|
betas, err := anthropicBetaValues(r.Header.Values(anthropicBetaHeader))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if len(betas) > 0 {
|
|
headers[anthropicBetaHeader] = strings.Join(betas, ",")
|
|
}
|
|
}
|
|
credentialHeaders, err := s.providerTunnelAuthHeaders(r)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, value := range credentialHeaders {
|
|
credential := strings.TrimSpace(value)
|
|
if fields := strings.Fields(credential); len(fields) > 1 {
|
|
credential = strings.Join(fields[1:], " ")
|
|
}
|
|
if scheme := strings.TrimSpace(profile.Auth.Scheme); scheme != "" {
|
|
credential = scheme + " " + credential
|
|
}
|
|
if credential != "" {
|
|
headers[http.CanonicalHeaderKey(profile.Auth.Header)] = credential
|
|
}
|
|
break
|
|
}
|
|
return headers, nil
|
|
}
|
|
|
|
func (s *Server) writeAnthropicDispatchError(w http.ResponseWriter, err error) {
|
|
var clientErr *anthropicClientError
|
|
if errors.As(err, &clientErr) {
|
|
writeAnthropicError(w, http.StatusBadRequest, clientErr.errorType, clientErr.message)
|
|
return
|
|
}
|
|
if errors.Is(err, edgeservice.ErrProviderPoolCandidateRejected) {
|
|
writeAnthropicError(w, http.StatusBadRequest, "not_supported_error", "no provider profile supports the requested Messages operation")
|
|
return
|
|
}
|
|
if errors.Is(err, errProviderAuthRequired) {
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", "provider auth token is required")
|
|
return
|
|
}
|
|
writeAnthropicError(w, http.StatusBadGateway, "api_error", "provider dispatch failed")
|
|
}
|
|
|
|
func writeAnthropicIngressError(w http.ResponseWriter, err error) {
|
|
if errors.Is(err, errOpenAIIngressTooLarge) {
|
|
writeAnthropicError(w, http.StatusRequestEntityTooLarge, "invalid_request_error", "request body is too large")
|
|
return
|
|
}
|
|
writeAnthropicError(w, http.StatusBadRequest, "invalid_request_error", "request body could not be read")
|
|
}
|
|
|
|
func countAnthropicInputTokens(req anthropicMessageRequest, counter config.TokenCounterConf) (int, error) {
|
|
payload := map[string]any{"messages": req.Messages}
|
|
if len(req.System) > 0 {
|
|
payload["system"] = json.RawMessage(req.System)
|
|
}
|
|
if len(req.Tools) > 0 {
|
|
payload["tools"] = req.Tools
|
|
}
|
|
encoded, err := json.Marshal(payload)
|
|
if err != nil {
|
|
return 0, fmt.Errorf("encode token counter input: %w", err)
|
|
}
|
|
runes := utf8.RuneCount(encoded)
|
|
var count int
|
|
switch counter.Mode {
|
|
case config.TokenCounterDeterministic:
|
|
count = (runes + 3) / 4
|
|
case config.TokenCounterEstimate:
|
|
count = (runes*counter.Per1kInput + 999) / 1000
|
|
default:
|
|
return 0, fmt.Errorf("unsupported token counter mode %q", counter.Mode)
|
|
}
|
|
if count < 1 {
|
|
count = 1
|
|
}
|
|
return count, nil
|
|
}
|