A production pattern for replayable, multilingual chat enrichment
Ship an AI agent and the analytics questions arrive immediately.
What do users ask? Which languages do they use? Which topics grow fastest?
Where does the agent fail? Which product area creates the most demand?
The naive implementation puts another LLM call inside the chat request. That
is usually the wrong boundary.
Analytics does not belong on the agent's hot path. A classifier adds latency,
cost, and a new failure mode to every user turn.
The agent must answer the user. The analytics system can work after the turn.
This separation changes the problem. Chat enrichment becomes a data pipeline
with AI stages, not an AI feature with some database writes.
That pipeline must support replay, partial failure, idempotency, rate limits,
schema changes, and reporting. The prompt is only one component.
The classifier is the easy part. The cursor is the hard part.
Keep enrichment off the hot path
The live agent writes the original conversation to its operational store. A
separate worker reads completed or updated sessions and creates analytics data.
The user-facing request ends before enrichment begins.
This boundary gives the system five useful properties:
- Enrichment cannot slow the user response.
- A provider outage does not break chat.
- The pipeline can replay old sessions after a taxonomy change.
- The analytics model can change without a live-agent deployment.
- The source conversation remains the authoritative record.
The worker can run as a scheduled container job. It starts, drains unseen
changes, saves a checkpoint, and exits.
A long-lived service also works. The scheduled-job model is easier to operate
when minute-level freshness is sufficient.
Start with a small source contract
The source document needs a stable session identifier and a list of messages.
Each message needs its own stable identifier.
{
"id": "session-123",
"updatedAt": "2026-04-14T10:42:00Z",
"userSource": "product-a",
"messages": [
{
"id": "message-001",
"createdAt": "2026-04-14T10:41:58Z",
"prompt": "Quelle pompe convient a un fluide visqueux ?",
"response": "...",
"responseSource": "retrieval",
"traceId": "trace-abc"
}
]
}
Do not couple the pipeline to the complete operational document shape. Copy
only the fields that serve enrichment, lineage, or reporting.
The source remains operational data. The destination becomes an analytics
projection.
Make the enrichment stages explicit
Each message moves through three stages in a fixed order:
- Detect the language of the prompt.
- Translate the prompt and response into one analysis language.
- Classify the translated prompt into a controlled taxonomy.
def enrich_session(session: dict) -> dict:
enriched = copy_for_processing(session)
messages = enriched["messages"]
language_detector.detect_messages(messages)
translator.translate_messages(messages, to_language="en")
classifier.classify_messages(messages)
enriched["processed"] = True
return enriched
The explicit order matters. A shared analysis language makes taxonomy results
more comparable across regions.
Translation is not always necessary. If the detected language already matches
the target, the translator returns the original text.
The classifier prefers the translated prompt. It uses the original prompt if
translation returns no usable text.
Keep the original prompt and response. Translation is a derived field, not a
replacement for source data.
Treat structured output as an API contract
Free-form labels become dirty data fast. The model changes capitalization,
invented labels appear, and parsers accept malformed output.
Use structured output with a schema that rejects invalid ranges and shapes.
from pydantic import BaseModel, Field, field_validator
class Category(BaseModel):
name: str
subcategory: str
relevance: float = Field(ge=0.1, le=1.0)
confidence: int = Field(ge=0, le=10)
class Classification(BaseModel):
categories: list[Category]
intent: str
industry: str
@field_validator("categories")
@classmethod
def order_by_relevance(cls, value: list[Category]) -> list[Category]:
if not value:
raise ValueError("at least one category is required")
return sorted(value, key=lambda item: item.relevance, reverse=True)
class ClassificationBatch(BaseModel):
items: list[Classification]
The schema validates syntax. The system prompt defines semantic constraints.
That prompt must contain the allowed categories, intent rules, industry rules,
and score meanings. It must also define the fallback behavior for ambiguity.
Do not let downstream dashboards infer which label variant the model meant.
Normalize at the model boundary.
Store errors as data
An analytics pipeline must not erase a chat because one enrichment provider
fails.
Each stage needs a local fallback:
| Stage | Fallback output | Error field |
|---|---|---|
| Language detection | unknown with zero confidence |
Detection result shows uncertainty |
| Translation | Original text | translationError=true |
| Classification | A controlled Other result |
classificationError=true plus the reason |
This is deliberate partial success. The row still contains the original
conversation, lineage fields, and the successful enrichment fields.
Analysts can exclude failed classifications. Operators can measure failure
rates. A replay can repair affected rows later.
Do not hide errors in logs only. Logs expire and dashboards cannot join them
to the affected message.
Use at-least-once processing
Exactly-once delivery is usually the wrong goal for this pipeline. It adds
coordination while external model calls remain outside the database transaction.
Use at-least-once reads with idempotent writes.
The source change feed returns documents in pages. Each page has a continuation
token before and after the page.
The worker processes documents concurrently, but it advances the durable
checkpoint only after every document in a page reaches a durable state.
GOOD = {"processed", "skipped_invalid", "skipped_unchanged"}
checkpoint_to_save = None
for index, change in enumerate(changes):
result = results.get(index)
if result is None or result.status not in GOOD:
break
next_change = changes[index + 1] if index + 1 < len(changes) else None
at_page_boundary = (
next_change is None
or next_change.continuation_before != change.continuation_before
)
if at_page_boundary:
checkpoint_to_save = change.continuation_after
if checkpoint_to_save is not None:
checkpoint_store.save(checkpoint_to_save)
If one write fails, the checkpoint stays before that page. The next run reads
the page again.
This creates duplicate work, not lost work. Idempotent destination writes make
that trade acceptable.
Keep the checkpoint outside the source store when possible. Then the worker
needs read access to the operational database, not write access.
Define idempotency from content
A source timestamp is useful for progress. It is weak evidence that the useful
content changed.
System fields can change without a semantic document change. Some migrations
also add destination-only metadata.
Create a stable fingerprint from the source content instead.
VOLATILE_FIELDS = {
"_etag",
"_rid",
"_ts",
"_lsn",
"processed",
"processedAt",
"sourceFingerprint",
}
def source_fingerprint(document: dict) -> str:
stable = {
key: value
for key, value in document.items()
if key not in VOLATILE_FIELDS
}
payload = json.dumps(
stable,
sort_keys=True,
separators=(",", ":"),
ensure_ascii=False,
)
return hashlib.sha256(payload.encode("utf-8")).hexdigest()
Before enrichment, read the destination metadata by session identifier. Skip
the expensive stages when the stored fingerprint matches.
The fingerprint reduces cost during retries, drains, and scheduled catch-up
runs. It also separates content identity from database implementation details.
Store the fingerprint beside the output session. Make the fingerprint version
explicit if its canonicalization rules can change.
Make writes idempotent and atomic per session
Use a parent row for the session and one child row per message.
CREATE TABLE ChatSession (
SessionId NVARCHAR(128) NOT NULL PRIMARY KEY,
SourceFingerprint NVARCHAR(64) NULL,
SourceUpdatedAt DATETIME2 NULL,
ProcessedAt DATETIME2 NOT NULL
);
CREATE TABLE ChatMessageEnrichment (
SessionId NVARCHAR(128) NOT NULL,
MessageId NVARCHAR(128) NOT NULL,
DetectedLanguage NVARCHAR(16) NULL,
TranslatedPrompt NVARCHAR(MAX) NULL,
TranslatedResponse NVARCHAR(MAX) NULL,
TranslationError BIT NOT NULL,
ClassificationJson NVARCHAR(MAX) NULL,
ClassificationError BIT NOT NULL,
ClassificationErrorText NVARCHAR(4000) NULL,
TraceId NVARCHAR(128) NULL,
PRIMARY KEY (SessionId, MessageId),
FOREIGN KEY (SessionId) REFERENCES ChatSession(SessionId)
);
Upsert the session row and all message rows in one transaction. Roll back the
complete session write if one message fails.
This transaction does not cover the external enrichment calls. It covers the
destination projection, which is the boundary that SQL can protect.
Use one database connection per worker thread. Many database clients do not
support concurrent use of one connection.
Close connections after worker threads finish. Thread pools can otherwise
leave dead-thread connections in a long-lived cache.
Normalize the stable dimensions
Do not normalize every model output on day one. That creates a large schema
before query patterns are clear.
A useful middle ground has three parts:
- Session fields live in the session table.
- Message fields live in the message table.
- Nested arrays and evolving model output stay as valid JSON.
Then expose a reporting view that extracts the stable dimensions.
CREATE VIEW ChatMessageAnalytics AS
SELECT
s.SessionId,
m.MessageId,
m.DetectedLanguage,
JSON_VALUE(m.ClassificationJson, '$.intent') AS Intent,
JSON_VALUE(m.ClassificationJson, '$.industry') AS Industry,
JSON_VALUE(
m.ClassificationJson,
'$.categories[0].name'
) AS PrimaryCategory,
m.TranslationError,
m.ClassificationError,
m.TraceId
FROM ChatMessageEnrichment AS m
JOIN ChatSession AS s ON s.SessionId = m.SessionId;
This model keeps ingestion flexible and gives analysts a flat contract. Add
columns when a field becomes stable and heavily queried.
Separate metadata backfills from AI reprocessing
Schema changes do not always require new model calls.
Suppose the source adds traceId, responseSource, or a richer document
reference. The fingerprint can identify the session as unchanged.
The pipeline can patch those destination columns without language detection,
translation, or classification.
This light backfill path matters at scale. A metadata migration must not create
a large model bill by default.
Use separate methods for full enrichment and metadata-only updates. The code
path then makes cost visible.
Bound concurrency at both levels
There are two natural units of parallel work:
- Sessions inside a source batch.
- Messages inside one session.
That creates multiplicative concurrency.
If the worker runs ten sessions and each session classifies ten messages, it
can create 100 concurrent model calls.
Treat these limits as one capacity budget.
maximum model concurrency
~= document workers x message workers
Use a smaller inner limit during large historical drains. Daily catch-up runs
can use a higher limit because their batches stay small.
Language detection and translation can remain sequential inside a session if
provider limits dominate. Optimize the measured bottleneck, not the diagram.
Rotate providers only for retryable failures
A provider pool can absorb rate limits. It must not hide permanent errors.
For an HTTP 429 response, move to the next configured client. If every client
returns 429, wait with exponential backoff and retry for a bounded number of
rounds.
For authentication errors, invalid payloads, and other permanent failures,
record the fallback result immediately.
This distinction avoids two bad outcomes:
- One throttled endpoint does not block the complete batch.
- A broken credential does not create a long retry storm.
Keep retry limits in code and expose provider-specific counters. A log line is
not a capacity plan.
Use three modes for three jobs
A production pipeline needs more than one run command.
Incremental catch-up
Read from the saved change-feed checkpoint. Process all unseen pages, save the
new checkpoint, and exit when the feed reaches its current boundary.
This is the normal scheduled mode.
Historical drain
A full replay needs stronger boundaries:
- Record a source timestamp as the snapshot cutoff.
- Process every source document at or before that cutoff.
- Process documents that changed after the cutoff.
- Compare source identifiers with destination identifiers.
- Retry any destination gaps.
A replay is complete only after identifier-level reconciliation.
The reconciliation phase is important. Equal row counts do not prove that both
sides contain the same sessions.
One-batch mode
Process one small batch and stop. This mode is useful for deployment smoke
checks and local debugging.
Keep separate checkpoints for incremental and historical runs. A backfill must
not corrupt the cursor for daily processing.
Preserve lineage into the reporting layer
Every message row needs enough context for a developer to trace it back.
Useful lineage fields include:
- The source session identifier.
- The source message identifier.
- The source update timestamp.
- The source fingerprint.
- The agent response source.
- The upstream trace identifier.
- The pipeline processing timestamp.
- Translation and classification error fields.
The trace identifier connects a dashboard anomaly to the original agent trace.
The fingerprint explains why the pipeline skipped or reprocessed a session.
Without lineage, an analytics table becomes a pile of plausible labels.
Treat chat analytics as sensitive data
The destination contains user prompts, agent responses, identity fields, and
possibly document references. It can be more sensitive than the operational
store because it is easier to query in bulk.
Apply the same controls as any other user-data system:
- Use least-privilege identities for the source and destination.
- Put checkpoints in the destination so the source can stay read-only.
- Define retention for raw text and derived labels.
- Restrict access to identity columns.
- Do not copy prompts into routine logs.
- Record taxonomy and prompt versions for later audits.
- Define a deletion path that removes session and message rows.
Derived labels are still user data. A category inferred by a model can reveal
more than the original dashboard owner expects.
Test invariants, not only examples
Example classifications are useful, but the pipeline fails at boundaries.
Test these invariants:
- The source document stays unchanged during enrichment.
- A stable fingerprint skips all expensive providers.
- A failed document blocks checkpoint movement past its page.
- A successful replay produces the same destination keys.
- One message write failure rolls back the complete session.
- One classification failure does not stop other messages.
- A 429 response rotates to the next provider.
- A permanent provider error creates a flagged fallback.
- A historical drain resumes from its saved phase.
- Reconciliation finds identifier gaps even when row counts match.
- Worker-thread database connections close after the batch.
- Structured output rejects invalid ranges and empty category lists.
Mock external providers for the main suite. Keep live model tests opt-in and
small.
The design in one sentence
Write chats once, enrich them asynchronously, and make every expensive step
safe to replay.
The model adds labels. The pipeline makes those labels believable.
That trust comes from page-boundary checkpoints, content fingerprints,
transactional upserts, explicit fallbacks, bounded concurrency, and complete
lineage.
Without those parts, chat analytics is a prompt attached to a cron job.
With them, it becomes infrastructure.