Structured extraction from emails into a database

I never let the model write directly. The Model component only produces JSON against a schema, and a custom validator component sits between it and the database. That validator parses the JSON with Pydantic, then applies business rules: quantities positive, units from an allowed list, dates in the future, sender domain mapped to a known customer, and every SKU confirmed against the catalog. The prompt tells the model to return null instead of guessing, and a null in a required field goes to the failure queue, not to a default value. The insert itself runs in a transaction with a unique constraint on Message-ID and customer reference. So the model can be wrong, but a wrong answer ends up in review, not in the orders table.

The mail provider's inbound-parse feature receives an email at a dedicated address and posts sender, subject, body, attachment metadata and Message-ID as JSON to a Langflow webhook. The flow first checks the Message-ID against an idempotency ledger. A custom Python component then cleans the body. A Prompt component holding the schema and a few short examples feeds a Model component set to structured JSON output. A custom validator checks the result against a Pydantic model, applies business rules and confirms SKUs against the catalog API. Valid records are upserted in one transaction, and only then is the ledger marked processed. Every failure goes to a queue for review and replay, and every run is traced in Langfuse or LangSmith.

An unknown URL is not security, because anyone who learns it could inject fake orders. I would require authentication on the Langflow endpoint using an API key, and run it behind HTTPS. If the mail provider signs its inbound requests, I would verify that signature at a gateway in front of Langflow and reject anything that fails. I would allow only the expected payload shape and size, and rate-limit at the gateway. Inside the flow, the trust boundary stays in code: even an authenticated request only produces an order if the sender domain maps to a known customer and every rule passes. Credentials for the database and catalog live in secrets, never in the flow definition.

The key point is that the model has no power to write anything. It only produces JSON, and it has no tools, no database access and no agent loop. Whatever the email says, the output still has to pass the Pydantic schema, the business rules and the catalog SKU check, and only code performs the upsert. An injected instruction can at worst produce a wrong but valid-looking order. So I would add rules such as quantity limits per SKU or per customer, which send outliers to review. The prompt also clearly separates instructions from email content and tells the model to treat the body as data. Traces let us find injection attempts afterwards.

There are four main places. First, the provider fails to deliver the webhook. Provider retries help, and reconciling the provider's inbound logs against the ledger catches gaps. Second, the flow crashes mid-run. Recording the Message-ID as received early, plus alerts on entries that never reach a final state, catches this. Third, a step fails without routing to the queue. The rule is that every error path, including parse, validation, SKU and database errors, publishes to the failure queue. Unexpected exceptions get a generic reason code rather than being swallowed. Fourth, the ledger is marked processed before the commit. Marking it only after the commit prevents this. Publishing to the failure queue must itself be reliable, with retries.

Email and webhooks are both at-least-once, so I assume every message may arrive twice. The first step after the webhook checks the Message-ID in an idempotency ledger and stops if it is already processed. The database also has a unique key, so even a race between two deliveries cannot create two orders; the second insert becomes a no-op upsert. I only mark the ledger processed after the commit succeeds. For failures, I separate transient errors like model timeouts or database locks, which get automatic retries with backoff, from content errors like a missing address, which go to the failure queue with a reason code for a human. Replays go through the run API and the same validator, so they are also idempotent.

I start by building a labelled set from real emails, especially ones that landed in the failure queue, with the correct fields recorded by reviewers. Tracing in Langfuse gives me the prompt, raw output and validation result for each run, so I can see which fields fail most. When I change the prompt, examples or model, I run that labelled set through the flow first, using API tweaks to point at the new model or prompt without editing the production flow, and compare field-level accuracy and straight-through rate. I only promote a change if it improves the weak fields without regressing others. I also watch reason codes per sender, because one customer's new email template often explains a sudden spike.

If the ledger were marked first and the insert then failed, a retried delivery would be skipped and the order lost, which breaks the rule against silently dropping an email. Marking it after the commit means the worst case is a crash between commit and ledger update. When the webhook retries, the ledger still says unprocessed, so the flow runs again. The upsert keyed on Message-ID plus customer reference then finds the existing row and does not create a second order. If the ledger lives in the same database, I would update it inside the same transaction, which closes the gap entirely.

