DEV Community

Trkfpn392751
Trkfpn392751

Posted on

Tenant-Auditable CRM Topic Classification — Embeddings, Semantic Search, Reranking, and LLM Evidence

#ai #search #go

Short answer: embed the policy and taxonomy docs, retrieve a small candidate set for each sales-call summary, rerank those snippets, and ask an LLM to classify only from that evidence. For a healthtech CRM, record the evidence IDs, taxonomy version, tenant ID, and metered cost beside every proposed action. That makes a wrong tag explainable and a tenant bill reconcilable.

Don't put the entire taxonomy handbook in every prompt. The larger prompt costs more to inspect, mixes irrelevant definitions into the decision, and makes a taxonomy update harder to trace. Retrieval narrows the field; reranking decides which definitions deserve the classifier's attention.

The operational rule is blunt: no adequate evidence, no automatic tag.

Start with the tenant cost ledger

A topic such as renewal_risk is a business definition, not a universal semantic fact. A sales rep saying “legal still has the amendment” may imply renewal risk for one tenant's workflow and contract review for another. The classifier needs the current label definition, its exclusions, and perhaps a tenant-specific rule. Store those short guidance passages as embeddings with stable document IDs, tenant scope, taxonomy version, and effective date. At request time, semantic search should return a deliberately broad candidate set; reranking then orders that set against the actual call summary.

The ledger needs one record per stage, not one blended “AI charge” per call. Keep the tenant ID, call ID, taxonomy version, operation (embed, rerank, or classify), model ID, request ID, evidence IDs, and reported call cost together. This separation creates two useful failure domains. If retrieval misses the right policy, inspect embedding coverage and metadata filters. If the correct policy reaches the final prompt but the label is wrong, inspect the classification instruction and output. One opaque prompt gives an on-call engineer no such boundary — only a bad CRM mutation, an irreconcilable tenant total, and a vague model transcript.

There is another reason to keep evidence small. A classifier should see the definitions most relevant to this call, not dozens of near-neighbor labels that happen to share words like “clinical,” “security,” or “review.” Reranking is the precision stage. It doesn't replace tenant filters: apply access and version constraints before vector search, then rerank only the permitted candidates.

How should semantic search embeddings rerank docs before LLM topic classification?

Use a four-step request path. First, normalize the call summary and attach the tenant and taxonomy version. Second, create its embedding and query the vector store with tenant filters. Third, send perhaps the best 10–20 permitted snippets to a reranker and retain a much smaller evidence set. Fourth, ask the chat model for a JSON object containing the chosen topic, confidence, evidence IDs, and CRM actions. Those candidate counts are tuning starting points, not measured thresholds; your mileage may vary with taxonomy overlap and snippet size.

The classifier must be allowed to abstain. A low retrieval score is not automatically comparable across embedding models, and a rerank score is not a calibrated probability. Choose acceptance thresholds from a labeled validation set, version them, and send uncertain cases to review. I'm not sure any vendor default can resolve that policy choice because the cost of a false security_review tag belongs to the application, not the model provider.

The provider decision is secondary to those contracts:

Option Where it fits The catch
OpenAI One provider for embeddings and structured chat output It does not provide the separate reranking stage described here; pair it with another reranker or own that stage
Cohere Embedding and reranking are the center of the design Keep the final structured classification contract portable if chat is supplied elsewhere
Pinecone Managed vector search with integrated inference options It adds a managed retrieval system; teams already operating Postgres may prefer fewer data stores
pgvector Tenant-filtered vectors should stay with relational CRM metadata The team owns indexing, tuning, scaling, and operational response
Anthropic Claude is under consideration for the final classification step Embeddings, vector search, and reranking remain separate contracts
Gemini A Google model is already approved for classification Validate structured output against the application's schema and keep retrieval portable
OpenRouter One chat integration must reach multiple model providers It does not remove the need to own tenant-scoped retrieval and evidence accounting
Together AI The team wants another direct inference-provider option Verify the required embedding, reranking, and JSON behavior before selecting the whole pipeline
Infrai A plain REST API, with no required SDK, can put embeddings, native rerank, and OpenAI-compatible chat behind one key; consistent per-call cost metadata supports tenant attribution Check discovery for model and region readiness; stick with direct providers when their specialized controls or an existing contract matter more

