Chapter 3.2 AI Data Pipelines

The Retrieve → Normalize Context → Build Prompt sequence introduced in Chapter 3.1 is a pipeline in practice but not in architecture. The three nodes transform data sequentially, but there are no formal schema contracts at the boundaries between them, no stage-level error handling, and no defined separation between the transformation stages and the AI assessment stage.

Chapter 3.2 makes that structure explicit. A pipeline is not a workflow the distinction is architectural, not cosmetic. Understanding it is the prerequisite for Chapter 3.3, where the AI assessment stage itself becomes a coordinated sequence of multiple calls.

Learning Objectives

After completing this chapter, you will be able to:

Describe the five-stage AI data pipeline (Ingest → Normalize → Enrich → Assess → Route + Store) and explain the responsibility, input schema, and output schema of each stage.
Design schema contracts at each stage boundary and implement IF validation nodes that enforce contract compliance before passing data to the next stage.
Implement stage-level error handling that routes contract violations to a defined fallback path rather than allowing schema errors to propagate silently to the Assess stage.
Explain the architectural benefit of separating the Assess stage from the data preparation stages, using the Chapter 3.3 upgrade path (single Assess → multi-agent Assess) as a concrete example.
Build the complete five-stage pipeline for the Meridian deal assessment workflow, applying the pipeline discipline to a real multi-field deal submission schema.
Troubleshoot a pipeline where the AI assessment produces inconsistent outputs for semantically similar inputs, by isolating the failure to the Normalize or Enrich stage’s field mapping logic.

3.2.1 Workflow vs. Pipeline

Workflow

A workflow executes a business process. It has business logic, conditional decisions, side effects (writing to HubSpot, sending Slack messages), and a defined start and end state. The Part II four-workflow platform is a set of workflows in this sense: Workflow A intakes a contact and produces a scored, routed record; Workflow B enforces lifecycle transitions; Workflow C manages follow-up cadences; Workflow D handles survey integration.

Pipeline

A pipeline transforms data. Each stage has a defined input schema, applies a transformation, and produces an output schema. The output of stage N is the input of stage N+1. No stage has side effects no CRM writes, no notifications until the terminal stage, where the transformed data is delivered to its destination.

The distinction matters because pipelines and workflows have different failure modes, different testing strategies, and different extension patterns.

Property	Workflow	Pipeline
Primary purpose	Execute a business process	Transform data through defined stages
Stage side effects	Expected (CRM writes, notifications)	None until terminal stage
Failure mode	Process halts; side effects may be partial	Stage fails; no side effects propagate
Testing	Path coverage (did the right branch fire?)	Schema validation at each stage boundary
Extension	Add new workflow or branch	Add new pipeline stage

The Part II four workflows are business process workflows. Chapter 3.2 introduces the pipeline as a complementary architectural pattern that sits between data arrival and the assessment workflow.

Design Practice

When you add enrichment stages before an AI call, you are building a pipeline even if you implement it as nodes in an n8n workflow. Naming it as a pipeline and applying pipeline discipline to it makes it testable, extensible, and independently deployable from the surrounding workflow logic.

3.2.2 AI Pipeline Anatomy

An AI data pipeline has five canonical stages. The pipeline architecture is the same whether the Assess stage contains one AI call or ten.

Ingest → Normalize → Enrich → Assess → Route + Store

Ingest → Normalize → Enrich → Assess → Route + Store

Stage	Responsibility	Input	Output
Ingest	Receive raw data from one or more sources	Webhook payload, API response, file upload	Validated raw record
Normalize	Transform all source formats to a shared internal schema	Raw record (source-specific structure)	Canonical contact schema
Enrich	Augment with additional signals from external sources	Canonical contact schema	Enriched contact schema
Assess	Apply AI assessment to the enriched, normalized context	Enriched contact schema	Advisory output contract
Route + Store	Post-processing, property writes, downstream triggers	Advisory output contract	HubSpot properties written; notifications sent

Every Part III practical from Chapter 3.1 onward implements some or all of these stages. Chapter 3.1’s Retrieve + Normalize Context nodes are the Enrich and partial Normalize stages of the pipeline, informally. Chapter 3.2 formalizes the full five-stage structure.

Design Practice

The annotation in Figure 34.1 reads: “In Chapter 3.3, the Assess stage becomes a coordinated multi-agent sequence.” That is the structural insight the pipeline model enables you can upgrade the Assess stage from one AI call to three without changing the Ingest, Normalize, Enrich, or Route + Store stages. The pipeline isolates the assessment from the data preparation and the downstream delivery.