Raw order emails carry a lot of noise: HTML markup, signatures, legal disclaimers and, most dangerously, quoted reply chains that may contain older orders. If the model sees the quoted history, it can extract last week's line items as if they were new. Cleaning in a deterministic Python component is cheap, testable and repeatable, and it cuts token cost and latency. I keep the sender domain and received time because the validator needs them for customer mapping and date checks. I also truncate very long bodies so the prompt stays bounded. The cleaned text and a flag saying whether truncation happened both go into the trace.

An inferred value looks exactly like a real one. If the model guesses a delivery date or fills in a likely unit, the JSON passes the schema and the wrong order goes to the warehouse. Null makes missing data explicit. The validator can then decide in code whether that field is required. A missing required field becomes a failure with a clear reason code and goes to a reviewer. A missing optional field is stored as empty. This keeps the model doing extraction only, not business decisions. It also makes accuracy measurable, because we can count how often the model returned null for a field that the reviewer later filled in from the email.

I would use four main stores. The first is an orders table with an internal ID, the Message-ID, customer ID, customer reference, delivery address, requested date, status, and the flow and prompt version that produced it. It has a unique constraint on Message-ID plus customer reference. The second is an order_lines table with order ID, SKU, quantity and unit, plus foreign keys and check constraints such as quantity greater than zero. The third is an idempotency ledger keyed on Message-ID, holding state, timestamps and the trace ID. The fourth is a failures record, alongside the queue, holding the email reference, raw model output, reason code and review outcome. Database constraints repeat the most critical rules as a last line of defence.

I would not quote numbers I haven't been given. Instead, I would ask for daily email volume, peak-hour share and average email size, then reason with variables. Peak arrival rate is the peak-hour volume divided by 3,600 seconds. By Little's law, the concurrent runs needed equal the arrival rate times the end-to-end latency per email. Model latency usually dominates that latency, followed by the catalog lookup. That figure tells me how many Langflow workers I need and whether the model provider's rate limits are a constraint. Token cost is emails times the average prompt plus output tokens. Cleaning and truncation directly reduce that. Database load is small, at one transaction per email, so the model and catalog API are the real bottlenecks.

The validator must not pass SKUs it could not check, so an order is never inserted on the assumption that its SKUs are fine. I would retry a few times with a short timeout and backoff. If the API is still unavailable, the record goes to the failure queue with a distinct, retryable reason code such as a catalog-unavailable code. That is different from an unknown-SKU code, which needs a human. Because the reason is machine-readable, an automated replay job can resubmit these records once the API recovers, and reviewers only see genuine data problems. I would also alert on a spike in that reason code, because it signals an outage, not bad emails.

Traces in Langfuse or LangSmith are very useful but they copy personal data into another system, so I treat them as sensitive. I would mask or redact fields like phone numbers and personal names where the trace does not need them. I would set a retention period that matches company policy and limit access to the engineering and ops roles that need it. If the team needs data to stay within a region or on its own infrastructure, a self-hosted tracing deployment is an option to consider. The failure queue and review console hold raw email references, so they get the same access controls and retention rules. I would also review the model provider's data-use terms before sending emails to it.

Each message carries the original email reference, meaning the Message-ID and a pointer to the stored raw payload. It also carries the cleaned text, the raw model output, any validation errors, and a machine-readable reason code such as parse error, schema violation, unknown SKU, unmapped sender domain, catalog unavailable or database error. The trace ID and flow version are included too. The raw output lets a reviewer see what the model actually produced. The reason code lets us route and count failures and replay the retryable ones automatically. Without the original reference, a reviewer could not check the source email, and replay could not reuse the same idempotency key.

A replay carries the original Message-ID, so it uses the same idempotency key as the first attempt. The upsert is keyed on Message-ID plus customer reference, and the database enforces a unique constraint on that pair. If the original actually committed, for example after a timeout that looked like a failure, the replay updates the same row instead of adding a second one. The review console should also mark the failure as resolved as soon as it is approved, and refuse a second approval. That stops two reviewers replaying the same item. Finally, the ledger moves to processed only after a successful commit, whether that commit came from the first run or a replay.

Reviewers are human and make the same kinds of mistakes the model does, such as a mistyped SKU, a wrong unit or an impossible date. If corrections skipped validation, the review console would become a second write path with weaker guarantees. That breaks the rule that nothing unvalidated is ever inserted. Sending the record through the same validator means there is exactly one definition of a valid order and one insert path. It includes the Pydantic schema, business rules, the catalog check and the transactional upsert. If the correction still fails, it goes back to the queue with a new reason code. This also keeps audit and tracing consistent, because every insert has a trace.

