277 lines
11 KiB
Go
277 lines
11 KiB
Go
package llm
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"sync"
|
|
"time"
|
|
|
|
"gitea.maximumdirect.net/eric/audita/v3/internal/core/config"
|
|
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/outcomes"
|
|
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/responseschema"
|
|
"gitea.maximumdirect.net/eric/audita/v3/internal/framework/structuredoutput"
|
|
"gitea.maximumdirect.net/eric/audita/v3/internal/prompts"
|
|
pk "go.maximumdirect.net/promptkit"
|
|
)
|
|
|
|
// CallHooks receive one logical start and finish record and the execution
|
|
// events for this call. Call records can contain transcript content.
|
|
type CallHooks struct {
|
|
OnCall func(structuredoutput.CallRecord)
|
|
OnEvent pk.ExecutionObserver
|
|
OnEventWithCall func(uint64, structuredoutput.DiagnosticContext, pk.ExecutionEvent)
|
|
}
|
|
|
|
type RoleClient struct {
|
|
runtime *Runtime
|
|
role string
|
|
profileID string
|
|
settings config.LLM
|
|
hooks CallHooks
|
|
credentialMu sync.Mutex
|
|
credentialNames map[string]struct{}
|
|
}
|
|
|
|
var _ structuredoutput.Client = (*RoleClient)(nil)
|
|
|
|
func (r *Runtime) NewRoleClient(role, profileID string, settings config.LLM, hooks CallHooks) (*RoleClient, error) {
|
|
if r == nil || r.Engine == nil {
|
|
return nil, fmt.Errorf("LLM engine is unavailable")
|
|
}
|
|
if profileID == "" {
|
|
return nil, fmt.Errorf("profile selection is required")
|
|
}
|
|
return &RoleClient{runtime: r, role: role, profileID: profileID, settings: settings, hooks: hooks}, nil
|
|
}
|
|
|
|
func (c *RoleClient) CompleteStructured(parent context.Context, req structuredoutput.Request, out any) (response structuredoutput.Response, err error) {
|
|
ctx, cancel := context.WithTimeout(parent, c.settings.OperationTimeout)
|
|
defer cancel()
|
|
var terminalMu sync.Mutex
|
|
var terminal pk.ExecutionEvent
|
|
terminalAvailable := false
|
|
defer func() {
|
|
terminalMu.Lock()
|
|
if terminalAvailable {
|
|
response.Execution = outcomes.ExecutionCounts{
|
|
Available: true,
|
|
ProviderAttempts: int(terminal.Attempts),
|
|
TransientRetries: int(terminal.TransientRetries),
|
|
StructuralRepairs: int(terminal.RepairCalls),
|
|
RetriesExhausted: terminal.RetriesExhausted,
|
|
}
|
|
if terminal.StatusCode != 0 {
|
|
status := terminal.StatusCode
|
|
response.Execution.HTTPStatus = &status
|
|
}
|
|
}
|
|
terminalMu.Unlock()
|
|
}()
|
|
record := structuredoutput.CallRecord{CallID: c.runtime.nextCallID.Add(1), Context: copyContext(req.Context), Started: time.Now().UTC()}
|
|
record.Context.Role = c.role
|
|
record.Prompt = structuredoutput.PromptMetadata{PromptID: req.PromptID, ProfileID: c.profileID}
|
|
if c.hooks.OnCall != nil {
|
|
c.hooks.OnCall(record)
|
|
}
|
|
defer func() {
|
|
record.Finished = time.Now().UTC()
|
|
if err != nil {
|
|
var failure *structuredoutput.Failure
|
|
if errors.As(err, &failure) && failure.Reason == outcomes.ReasonInvalidOutputExhausted {
|
|
record.Failure = structuredoutput.FailureMalformedOutput
|
|
} else {
|
|
record.Failure = structuredoutput.FailureRuntime
|
|
}
|
|
}
|
|
if c.hooks.OnCall != nil {
|
|
c.hooks.OnCall(record)
|
|
}
|
|
}()
|
|
if parent.Err() != nil {
|
|
return response, classifiedFailure(parent, ctx, parent.Err(), nil)
|
|
}
|
|
if err = validateOutputTarget(out); err != nil {
|
|
return response, &structuredoutput.Failure{Class: structuredoutput.FailureFatal, Reason: outcomes.ReasonSetupFailure, Message: "invalid structured output target", Cause: err}
|
|
}
|
|
contract, contractErr := OutputContract(req.Schema, c.settings.RepairAttempts)
|
|
if contractErr != nil {
|
|
return response, &structuredoutput.Failure{Class: structuredoutput.FailureFatal, Reason: outcomes.ReasonSetupFailure, Message: "unknown structured response schema", Cause: contractErr}
|
|
}
|
|
vars := prompts.WithSharedVariables(req.Variables)
|
|
for name, value := range prompts.SharedVariables() {
|
|
vars[name] = value
|
|
}
|
|
handle, prepErr := c.runtime.Engine.PrepareExecution(ctx, pk.RunRequest{
|
|
PromptID: req.PromptID, ProfileID: c.profileID, Vars: vars,
|
|
TransientRetries: &c.settings.TransientRetries,
|
|
Validation: contract,
|
|
Observer: func(event pk.ExecutionEvent) {
|
|
if event.Kind == pk.ExecutionFinished {
|
|
terminalMu.Lock()
|
|
terminal = event
|
|
terminalAvailable = true
|
|
terminalMu.Unlock()
|
|
}
|
|
if c.hooks.OnEvent != nil {
|
|
c.hooks.OnEvent(event)
|
|
}
|
|
if c.hooks.OnEventWithCall != nil {
|
|
c.hooks.OnEventWithCall(record.CallID, record.Context, event)
|
|
}
|
|
},
|
|
})
|
|
if prepErr != nil {
|
|
return response, classifiedFailure(parent, ctx, prepErr, nil)
|
|
}
|
|
defer handle.Discard()
|
|
details := handle.Details()
|
|
if name := details.EffectiveModelParams.APIKeyEnv; name != "" {
|
|
c.credentialMu.Lock()
|
|
if c.credentialNames == nil {
|
|
c.credentialNames = make(map[string]struct{})
|
|
}
|
|
c.credentialNames[name] = struct{}{}
|
|
c.credentialMu.Unlock()
|
|
}
|
|
response.Prompt = promptMetadata(details)
|
|
record.Prompt = response.Prompt
|
|
response.Model = details.EffectiveModelParams.Model
|
|
response.Prepared = preparedSnapshot(details)
|
|
record.Prepared = response.Prepared
|
|
result, runErr := c.runtime.Engine.RunPrepared(ctx, handle)
|
|
if result != nil {
|
|
response.FinalContent = result.RawOutput
|
|
record.FinalContent = result.RawOutput
|
|
response.Model = result.ModelName
|
|
response.PromptTokens = result.Usage.PromptTokens
|
|
response.CompletionTokens = result.Usage.CompletionTokens
|
|
response.TotalTokens = result.Usage.TotalTokens
|
|
}
|
|
if runErr != nil {
|
|
return response, classifiedFailure(parent, ctx, runErr, result)
|
|
}
|
|
if result == nil {
|
|
return response, &structuredoutput.Failure{Class: structuredoutput.FailureFatal, Reason: outcomes.ReasonInternalFailure, Message: "structured model call returned no result"}
|
|
}
|
|
if !result.Validation.IsValid {
|
|
return response, classifiedFailure(parent, ctx, nil, result)
|
|
}
|
|
if decodeErr := json.Unmarshal([]byte(result.RawOutput), out); decodeErr != nil {
|
|
var typeError *json.UnmarshalTypeError
|
|
if errors.As(decodeErr, &typeError) {
|
|
return response, &structuredoutput.Failure{Class: structuredoutput.FailureRecoverable, Reason: outcomes.ReasonInvalidOutput, Message: "structured output contains an unusable value", Cause: decodeErr}
|
|
}
|
|
return response, &structuredoutput.Failure{Class: structuredoutput.FailureFatal, Reason: outcomes.ReasonInternalFailure, Message: "validated structured output could not be decoded", Cause: decodeErr}
|
|
}
|
|
response.Content = json.RawMessage(result.RawOutput)
|
|
response.Provider = "openai-compatible"
|
|
return response, nil
|
|
}
|
|
|
|
// OutputContract maps an Audita domain schema to the native request contract.
|
|
func OutputContract(schema responseschema.Key, repairs int) (*pk.OutputContract, error) {
|
|
path, ok := responseschema.AssetPath(schema)
|
|
if !ok {
|
|
return nil, fmt.Errorf("schema %q", schema)
|
|
}
|
|
return &pk.OutputContract{Format: pk.FormatJSON, ValidationMode: pk.ValidationJSONSchema, SchemaPath: path, RepairAttempts: repairs}, nil
|
|
}
|
|
|
|
// classifiedFailure deliberately checks setup defects before broad PromptKit
|
|
// validation/generation wrappers. Provider text is kept only in the typed cause.
|
|
func classifiedFailure(parent, local context.Context, cause error, result *pk.RunResult) *structuredoutput.Failure {
|
|
failure := &structuredoutput.Failure{Class: structuredoutput.FailureFatal, Reason: outcomes.ReasonInternalFailure, Message: "structured model call failed", Cause: cause}
|
|
if parent.Err() != nil {
|
|
failure.Class = structuredoutput.FailureCanceled
|
|
failure.Reason = outcomes.ReasonCanceled
|
|
failure.Message = "processing canceled"
|
|
failure.Cause = parent.Err()
|
|
return failure
|
|
}
|
|
if isSetupFailure(cause) {
|
|
failure.Reason = outcomes.ReasonSetupFailure
|
|
failure.Message = "structured model setup failed"
|
|
return failure
|
|
}
|
|
if errors.Is(local.Err(), context.DeadlineExceeded) || errors.Is(cause, context.DeadlineExceeded) {
|
|
failure.Class = structuredoutput.FailureRecoverable
|
|
failure.Reason = outcomes.ReasonOperationTimeout
|
|
failure.Message = "model operation timed out"
|
|
return failure
|
|
}
|
|
if errors.Is(cause, pk.ErrCapacityExceeded) {
|
|
failure.Class = structuredoutput.FailureRecoverable
|
|
failure.Reason = outcomes.ReasonCapacityExceeded
|
|
failure.Message = "model capacity was unavailable"
|
|
return failure
|
|
}
|
|
if errors.Is(cause, pk.ErrTransport) {
|
|
failure.Class = structuredoutput.FailureRecoverable
|
|
failure.Reason = outcomes.ReasonTransportFailed
|
|
failure.Message = "model transport failed"
|
|
return failure
|
|
}
|
|
if errors.Is(cause, pk.ErrMalformedResponse) {
|
|
failure.Class = structuredoutput.FailureRecoverable
|
|
failure.Reason = outcomes.ReasonProviderResponseInvalid
|
|
failure.Message = "model provider response was invalid"
|
|
return failure
|
|
}
|
|
var provider *pk.GenerationError
|
|
if errors.As(cause, &provider) && provider.StatusCode() != 0 {
|
|
failure.Class = structuredoutput.FailureRecoverable
|
|
failure.Reason = outcomes.ReasonProviderFailed
|
|
failure.Message = "model provider failed"
|
|
return failure
|
|
}
|
|
if result != nil && !result.Validation.IsValid {
|
|
failure.Class = structuredoutput.FailureRecoverable
|
|
failure.Kind = structuredoutput.FailureMalformedOutput
|
|
failure.Reason = outcomes.ReasonInvalidOutputExhausted
|
|
failure.Message = "structured output remained invalid"
|
|
return failure
|
|
}
|
|
if errors.Is(cause, pk.ErrValidation) {
|
|
failure.Reason = outcomes.ReasonSetupFailure
|
|
failure.Message = "structured validation contract failed"
|
|
}
|
|
return failure
|
|
}
|
|
|
|
func isSetupFailure(err error) bool {
|
|
return errors.Is(err, pk.ErrInvalidConfig) || errors.Is(err, pk.ErrInvalidRequest) ||
|
|
errors.Is(err, pk.ErrPromptNotFound) || errors.Is(err, pk.ErrProfileNotFound) ||
|
|
errors.Is(err, pk.ErrPromptLoad) || errors.Is(err, pk.ErrProfileLoad) ||
|
|
errors.Is(err, pk.ErrArtifactLoad) || errors.Is(err, pk.ErrPromptRender)
|
|
}
|
|
|
|
func promptMetadata(details pk.PreparedRun) structuredoutput.PromptMetadata {
|
|
return structuredoutput.PromptMetadata{
|
|
PromptID: details.PromptID,
|
|
DefinitionHash: details.PromptHash, RenderedHash: details.RenderedPromptHash,
|
|
ProfileID: details.SelectedProfileID, BackendID: details.SelectedBackendID,
|
|
Model: details.EffectiveModelParams.Model,
|
|
}
|
|
}
|
|
|
|
func preparedSnapshot(details pk.PreparedRun) *structuredoutput.PreparedSnapshot {
|
|
messages := make([]structuredoutput.Message, len(details.Messages))
|
|
for i, message := range details.Messages {
|
|
messages[i] = structuredoutput.Message{Role: string(message.Role), Content: message.Content}
|
|
}
|
|
return &structuredoutput.PreparedSnapshot{Messages: messages, Prompt: promptMetadata(details)}
|
|
}
|
|
|
|
// ConfiguredSecrets supplies artifact redaction with the engine's referenced credentials.
|
|
func (c *RoleClient) ConfiguredSecrets(ctx context.Context) []string {
|
|
c.credentialMu.Lock()
|
|
names := make([]string, 0, len(c.credentialNames))
|
|
for name := range c.credentialNames {
|
|
names = append(names, name)
|
|
}
|
|
c.credentialMu.Unlock()
|
|
return c.runtime.ConfiguredSecrets(ctx, names...)
|
|
}
|