3.2.3 Schema Contracts at Stage Boundaries

Each stage boundary in the pipeline is a validation point. The output schema of stage N must satisfy the input schema of stage N+1 before execution proceeds.

This is the Part I output contract discipline (Section 1.5) applied between pipeline stages rather than between the AI call and downstream workflow logic.

Normalize

The canonical contact schema the output of Normalize and the input of Enrich must define:

Field	Type	Required	Notes
`contact_id`	String	Yes	HubSpot contact ID
`submission_id`	String	Yes	Unique identifier for this intake event
`inquiry_text`	String	Yes	Free-text field for AI assessment
`engagement_type`	Enum	Yes	Values defined in domain spec
`company_name`	String	Yes	Used for firmographic enrichment lookup
`source`	Enum	Yes	Intake channel
`normalized_at`	ISO8601 datetime	Yes	Stage completion timestamp

Enrich

The enriched contact schema the output of Enrich and the input of Assess extends the canonical schema with enrichment fields:

Field	Type	Required at Enrich exit	Notes
All canonical fields		Yes	Passed through unchanged
`firmographic_source`	Enum: `api` \| `mock` \| `unavailable`	Yes	Indicates enrichment source
`company_size_category`	Enum	Conditional	Required unless `firmographic_source = unavailable`
`industry_vertical`	String	Conditional	Required unless `firmographic_source = unavailable`
`enriched_at`	ISO8601 datetime	Yes	Stage completion timestamp

The IF node at the Normalize/Enrich boundary validates that all required canonical fields are present before the Enrich stage executes. A missing required field at Normalize exit is a data quality failure that must be handled at the Normalize stage not silently passed to Enrich with empty values. A pipeline that catches the failure at Normalize produces a recoverable error. A pipeline that passes empty values to Enrich produces a miscalibrated AI output with no visible error.

Production Risk

A schema violation at the Normalize/Enrich boundary caught by an IF validation node is a recoverable error route to a fallback that logs the failure and skips enrichment. A schema violation that is not caught propagates to the Assess stage, where the AI receives an incomplete context. The AI will produce an output regardless often a plausible-sounding but miscalibrated one. Silent corruption is harder to diagnose than a caught validation failure.

3.2.4 The Mock Enrichment Stage

Mock Enrichment Stage

In production, the Enrich stage calls an external API a firmographic data provider (company size, industry classification, funding history) before passing the enriched contact to the Assess stage. In Practical 3.2, a Code node returns mock firmographic data in the same schema a real API would return.

The pipeline architecture is unchanged when the mock Code node is replaced with a real HTTP Request node only the node type changes.

The mock approach lets you build and test the full five-stage pipeline without dealing with API authentication, rate limits, or connectivity dependencies. The schema of the Enrich stage output is identical.

Mock enrichment Code node example output:

// Returns mock firmographic data in the same schema as a live firmographic API
return {
  json: {
    firmographic_source: "mock",
    company_size_category: "mid-market",    // 50–500 employees
    industry_vertical: "professional-services",
    recent_funding_signals: false,
    headquarter_region: "northeast-us",
    data_confidence: 1.0                    // Mock data is always "confident"
  }
};

Engineering Rationale

When replacing the mock Code node with a live firmographic API, use the same output field names and types. Downstream nodes reference $json.company_size_category if the real API returns company_size instead, a Normalize output step inside the Enrich stage maps the API’s field names to the canonical schema. The pipeline’s Assess stage never needs to know which field name the API used.

3.2.5 Stage-Level Error Handling

Stage-Level Error Handling

Each stage boundary is a potential failure point. The Part I five-element reliability model (Section 1.5) is extended here across multiple stages rather than applied to a single AI call.

Stage	Failure Mode	On Error Response
Ingest	Malformed webhook payload; missing required field	Log failure; reject record; Slack alert
Normalize	Required canonical field missing	Log failure; skip enrichment and assess with available fields; flag `normalization_incomplete`
Enrich	API unavailable; mock returns unexpected schema	Set `firmographic_source: "unavailable"`; proceed to Assess with enrichment fields empty
Assess	AI API unavailable; parse error; confidence below threshold	Rule fallback path (same as Chapter 2.5); set `p6_last_execution_source: "rule_fallback"`
Route + Store	HubSpot write failure	Retry once; if retry fails, log to Slack with contact ID