The From address alone is weak evidence. I would use the authentication results that the mail provider includes with inbound mail, such as SPF and DKIM checks, if they are available in the payload. Mail that fails those checks goes to review instead of being processed. Domain mapping should be an allow-list kept in a table, not something the model decides. For higher-risk cases, I would add rules such as a delivery address that doesn't match the customer's known addresses, or an unusually large order, and send those to review. The design limits the damage because nothing is inserted without passing these checks.

The best ground truth comes from reviewed records. Every run is traced with the prompt, model output, validation result and insert outcome. For failures, the reviewer's approved correction shows what the right values were. I compare the model's output with the final record field by field, covering customer reference, each line item's SKU, quantity and unit, the address and the date. That gives per-field error rates, along with how often the model wrongly returned null or guessed. Records that passed without review are not proof of correctness, so I also sample some of them for human checking. This keeps the measurement from relying only on the failures.

An order is only meaningful as a whole. If the header commits but two of five line items fail, the warehouse sees a partial order that looks real. One transaction gives all-or-nothing behaviour: either the order and every line item are written, or nothing is. If any step fails, the whole record goes to the failure queue with a database-error reason. The upsert inside the transaction also replaces line items as a set on reprocessing, so lines are not duplicated. Finally, the ledger update follows the commit. If it is in the same database, it should be in the same transaction.

First, I would watch for completeness. Every received Message-ID should reach a final state, either processed or in the failure queue, within a set time. An alert on entries stuck in received is how we catch silent drops. Second, I would track the failure rate by reason code. A spike in catalog-unavailable means an outage, while a spike in schema violations suggests a model or prompt regression. Third, I would track queue depth and how old the oldest unreviewed failure is, so the review backlog doesn't grow unnoticed. Fourth, I would track latency and token use per run from the traces. Finally, I would track field accuracy over time from reviewed records.

Langflow gives a visual graph that both engineers and ops can read. It also offers fast iteration on prompts in the Playground, run and webhook endpoints without writing a server, and built-in tracing to Langfuse or LangSmith. The parts that need strict guarantees are still plain Python inside custom components: cleaning, validation and the transactional upsert. Those parts stay testable. The trade-offs are that flow definitions need deliberate version control and review, and that teams must resist putting business logic into prompts. If volume or strict operational needs outgrow it, the validator and upsert code can move into a service unchanged, because they're already ordinary Python.

I treat them as retryable and keep them separate from content failures. The Model component call gets a timeout and a small number of retries with backoff. If it still fails, the record goes to the failure queue with a retryable reason code and the original email reference, so an automated replay can resubmit it later without a human. Concurrency is capped below the provider's rate limits. Because of the ledger and the keyed upsert, retries and replays are safe even if an earlier attempt partly succeeded. Persistent errors trigger an alert so the team can switch to a fallback model, if one has been tested on the same evaluation set.

First, I make the webhook path durable. The Message-ID and raw payload are recorded in the ledger or a staging store early, so a burst cannot lose emails even if processing falls behind. Then I scale Langflow workers horizontally, since each email is independent and the flow is stateless apart from the ledger and database. The real limits are the model provider's rate limits and the catalog API. I would cap concurrency to stay under them and let work queue up, rather than fail. Any run that hits a rate-limit error goes to the failure queue with a retryable reason code. The mail provider's retries are safe because of the idempotency check.

Today the webhook passes only attachment metadata, so the model never sees attachment content. The flow must not quietly ignore it. If the body has no line items but there is an attachment, the email goes to the failure queue with an attachment-only reason code, so nothing is silently dropped. To support attachments properly, I would add a step before the prompt that fetches the file from the provider and extracts text from it. For spreadsheets, I would parse rows deterministically in code rather than with the model, because tables are already structured. The validator stays the same. That is the benefit of separating extraction from validation.

The schema should decide this explicitly, because otherwise the model may merge them. I would let the output be a list of orders, each with its own customer reference, delivery address and line items. The upsert key of Message-ID plus customer reference supports this, because each order in the email has a different reference. If an email has two orders but no distinct references, the validator cannot key them safely. In that case I would send it to review with a specific reason code rather than guess. All orders from one email should commit in one transaction, so the email is either fully processed or fully failed.

