|
| 1 | +package connector |
| 2 | + |
| 3 | +import ( |
| 4 | + "context" |
| 5 | + "fmt" |
| 6 | + "net/http" |
| 7 | + "strconv" |
| 8 | + "strings" |
| 9 | + "time" |
| 10 | + |
| 11 | + "github.com/dvflw/mantle/internal/metrics" |
| 12 | +) |
| 13 | + |
| 14 | +// EmbeddingConnector implements the ai/embed action: it turns text into vector |
| 15 | +// embeddings via a provider's embeddings API. It mirrors AIConnector's provider |
| 16 | +// selection, base-URL/model allowlists, and metrics. Because the action lives |
| 17 | +// under the "ai/" prefix, the engine's token-budget accounting applies to it |
| 18 | +// automatically (via the usage.total_tokens it returns). |
| 19 | +type EmbeddingConnector struct { |
| 20 | + Client *http.Client |
| 21 | + AllowedBaseURLs []string |
| 22 | + AllowedModels []string // empty = all models allowed |
| 23 | +} |
| 24 | + |
| 25 | +// Execute embeds one or more input strings and returns the vectors, including |
| 26 | +// pgvector text literals for direct use in a postgres/query arg (e.g. |
| 27 | +// INSERT ... VALUES ($1::vector)). |
| 28 | +func (c *EmbeddingConnector) Execute(ctx context.Context, params map[string]any) (map[string]any, error) { |
| 29 | + providerName, _ := params["provider"].(string) |
| 30 | + if providerName == "" { |
| 31 | + providerName = "openai" |
| 32 | + } |
| 33 | + |
| 34 | + model, _ := params["model"].(string) |
| 35 | + if model == "" { |
| 36 | + return nil, fmt.Errorf("ai/embed: model is required") |
| 37 | + } |
| 38 | + if len(c.AllowedModels) > 0 && !stringInSlice(model, c.AllowedModels) { |
| 39 | + return nil, fmt.Errorf("ai/embed: model %q not in allowed list", model) |
| 40 | + } |
| 41 | + |
| 42 | + inputs, err := extractEmbedInputs(params) |
| 43 | + if err != nil { |
| 44 | + return nil, fmt.Errorf("ai/embed: %w", err) |
| 45 | + } |
| 46 | + |
| 47 | + provider, err := c.getProvider(providerName, params) |
| 48 | + if err != nil { |
| 49 | + return nil, fmt.Errorf("ai/embed: %w", err) |
| 50 | + } |
| 51 | + |
| 52 | + req := &EmbeddingRequest{Model: model, Inputs: inputs} |
| 53 | + if cred, ok := params["_credential"].(map[string]string); ok { |
| 54 | + req.Credential = cred |
| 55 | + } |
| 56 | + if d, ok := extractInt(params["dimensions"]); ok && d > 0 { |
| 57 | + req.Dimensions = d |
| 58 | + } |
| 59 | + |
| 60 | + workflow, _ := params["_workflow"].(string) |
| 61 | + step, _ := params["_step"].(string) |
| 62 | + |
| 63 | + start := time.Now() |
| 64 | + resp, err := provider.Embeddings(ctx, req) |
| 65 | + duration := time.Since(start).Seconds() |
| 66 | + |
| 67 | + metrics.AIRequestDuration.WithLabelValues(workflow, step, model, providerName).Observe(duration) |
| 68 | + if err != nil { |
| 69 | + metrics.AIRequestsTotal.WithLabelValues(workflow, step, model, providerName, "error").Inc() |
| 70 | + return nil, fmt.Errorf("ai/embed: %w", err) |
| 71 | + } |
| 72 | + metrics.AITokensTotal.WithLabelValues(workflow, step, model, providerName, "prompt").Add(float64(resp.Usage.PromptTokens)) |
| 73 | + metrics.AIRequestsTotal.WithLabelValues(workflow, step, model, providerName, "success").Inc() |
| 74 | + |
| 75 | + return embeddingResponseToOutput(resp), nil |
| 76 | +} |
| 77 | + |
| 78 | +// getProvider returns the EmbeddingProvider for the given provider name. |
| 79 | +func (c *EmbeddingConnector) getProvider(name string, params map[string]any) (EmbeddingProvider, error) { |
| 80 | + switch name { |
| 81 | + case "openai": |
| 82 | + baseURL := "https://api.openai.com/v1" |
| 83 | + if u, ok := params["base_url"].(string); ok && u != "" { |
| 84 | + baseURL = u |
| 85 | + } |
| 86 | + if len(c.AllowedBaseURLs) > 0 && !stringInSlice(baseURL, c.AllowedBaseURLs) { |
| 87 | + return nil, fmt.Errorf("base_url %q not in allowed list", baseURL) |
| 88 | + } |
| 89 | + return &OpenAIProvider{Client: c.Client, BaseURL: baseURL}, nil |
| 90 | + case "bedrock": |
| 91 | + // Bedrock embeddings use InvokeModel with per-model request shapes |
| 92 | + // (Titan, Cohere), distinct from the Converse API used for chat. |
| 93 | + // Tracked as a follow-up; OpenAI-compatible endpoints (including Azure |
| 94 | + // OpenAI and local servers) are reachable today via base_url. |
| 95 | + return nil, fmt.Errorf("bedrock embeddings are not yet supported; use provider: openai (optionally with base_url)") |
| 96 | + default: |
| 97 | + return nil, fmt.Errorf("unknown provider %q (available: openai)", name) |
| 98 | + } |
| 99 | +} |
| 100 | + |
| 101 | +// extractEmbedInputs reads the `input` param, accepting a single string or an |
| 102 | +// array of strings. |
| 103 | +func extractEmbedInputs(params map[string]any) ([]string, error) { |
| 104 | + raw, ok := params["input"] |
| 105 | + if !ok { |
| 106 | + return nil, fmt.Errorf("input is required") |
| 107 | + } |
| 108 | + switch v := raw.(type) { |
| 109 | + case string: |
| 110 | + if v == "" { |
| 111 | + return nil, fmt.Errorf("input must not be empty") |
| 112 | + } |
| 113 | + return []string{v}, nil |
| 114 | + case []string: |
| 115 | + if len(v) == 0 { |
| 116 | + return nil, fmt.Errorf("input must not be empty") |
| 117 | + } |
| 118 | + return v, nil |
| 119 | + case []any: |
| 120 | + out := make([]string, 0, len(v)) |
| 121 | + for i, item := range v { |
| 122 | + s, ok := item.(string) |
| 123 | + if !ok { |
| 124 | + return nil, fmt.Errorf("input[%d] must be a string, got %T", i, item) |
| 125 | + } |
| 126 | + out = append(out, s) |
| 127 | + } |
| 128 | + if len(out) == 0 { |
| 129 | + return nil, fmt.Errorf("input must not be empty") |
| 130 | + } |
| 131 | + return out, nil |
| 132 | + default: |
| 133 | + return nil, fmt.Errorf("input must be a string or array of strings, got %T", raw) |
| 134 | + } |
| 135 | +} |
| 136 | + |
| 137 | +// embeddingResponseToOutput converts an EmbeddingResponse to the connector's |
| 138 | +// output map. It always includes the full `embeddings`/`vectors` arrays, and |
| 139 | +// for the common single-input case surfaces `embedding`/`vector`/`dimensions`. |
| 140 | +func embeddingResponseToOutput(resp *EmbeddingResponse) map[string]any { |
| 141 | + vectors := make([]string, len(resp.Embeddings)) |
| 142 | + for i, e := range resp.Embeddings { |
| 143 | + vectors[i] = formatPGVector(e) |
| 144 | + } |
| 145 | + |
| 146 | + out := map[string]any{ |
| 147 | + "model": resp.Model, |
| 148 | + "embeddings": resp.Embeddings, |
| 149 | + "vectors": vectors, |
| 150 | + "count": len(resp.Embeddings), |
| 151 | + "usage": map[string]any{ |
| 152 | + "prompt_tokens": resp.Usage.PromptTokens, |
| 153 | + "total_tokens": resp.Usage.TotalTokens, |
| 154 | + }, |
| 155 | + } |
| 156 | + if len(resp.Embeddings) > 0 { |
| 157 | + out["embedding"] = resp.Embeddings[0] |
| 158 | + out["vector"] = vectors[0] |
| 159 | + out["dimensions"] = len(resp.Embeddings[0]) |
| 160 | + } |
| 161 | + return out |
| 162 | +} |
| 163 | + |
| 164 | +// formatPGVector renders a float slice as a pgvector text literal: "[1,2,3]". |
| 165 | +// This can be passed straight into a postgres/query arg and cast with ::vector. |
| 166 | +func formatPGVector(v []float64) string { |
| 167 | + var b strings.Builder |
| 168 | + b.WriteByte('[') |
| 169 | + for i, f := range v { |
| 170 | + if i > 0 { |
| 171 | + b.WriteByte(',') |
| 172 | + } |
| 173 | + b.WriteString(strconv.FormatFloat(f, 'g', -1, 64)) |
| 174 | + } |
| 175 | + b.WriteByte(']') |
| 176 | + return b.String() |
| 177 | +} |
| 178 | + |
| 179 | +func stringInSlice(s string, list []string) bool { |
| 180 | + for _, v := range list { |
| 181 | + if v == s { |
| 182 | + return true |
| 183 | + } |
| 184 | + } |
| 185 | + return false |
| 186 | +} |
0 commit comments