internal/pluginhost/adapters_usage_translation.go
Rama main · réplica de lectura del repositorio original.
package pluginhost
import (
"bytes"
"context"
"fmt"
"runtime/debug"
"strings"
"sync"
"github.com/router-for-me/CLIProxyAPI/v8/internal/logging"
"github.com/router-for-me/CLIProxyAPI/v8/internal/registry"
"github.com/router-for-me/CLIProxyAPI/v8/internal/thinking"
coreusage "github.com/router-for-me/CLIProxyAPI/v8/sdk/cliproxy/usage"
"github.com/router-for-me/CLIProxyAPI/v8/sdk/pluginapi"
sdktranslator "github.com/router-for-me/CLIProxyAPI/v8/sdk/translator"
log "github.com/sirupsen/logrus"
)
func (h *Host) RegisterUsagePlugins() {
if h == nil {
return
}
for _, record := range h.activeRecords() {
plugin := record.plugin.Capabilities.UsagePlugin
if plugin == nil || h.isPluginFused(record.id) {
continue
}
coreusage.RegisterNamedPlugin("plugin:"+record.id, &usageAdapter{
host: h,
pluginID: record.id,
plugin: plugin,
})
}
}
func (h *Host) refreshThinkingProviders(records []capabilityRecord) {
thinking.ClearPluginProviders()
if h == nil {
return
}
for _, record := range records {
applier := record.plugin.Capabilities.ThinkingApplier
if applier == nil || h.isPluginFused(record.id) {
continue
}
provider, okProvider := h.callThinkingIdentifier(record, applier)
if !okProvider {
continue
}
thinking.RegisterPluginProvider(record.id, provider, record.priority, &thinkingAdapter{
host: h,
pluginID: record.id,
path: record.path,
version: record.version,
provider: provider,
applier: applier,
})
}
}
func (h *Host) callThinkingIdentifier(record capabilityRecord, applier pluginapi.ThinkingApplier) (provider string, ok bool) {
if h == nil || applier == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) {
return "", false
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(record.id, "ThinkingApplier.Identifier", recovered)
provider = ""
ok = false
}
}()
provider = strings.ToLower(strings.TrimSpace(applier.Identifier()))
if provider == "" {
return "", false
}
return provider, true
}
func (h *Host) currentUsagePlugin(pluginID string) pluginapi.UsagePlugin {
if h == nil || strings.TrimSpace(pluginID) == "" {
return nil
}
for _, record := range h.activeRecords() {
if record.id != pluginID {
continue
}
if h.isPluginFused(record.id) {
return nil
}
return record.plugin.Capabilities.UsagePlugin
}
return nil
}
func (h *Host) fusePlugin(id, method string, recovered any) {
if h == nil {
return
}
h.mu.Lock()
h.fused[id] = fmt.Sprintf("%s panic: %v", method, recovered)
h.mu.Unlock()
thinking.UnregisterPluginProviders(id)
log.WithField("plugin_id", id).WithField("method", method).Errorf("pluginhost: plugin panic recovered: %v\n%s", recovered, debug.Stack())
}
func (h *Host) isPluginFused(id string) bool {
if h == nil {
return false
}
h.mu.Lock()
_, fused := h.fused[id]
h.mu.Unlock()
return fused
}
type usageAdapter struct {
host *Host
pluginID string
plugin pluginapi.UsagePlugin
}
type thinkingAdapter struct {
host *Host
pluginID string
path string
version string
provider string
applier pluginapi.ThinkingApplier
}
const (
defaultMaxPendingPluginUsageRecords = 4096
maxPluginUsageReplayAttempts = 2
)
type pendingPluginUsageRecord struct {
ctx context.Context
record pluginapi.UsageRecord
attempts int
}
type pluginUsageState struct {
mu sync.Mutex
cond *sync.Cond
inFlight int
quiescing bool
replaying bool
asyncDraining bool
closed bool
quiesceWaiters int
waitingProducers int
pending []pendingPluginUsageRecord
}
func (h *Host) usageStateForPlugin(pluginID string) *pluginUsageState {
pluginID = strings.TrimSpace(pluginID)
if h == nil || pluginID == "" {
return nil
}
h.mu.Lock()
defer h.mu.Unlock()
if h.usageStates == nil {
h.usageStates = make(map[string]*pluginUsageState)
}
state := h.usageStates[pluginID]
if state == nil {
state = &pluginUsageState{}
state.cond = sync.NewCond(&state.mu)
h.usageStates[pluginID] = state
}
return state
}
func (h *Host) pendingUsageLimit() int {
if h == nil {
return defaultMaxPendingPluginUsageRecords
}
h.mu.Lock()
limit := h.maxPendingUsageRecords
h.mu.Unlock()
if limit <= 0 {
return defaultMaxPendingPluginUsageRecords
}
return limit
}
func contextDone(ctx context.Context) bool {
if ctx == nil {
return false
}
select {
case <-ctx.Done():
return true
default:
return false
}
}
func (h *Host) beginUsageReplacement(ctx context.Context, pluginID string) bool {
state := h.usageStateForPlugin(pluginID)
if state == nil {
return true
}
if ctx == nil {
ctx = context.Background()
}
h.mu.Lock()
waitHook := h.usageDrainWaitHook
h.mu.Unlock()
state.mu.Lock()
state.closed = false
state.quiescing = true
if state.inFlight == 0 && !state.asyncDraining {
state.mu.Unlock()
return !contextDone(ctx)
}
stopWake := context.AfterFunc(ctx, func() {
state.mu.Lock()
state.cond.Broadcast()
state.mu.Unlock()
})
defer stopWake()
notifiedWait := false
for (state.inFlight > 0 || state.asyncDraining) && !contextDone(ctx) {
if !notifiedWait && waitHook != nil {
notifiedWait = true
waitHook(pluginID)
}
state.cond.Wait()
}
ok := state.inFlight == 0 && !state.asyncDraining && !contextDone(ctx)
state.mu.Unlock()
return ok
}
func (h *Host) finishUsageReplacement(pluginID string) {
state := h.usageStateForPlugin(pluginID)
if state == nil || h.pluginNeedsUsageRecovery(pluginID) {
return
}
state.mu.Lock()
if !state.quiescing {
state.mu.Unlock()
return
}
state.quiescing = false
state.replaying = true
remainingSync := len(state.pending) + state.quiesceWaiters
for remainingSync > 0 {
if state.closed {
state.pending = nil
state.quiescing = false
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
if len(state.pending) == 0 {
if state.quiesceWaiters > 0 {
state.cond.Broadcast()
state.cond.Wait()
continue
}
break
}
plugin := h.currentUsagePlugin(pluginID)
if plugin == nil {
if h.pluginNeedsUsageRecovery(pluginID) {
state.quiescing = true
} else {
state.pending = nil
}
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
item := state.pending[0]
state.pending[0] = pendingPluginUsageRecord{}
state.pending = state.pending[1:]
remainingSync--
state.cond.Broadcast()
state.inFlight++
state.mu.Unlock()
errDeliver := h.invokeUsagePlugin(item.ctx, pluginID, plugin, item.record)
state.mu.Lock()
state.inFlight--
if state.inFlight == 0 {
state.cond.Broadcast()
}
if errDeliver != nil && !state.closed {
if state.quiescing || h.pluginNeedsUsageRecovery(pluginID) {
state.pending = append([]pendingPluginUsageRecord{item}, state.pending...)
state.quiescing = true
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
item.attempts++
if item.attempts < maxPluginUsageReplayAttempts && h.currentUsagePlugin(pluginID) != nil {
state.pending = append([]pendingPluginUsageRecord{item}, state.pending...)
break
}
}
}
if state.closed {
state.pending = nil
state.quiescing = false
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
if len(state.pending) == 0 && state.waitingProducers == 0 {
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
if !state.asyncDraining {
state.asyncDraining = true
go h.drainPostReplayUsage(pluginID, state)
}
state.mu.Unlock()
}
func (h *Host) drainPostReplayUsage(pluginID string, state *pluginUsageState) {
if h == nil || state == nil {
return
}
state.mu.Lock()
for {
if state.closed || state.quiescing {
state.asyncDraining = false
if state.closed {
state.pending = nil
state.replaying = false
}
state.cond.Broadcast()
state.mu.Unlock()
return
}
if len(state.pending) == 0 {
if state.waitingProducers > 0 {
state.cond.Broadcast()
state.cond.Wait()
continue
}
state.replaying = false
state.asyncDraining = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
plugin := h.currentUsagePlugin(pluginID)
if plugin == nil {
if h.pluginNeedsUsageRecovery(pluginID) {
state.quiescing = true
} else {
state.pending = nil
}
state.replaying = false
state.asyncDraining = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
item := state.pending[0]
state.pending[0] = pendingPluginUsageRecord{}
state.pending = state.pending[1:]
state.cond.Broadcast()
state.inFlight++
state.mu.Unlock()
errDeliver := h.invokeUsagePlugin(item.ctx, pluginID, plugin, item.record)
state.mu.Lock()
state.inFlight--
if state.inFlight == 0 {
state.cond.Broadcast()
}
if errDeliver != nil && !state.closed {
if state.quiescing || h.pluginNeedsUsageRecovery(pluginID) {
state.pending = append([]pendingPluginUsageRecord{item}, state.pending...)
state.quiescing = true
state.replaying = false
state.asyncDraining = false
state.cond.Broadcast()
state.mu.Unlock()
return
}
item.attempts++
if item.attempts < maxPluginUsageReplayAttempts && h.currentUsagePlugin(pluginID) != nil {
state.pending = append([]pendingPluginUsageRecord{item}, state.pending...)
}
}
}
}
func (h *Host) clearUsageReplacement(pluginID string) {
pluginID = strings.TrimSpace(pluginID)
if h == nil || pluginID == "" {
return
}
h.mu.Lock()
state := h.usageStates[pluginID]
delete(h.usageStates, pluginID)
h.mu.Unlock()
if state == nil {
return
}
state.mu.Lock()
state.closed = true
state.pending = nil
state.quiescing = false
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
}
func (h *Host) clearAllUsageReplacements() {
if h == nil {
return
}
h.mu.Lock()
states := make([]*pluginUsageState, 0, len(h.usageStates))
for _, state := range h.usageStates {
if state != nil {
states = append(states, state)
}
}
h.usageStates = make(map[string]*pluginUsageState)
h.mu.Unlock()
for _, state := range states {
state.mu.Lock()
state.closed = true
state.pending = nil
state.quiescing = false
state.replaying = false
state.cond.Broadcast()
state.mu.Unlock()
}
}
func (h *Host) dispatchPluginUsageRecord(ctx context.Context, pluginID string, record pluginapi.UsageRecord) {
state := h.usageStateForPlugin(pluginID)
if state == nil {
return
}
if ctx == nil {
ctx = context.Background()
} else {
ctx = context.WithoutCancel(ctx)
}
limit := h.pendingUsageLimit()
h.mu.Lock()
backpressureHook := h.usageBackpressureWaitHook
h.mu.Unlock()
state.mu.Lock()
if state.closed {
state.mu.Unlock()
return
}
if state.quiescing || state.replaying || len(state.pending) > 0 {
quiesceWaiter := state.quiescing
if quiesceWaiter {
state.quiesceWaiters++
}
state.waitingProducers++
notifiedBackpressure := false
for !state.closed && (state.quiescing || state.replaying || len(state.pending) > 0) && len(state.pending) >= limit {
if !notifiedBackpressure && backpressureHook != nil {
notifiedBackpressure = true
backpressureHook(pluginID)
}
state.cond.Wait()
}
if quiesceWaiter {
state.quiesceWaiters--
}
state.waitingProducers--
if state.closed {
state.cond.Broadcast()
state.mu.Unlock()
return
}
if state.quiescing || state.replaying || len(state.pending) > 0 {
state.pending = append(state.pending, pendingPluginUsageRecord{
ctx: ctx,
record: record,
})
state.cond.Broadcast()
state.mu.Unlock()
return
}
state.cond.Broadcast()
}
plugin := h.currentUsagePlugin(pluginID)
if plugin == nil {
state.mu.Unlock()
return
}
state.inFlight++
state.mu.Unlock()
var errDeliver error
defer func() {
state.mu.Lock()
if errDeliver != nil && !state.closed && (state.quiescing || state.replaying) {
state.pending = append([]pendingPluginUsageRecord{{
ctx: ctx,
record: record,
}}, state.pending...)
}
state.inFlight--
if state.inFlight == 0 {
state.cond.Broadcast()
}
state.mu.Unlock()
}()
errDeliver = h.invokeUsagePlugin(ctx, pluginID, plugin, record)
}
func (h *Host) invokeUsagePlugin(ctx context.Context, pluginID string, plugin pluginapi.UsagePlugin, record pluginapi.UsageRecord) (err error) {
if h == nil || plugin == nil {
return nil
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(pluginID, "UsagePlugin.HandleUsage", recovered)
err = nil
}
}()
if rpcAdapter, ok := plugin.(*rpcPluginAdapter); ok {
return rpcAdapter.deliverUsage(ctx, record)
}
plugin.HandleUsage(ctx, record)
return nil
}
func (a *usageAdapter) HandleUsage(ctx context.Context, record coreusage.Record) {
if a == nil || a.host == nil {
return
}
if ctx == nil {
ctx = context.Background()
} else {
ctx = context.WithoutCancel(ctx)
}
sessionID := strings.TrimSpace(record.SessionID)
parentSessionID := strings.TrimSpace(record.ParentSessionID)
if sessionID == "" {
clientMeta := logging.GetClientRequestMetadata(ctx)
sessionID = strings.TrimSpace(clientMeta.SessionID)
parentSessionID = strings.TrimSpace(clientMeta.ParentSessionID)
} else if parentSessionID == "" {
clientMeta := logging.GetClientRequestMetadata(ctx)
if sessionID == strings.TrimSpace(clientMeta.SessionID) {
parentSessionID = strings.TrimSpace(clientMeta.ParentSessionID)
}
}
if sessionID == "" || sessionID == parentSessionID {
parentSessionID = ""
}
requestID := strings.TrimSpace(record.RequestID)
traceID := strings.TrimSpace(record.TraceID)
if traceID == "" {
traceID = strings.TrimSpace(logging.GetRequestID(ctx))
}
selectionID := strings.TrimSpace(record.SelectionID)
if selectionID == "" {
selectionID = strings.TrimSpace(coreusage.SelectionIDFromContext(ctx))
}
accessProvider := strings.TrimSpace(record.AccessProvider)
if accessProvider == "" {
accessProvider = strings.TrimSpace(coreusage.AccessProviderFromContext(ctx))
}
isNativeKey := record.IsNativeKey
if !isNativeKey {
if nativeExplicit, ok := coreusage.IsNativeKeyFromContext(ctx); ok {
isNativeKey = nativeExplicit
} else if accessProvider != "" {
isNativeKey = coreusage.IsNativeAccessProvider(accessProvider)
}
}
apiKey := strings.TrimSpace(record.APIKey)
if apiKey == "" {
apiKey = strings.TrimSpace(coreusage.APIKeyFromContext(ctx))
}
a.host.dispatchPluginUsageRecord(ctx, a.pluginID, pluginapi.UsageRecord{
RequestID: requestID,
TraceID: traceID,
SelectionID: selectionID,
Provider: record.Provider,
BaseURL: record.BaseURL,
ExecutorType: record.ExecutorType,
Model: record.Model,
Alias: record.Alias,
APIKey: apiKey,
IsNativeKey: isNativeKey,
AccessProvider: accessProvider,
SessionID: sessionID,
ParentSessionID: parentSessionID,
AuthID: record.AuthID,
AuthIndex: record.AuthIndex,
AuthType: record.AuthType,
Source: record.Source,
ReasoningEffort: record.ReasoningEffort,
ServiceTier: record.ServiceTier,
EffectiveServiceTier: record.EffectiveServiceTier,
ResponseServiceTier: record.ResponseServiceTier,
ResponseModel: record.ResponseModel,
Generate: coreusage.GenerateEnabled(record.Generate),
Stream: record.Stream,
RequestedAt: record.RequestedAt,
Latency: record.Latency,
TTFT: record.TTFT,
Failed: record.Failed,
Failure: pluginapi.UsageFailure{
StatusCode: record.Fail.StatusCode,
Body: record.Fail.Body,
},
Detail: pluginapi.UsageDetail{
InputTokens: record.Detail.InputTokens,
OutputTokens: record.Detail.OutputTokens,
ReasoningTokens: record.Detail.ReasoningTokens,
CachedTokens: record.Detail.CachedTokens,
CacheReadTokens: record.Detail.CacheReadTokens,
CacheCreationTokens: record.Detail.CacheCreationTokens,
TotalTokens: record.Detail.TotalTokens,
TextInputTokens: coreusage.CloneInt64Ptr(record.Detail.TextInputTokens),
ImageInputTokens: coreusage.CloneInt64Ptr(record.Detail.ImageInputTokens),
AudioInputTokens: coreusage.CloneInt64Ptr(record.Detail.AudioInputTokens),
VideoInputTokens: coreusage.CloneInt64Ptr(record.Detail.VideoInputTokens),
FileInputTokens: coreusage.CloneInt64Ptr(record.Detail.FileInputTokens),
CachedTextTokens: coreusage.CloneInt64Ptr(record.Detail.CachedTextTokens),
CachedImageTokens: coreusage.CloneInt64Ptr(record.Detail.CachedImageTokens),
CachedAudioTokens: coreusage.CloneInt64Ptr(record.Detail.CachedAudioTokens),
TextOutputTokens: coreusage.CloneInt64Ptr(record.Detail.TextOutputTokens),
ImageOutputTokens: coreusage.CloneInt64Ptr(record.Detail.ImageOutputTokens),
AudioOutputTokens: coreusage.CloneInt64Ptr(record.Detail.AudioOutputTokens),
},
ResponseHeaders: cloneHeader(record.ResponseHeaders),
})
}
func (a *thinkingAdapter) Apply(body []byte, config thinking.ThinkingConfig, modelInfo *registry.ModelInfo) (out []byte, err error) {
if a == nil || a.applier == nil || a.host == nil || a.host.isPluginFused(a.pluginID) || !a.host.pluginIdentityCurrent(a.pluginID, a.path, a.version) {
return bytes.Clone(body), nil
}
defer func() {
if recovered := recover(); recovered != nil {
a.host.fusePlugin(a.pluginID, "ThinkingApplier.ApplyThinking", recovered)
out = bytes.Clone(body)
err = nil
}
}()
resp, errApply := a.applier.ApplyThinking(context.Background(), pluginapi.ThinkingApplyRequest{
Provider: a.provider,
Model: registryModelInfoToPluginModelInfo(modelInfo),
Config: pluginapi.ThinkingConfig{
Mode: config.Mode.String(),
Budget: config.Budget,
Level: string(config.Level),
},
Body: bytes.Clone(body),
})
if errApply != nil || len(resp.Body) == 0 {
return bytes.Clone(body), nil
}
return bytes.Clone(resp.Body), nil
}
func (h *Host) NormalizeRequest(ctx context.Context, from, to sdktranslator.Format, model string, body []byte, stream bool) []byte {
current := bytes.Clone(body)
for _, record := range h.activeRecords() {
if h.isPluginFused(record.id) || record.plugin.Capabilities.RequestNormalizer == nil {
continue
}
if normalized, ok := h.callRequestNormalizer(ctx, record, from, to, model, current, stream); ok {
current = normalized
}
}
return current
}
func (h *Host) TranslateRequest(ctx context.Context, from, to sdktranslator.Format, model string, body []byte, stream bool) ([]byte, bool) {
for _, record := range h.activeRecords() {
if h.isPluginFused(record.id) || record.plugin.Capabilities.RequestTranslator == nil {
continue
}
if translated, ok := h.callRequestTranslator(ctx, record, from, to, model, body, stream); ok {
return translated, true
}
}
return bytes.Clone(body), false
}
func (h *Host) NormalizeResponseBefore(ctx context.Context, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) []byte {
current := bytes.Clone(body)
for _, record := range h.activeRecords() {
normalizer := record.plugin.Capabilities.ResponseBeforeTranslator
if h.isPluginFused(record.id) || normalizer == nil {
continue
}
if normalized, ok := h.callResponseNormalizer(ctx, record, "ResponseBeforeTranslator.NormalizeResponse", normalizer, from, to, model, originalRequestRawJSON, requestRawJSON, current, stream); ok {
current = normalized
}
}
return current
}
func (h *Host) TranslateResponse(ctx context.Context, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) ([]byte, bool) {
for _, record := range h.activeRecords() {
translator := record.plugin.Capabilities.ResponseTranslator
if h.isPluginFused(record.id) || translator == nil {
continue
}
if translated, ok := h.callResponseTranslator(ctx, record, translator, from, to, model, originalRequestRawJSON, requestRawJSON, body, stream); ok {
return translated, true
}
}
return bytes.Clone(body), false
}
func (h *Host) NormalizeResponseAfter(ctx context.Context, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) []byte {
current := bytes.Clone(body)
for _, record := range h.activeRecords() {
normalizer := record.plugin.Capabilities.ResponseAfterTranslator
if h.isPluginFused(record.id) || normalizer == nil {
continue
}
if normalized, ok := h.callResponseNormalizer(ctx, record, "ResponseAfterTranslator.NormalizeResponse", normalizer, from, to, model, originalRequestRawJSON, requestRawJSON, current, stream); ok {
current = normalized
}
}
return current
}
func (h *Host) callRequestNormalizer(ctx context.Context, record capabilityRecord, from, to sdktranslator.Format, model string, body []byte, stream bool) (out []byte, ok bool) {
if h == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) || record.plugin.Capabilities.RequestNormalizer == nil {
return nil, false
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(record.id, "RequestNormalizer.NormalizeRequest", recovered)
out = nil
ok = false
}
}()
resp, errNormalizeRequest := record.plugin.Capabilities.RequestNormalizer.NormalizeRequest(ctx, pluginapi.RequestTransformRequest{
FromFormat: from.String(),
ToFormat: to.String(),
Model: model,
Stream: stream,
Body: bytes.Clone(body),
})
if errNormalizeRequest != nil || len(resp.Body) == 0 {
return nil, false
}
return bytes.Clone(resp.Body), true
}
func (h *Host) callRequestTranslator(ctx context.Context, record capabilityRecord, from, to sdktranslator.Format, model string, body []byte, stream bool) (out []byte, ok bool) {
if h == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) || record.plugin.Capabilities.RequestTranslator == nil {
return nil, false
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(record.id, "RequestTranslator.TranslateRequest", recovered)
out = nil
ok = false
}
}()
resp, errTranslateRequest := record.plugin.Capabilities.RequestTranslator.TranslateRequest(ctx, pluginapi.RequestTransformRequest{
FromFormat: from.String(),
ToFormat: to.String(),
Model: model,
Stream: stream,
Body: bytes.Clone(body),
})
if errTranslateRequest != nil || len(resp.Body) == 0 {
return nil, false
}
return bytes.Clone(resp.Body), true
}
func (h *Host) callResponseNormalizer(ctx context.Context, record capabilityRecord, method string, normalizer pluginapi.ResponseNormalizer, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) (out []byte, ok bool) {
if h == nil || normalizer == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) {
return nil, false
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(record.id, method, recovered)
out = nil
ok = false
}
}()
resp, errNormalizeResponse := normalizer.NormalizeResponse(ctx, pluginapi.ResponseTransformRequest{
FromFormat: from.String(),
ToFormat: to.String(),
Model: model,
Stream: stream,
OriginalRequest: bytes.Clone(originalRequestRawJSON),
TranslatedRequest: bytes.Clone(requestRawJSON),
Body: bytes.Clone(body),
})
if errNormalizeResponse != nil || len(resp.Body) == 0 {
return nil, false
}
return bytes.Clone(resp.Body), true
}
func (h *Host) callResponseTranslator(ctx context.Context, record capabilityRecord, translator pluginapi.ResponseTranslator, from, to sdktranslator.Format, model string, originalRequestRawJSON, requestRawJSON, body []byte, stream bool) (out []byte, ok bool) {
if h == nil || translator == nil || h.isPluginFused(record.id) || !h.recordCurrent(record) {
return nil, false
}
defer func() {
if recovered := recover(); recovered != nil {
h.fusePlugin(record.id, "ResponseTranslator.TranslateResponse", recovered)
out = nil
ok = false
}
}()
resp, errTranslateResponse := translator.TranslateResponse(ctx, pluginapi.ResponseTransformRequest{
FromFormat: from.String(),
ToFormat: to.String(),
Model: model,
Stream: stream,
OriginalRequest: bytes.Clone(originalRequestRawJSON),
TranslatedRequest: bytes.Clone(requestRawJSON),
Body: bytes.Clone(body),
})
if errTranslateResponse != nil || len(resp.Body) == 0 {
return nil, false
}
return bytes.Clone(resp.Body), true
}