A failure in stage N should not halt the entire pipeline unless the stage N output is strictly required for stage N+1.

The critical design principle: a failure in stage N should not halt the entire pipeline unless the stage N output is strictly required for stage N+1.

The Enrich stage can fail gracefully (returning firmographic_source: "unavailable") and the Assess stage can proceed with reduced context. The Normalize stage failing on a required field like inquiry_text cannot be recovered gracefully there is nothing for the AI to assess. A pipeline that distinguishes recoverable failures from unrecoverable ones degrades predictably. A pipeline that treats all failures identically either halts unnecessarily or proceeds with corrupted data.

Engineering Rationale

Add an On Error connection at the Enrich stage that routes to a Code node setting firmographic_source: "unavailable" and passing the canonical schema unchanged. This mirrors the Part II AI fallback path: when the external enrichment source is unavailable, the pipeline continues with reduced context rather than failing entirely.

Reference Diagrams

Figure 3.2.1 AI Pipeline Anatomy

Figure 34.1 shows the five canonical pipeline stages with schema labels at each boundary and the On Error path from the Enrich stage. The annotation on the Assess stage previews Chapter 3.3.

%%{init:{"theme":"base","themeVariables":{"primaryColor":"#eef2ff","primaryTextColor":"#1e1b4b","primaryBorderColor":"#6366f1","lineColor":"#6366f1","clusterBkg":"#f8f9ff","clusterBorder":"#6366f1","titleColor":"#1e1b4b","edgeLabelBackground":"#f6f4ef","fontFamily":"system-ui,sans-serif","fontSize":"13px"}}}%%

flowchart TD
    IN["INGEST Webhook payload / API event → raw record"]:::process
    NR["NORMALIZE Required canonical fields → canonical contact schema"]:::process
    NE["On Error: log + flag normalization_incomplete"]:::fallback
    EN["ENRICH Firmographic data (API or mock) → enriched contact schema"]:::process
    EE["On Error: firmographic_source unavailable continue"]:::fallback
    AS["ASSESS (AI Call) In Ch 3.3: coordinated multi-agent sequence → advisory output contract"]:::success
    AE["On Error: rule fallback (Part II pattern)"]:::fallback
    RS["ROUTE + STORE HubSpot property writes Slack notifications p6_ audit properties"]:::process

    IN --> NR
    NR -->|"validation failure"| NE
    NR -->|"canonical schema"| EN
    EN -->|"enrichment error"| EE
    EN -->|"enriched schema"| AS
    EE --> AS
    AS -->|"AI error"| AE
    AS -->|"advisory output contract"| RS
    classDef trigger  fill:#dcfce7,stroke:#16a34a,color:#14532d,font-weight:600
    classDef process  fill:#eef2ff,stroke:#6366f1,color:#1e1b4b
    classDef decision fill:#fef9c3,stroke:#ca8a04,color:#78350f,font-weight:600
    classDef success  fill:#d1fae5,stroke:#059669,color:#064e3b,font-weight:700
    classDef fallback fill:#fee2e2,stroke:#dc2626,color:#7f1d1d

Figure 34.1: AI Pipeline Anatomy. Five-stage horizontal flow: Ingest, Normalize, Enrich, Assess, Route + Store. Schema labels at each stage boundary. On Error paths shown at Normalize and Enrich boundaries. Assess stage annotated: “In Chapter 3.3, this single stage becomes a coordinated multi-agent sequence.”

Practical Exercise 3.2 Five-Stage Pipeline with Mock Enrichment

Business Scenario

Vantage Advisory Partners is expanding its assessment criteria. In addition to the intake form data and peer context introduced in Chapter 3.1, the advisors want the AI to factor in firmographic signals: company size, industry vertical, and funding history. These signals are available from an external data provider but the assessment pipeline needs a formal structure before that integration can be added reliably.

The Problem

The Retrieve → Normalize Context → Build Prompt sequence from Chapter 3.1 has no formal boundary discipline. If a field is missing or misnamed at one stage, it propagates silently to the next. There is no defined place to add enrichment, no validation gate, and no defined fallback when an external data source is unavailable.

The Architectural Solution

Restructure Workflow A as a five-stage pipeline with named stages and validation gates. The Enrich stage receives the canonical contact schema, augments it with firmographic data, and passes the enriched schema to the Assess stage. A mock Code node at Enrich produces the correct schema without requiring a live API connection.

Updated Workflow

Figure 34.2 shows the five named pipeline stages with validation gates and the On Error fallback path at the Enrich stage.

