建议用最窄的多模型抽象层+严格eval验证+显式escape hatch,让换模型成为配置事件而非代码重写。
Short answer: choose the narrowest multi-model API contract that your evals can verify, then keep provider-specific features behind explicit escape hatches. For a fintech team turning sales-call transcripts into CRM actions, this makes a model change a controlled configuration event instead of a rewrite, while still leaving room for a native capability when quality justifies its operational cost.
The hard part is not sending a prompt. It is proving that a changed model still produces safe, useful CRM work under the same latency budget. A small team needs one request shape, one audit trail, and one rollback switch before it needs a perfect abstraction.
Treat each transcript as an untrusted document. The pipeline should persist the source, redact or tokenize sensitive fields, ask for a typed action proposal, validate that proposal, and only then write to the CRM. The model never gets direct write authority. A human or deterministic policy approves changes such as a follow-up date, account owner, or deal stage.
That ordering changes the provider decision. A model that is 80 milliseconds faster is irrelevant if a schema drift silently turns "next Tuesday" into an invalid date. I keep the model response, normalized request, policy result, and final CRM mutation under one correlation ID. It gives the eval harness something concrete to compare and gives an on-call engineer a reversible unit of work.
The first version can be small:
from dataclasses import dataclass
from typing import Any, Protocol
@dataclass
class ActionProposal:
action: str
confidence: float
fields: dict[str, Any]
class ModelGateway(Protocol):
def complete(self, *, model: str, messages: list[dict[str, str]]) -> str:
...
def summarize_call(gateway: ModelGateway, transcript: str, model: str) -> ActionProposal:
prompt = (
"Extract CRM actions from this sales call. Return JSON with action, "
"confidence, and fields. Do not invent values.\n\n" + transcript
)
raw = gateway.complete(
model=model,
messages=[{"role": "user", "content": prompt}],
)
proposal = parse_and_validate(raw)
if proposal.confidence < 0.85:
raise ValueError("requires human review")
return proposal
def parse_and_validate(raw: str) -> ActionProposal:
# Replace this with a strict JSON parser and a schema library in production.
raise NotImplementedError
The interface intentionally says nothing about streaming, tool names, or vendor metadata. Those belong in an adapter that converts a provider response into this contract. I started with a larger interface once. It became a museum of optional flags, and every flag became another thing the eval suite had to understand.
Standardize the parts that affect safety and measurement: message roles, structured output rules, timeout semantics, retry classification, token accounting, and a deadline propagated from the HTTP request. Do not pretend that temperature, context windows, or tool-calling behavior are equivalent; record them as capabilities and test them explicitly.
The adapter should also normalize errors into a small taxonomy: caller input, authentication, rate limit, transient upstream, and policy rejection. A retry is sensible for a transient response and dangerous for a duplicate CRM mutation. Idempotency keys on the write side matter more than clever backoff on the read side. I've found that writing these categories into the event schema early saves a surprisingly long argument later, because the on-call dashboard can answer "what failed?" without parsing provider-specific text.
Here is a provider-neutral retry boundary. It is deliberately boring.
import random
import time
def call_with_deadline(call, *, attempts: int = 3, deadline_s: float = 8.0):
started = time.monotonic()
for attempt in range(attempts):
remaining = deadline_s - (time.monotonic() - started)
if remaining <= 0:
raise TimeoutError("model deadline exceeded")
try:
return call(timeout=remaining)
except RateLimited as exc:
if attempt == attempts - 1:
raise
delay = min(exc.retry_after_s, remaining / 2)
time.sleep(delay + random.random() * 0.05)
except TransientUpstream:
if attempt == attempts - 1:
raise
time.sleep(min(2 ** attempt, remaining / 2))
The names in this snippet are application exceptions, not claims about any one API. In tests, inject fake adapters and replay a fixed corpus of calls. Measure action precision, unsafe-action rate, review rate, p50/p95 latency, and token usage together. A single composite score hides the trade-off we actually need to debate.
Use each product as an interchangeable evidence source, not as the architecture. OpenAI, Anthropic Claude, and Google Gemini all expose language-model APIs, but their native request fields, tool schemas, safety controls, and usage metadata differ. A gateway can offer a common chat-and-JSON subset; it cannot erase those differences without either dropping features or leaking provider concepts into application code.
The table is a decision aid, not a ranking. Your mileage may vary when data residency, procurement, or an existing platform team changes the setup cost.
That is why I keep a capability matrix beside the adapter tests. One row says "strict JSON"; another says "tool calls"; another says "maximum input size." The matrix drives routing and rejects an unsupported request before it reaches production. It also makes a migration reviewable: the diff is a changed capability declaration and eval result, not a surprise in a shared helper.
For a small team, the practical split is 90 percent portable path and 10 percent explicit native path. The native path carries a provider name in configuration, has its own fixture set, and cannot bypass the same redaction and approval policy. This preserves an escape hatch without making every call vendor-shaped.
Portability is a hypothesis until the harness reruns it. Keep a versioned set of de-identified transcripts with expected actions and adjudicated edge cases: ambiguous dates, competitors mentioned in passing, missing account IDs, and a caller asking for an action the CRM cannot represent.
Run candidates in shadow mode first. Compare normalized proposals, not raw text, and inspect disagreements by category. I once treated a lower average latency as a win; the tail was hiding a timeout cluster after long transcripts. The fix was a prompt budget and a deadline, not another provider.
Token counts need a declared method. Provider-reported usage is the accounting source when available; a local tokenizer such as tiktoken can estimate prompt size for experiments, but different tokenizers make cross-provider totals approximate. Store both the estimate and the reported value so a cost alert does not masquerade as a quality regression.
The catch is that a normalized API is not suitable when a product depends on a provider-native modality or control that the common contract cannot express. Keep a direct integration for realtime voice, unusual file inputs, or a tool protocol whose semantics are central to the user experience. Budget the extra auth, observability, and fixture work instead of hiding it.
It is also a poor fit when compliance requires every prompt and response to stay within a particular cloud boundary. HIPAA's Security and Privacy Rules are a useful reminder to inventory every processor and retention point; a gateway adds another party to that review. In that case, a cloud-native service with the required contractual controls may be the defensible choice, even if its SDK ties you to one environment.
Finally, a one-model product with no credible switching requirement should use the vendor SDK until the second-model work is real. Abstraction has a carrying cost. Revisit it when an eval shows a meaningful quality or latency alternative, then migrate behind the contract above.
The operational checklist is short: pin adapter versions, log correlation IDs and normalized usage, enforce deadlines, validate before CRM writes, replay the eval corpus on every model change, and keep a tested rollback model. Those controls are what let a small team move quickly without confusing portability with a pile of wrappers.