fix: update serialization so example works

examples/cloud/README.md

@@ -31,3 +31,11 @@ This example shows how to use the managed index with a query engine.
 ```shell
 pnpx ts-node cloud/query.ts
 ```
+
+## Pipeline
+
+This example shows how to create a managed index with a pipeline.
+
+```shell
+pnpx ts-node cloud/pipeline.ts
+```

examples/cloud/pipeline.ts

@@ -0,0 +1,34 @@
+import fs from "node:fs/promises";
+
+import {
+  Document,
+  IngestionPipeline,
+  OpenAIEmbedding,
+  SimpleNodeParser,
+} from "llamaindex";
+
+async function main() {
+  // Load essay from abramov.txt in Node
+  const path = "node_modules/llamaindex/examples/abramov.txt";
+
+  const essay = await fs.readFile(path, "utf-8");
+
+  // Create Document object with essay
+  const document = new Document({ text: essay, id_: path });
+  const pipeline = new IngestionPipeline({
+    name: "pipeline",
+    transformations: [
+      new SimpleNodeParser({ chunkSize: 1024, chunkOverlap: 20 }),
+      new OpenAIEmbedding({ apiKey: "api-key" }),
+    ],
+  });
+
+  const pipelineId = await pipeline.register({
+    documents: [document],
+    verbose: true,
+  });
+
+  console.log(`Pipeline with id ${pipelineId} successfully created.`);
+}
+
+main().catch(console.error);

packages/core/src/cloud/config.ts

@@ -0,0 +1,81 @@
+import type { PlatformApi } from "@llamaindex/cloud";
+import { BaseNode, Document } from "../Node.js";
+import { OpenAIEmbedding } from "../embeddings/OpenAIEmbedding.js";
+import type { TransformComponent } from "../ingestion/types.js";
+import { SimpleNodeParser } from "../nodeParsers/SimpleNodeParser.js";
+
+export type GetPipelineCreateParams = {
+  pipelineName: string;
+  pipelineType: any; // TODO: PlatformApi.PipelineType is not exported
+  transformations?: TransformComponent[];
+  inputNodes?: BaseNode[];
+};
+
+function getTransformationConfig(
+  transformation: TransformComponent,
+): PlatformApi.ConfiguredTransformationItem {
+  if (transformation instanceof SimpleNodeParser) {
+    return {
+      configurableTransformationType: "SENTENCE_AWARE_NODE_PARSER",
+      component: {
+        // TODO: API returns 422 if these parameters are included
+        // chunkSize: transformation.textSplitter.chunkSize, // TODO: set to public in SentenceSplitter
+        // chunkOverlap: transformation.textSplitter.chunkOverlap, // TODO: set to public in SentenceSplitter
+        // includeMetadata: transformation.includeMetadata,
+        // includePrevNextRel: transformation.includePrevNextRel,
+      },
+    };
+  }
+  if (transformation instanceof OpenAIEmbedding) {
+    return {
+      configurableTransformationType: "OPENAI_EMBEDDING",
+      component: {
+        modelName: transformation.model,
+        apiKey: transformation.apiKey,
+        embedBatchSize: transformation.embedBatchSize,
+        dimensions: transformation.dimensions,
+      },
+    };
+  }
+  throw new Error(`Unsupported transformation: ${typeof transformation}`);
+}
+
+function getDataSourceConfig(node: BaseNode): PlatformApi.DataSourceCreate {
+  if (node instanceof Document) {
+    return {
+      name: node.id_,
+      sourceType: "DOCUMENT",
+      component: {
+        id: node.id_,
+        text: node.text,
+        textTemplate: node.textTemplate,
+        startCharIdx: node.startCharIdx,
+        endCharIdx: node.endCharIdx,
+        metadataSeparator: node.metadataSeparator,
+        excludedEmbedMetadataKeys: node.excludedEmbedMetadataKeys,
+        excludedLlmMetadataKeys: node.excludedLlmMetadataKeys,
+        extraInfo: node.metadata,
+      },
+    };
+  }
+  throw new Error(`Unsupported node: ${typeof node}`);
+}
+
+export async function getPipelineCreate(
+  params: GetPipelineCreateParams,
+): Promise<PlatformApi.PipelineCreate> {
+  const {
+    pipelineName,
+    pipelineType,
+    transformations = [],
+    inputNodes = [],
+  } = params;
+
+  return {
+    name: pipelineName,
+    configuredTransformations: transformations.map(getTransformationConfig),
+    dataSources: inputNodes.map(getDataSourceConfig),
+    dataSinks: [],
+    pipelineType,
+  };
+}