%%{init:{"theme":"base","themeVariables":{"primaryColor":"#eef2ff","primaryTextColor":"#1e1b4b","primaryBorderColor":"#6366f1","lineColor":"#6366f1","clusterBkg":"#f8f9ff","clusterBorder":"#6366f1","titleColor":"#1e1b4b","edgeLabelBackground":"#f6f4ef","fontFamily":"system-ui,sans-serif","fontSize":"13px"}}}%%

flowchart TD
    A["Webhook Trigger (raw payload)"]:::trigger --> B["[INGEST] Webhook Node"]:::trigger
    B --> C["[NORMALIZE] Code Node → Canonical Contact Schema"]:::process
    C --> D{"[NORMALIZE GATE] IF Node (required fields present?)"}:::decision
    D -->|"True"| E["[ENRICH] Code Node Firmographic (Mock) → Enriched Schema"]:::process
    E -->|"On Error: firmographic_source: unavailable"| F["[ASSESS] Build Prompt → HTTP Request (LLM) → Parse Response"]:::process
    D -->|"False"| G[["Route to Error Handler"]]
    E --> F
    F --> H["[ROUTE + STORE] HubSpot Property Writes + p6_ Audit Fields"]:::success
    classDef trigger  fill:#dcfce7,stroke:#16a34a,color:#14532d,font-weight:600
    classDef process  fill:#eef2ff,stroke:#6366f1,color:#1e1b4b
    classDef decision fill:#fef9c3,stroke:#ca8a04,color:#78350f,font-weight:600
    classDef success  fill:#d1fae5,stroke:#059669,color:#064e3b,font-weight:700
    classDef fallback fill:#fee2e2,stroke:#dc2626,color:#7f1d1d

Figure 34.2: Five-Stage Pipeline with Validation Gates. Five-stage pipeline for Chapter 3.2 with explicit stage names and validation gates. The Normalize gate validates required fields before enrichment; the Enrich stage falls back to firmographic_source: unavailable on error rather than halting the pipeline.

Step 1 Restructure Workflow A as a Pipeline

Purpose

The five-stage pipeline naming convention exists to make each stage’s architectural role explicit and testable in isolation. Renaming nodes to match stage names Ingest, Normalize, Enrich, Assess, Route + Store aligns the visual representation in n8n with the schema contracts at each boundary. This is not cosmetic: when a downstream failure occurs, stage naming tells you precisely which transformation produced the bad output. A workflow where every node is labeled with what it does (“Webhook”, “Build Prompt”) makes boundary failures harder to diagnose than a workflow where every node is labeled with which stage it belongs to.

Operation Summary

Property	Value
Node Type	Rename / relabel (no new nodes added in this step)
Primary Function	Align node labels to five-stage pipeline naming convention
Input	Existing Workflow A from Practical 3.1
Output	Same workflow with stage-labeled nodes

Configuration

Rename nodes in n8n using the following mapping:

Current Label	New Stage Label
Webhook	`[INGEST] Webhook`
Intake validation / Code node	`[NORMALIZE] Canonical Schema`
(new Step 3)	`[ENRICH] Firmographic (Mock)`
Build Prompt	`[ASSESS] Build Prompt`
HTTP Request (LLM)	`[ASSESS] HTTP Request LLM`
Parse Response	`[ASSESS] Parse Response`
HubSpot property write nodes	`[ROUTE + STORE] HubSpot Property Write`

Stage prefix brackets [INGEST], [NORMALIZE], etc. make stage boundaries visible in the n8n canvas without requiring a diagram.

Output Table

Output	Description
Relabeled workflow	Node names match five-stage pipeline convention
Unchanged data flow	No schema or logic changes in this step

Engineering Rationale

Stage-labeled nodes enforce a convention that the remaining steps depend on. When Step 2 inserts an IF gate, it belongs in the [NORMALIZE] stage zone of the canvas. When Step 3 inserts the Enrich node, it belongs between [NORMALIZE] and [ASSESS]. The naming convention makes misplacement visible before it causes a schema contract violation.

Step 2 Add the Normalize Validation Gate

Purpose

The Normalize stage produces the canonical contact schema that every downstream stage depends on. A missing inquiry_text field at this stage will cause the AI prompt to be assembled without the primary assessment signal producing a plausible-sounding but indefensible advisory output with no error logged. The Normalize Validation Gate catches this failure at its source: immediately after the Normalize Code node, before the Enrich stage executes. A schema violation caught here produces a recoverable, logged error. A schema violation that propagates silently to the Assess stage produces a corrupted advisory output.

