227 lines
11 KiB
Go
227 lines
11 KiB
Go
package main
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"io"
|
|
"net/http"
|
|
"net/url"
|
|
"time"
|
|
|
|
"git.cin.su/m/Assistant/adapters/models"
|
|
"git.cin.su/m/Assistant/contracts/capability"
|
|
"git.cin.su/m/Assistant/contracts/cognitive"
|
|
"git.cin.su/m/Assistant/platform/ids"
|
|
)
|
|
|
|
const creatorAdvisoryCapabilityID capability.CapabilityID = "creator-advisory"
|
|
|
|
type bindingResolver interface {
|
|
Resolve(context.Context, capability.CognitiveRequirement) (capability.ResolvedCognitiveBinding, error)
|
|
Observe(context.Context, capability.CognitiveUsageObservation) error
|
|
RegisterDiscovery(context.Context, capability.CognitiveDiscoverySnapshot) (capability.CognitiveDiscoveryRegistration, error)
|
|
}
|
|
|
|
type providerRegistration struct {
|
|
ProviderInstanceID capability.ProviderInstanceID
|
|
Kind capability.ProviderKind
|
|
EndpointRef capability.EndpointBindingRef
|
|
CredentialRef capability.CredentialRef
|
|
Provider advisoryProvider
|
|
Discovery models.LocalProvider
|
|
}
|
|
|
|
type cognitiveGateway struct {
|
|
resolver bindingResolver
|
|
providers map[capability.ProviderKind][]providerRegistration
|
|
now func() time.Time
|
|
allowCanary bool
|
|
canaryMaxOutputBytes int
|
|
}
|
|
|
|
func (g cognitiveGateway) policyHandler(w http.ResponseWriter, r *http.Request) {
|
|
var request cognitive.GatewayAdvisoryPolicyRequest
|
|
if !decodeJSON(w, r, 8<<10, &request) || request.PrincipalID == "" {
|
|
return
|
|
}
|
|
requirement := capability.CognitiveRequirement{CapabilityID: creatorAdvisoryCapabilityID, OwnerPrincipalID: request.PrincipalID, DataPolicy: request.DataPolicy, RequiredModalities: []capability.ModelModality{capability.ModalityText}, RequiredCapabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, AllowCanary: g.allowCanary, MaxOutputBytes: 1}
|
|
selected, err := g.resolver.Resolve(r.Context(), requirement)
|
|
policy := cognitive.GatewayAdvisoryPolicy{}
|
|
if err == nil && selected.Binding.DataPolicy != capability.DataPolicyLocalOnly {
|
|
policy.ExternalEgress = true
|
|
policy.AcknowledgementRequired = true
|
|
policy.Disclosure = "Текст вопроса будет передан внешнему когнитивному провайдеру; секреты и внутренний контекст не передаются."
|
|
}
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(policy)
|
|
}
|
|
|
|
func (g cognitiveGateway) advisoryHandler(w http.ResponseWriter, r *http.Request) {
|
|
var request cognitive.GatewayAdvisoryRequest
|
|
if !decodeJSON(w, r, 80<<10, &request) {
|
|
return
|
|
}
|
|
if request.Version != cognitive.CreatorAdvisoryV1 || request.RequestID == "" || request.ActorID == "" || request.PrincipalID == "" || request.ScopeID == "" || request.Message == "" || request.MaxOutputBytes < 1 {
|
|
http.Error(w, "invalid advisory request", http.StatusBadRequest)
|
|
return
|
|
}
|
|
maxOutputBytes := request.MaxOutputBytes
|
|
if g.allowCanary && g.canaryMaxOutputBytes > 0 && maxOutputBytes > g.canaryMaxOutputBytes {
|
|
maxOutputBytes = g.canaryMaxOutputBytes
|
|
}
|
|
requirement := capability.CognitiveRequirement{CapabilityID: creatorAdvisoryCapabilityID, OwnerPrincipalID: request.PrincipalID, DataPolicy: request.DataPolicy, RequiredModalities: []capability.ModelModality{capability.ModalityText}, RequiredCapabilities: []capability.ModelCapability{capability.CapabilityAdvisory}, AllowCanary: g.allowCanary, InputBytes: int64(len(request.Message)), MaxOutputBytes: int64(maxOutputBytes)}
|
|
selected, err := g.resolver.Resolve(r.Context(), requirement)
|
|
if err != nil {
|
|
writeGatewayResponse(w, cognitive.GatewayAdvisoryResponse{Version: request.Version, RequestID: request.RequestID, FinishReason: cognitive.FinishProviderUnavailable})
|
|
return
|
|
}
|
|
if selected.Binding.DataPolicy != capability.DataPolicyLocalOnly && !request.EgressAcknowledged {
|
|
http.Error(w, "external egress acknowledgement required", http.StatusBadRequest)
|
|
return
|
|
}
|
|
provider := g.provider(selected)
|
|
if provider == nil {
|
|
writeGatewayResponse(w, cognitive.GatewayAdvisoryResponse{Version: request.Version, RequestID: request.RequestID, FinishReason: cognitive.FinishProviderUnavailable})
|
|
return
|
|
}
|
|
invocation := cognitive.AdvisoryRequest{Version: request.Version, RequestID: request.RequestID, ActorID: request.ActorID, PrincipalID: request.PrincipalID, ScopeID: request.ScopeID, Message: request.Message, Context: request.Context, Locale: request.Locale, MaxOutputBytes: maxOutputBytes, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, ProviderInstanceID: selected.Provider.ID, ModelDescriptorID: selected.Model.ID, ProviderModelID: selected.Model.ProviderModelID}
|
|
started := g.now()
|
|
response, invokeErr := provider.Advise(r.Context(), invocation)
|
|
if invokeErr != nil {
|
|
response = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: selected.Binding.ID, ProviderBindingRevision: selected.Binding.Revision, FinishReason: cognitive.FinishInvalidProviderResponse}
|
|
}
|
|
g.observe(r.Context(), selected, response, started)
|
|
if fallbackEligible(response.FinishReason) {
|
|
requirement.ExcludedBindingIDs = append(requirement.ExcludedBindingIDs, selected.Binding.ID)
|
|
fallback, resolveErr := g.resolver.Resolve(r.Context(), requirement)
|
|
fallbackProvider := g.provider(fallback)
|
|
if resolveErr == nil && fallbackProvider != nil && (fallback.Binding.DataPolicy == capability.DataPolicyLocalOnly || request.EgressAcknowledged) {
|
|
fallbackRequest := invocation
|
|
fallbackRequest.ProviderBindingID, fallbackRequest.ProviderBindingRevision = fallback.Binding.ID, fallback.Binding.Revision
|
|
fallbackRequest.ProviderInstanceID, fallbackRequest.ModelDescriptorID, fallbackRequest.ProviderModelID = fallback.Provider.ID, fallback.Model.ID, fallback.Model.ProviderModelID
|
|
fallbackStarted := g.now()
|
|
fallbackResponse, fallbackErr := fallbackProvider.Advise(r.Context(), fallbackRequest)
|
|
if fallbackErr != nil {
|
|
fallbackResponse = cognitive.AdvisoryResponse{Version: request.Version, RequestID: request.RequestID, ProviderBindingID: fallback.Binding.ID, ProviderBindingRevision: fallback.Binding.Revision, FinishReason: cognitive.FinishInvalidProviderResponse}
|
|
}
|
|
g.observe(r.Context(), fallback, fallbackResponse, fallbackStarted)
|
|
response = fallbackResponse
|
|
}
|
|
}
|
|
writeGatewayResponse(w, cognitive.GatewayAdvisoryResponse{Version: response.Version, RequestID: response.RequestID, AdvisoryText: response.AdvisoryText, ReferencedExcerptIDs: response.ReferencedExcerptIDs, FinishReason: response.FinishReason, Usage: response.Usage})
|
|
}
|
|
|
|
func (g cognitiveGateway) observe(ctx context.Context, selected capability.ResolvedCognitiveBinding, response cognitive.AdvisoryResponse, started time.Time) {
|
|
usage := response.Usage
|
|
observationID, idErr := ids.New()
|
|
if idErr == nil {
|
|
observation := capability.CognitiveUsageObservation{ObservationID: observationID, RequestID: ids.ID(response.RequestID), BindingID: selected.Binding.ID, BindingRevision: selected.Binding.Revision, ProviderInstanceID: selected.Provider.ID, ModelDescriptorID: selected.Model.ID, ObservedAtUnixMillis: g.now().UnixMilli(), LatencyMillis: g.now().Sub(started).Milliseconds(), ResultClass: capability.CognitiveResultClass(response.FinishReason)}
|
|
if usage != nil {
|
|
observation.InputUnits, observation.OutputUnits = usage.InputUnits, usage.OutputUnits
|
|
}
|
|
if response.FinishReason != cognitive.FinishCompleted {
|
|
observation.FailureType = capability.CognitiveFailureType(response.FinishReason)
|
|
}
|
|
if selected.Provider.Kind == "openai-compatible-chat-v1" || selected.Provider.Kind == "ollama-v1" {
|
|
zero := int64(0)
|
|
observation.ActualCostMicros = &zero
|
|
}
|
|
_ = g.resolver.Observe(ctx, observation)
|
|
}
|
|
}
|
|
|
|
func fallbackEligible(reason cognitive.FinishReason) bool {
|
|
switch reason {
|
|
case cognitive.FinishProviderUnavailable, cognitive.FinishModelUnavailable, cognitive.FinishTimeout, cognitive.FinishDegraded:
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (g cognitiveGateway) provider(selected capability.ResolvedCognitiveBinding) advisoryProvider {
|
|
for _, registration := range g.providers[selected.Provider.Kind] {
|
|
if registration.EndpointRef == selected.Provider.EndpointRef && registration.CredentialRef == selected.Provider.CredentialRef {
|
|
return registration.Provider
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func writeGatewayResponse(w http.ResponseWriter, response cognitive.GatewayAdvisoryResponse) {
|
|
w.Header().Set("Content-Type", "application/json")
|
|
_ = json.NewEncoder(w).Encode(response)
|
|
}
|
|
|
|
func decodeJSON(w http.ResponseWriter, r *http.Request, limit int64, destination any) bool {
|
|
r.Body = http.MaxBytesReader(w, r.Body, limit)
|
|
decoder := json.NewDecoder(r.Body)
|
|
decoder.DisallowUnknownFields()
|
|
if err := decoder.Decode(destination); err != nil {
|
|
http.Error(w, "invalid request", http.StatusBadRequest)
|
|
return false
|
|
}
|
|
if err := decoder.Decode(&struct{}{}); err != io.EOF {
|
|
http.Error(w, "invalid request", http.StatusBadRequest)
|
|
return false
|
|
}
|
|
return true
|
|
}
|
|
|
|
type httpBindingResolver struct {
|
|
base *url.URL
|
|
client *http.Client
|
|
}
|
|
|
|
func newHTTPBindingResolver(raw string, client *http.Client) (*httpBindingResolver, error) {
|
|
base, err := url.Parse(raw)
|
|
if err != nil || base.Scheme == "" || base.Host == "" {
|
|
return nil, fmt.Errorf("invalid Core Runtime URL")
|
|
}
|
|
return &httpBindingResolver{base: base, client: client}, nil
|
|
}
|
|
|
|
func (r *httpBindingResolver) Resolve(ctx context.Context, value capability.CognitiveRequirement) (capability.ResolvedCognitiveBinding, error) {
|
|
var result capability.ResolvedCognitiveBinding
|
|
err := r.post(ctx, "/v1/cognitive-bindings/resolve", value, &result)
|
|
return result, err
|
|
}
|
|
|
|
func (r *httpBindingResolver) Observe(ctx context.Context, value capability.CognitiveUsageObservation) error {
|
|
return r.post(ctx, "/v1/cognitive-usage-observations", value, nil)
|
|
}
|
|
|
|
func (r *httpBindingResolver) RegisterDiscovery(ctx context.Context, value capability.CognitiveDiscoverySnapshot) (capability.CognitiveDiscoveryRegistration, error) {
|
|
var result capability.CognitiveDiscoveryRegistration
|
|
err := r.post(ctx, "/v1/cognitive-registry/discoveries", value, &result)
|
|
return result, err
|
|
}
|
|
|
|
func (r *httpBindingResolver) post(ctx context.Context, path string, value, destination any) error {
|
|
payload, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
target := *r.base
|
|
target.Path = path
|
|
request, err := http.NewRequestWithContext(ctx, http.MethodPost, target.String(), bytes.NewReader(payload))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
request.Header.Set("Content-Type", "application/json")
|
|
response, err := r.client.Do(request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer response.Body.Close()
|
|
if response.StatusCode < 200 || response.StatusCode >= 300 {
|
|
_, _ = io.Copy(io.Discard, io.LimitReader(response.Body, 4096))
|
|
return fmt.Errorf("registry status %d", response.StatusCode)
|
|
}
|
|
if destination != nil {
|
|
return json.NewDecoder(io.LimitReader(response.Body, 64<<10)).Decode(destination)
|
|
}
|
|
return nil
|
|
}
|