No row wins everywhere. A team with mature Postgres operations and strict data-locality controls should start with pgvector. A team standardized on Cohere's retrieval stack should keep it. The unified REST option becomes attractive when language-neutral HTTP, one authentication boundary, and consistent metering reduce integration work, but it is not suitable when procurement mandates a direct vendor relationship or when a required model is unavailable in the target region.

Wire the evidence boundary

The following Go program shows the orchestration boundary without fixing a model name that may change. It expects INFRAI_BASE_URL, INFRAI_API_KEY, EMBED_MODEL, RERANK_MODEL, and CHAT_MODEL; it calls only the three verified paths. The vector-store query is represented as an injected result because its schema belongs to your database, not the AI API. In production, fetch those candidates with a parameterized tenant and taxonomy-version predicate before calling classify.

package main

import (
    "bytes"
    "context"
    "encoding/json"
    "errors"
    "fmt"
    "io"
    "net/http"
    "os"
    "strconv"
    "strings"
    "time"
)

type candidate struct {
    ID   string `json:"id"`
    Text string `json:"text"`
}

type label struct {
    Topic      string   `json:"topic"`
    Confidence float64  `json:"confidence"`
    Evidence   []string `json:"evidence_ids"`
    Actions    []string `json:"crm_actions"`
    Abstain    bool     `json:"abstain"`
}

type client struct {
    baseURL string
    key     string
    http    *http.Client
}

func (c client) post(ctx context.Context, path string, in, out any) error {
    body, err := json.Marshal(in)
    if err != nil {
        return err
    }

    for attempt := 0; attempt < 4; attempt++ {
        req, err := http.NewRequestWithContext(ctx, http.MethodPost, c.baseURL+path, bytes.NewReader(body))
        if err != nil {
            return err
        }
        req.Header.Set("Authorization", "Bearer "+c.key)
        req.Header.Set("Content-Type", "application/json")

        resp, err := c.http.Do(req)
        if err != nil {
            return err
        }
        responseBody, readErr := io.ReadAll(resp.Body)
        resp.Body.Close()
        if readErr != nil {
            return readErr
        }

        if resp.StatusCode == http.StatusTooManyRequests {
            delay := time.Duration(1<<attempt) * time.Second
            if seconds, parseErr := strconv.Atoi(resp.Header.Get("Retry-After")); parseErr == nil {
                delay = time.Duration(seconds) * time.Second
            }
            select {
            case <-ctx.Done():
                return ctx.Err()
            case <-time.After(delay):
                continue
            }
        }
        if resp.StatusCode < 200 || resp.StatusCode >= 300 {
            return fmt.Errorf("POST %s returned %d: %s", path, resp.StatusCode, strings.TrimSpace(string(responseBody)))
        }
        return json.Unmarshal(responseBody, out)
    }
    return errors.New("rate limit retry budget exhausted")
}

func (c client) classify(ctx context.Context, summary string, docs []candidate) (label, error) {
    var embedding struct {
        Data []struct {
            Embedding []float64 `json:"embedding"`
        } `json:"data"`
    }
    if err := c.post(ctx, "/v1/embeddings", map[string]any{
        "model": os.Getenv("EMBED_MODEL"),
        "input": summary,
    }, &embedding); err != nil {
        return label{}, err
    }
    if len(embedding.Data) != 1 {
        return label{}, errors.New("embedding response did not contain one vector")
    }
    // Query the tenant-filtered vector store with embedding.Data[0].Embedding.

    texts := make([]string, len(docs))
    for i := range docs {
        texts[i] = docs[i].Text
    }
    var ranked struct {
        Results []struct {
            Index int `json:"index"`
        } `json:"results"`
    }
    if err := c.post(ctx, "/v1/ai/rerank", map[string]any{
        "model": os.Getenv("RERANK_MODEL"),
        "query": summary,
        "documents": texts,
        "top_n": 3,
    }, &ranked); err != nil {
        return label{}, err
    }

    evidence := make([]candidate, 0, len(ranked.Results))
    for _, result := range ranked.Results {
        if result.Index < 0 || result.Index >= len(docs) {
            return label{}, errors.New("reranker returned an invalid document index")
        }
        evidence = append(evidence, docs[result.Index])
    }
    prompt, err := json.Marshal(map[string]any{"summary": summary, "allowed_evidence": evidence})
    if err != nil {
        return label{}, err
    }

    var completion struct {
        Choices []struct {
            Message struct {
                Content string `json:"content"`
            } `json:"message"`
        } `json:"choices"`
    }
    schema := map[string]any{
        "type": "object",
        "properties": map[string]any{
            "topic": map[string]string{"type": "string"},
            "confidence": map[string]string{"type": "number"},
            "evidence_ids": map[string]any{"type": "array", "items": map[string]string{"type": "string"}},
            "crm_actions": map[string]any{"type": "array", "items": map[string]string{"type": "string"}},
            "abstain": map[string]string{"type": "boolean"},
        },
        "required": []string{"topic", "confidence", "evidence_ids", "crm_actions", "abstain"},
        "additionalProperties": false,
    }
    if err := c.post(ctx, "/v1/chat/completions", map[string]any{
        "model": os.Getenv("CHAT_MODEL"),
        "messages": []map[string]string{
            {"role": "system", "content": "Classify only from allowed evidence. Abstain when it is insufficient."},
            {"role": "user", "content": string(prompt)},
        },
        "response_format": map[string]any{
            "type": "json_schema",
            "json_schema": map[string]any{"name": "crm_topic", "strict": true, "schema": schema},
        },
    }, &completion); err != nil {
        return label{}, err
    }
    if len(completion.Choices) != 1 {
        return label{}, errors.New("classification response did not contain one choice")
    }

    var result label
    if err := json.Unmarshal([]byte(completion.Choices[0].Message.Content), &result); err != nil {
        return label{}, err
    }
    return result, nil
}