Operation Summary

Property	Value
Node Type	IF
Primary Function	Validate required canonical fields before Enrich executes
Input	Output of `[NORMALIZE] Canonical Schema` node
Output (True)	Proceeds to `[ENRICH] Firmographic (Mock)`
Output (False)	Routes to On Error Code node

Decision Configuration

// n8n IF node condition all three must be true
{{ $json.inquiry_text && $json.inquiry_text.length > 0 }}
AND
{{ $json.engagement_type !== undefined && $json.engagement_type !== null }}
AND
{{ $json.contact_id !== undefined && $json.contact_id !== null }}

The three conditions are AND-joined: all required fields must be present for the True branch to fire. inquiry_text validation includes a length check an empty string passes a presence check but fails as an assessable input. engagement_type and contact_id require only presence; their value validation is the responsibility of the Normalize Code node upstream.

On Error Path Configuration

On the False branch, insert a Code node named [NORMALIZE] On Error Flag Incomplete. The node returns:

return [{
  json: {
    normalization_incomplete: true,
    missing_fields: [
      ...($json.inquiry_text && $json.inquiry_text.length > 0 ? [] : ['inquiry_text']),
      ...($json.engagement_type ? [] : ['engagement_type']),
      ...($json.contact_id ? [] : ['contact_id'])
    ],
    contact_id: $json.contact_id ?? 'unknown',
    failed_at: new Date().toISOString()
  }
}];

After this Code node, insert an HTTP Request node that posts a Slack message to #vap-ops with normalization_incomplete: true, the list of missing fields, and contact_id. This is a hard stop the pipeline does not proceed to Enrich or Assess with incomplete normalization.

Output Table

Output (True branch)	Description
Canonical contact schema	All required fields present; passes to Enrich

Output (False branch)	Description
`normalization_incomplete`	Boolean `true`
`missing_fields`	Array of field names that failed validation
`contact_id`	Contact ID for Slack alert (or `"unknown"`)
`failed_at`	ISO8601 timestamp of validation failure

Engineering Rationale

The validation gate applies the Part I output contract discipline to inter-stage data flow. In the Part I three-node pattern, Parse Response validates the AI’s output before it reaches downstream routing. The Normalize Validation Gate applies the same principle one stage earlier: validate the pipeline’s internal data product before passing it to the next transformation stage. Both are contract enforcement at a boundary the same pattern at different positions in the pipeline.

Step 3 Add the Enrich Stage (Mock)

Purpose

The Enrich stage augments the canonical contact schema with firmographic signals company size, industry classification, funding history that the AI uses as additional assessment context. In production, this stage calls an external firmographic data provider API. In this practical, a Code node returns mock data in the identical schema the live API would return. This approach lets you build and validate the full five-stage pipeline including the On Error fallback path without API credentials, rate limit dependencies, or network connectivity requirements. The mock node is structurally identical to the live integration: when the live API is ready, only the node type changes. The schema, stage position, and On Error path are unchanged.

Operation Summary

Property	Value
Node Type	Code (JavaScript) mock; HTTP Request in production
Primary Function	Return mock firmographic enrichment data in canonical schema
Input	Canonical contact schema from `[NORMALIZE]` gate (True branch)
Output	Enriched contact schema with firmographic fields added

Implementation Logic

// [ENRICH] Firmographic (Mock) Code node
// Returns mock data in the same schema as a live firmographic API
const canonicalSchema = $input.first().json;

return [{
  json: {
    // Pass all canonical schema fields through unchanged
    ...canonicalSchema,
    // Add firmographic enrichment fields
    firmographic_source: "mock",
    company_size_category: "mid-market",      // 50–500 employees
    industry_vertical: "professional-services",
    recent_funding_signals: false,
    headquarter_region: "northeast-us",
    data_confidence: 1.0,                     // Mock is always "confident"
    enriched_at: new Date().toISOString()
  }
}];

The spread operator (...canonicalSchema) passes all canonical fields through unchanged. This pattern ensures the enriched schema is a strict superset of the canonical schema every field the canonical schema guarantees is still present after enrichment. Enrichment fields are additive; they do not replace canonical fields.

Enrich Field Table

