Links

Assets

Assets are external resources handled by LangStream at deployment time. You don't want to deal with creating Kafka topics or vector database tables every time you change your application, so LangStream allows you to automate the behavior of assets at deployment, update, and deletion.
The LangStream deployer creates resources during the app-setup job. If the job completes, the deployer creates another job called runtime-deployer which creates the agent CRDs. When runtime-deployer completes successfully, the application status changes to DEPLOYED. (If you get the ERROR_DEPLOYING status instead, check the logs with langstream apps logs <my-app-id>).
For example, when you initialize an Astra DB database, you might want to also create a keyspace and a table to save time. To do this, add the astra-keyspace and cassandra-table as assets to your pipeline.yaml:
name: "Write to Astra DB"
topics:
- name: "input-topic"
creation-mode: create-if-not-exists
assets:
- name: "langstream-keyspace"
asset-type: "astra-keyspace"
creation-mode: create-if-not-exists
config:
keyspace: "langstream"
datasource: "AstraDatasource"
- name: "products-table"
asset-type: "cassandra-table"
creation-mode: create-if-not-exists
config:
table-name: "products"
keyspace: "langstream"
datasource: "AstraDatasource"
create-statements:
- "CREATE TABLE IF NOT EXISTS langstream.products (id int PRIMARY KEY,name TEXT,description TEXT);"
pipeline:
- name: "Write to Astra"
type: "vector-db-sink"
input: "input-topic"
configuration:
datasource: "AstraDatasource"
table-name: "products"
keyspace: "langstream"
mapping: "id=value.id,description=value.description,name=value.name"
Notice this pipeline is using datasource type AstraDatasource. The assets behavior described here is currently only available for AstraDatasource and CassandraDataSource.
Other commands available for assets include creating indexes within your tables, and truncating tables, as below.
config:
table-name: "products"
keyspace: "langstream"
datasource: "AstraDatasource"
create-statements:
- "CREATE TABLE IF NOT EXISTS langstream.products (id int PRIMARY KEY,name TEXT,description TEXT, embeddings VECTOR<FLOAT,1536>);"
- "CREATE CUSTOM INDEX IF NOT EXISTS documents_ann_index ON documents.documents(embeddings) USING 'StorageAttachedIndex';"
delete-statements:
- "TRUNCATE TABLE langstream.products;"
For more on how AstraDatasource, CassandraDataSource, and LangStream assets interact at application deployment, see Data storage.

Configuration

Asset Type
Field
Type
Description
cassandra-table
table-name
String
Name of the table.
keyspace
String
Name of the keyspace.
create-statements
List
List of statements to create the table.
delete-statements
List
List of statements to delete the table. Required only if deletion-mode is set to "delete".
cassandra-keyspace
keyspace
String
Name of the keyspace.
create-statements
List
List of statements to create the keyspace.
delete-statements
List
List of statements to delete the keyspace. Required only if deletion-mode is set to "delete".
secureBundle
N/A
If present, an exception is thrown indicating that "astra-keyspace" should be used.
database
N/A
If present, an exception is thrown indicating that "astra-keyspace" should be used.
astra-keyspace
keyspace
String
Name of the keyspace.
token
String
Token for accessing AstraDB.
database
String
Name of the database in AstraDB.
All Asset Types
datasource
Map
Nested configuration containing a configuration field with datasource-specific configurations.
Last modified 4mo ago