I keep the flow, including the Prompt component text and the Model component settings, under version control by exporting the flow definition and treating changes like code changes. Each run records the flow version, prompt version and model identifier in the trace and on the order row. When a reviewer finds a bad extraction, we can open the trace and see exactly which prompt and model ran. Then we can tell whether the issue started after a specific change. That also makes rollback possible, because we know which version was last good. It also lets us compare versions on the same set of real failed emails.

Truncation keeps prompts bounded and avoids paying for long disclaimers and history. But if an order really is long, the line items after the cutoff disappear, and the model returns a valid-looking but incomplete order. That would break the goal of a correct database even though nothing failed. So the cleaning component records whether it truncated and how much was cut, and passes that flag to the validator. A truncated email that yields line items should go to review, or be handled by a path that processes the body in chunks. Cleaning out quoted replies first means truncation rarely removes real order content.

Not for the main path. This task is a fixed pipeline with strict guarantees, and a fixed graph makes every run predictable, testable and easy to trace. An agent decides its own steps, so it might skip the catalog check or call a write tool on an unvalidated record. That breaks the core rule. It also costs more and adds latency per email. An agent could be useful in a side role. One example is helping a reviewer by looking up catalog matches for an unknown SKU, using tools such as a catalog search component in tool mode. Its suggestions would still go through the same validator before any insert.

These are a different intent, so I would add an intent field to the schema: new, amendment or cancellation. Each intent gets its own validation path. An amendment usually has a new Message-ID, so idempotency still works per email. But it must match an existing order by customer reference, and the validator checks that the order exists and can still be changed. I would not let amendments overwrite orders automatically at first. Sending them to review until accuracy is proven is safer, because a wrong cancellation is costly. Each change should be recorded as a new version or event on the order, not a silent overwrite.

I give the model the received time from the cleaning step, so relative dates have an anchor. I also ask for ISO format output. In the prompt examples, I show that ambiguous or unstated dates should be null rather than guessed. The validator then applies rules in code. The date must parse, must not be in the past relative to the received time, and must fall within an allowed horizon. Where a customer's date format is known, such as day-first, I store that per customer and apply it in code. Anything ambiguous goes to review with a date-specific reason code, and I store dates with explicit time zones.

I keep a regression set of real, anonymised emails with known correct outputs, mostly built from past reviewed failures plus typical clean orders. I unit-test the custom cleaning and validator components like normal Python code, since they're deterministic. For prompt or model changes, I iterate in the Playground first. Then I run the whole set through the candidate flow via the API and compare field-level accuracy and failure reason codes against the current version, using traces. I check that nothing that used to pass now fails. I also test the failure paths on purpose, such as malformed JSON, catalog down and a duplicate Message-ID, to confirm each one reaches the queue.

Only authorised ops users should see failures, because they contain customer data. Each item is locked to one reviewer at a time and can be approved only once, which prevents double replays. The console records who changed which fields, the before and after values, and when. That gives an audit trail and also produces labelled data for accuracy measurement. Reviewers can also reject an email as not an order. That is a recorded final state, so it does not count as a silent drop. The console calls the flow's run endpoint with a service credential that is not exposed to the browser, and it shows validator errors back to the reviewer.

1. AIn the review console
2. BAt the database insert, by catching the error
3. CAt the first check, by looking up the Message-ID in the idempotency ledgerCorrect answer
4. DIn the validator, by comparing line items

Why: The first step after the webhook checks the Message-ID in the ledger, so duplicates stop before any cleaning or model cost.

1. AAfter the model returns JSON
2. BAfter validation passes but before the insert
3. COnly after the database commit succeedsCorrect answer
4. DAs soon as the webhook is received

Why: Marking it before the commit could make a failed insert look done, so a retry would be skipped and the email silently lost.

1. ATurning off structured output
2. BThe model has no write access, and code validation with business rules gates every insertCorrect answer
3. CA stronger system prompt alone
4. DLetting the model decide whether the email is trustworthy

Why: The model only produces JSON, and the schema, rules, catalog check and code-only upsert limit what injected text can achieve.