Field	Type	Value (Mock)	Description
`firmographic_source`	Enum	`"mock"`	Source indicator: `mock`, `api`, `unavailable`
`company_size_category`	Enum	`"mid-market"`	Firm size bracket for AI context
`industry_vertical`	String	`"professional-services"`	Industry classification
`recent_funding_signals`	Boolean	`false`	Whether recent funding activity is present
`headquarter_region`	Enum	`"northeast-us"`	Regional context for advisory calibration
`data_confidence`	Float 0–1	`1.0`	Enrichment data reliability score
`enriched_at`	ISO8601	Current timestamp	Stage completion timestamp

On Error Path Configuration

Add an On Error connection from the Enrich Code node. The On Error path connects to a Code node named [ENRICH] On Error Unavailable:

// On Error fallback return canonical schema with unavailable marker
const canonicalSchema = $input.first().json;
return [{
  json: {
    ...canonicalSchema,
    firmographic_source: "unavailable",
    company_size_category: null,
    industry_vertical: null,
    recent_funding_signals: null,
    headquarter_region: null,
    data_confidence: 0,
    enriched_at: new Date().toISOString()
  }
}];

This fallback allows the Assess stage to proceed with reduced context rather than halting the pipeline. The firmographic_source: "unavailable" value is the explicit signal to the Build Prompt node that firmographic fields should not be factored into the assessment.

Output Table

Output	Description
All canonical fields	Passed through unchanged from `[NORMALIZE]`
`firmographic_source`	`"mock"`, `"api"`, or `"unavailable"`
`company_size_category`	Firm size enum or `null` if unavailable
`industry_vertical`	Industry string or `null` if unavailable
`recent_funding_signals`	Boolean or `null` if unavailable
`enriched_at`	ISO8601 stage completion timestamp

Engineering Rationale

The mock Enrich node is a deliberate scoping pattern, not a shortcut. Any integration that returns data to a pipeline stage should be testable without live connectivity. A Code node that returns the correct schema lets you validate every downstream node including the On Error fallback and the Build Prompt enrichment context block before touching external systems. When the production API is integrated, the schema at the Enrich boundary is already correct. Integration testing then validates the API call itself, not the downstream schema.

Step 4 Update the Assess Stage (Build Prompt)

Purpose

The Build Prompt node assembles the AI prompt from all available context: peer scores from Practical 3.1 and firmographic signals from Step 3. This step adds a named firmographic context block to the system message, positioned after the peer context block. The AI uses firmographic signals as assessment calibration: a professional services firm in the mid-market category has a different risk profile than a seed-stage startup in an unfamiliar vertical. The explicit instruction for the firmographic_source: "unavailable" case prevents the AI from hallucinating firmographic context when enrichment failed.

Operation Summary

Property	Value
Node Type	Code (JavaScript) or Set node
Primary Function	Add firmographic context block to enriched advisory prompt
Input	Enriched contact schema from `[ENRICH]` stage
Output	`prompt_payload` with peer context + firmographic context

Implementation Logic

const enriched = $input.first().json;
const peerContext = enriched.peer_context ?? { peer_count: 0 };

const firmographicBlock = enriched.firmographic_source === 'unavailable'
  ? `FIRMOGRAPHIC CONTEXT: Unavailable. Do not factor company size or industry into this assessment.`
  : `FIRMOGRAPHIC CONTEXT:
Company size: ${enriched.company_size_category}
Industry: ${enriched.industry_vertical}
Recent funding signals: ${enriched.recent_funding_signals}
Region: ${enriched.headquarter_region}
Enrichment confidence: ${enriched.data_confidence}`;

const systemMessage = `
You are an advisory assessment specialist for Vantage Advisory Partners.

PEER CONTEXT (last ${peerContext.peer_count} assessed contacts):
${JSON.stringify(peerContext, null, 2)}

${firmographicBlock}

CURRENT SUBMISSION:
Inquiry text: ${enriched.inquiry_text}
Engagement type: ${enriched.engagement_type}
Company: ${enriched.company_name}
Source: ${enriched.source}

Assess the current submission and return the advisory output in the required JSON schema.
`.trim();

return [{
  json: {
    prompt_payload: {
      model: "gpt-4o-mini",
      messages: [
        { role: "system", content: systemMessage },
        { role: "user", content: "Provide the advisory assessment." }
      ],
      max_tokens: 400,
      temperature: 0.2
    }
  }
}];

The firmographic block uses a ternary to produce either an explicit unavailability notice or the populated context block. The AI instruction for the unavailable case is directive “do not factor” rather than permissive (“firmographic data may not be available”). A directive instruction produces more consistent behavior across repeated calls than a permissive one.

