> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.runtype.com/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.runtype.com/_mcp/server.

# Using records in flows

Use Records in Flows to retrieve structured context, personalize responses, cache data between runs, and build retrieval-augmented generation (RAG) flows.

## Common patterns

Use one of these patterns to add Records to a Flow.

### Build a retrieval-augmented generation flow

A RAG flow finds Records that match a question with **Vector Search**, formats the matching content with **Run Code**, and generates an answer from that content with **Run Task**. For the full configuration, see [Build a complete RAG flow](#build-a-complete-rag-flow).

**Vector Search** returns only Records that have an embedding created with the same embedding model that the search uses. Records that you create in the dashboard or through the API have no embedding. Create embeddings first in an ingest Flow with **Generate Embedding** and **Store Vector**.

### Personalize responses

To pass customer data to **Run Task**, follow these steps:

1. Add a **Get Record** step that looks up the customer Record, and set its output variable to `customer`.
2. Add a **Run Task** step that references the customer metadata in its prompt.

Use a prompt like this:

```plaintext
Customer message: {{message}}

Customer information:
- Name: {{customer.metadata.name}}
- Tier: {{customer.metadata.tier}}
- Order history: {{customer.metadata.order_count}} orders

Provide a response that matches the customer's tier.
```

Use snake_case metadata keys such as `order_count`. When you create a Record in the dashboard or through the API, Runtype converts each top-level metadata key to lowercase snake_case, so a key you enter as `orderCount` is stored as `ordercount`.

### Cache API results

To reuse an API response across runs, follow these steps:

1. Add a **List Records** step that filters on the cache Record type and name. Set its output variable to `cachedResults`, and keep **If no records match** set to **Succeed with an empty list**.
2. Add a **Conditional Logic** step with the condition `{{cachedResults.length}} === 0`.
3. In **If true**, add a **Fetch URL** or **Make API Call** step that calls the external API, and set its output variable to `apiData`.
4. In **If true**, after the API step, add an **Upsert Record** step that stores `apiData` with the same Record type and name that the **List Records** step filters on.
5. In **Otherwise**, add a **Run Code** step that returns `cachedResults[0].metadata`, and set its output variable to `apiData`.

Both paths write `apiData`, so later steps reference `{{apiData}}` whichever path ran. The data that **Upsert Record** stores must be a JSON object.

### Store conversation history

To keep a conversation history in one Record, follow these steps:

1. Add a **List Records** step that finds the conversation Record, with **If no records match** set to **Succeed with an empty list**. The first message of a conversation has no Record yet.
2. Add a **Run Task** step that generates a response from the stored history.
3. Add a **Run Code** step that returns an object with the complete updated history, for example `{ history: [...previousHistory, newTurn] }`.
4. Add an **Upsert Record** step that stores the **Run Code** output. Name the Record with a stable value, such as a conversation ID that you pass to the Flow as a variable.

**Upsert Record** replaces each top-level key that it writes, so it replaces the stored `history` array instead of appending to it. Build the full array in **Run Code** before the upsert.

### Process each Record in a batch

When you run a Flow in batch, each Record becomes one execution, and the Flow reads the current Record from `_record`. Reference its fields with `{{_record.metadata.FIELD_NAME}}`, where `FIELD_NAME` is a metadata key.

To write results back to the Record that the execution processes, add an **Update Record** step and set **Find record** to **From context**. The step updates the current Record, so you don't configure an ID or a filter. For more information, see [Running flows in batch](/user-guide/flows/running-flows-in-batch).

## Choose a read step

Choose the step that matches the result that you need:

* Use **Get Record** to return one Record for an ID or a structured lookup.
* Use **List Records** to return an array for a structured lookup with zero or more matches.
* Use **Vector Search** to find Records by semantic similarity.

**List Records** and **Vector Search** both return arrays. Pass the array to **Run Code** to select fields and build context for **Run Task**.

## Look up one Record

Use **Get Record** when you need one Record. Choose one of these lookup methods on the step card:

* **By ID**: Enter a Record ID, such as `{{customerId}}`.
* **By query**: Select a Record type and add filter conditions on the Record name or metadata fields.

When several Records match, **Get Record** returns the most recently updated one. When no Record matches, the step fails and writes no output variable.

In the API or SDK, this `recordFilter` config finds a customer by email:

```json
{
  "type": "customers",
  "where": {
    "op": "and",
    "conditions": [
      {
        "field": "email",
        "op": "eq",
        "value": "{{customerEmail}}"
      }
    ]
  }
}
```

In this example, `customerEmail` is the Flow variable that contains the email address. For the available operators, see [Filtering and searching records](/user-guide/records/filtering-and-searching-records).

## Retrieve multiple Records

Use **List Records** when a lookup can match zero or more Records. The step returns an array, newest update first. Configure these fields on the step card:

* **Maximum records**: The number of Records to return. The default is 50, and the maximum is 1000.
* **If no records match**: **Succeed with an empty list** returns `[]`, which is the default. **Fail the step** fails the step instead.

After **List Records** writes to `customerMatches`, check the array length with a **Conditional Logic** step:

```plaintext
Conditional Logic: {{customerMatches.length}} > 0
  If true: Use existing customer data
  Otherwise: Create a customer Record
```

## Create and update Records

Use **Upsert Record** to create a Record or update an existing Record. The step matches on the Record type and name. It creates a Record when no Record matches and updates the matching Record otherwise.

This configuration stores an analysis object:

```plaintext
Step: Upsert Record
from variable: analysisResult
with type: analytics
named: {{eventId}}
Output variable: upsertedRecord
```

In this example, `analysisResult` must resolve to a JSON object, and `eventId` identifies the Record. To store text, add a **Run Code** step that wraps the text in an object. In the API or SDK, you can instead set `contentField` to store the text under that key. If a **Run Task** step produces the source, set its **Response Format** to **JSON**.

Set a stable Record name to update the same Record on every run. If you leave the name empty, **Upsert Record** creates a Record with a generated name on every run.

**Upsert Record** has these behaviors to plan for:

* Under **Options**, **Merge Strategy** is **Merge** by default. **Merge** keeps the stored top-level keys that the source doesn't contain and replaces each key that it does. It doesn't combine nested objects or arrays. **Replace** overwrites all metadata.
* The output variable contains `id`, `name`, `type`, `createdAt`, and `updatedAt`. It doesn't contain the metadata.
* To write one Record for each item in an array, set **Create or update records** to **Batch**. For more information, see [Upsert several Records at once](/user-guide/flows/using-record-steps-upsert-retrieve#upsert-several-records-at-once).

Use **Update Record** to change metadata on an existing Record. Set **Find record** to **From context**, **By ID**, or **By filter**, and enter the fields to change. Its **Merge** strategy combines nested objects. By default, the output variable contains a summary of the update. To read the updated metadata in a later step, turn on **Return full record**.

A write fails when the metadata exceeds the Record limits, such as 256 KB of metadata, or when the Record type has a schema in `enforce` mode and the metadata doesn't match it. Conversation history and cache Records can grow past the size limit. With no error handling set, a rejected **Upsert Record** write lets the Flow continue, sets the output variable to `null`, and saves nothing. For the limits and how each error handling setting responds, see [Write limits and schema validation](/user-guide/flows/using-record-steps-upsert-retrieve#write-limits-and-schema-validation) and [Defining a schema](/user-guide/records/defining-a-schema).

## Handle missing Records

A step that fails writes no output variable. The Flow stops only when the step's error handling is **Stop on error**. Otherwise, later steps run, and a reference such as `{{customer.id}}` stays in their input as literal text.

Handle missing Records according to the step:

* **Get Record** fails when no Record matches.
* **List Records** returns `[]` when no Record matches. Set **If no records match** to **Fail the step** when an empty result is an error.
* **Update Record** fails when no Record matches its target. An **Update Record** step that you add in the dashboard starts with **Stop on error**, so a missing target stops the Flow. To continue the Flow instead, set its error handling to **Continue on error**.

After a **List Records** step writes to `recordMatches`, use this condition to detect an empty result:

```plaintext
Conditional Logic: {{recordMatches.length}} === 0
  If true: Handle the missing Records
  Otherwise: Continue with the matching Records
```

## Design Records for Flows

Follow these guidelines when you build Flows with Records:

* Store only metadata fields that your Flows query or pass to **Run Task**.
* Use stable Record types and names when repeated runs must target the same Record.
* Set **Maximum records** to the number of Records that the Flow needs.
* Set a Record type in **Vector Search** to scope the search to one Record type.
* Test semantic search with representative questions and review the returned Records.
* Keep historical Records and mark them inactive in metadata when you need historical state.

## Build a complete RAG flow

This Flow searches Records of type `documentation`, combines the matching content, and answers a question. Before you build it, store embeddings for the `documentation` Records with **Generate Embedding** and **Store Vector**.

To build the Flow, follow these steps:

1. Add a **Vector Search** step and configure it:
   * In **Search for**, enter `{{userQuestion}}`.
   * After **in**, select **PostgreSQL (pgvector)**.
   * In **filter by type**, enter `documentation`.
   * After **returning**, enter `3`. Keep the similarity after **results with similarity ≥** at `0.7`.
   * After **using**, select the embedding model that created the stored embeddings.
   * Set the output variable to `searchResults`.

2. Add a **Run Code** step and use this script to combine the matching content:

   ```javascript
   const context = searchResults
     .map((record) => record.metadata?.content)
     .filter(Boolean)
     .join('\n\n')

   return { context }
   ```

3. Set the **Run Code** output variable to `formatContext`.

4. Add a **Run Task** step and use this prompt:

   ```plaintext
   Answer the question using only this context.

   Context:
   {{formatContext.context}}

   Question:
   {{userQuestion}}

   If the context does not contain enough information, say that you do not have enough information to answer.

   Answer:
   ```

5. Set the **Run Task** output variable to `answerQuestion`.

Each item in `searchResults` contains `id`, `name`, `type`, `metadata`, and `similarity`, with the closest match first. The script reads `metadata.content`, so it works with fewer than three results. If you search a Weaviate collection instead of Records, each item holds its fields under `properties`. For each provider's result fields, see [Read search results](/user-guide/flows/using-record-steps-upsert-retrieve#read-search-results).

If **Vector Search** returns an empty array, check these causes:

* The Records have no stored embedding.
* The embedding model in **using** differs from the model that created the embeddings.
* No Record reaches the similarity threshold. Lower the value after **results with similarity ≥**.

## Next steps

* [Using record steps (get/list/upsert)](/user-guide/flows/using-record-steps-upsert-retrieve): configure each Record step.
* [Building retrieval that works](/user-guide/records/building-retrieval-that-works): tune chunking, result counts, and similarity thresholds.
* [Flow variables and templates](/user-guide/flows/flow-variables-and-templates): reference step outputs such as `{{searchResults[0].metadata.content}}`.
* [Creating and managing records](/user-guide/records/creating-and-managing-records): create and organize Records.
* [Using prompt steps](/user-guide/flows/using-prompt-steps): generate text or structured data from Record context.