Pipelines
Pipelines define the schema for the transformation of documents. Different Pipelines can be used for different tasks.
Defining Schema
New Pipelines require schema. Here are a few examples of variations of schema along with common use cases.
For the following section we will assume we have documents that have the structure:
{
"id": "Each document has a unique id",
"title": "Each document has a title",
"body": "Each document has some body text"
}
const pipeline = pgml.newPipeline("test_pipeline", {
title: {
full_text_search: { configuration: "english" },
},
body: {
splitter: { model: "recursive_character" },
semantic_search: {
model: "hkunlp/instructor-base",
parameters: {
instruction: "Represent the Wikipedia document for retrieval: ",
}
},
},
});
pipeline = Pipeline(
"test_pipeline",
{
"title": {
"full_text_search": {"configuration": "english"},
},
"body": {
"splitter": {"model": "recursive_character"},
"semantic_search": {
"model": "hkunlp/instructor-base",
"parameters": {
"instruction": "Represent the Wikipedia document for retrieval: ",
},
},
},
},
)
This Pipeline does two things. For each document in the Collection, it converts all titles into tsvectors enabling full text search, and splits and embeds the body text enabling semantic search using vectors. This kind of Pipeline would be great for site search utilizing hybrid keyword and semantic search.
For a more simple RAG use case, the following Pipeline would work well.
const pipeline = pgml.newPipeline("test_pipeline", {
body: {
splitter: { model: "recursive_character" },
semantic_search: {
model: "hkunlp/instructor-base",
parameters: {
instruction: "Represent the Wikipedia document for retrieval: ",
}
},
},
});
pipeline = Pipeline(
"test_pipeline",
{
"body": {
"splitter": {"model": "recursive_character"},
"semantic_search": {
"model": "hkunlp/instructor-base",
"parameters": {
"instruction": "Represent the Wikipedia document for retrieval: ",
},
},
},
},
)
This Pipeline splits and embeds the body text enabling semantic search using vectors. This is a very popular Pipeline for RAG.
We support most every open source model on Hugging Face, and OpenAI's embedding models. To use a model from OpenAI specify the source as openai, and make sure and set the environment variable OPENAI_API_KEY.
const pipeline = pgml.newPipeline("test_pipeline", {
body: {
splitter: { model: "recursive_character" },
semantic_search: {
model: "text-embedding-ada-002",
source: "openai"
},
},
});
pipeline = Pipeline(
"test_pipeline",
{
"body": {
"splitter": {"model": "recursive_character"},
"semantic_search": {"model": "text-embedding-ada-002", "source": "openai"},
},
},
)
Customizing the Indexes
By default the SDK uses HNSW indexes to efficiently perform vector recall. The default HNSW index sets m to 16 and ef_construction to 64. These defaults can be customized in the Pipeline schema. See pgvector for more information on vector indexes.
const pipeline = pgml.newPipeline("test_pipeline", {
body: {
splitter: { model: "recursive_character" },
semantic_search: {
model: "intfloat/e5-small",
hnsw: {
m: 100,
ef_construction: 200
}
},
},
});
pipeline = Pipeline(
"test_pipeline",
{
"body": {
"splitter": {"model": "recursive_character"},
"semantic_search": {
"model": "intfloat/e5-small",
"hnsw": {"m": 100, "ef_construction": 200},
},
},
},
)
Adding Pipelines to a Collection
The first time a Pipeline is added to a Collection it will automatically chunk and embed any documents already in that Collection.
await collection.add_pipeline(pipeline)
await collection.add_pipeline(pipeline)
Note: After a
Pipelinehas been added to aCollectioninstances of thePipelineobject can be created without specifying a schema:
const pipeline = pgml.newPipeline("test_pipeline")
pipeline = Pipeline("test_pipeline")
Searching with Pipelines
There are two different forms of search that can be done after adding a Pipeline to a Collection
See their respective pages for more information on searching.
Disable a Pipeline
Pipelines can be disabled or removed to prevent them from running automatically when documents are upserted.
const pipeline = pgml.newPipeline("test_pipeline")
const collection = pgml.newCollection("test_collection")
await collection.disable_pipeline(pipeline)
pipeline = Pipeline("test_pipeline")
collection = Collection("test_collection")
await collection.disable_pipeline(pipeline)
Disabling a Pipeline prevents it from running automatically, but leaves all tsvectors, chunks, and embeddings already created by that Pipeline in the database.
Enable a Pipeline
Disabled Pipelines can be re-enabled.
const pipeline = pgml.newPipeline("test_pipeline")
const collection = pgml.newCollection("test_collection")
await collection.enable_pipeline(pipeline)
pipeline = Pipeline("test_pipeline")
collection = Collection("test_collection")
await collection.enable_pipeline(pipeline)
Enabling a Pipeline will cause it to automatically run on all documents it may have missed while disabled.
Remove a Pipeline
const pipeline = pgml.newPipeline("test_pipeline")
const collection = pgml.newCollection("test_collection")
await collection.remove_pipeline(pipeline)
pipeline = Pipeline("test_pipeline")
collection = Collection("test_collection")
await collection.remove_pipeline(pipeline)
Removing a Pipeline deletes it and all associated data from the database. Removed Pipelines cannot be re-enabled but can be recreated.