Prompt Field Table

Field	Required	Description
`PEER CONTEXT` block	Yes	Normalized peer scoring summary from Practical 3.1
`FIRMOGRAPHIC CONTEXT` block	Yes	Enrichment data or explicit unavailability instruction
`CURRENT SUBMISSION` block	Yes	Core intake fields for AI assessment
`model`	Yes	LLM model identifier
`max_tokens`	Yes	Response length cap (400 for full advisory output)
`temperature`	Yes	Sampling temperature (0.2 for deterministic advisory)

Output Table

Output	Description
`prompt_payload`	Full prompt object with system and user messages
`model`	LLM identifier string
`max_tokens`	Token limit for the response
`temperature`	Sampling parameter

Engineering Rationale

Context block ordering in the system message reflects attention priority. Peer context and firmographic context are calibration frames they inform how the AI weights the current submission. Placing them before the current submission fields ensures the AI reads the reference frames before applying them to the specific case. Reversing the order (submission first, context after) produces weaker calibration because the AI has already begun forming an assessment before reading the reference data.

Step 5 Write Bootstrap Properties

Purpose

The Route + Store stage writes all four p6_ audit properties after every AI call including calls where enrichment fell back to "unavailable". Writing these properties unconditionally produces a complete dataset for the Chapter 3.5 observability layer. A dataset that only captures successful enrichment runs will undercount executions and produce misleading aggregate metrics. The firmographic_source value is not a p6_ property it is written to the contact’s standard audit fields but the four p6_ properties record the AI call’s outcome regardless of which enrichment path was taken.

Operation Summary

Property	Value
Node Type	HTTP Request
Method	PATCH
Endpoint	`/crm/v3/objects/contacts/{contactId}`
Primary Function	Write four `p6_` audit properties to the Contact record
Output	Updated contact object

Request Payload

{
  "properties": {
    "p6_last_execution_source": "ai",
    "p6_last_confidence_score": "{{ $('Parse Response').first().json.confidence_score }}",
    "p6_last_advisory_path": "{{ $('Parse Response').first().json.advisory_path }}",
    "p6_agent_call_count": "1"
  }
}

p6_agent_call_count remains 1 throughout Practical 3.2. The addition of the Enrich stage enriches the AI’s input context but does not add a second AI call. The transition to p6_agent_call_count: 2 occurs in Practical 3.3 when a second agent is introduced.

Output Table

Output	Description
`p6_last_execution_source`	`"ai"` for AI call completion; `"rule_fallback"` otherwise
`p6_last_confidence_score`	Float 0.00–1.00 from AI output
`p6_last_advisory_path`	Advisory path string matching VAP enum values
`p6_agent_call_count`	Integer `1` throughout Practical 3.2

Engineering Rationale

Write the four p6_ properties after the enriched AI call, not after the enrichment step. The properties record the AI call’s outcome not whether enrichment succeeded. An execution where enrichment fell back to "unavailable" and the AI produced a valid advisory output should still write p6_last_execution_source: "ai". An execution where the AI call failed and rule fallback was used should write p6_last_execution_source: "rule_fallback". The firmographic_source value from the Enrich stage is a separate data point relevant to analysis, but not a p6_ property.

Step 6 Test the Fallback Path

Purpose

The Enrich On Error fallback path is the most operationally critical path in the five-stage pipeline because it is the path most likely to be exercised in production when external enrichment is unavailable. Testing it before moving to the next chapter confirms two things: the fallback fires correctly when the Enrich node errors, and the pipeline completes with a valid advisory output based on reduced context. An untested fallback path that first fires in production will fail in a mode you have not observed and cannot diagnose quickly.

Operation Summary

Property	Value
Node Type	Code (test modification to `[ENRICH]` node)
Primary Function	Confirm On Error path fires and pipeline continues
Input	Any valid test contact payload
Output	Advisory output with `firmographic_source: "unavailable"`

Configuration

Temporarily modify the Enrich Code node by adding the following as the first line:

throw new Error("test simulated enrichment failure");

Run one test contact through the pipeline. Verify:

The Enrich node throws and routes to [ENRICH] On Error Unavailable
firmographic_source: "unavailable" is set in the fallback node output
Build Prompt renders the "unavailable" context block, not the populated firmographic block
The AI call completes and produces a valid advisory output
p6_last_execution_source: "ai" is written (not "rule_fallback")