1. ARecording each Message-ID on receipt and alerting on entries that never reach a final stateCorrect answer
2. BIncreasing the model's timeout
3. CTrusting the webhook's success response
4. DReviewing a random sample of orders weekly

Why: Comparing received entries against final states, processed or queued, is how the system proves nothing was silently dropped.

1. AAuthenticating requests, such as an API key or provider signature check at a gateway, plus customer mapping rulesCorrect answer
2. BUsing a long flow ID
3. CAsking the model whether the order looks real
4. DKeeping the URL secret

Why: A secret URL is not authentication, so requests must be verified, and business rules still gate every insert.

1. AIn a custom Python component after the Model component, using deterministic checksCorrect answer
2. BIn the mail provider's spam filter
3. CNowhere, because structured output guarantees correct values
4. DInside the prompt, by asking the model to double-check its own answer

Why: Structured output constrains shape, not truth, so deterministic code after the model is needed to enforce schema, business rules and catalog checks before anything is written.

1. AThe email subject line
2. BThe time the webhook was received
3. CThe email Message-ID, enforced with a unique constraintCorrect answer
4. DA random UUID generated per flow run

Why: The Message-ID stays the same across redeliveries, while subjects repeat across orders and timestamps or per-run UUIDs differ on every retry.

1. AAn empty string
2. BThe date one week after the received time
3. CToday's date
4. DnullCorrect answer

Why: The design asks for null for anything not stated, so the validator decides in code whether the missing field is acceptable.

1. AA Pydantic modelCorrect answer
2. BA second model call that grades the output
3. CA regular expression on the raw text
4. DThe database's error messages

Why: The validator parses the JSON against a Pydantic model, then applies business rules and the catalog SKU check.

1. AIt is emailed back to the inbound address
2. BIt is run through the same flow and validator before the upsertCorrect answer
3. CIt is stored in the failure queue as resolved, without insert
4. DIt is written directly to the orders table by the console

Why: Corrections re-enter the flow and pass the same validator, so no unvalidated record is ever inserted.

1. AThe cleaned body and the reviewer's name
2. BThe final database row ID
3. CThe original email reference, the raw model output and a machine-readable reason codeCorrect answer
4. DOnly the reason code

Why: These three let a reviewer see the source and the model's output, and let the system route and replay failures.

1. AOnly failures the model reports itself
2. BOnly JSON parsing errors
3. COnly database write errors
4. DParsing, validation, SKU lookup and database write failuresCorrect answer

Why: Every failure type is published to the queue, which is how the pipeline guarantees it never silently drops an email.

1. AInsert the order and check SKUs later
2. BSend the record to the failure queue with a retryable reason code and replay it after recoveryCorrect answer
3. CDrop the email and wait for the customer to resend
4. DAsk the model whether the SKUs look valid

Why: Unchecked SKUs must never be inserted, and a distinct retryable code allows automatic replay without losing the email.

1. AThe cleaning component drops repeated bodies
2. BNothing; a duplicate is unavoidable
3. CThe upsert keyed on Message-ID plus customer reference updates the existing rowCorrect answer
4. DThe model notices the order already exists

Why: The ledger still reads unprocessed, but the keyed upsert and unique constraint turn the rerun into an update, not a new insert.

1. ABecause Langflow requires it
2. BTo skip the ledger check
3. CTo make the model faster
4. DTo avoid partial orders if any part of the write failsCorrect answer

Why: All-or-nothing writes mean the warehouse never sees a header with missing lines, and failures go cleanly to the queue.

1. AInsert the order with a null customer
2. BFail the record to the queue with a reason code for reviewCorrect answer
3. CCreate a new customer automatically
4. DAsk the model to guess the customer

Why: Domain-to-customer mapping is a business rule, and failures must go to the queue rather than be guessed or dropped.

1. ATurn off tracing completely
2. BMake traces public for debugging
3. CStore traces in the orders table
4. DMask or redact personal fields where possible, limit access, and set a retention periodCorrect answer

Why: Traces are valuable for accuracy work but copy personal data, so they need masking, access control and retention limits.

1. AThe text is stored as the order notes
2. BThe email is discarded
3. CThe validator fails to parse it and the record goes to the failure queue with the raw outputCorrect answer
4. DThe flow retries forever until it gets valid JSON

Why: Parse failures are a defined failure type, and keeping the raw output helps reviewers and prompt debugging.