func main() {
    c := client{
        baseURL: strings.TrimRight(os.Getenv("INFRAI_BASE_URL"), "/"),
        key: os.Getenv("INFRAI_API_KEY"),
        http: &http.Client{Timeout: 30 * time.Second},
    }
    docs := []candidate{
        {ID: "taxonomy-v17-renewal", Text: "renewal_risk: evidence that an active renewal may be delayed or blocked"},
        {ID: "taxonomy-v17-security", Text: "security_review: a requested security assessment or security documentation"},
        {ID: "taxonomy-v17-contract", Text: "contract_review: legal review of agreement language or amendments"},
    }
    result, err := c.classify(context.Background(), "The hospital's legal team still has the renewal amendment.", docs)
    if err != nil {
        panic(err)
    }
    encoded, _ := json.MarshalIndent(result, "", "  ")
    fmt.Println(string(encoded))
}
Enter fullscreen mode Exit fullscreen mode

One correction matters here: the example's in-memory documents are there to make the control flow runnable, not to model production retrieval. Replace that slice with a pgvector query whose WHERE clause binds both tenant ID and taxonomy version. Never retrieve globally and try to remove other tenants after ranking.

For retries, these calls are read-like computations, so repeating the same request does not create a CRM action. Keep the actual CRM write downstream and idempotent, keyed by something stable such as tenant ID, call ID, taxonomy version, and action type. A timeout after the write must not create a duplicate task on retry. This is where classification demos tend to become production incidents.

Rehearse failure before enabling writes

Before enabling writes, run the pipeline in shadow mode on a labeled set. Record retrieval candidates, final evidence IDs, chosen label, abstention, taxonomy version, model IDs, request IDs, latency, and per-call cost. Aggregate cost by tenant and pipeline stage rather than dividing one invoice by call count; heavy calls and repeated reviews are exactly what that average conceals. Do not log raw health or sales content by default. Store hashes or references where the audit requirement permits, and apply the same retention policy to model traces as to the source call.

Watch three signals separately: evidence recall, accepted-label precision, and duplicate CRM actions. A rise in abstentions after a taxonomy release may be healthy; a fall in evidence recall is not. HTTP 429 belongs on its own dashboard because retry delay affects queue age even when every request eventually succeeds.

Rollback should be boring.

Keep the previous taxonomy embeddings addressable by version, gate automatic writes per tenant, and make the worker capable of switching back to review-only mode without redeploying. If a new embedding model changes nearest neighbors, rebuild a parallel index and compare it in shadow traffic; don't overwrite the active vectors in place. If a classifier release regresses, pin the prior model configuration and replay only from immutable call IDs. The idempotency key on the CRM mutation protects that replay from duplicate tasks, but it does not absolve you from checking that the earlier action is still valid.

The exit criterion is not “the model looks good.” It is a versioned evaluation showing acceptable evidence recall and accepted-label precision for each material tenant segment, plus a rollback exercise that leaves one CRM action per intended call. Ship review-only first. Then widen the write gate.

References

Top comments (0)