packages/core/src/cloud/types.ts

@@ -1,5 +1,6 @@
 import type { ServiceContext } from "../ServiceContext.js";
 
+export const DEFAULT_PIPELINE_NAME = "default";
 export const DEFAULT_PROJECT_NAME = "default";
 export const DEFAULT_BASE_URL = "https://api.cloud.llamaindex.ai";
 

packages/core/src/cloud/utils.ts

@@ -3,12 +3,20 @@ import { getEnv } from "@llamaindex/env";
 import type { ClientParams } from "./types.js";
 import { DEFAULT_BASE_URL } from "./types.js";
 
+function getBaseUrl(baseUrl?: string): string {
+  return baseUrl ?? getEnv("LLAMA_CLOUD_BASE_URL") ?? DEFAULT_BASE_URL;
+}
+
+export function getAppBaseUrl(baseUrl?: string): string {
+  return getBaseUrl(baseUrl).replace(/api\./, "");
+}
+
 export async function getClient({
   apiKey,
   baseUrl,
 }: ClientParams = {}): Promise<PlatformApiClient> {
   // Get the environment variables or use defaults
-  baseUrl = baseUrl ?? getEnv("LLAMA_CLOUD_BASE_URL") ?? DEFAULT_BASE_URL;
+  baseUrl = getBaseUrl(baseUrl);
   apiKey = apiKey ?? getEnv("LLAMA_CLOUD_API_KEY");
 
   const { PlatformApiClient } = await import("@llamaindex/cloud");

packages/core/src/ingestion/IngestionPipeline.ts

@@ -1,4 +1,12 @@
+import type { PlatformApiClient } from "@llamaindex/cloud";
 import type { BaseNode, Document } from "../Node.js";
+import { getPipelineCreate } from "../cloud/config.js";
+import {
+  DEFAULT_PIPELINE_NAME,
+  DEFAULT_PROJECT_NAME,
+  type ClientParams,
+} from "../cloud/types.js";
+import { getAppBaseUrl, getClient } from "../cloud/utils.js";
 import type { BaseReader } from "../readers/type.js";
 import type { BaseDocumentStore } from "../storage/docStore/types.js";
 import type { VectorStore } from "../storage/vectorStore/types.js";
@@ -55,11 +63,16 @@ export class IngestionPipeline {
   docStoreStrategy: DocStoreStrategy = DocStoreStrategy.UPSERTS;
   cache?: IngestionCache;
   disableCache: boolean = false;
+  client?: PlatformApiClient;
+  clientParams?: ClientParams;
+  projectName: string = DEFAULT_PROJECT_NAME;
+  name: string = DEFAULT_PIPELINE_NAME;
 
   private _docStoreStrategy?: TransformComponent;
 
-  constructor(init?: Partial<IngestionPipeline>) {
+  constructor(init?: Partial<IngestionPipeline> & ClientParams) {
     Object.assign(this, init);
+    this.clientParams = { apiKey: init?.apiKey, baseUrl: init?.baseUrl };
     this._docStoreStrategy = createDocStoreStrategy(
       this.docStoreStrategy,
       this.docStore,
@@ -115,4 +128,50 @@ export class IngestionPipeline {
     }
     return nodes;
   }
+
+  private async getClient(): Promise<PlatformApiClient> {
+    if (!this.client) {
+      this.client = await getClient(this.clientParams);
+    }
+    return this.client;
+  }
+
+  async register(params: {
+    documents?: Document[];
+    nodes?: BaseNode[];
+    verbose?: boolean;
+  }): Promise<string> {
+    const client = await this.getClient();
+
+    const inputNodes = await this.prepareInput(params.documents, params.nodes);
+    const project = await client.project.upsertProject({
+      name: this.projectName,
+    });
+    if (!project.id) {
+      throw new Error("Project ID should be defined");
+    }
+
+    // upload
+    const pipeline = await client.project.upsertPipelineForProject(
+      project.id,
+      await getPipelineCreate({
+        pipelineName: this.name,
+        pipelineType: "PLAYGROUND",
+        transformations: this.transformations,
+        inputNodes,
+      }),
+    );
+    if (!pipeline.id) {
+      throw new Error("Pipeline ID must be defined");
+    }
+
+    // Print playground URL if not running remote
+    if (params.verbose) {
+      console.log(
+        `Pipeline available at: ${getAppBaseUrl(this.clientParams?.baseUrl)}/project/${project.id}/playground/${pipeline.id}`,
+      );
+    }
+
+    return pipeline.id;
+  }
 }