1. ALangflow does not support defaults
2. BIt uses fewer tokens
3. CDatabases cannot store guesses
4. DA guess can pass schema checks while being wrong, but null lets code detect missing dataCorrect answer

Why: Explicit nulls let the validator enforce required fields instead of letting plausible inventions into orders.

1. AThe model's own confidence
2. BWhether the insert succeeded
3. CThe number of tokens used
4. DThe model output compared field by field with reviewer-approved records, plus sampled checks of passed recordsCorrect answer

Why: Reviewed corrections show correct values, and sampling passed records avoids measuring accuracy only on failures.

1. AThe model's deterministic output
2. BThe cleaning component
3. CThe review console
4. DA unique constraint on Message-ID plus customer reference in the orders tableCorrect answer

Why: A read-then-write ledger check can race, but the database unique constraint used by the upsert enforces one row per key.

1. ARestart the Langflow server with environment variables
2. BUse the Playground chat window for each record
3. CCall POST /api/v1/run/<flow-id> with tweaks that override component inputsCorrect answer
4. DExport the flow and re-import it with new defaults

Why: The run API accepts tweaks that override component settings per request, which suits programmatic replays while keeping the deployed flow unchanged.

1. AA scheduled job that polls the mailbox
2. BThe mail provider's inbound-parse feature posting JSON to the Langflow webhookCorrect answer
3. CA reviewer pressing run in the Playground
4. DAn Agent component reading the inbox

Why: The inbound-parse feature forwards sender, subject, body, attachment metadata and Message-ID as JSON to the flow's webhook.

1. ATo make the email look nicer in traces
2. BBecause the model cannot read quoted text
3. CSo the model doesn't extract older orders quoted in the thread as new onesCorrect answer
4. DTo remove the Message-ID

Why: Quoted history often contains earlier orders, and the model should see only the current order content.

1. AAs SQL stored procedures
2. BAs Python classes with inputs and outputsCorrect answer
3. CAs YAML files only
4. DAs JavaScript plugins

Why: Langflow custom components are Python classes that declare inputs and outputs, which suits the cleaning and validator steps.

1. ADaily volume divided by the number of workers
2. BNumber of SKUs times the number of customers
3. CPeak arrival rate multiplied by average end-to-end latency per emailCorrect answer
4. DModel context window divided by email size

Why: By Little's law, concurrency equals arrival rate times time in the system, and model latency usually dominates that time.

1. AIt shortens the prompt
2. BIt encrypts the failure
3. CIt replaces the need for traces
4. DFailures can be counted, routed, and replayed automatically by typeCorrect answer

Why: Codes let ops separate retryable outages from data problems and spot regressions as spikes in a specific code.

1. ABy the webhook
2. BBy the tracing tool
3. CBy the business rules in the validatorCorrect answer
4. DBy the cleaning component

Why: The validator enforces positive quantities, and a database check constraint can repeat that rule as a backstop.

1. AThe Message-ID is lost
2. BLine items after the cutoff are missed, giving an incomplete but valid-looking orderCorrect answer
3. CThe webhook rejects the request
4. DThe model refuses to answer

Why: Truncation can silently cut real content, so truncated emails should be flagged and sent to review or processed in chunks.

1. AYes; it replaces the idempotency ledger
2. BOnly for duplicate emails
3. CNo; each email is independent, and shared history could leak one email's data into anotherCorrect answer
4. DYes; it improves accuracy for all emails

Why: Message history gives per-session chat memory, which is not needed for stateless per-email extraction and could cross-contaminate outputs.

1. AOnly database connections
2. BOnly Prompt components
3. CComponents switched to tool mode, other flows, and MCP toolsCorrect answer
4. DOnly built-in web search

Why: The Agent component accepts tool-mode components, other flows and MCP tools.

1. AIt replaces the orders database
2. BIt hosts email for the inbound address
3. CIt exposes flows as tools that MCP clients can callCorrect answer
4. DIt stores message history for agents

Why: As an MCP server, Langflow exposes flows as tools to MCP clients, for example letting another team's agent call the extraction.

1. AThe orders database
2. BThe MCP server
3. CThe PlaygroundCorrect answer
4. DThe failure queue

Why: Flows are tested in the Playground before being run through the API or webhooks.