After confirming all five conditions, remove the throw new Error(...) line before running the six test contacts in production mode.

Output Table

Output	Description
`firmographic_source`	`"unavailable"` confirms fallback fired
`advisory_path`	Valid advisory path confirms AI completed despite fallback
`p6_last_execution_source`	`"ai"` confirms AI call succeeded on reduced context

Engineering Rationale

Design Practice

Always test the On Error path of every external call stage before moving to the next chapter. The mock Enrich node makes this straightforward you control when it fails. A live API does not give you that control, and a pipeline whose fallback path has never been tested in staging will fail in production in a way you have not designed for.

Deliverable: Five-stage pipeline in n8n with mock Enrich stage, Normalize validation gate, Enrich On Error fallback, enriched context in Build Prompt, and p6_ properties written. Fallback path tested. Six test contacts run end-to-end.

Estimated time: 2–3 hours.

Discussion Questions

The Normalize stage is defined as having no side effects no CRM writes, no notifications. If a contact normalization failure should alert the operations team, where does the Slack notification belong architecturally? In the Normalize stage, or in the On Error path that follows it?
The mock Enrich Code node always returns data_confidence: 1.0. When you replace it with a live firmographic API, the real API will sometimes return partial data or low-confidence signals. What changes in the Build Prompt instruction to handle a data_confidence < 0.6 enrichment result?
Chapter 3.3 replaces the Assess stage with a coordinated multi-agent sequence. Looking at Figure 34.1, which of the other four stages (Ingest, Normalize, Enrich, Route + Store) would need to change when the Assess stage goes multi-agent, and which would not?

Chapter Summary

A pipeline transforms data through stages with defined input and output schemas at each boundary. A workflow executes a business process with side effects and conditional decisions. Both patterns are used in Part III; they serve different architectural purposes and have different testing and failure characteristics.

The five canonical AI pipeline stages Ingest, Normalize, Enrich, Assess, Route + Store provide a named structure for the data transformation that precedes every AI call. Schema contracts at each stage boundary apply the Part I output contract discipline to inter-stage data flow: a validation failure at stage N is caught and handled at stage N, not propagated silently to stage N+1.

The mock Enrich stage is a deliberate scoping choice. It implements the full pipeline structure including the On Error fallback path without requiring a live external API. When production enrichment is needed, the mock Code node is replaced with an HTTP Request node. The schema at the Enrich stage output is unchanged. A system built with a mock Enrich stage that has the correct schema can swap in a live API without changing any downstream node. A system built without schema discipline at that boundary cannot.

Key Principle

The five canonical AI pipeline stages Ingest, Normalize, Enrich, Assess, Route + Store separate data transformation from AI assessment. Schema contracts at each boundary catch errors at their source rather than propagating them silently to the Assess stage.

Transition to Chapter 3.3

The annotation on the Assess stage in Figure 34.1 reads: “In Chapter 3.3, this single stage becomes a coordinated multi-agent sequence.” That is Chapter 3.3’s subject.

The pipeline architecture is the framework that makes multi-agent systems tractable. Each agent in a multi-agent Assess stage receives the same enriched contact schema that the Enrich stage produced. Each agent produces an output that conforms to a defined contract. The orchestrator merges or sequences those outputs before Route + Store receives them. The pipeline boundaries established in Chapter 3.2 do not move.

Chapter 3.3 introduces the three multi-agent coordination patterns (sequential, parallel, and conditional) and their implementation in n8n, and requires a mini-ADR documenting which pattern is most appropriate for the Meridian Venture Partners capstone.

Key Takeaways

A pipeline transforms data through stages with defined input/output schemas at each boundary. A workflow executes a business process with side effects. Both patterns appear in Part III; they serve different purposes.
The five canonical AI pipeline stages are: Ingest → Normalize → Enrich → Assess → Route + Store. The Assess stage is where the AI call (or calls) execute.
Schema contracts at each stage boundary apply the Part I output contract discipline to inter-stage data flow. Validate before stage N+1 executes, not after it fails.
The mock Enrich stage produces firmographic data in the same schema a live API would return. The pipeline architecture is unchanged when the mock Code node is replaced with an HTTP Request node.
Add an On Error connection at the Enrich stage that sets firmographic_source: "unavailable" and passes the canonical schema through unchanged. The Assess stage must handle reduced context without error.
Test every On Error fallback path before moving to the next chapter. The mock Enrich stage gives you full control over when the failure fires live APIs do not.

End of Chapter 3.2 AI Data Pipelines