chore: adds admin mode

integration-test/langfuse-integration-langchain.spec.ts

@@ -692,6 +692,131 @@ describe("Langchain", () => {
       expect((explainGeneration as any)["promptName"]).toBe(langfuseExplainPrompt.name);
       expect((explainGeneration as any)["promptVersion"]).toBe(langfuseExplainPrompt.version);
     });
+
+    it("should export events in admin mode", async () => {
+      process.env.LANGFUSE_SDK_ADMIN_ENABLED = "true";
+
+      try {
+        const handler = new CallbackHandler({
+          sessionId: "test-session",
+          userId: "test-user",
+          metadata: {
+            foo: "bar",
+            array: ["a", "b"],
+          },
+          tags: ["test-tag", "test-tag-2"],
+          version: "1.0.0",
+        });
+
+        handler.debug(true);
+
+        const messages = [new SystemMessage("You are an excellent Comedian"), new HumanMessage("Tell me a joke")];
+
+        const llm = new ChatOpenAI({ modelName: "gpt-4-turbo-preview" });
+        await llm.invoke(messages, { callbacks: [handler] });
+
+        await handler.flushAsync();
+        const shutdownResult = await handler.langfuse._shutdownAdmin();
+        if (!shutdownResult) {
+          throw new Error("No shutdown result");
+        }
+
+        console.log(JSON.stringify(shutdownResult, null, 2));
+
+        const events = shutdownResult;
+
+        expect(events.length).toBe(4);
+        const [traceCreate, generationCreate, generationUpdate, traceUpdate] = events;
+
+        // Check trace create event
+        expect(traceCreate.type).toBe("trace-create");
+        expect(traceCreate.body).toMatchObject({
+          name: "ChatOpenAI",
+          metadata: {
+            ls_provider: "openai",
+            ls_model_name: "gpt-4-turbo-preview",
+            ls_model_type: "chat",
+            ls_temperature: 1,
+            foo: "bar",
+            array: ["a", "b"],
+          },
+          userId: "test-user",
+          version: "1.0.0",
+          sessionId: "test-session",
+          input: [
+            {
+              content: "You are an excellent Comedian",
+              role: "system",
+            },
+            {
+              content: "Tell me a joke",
+              role: "user",
+            },
+          ],
+          tags: ["test-tag", "test-tag-2"],
+        });
+
+        // Check generation create event
+        expect(generationCreate.type).toBe("generation-create");
+        expect(generationCreate.body).toMatchObject({
+          name: "ChatOpenAI",
+          metadata: {
+            ls_provider: "openai",
+            ls_model_name: "gpt-4-turbo-preview",
+            ls_model_type: "chat",
+            ls_temperature: 1,
+          },
+          input: [
+            {
+              content: "You are an excellent Comedian",
+              role: "system",
+            },
+            {
+              content: "Tell me a joke",
+              role: "user",
+            },
+          ],
+          model: "gpt-4-turbo-preview",
+          modelParameters: {
+            temperature: 1,
+            top_p: 1,
+            frequency_penalty: 0,
+            presence_penalty: 0,
+          },
+          version: "1.0.0",
+        });
+
+        // Check generation update event
+        expect(generationUpdate.type).toBe("generation-update");
+        expect(generationUpdate.body).toMatchObject({
+          output: {
+            role: "assistant",
+          },
+          usage: {
+            completionTokens: expect.any(Number),
+            promptTokens: expect.any(Number),
+            totalTokens: expect.any(Number),
+          },
+          version: "1.0.0",
+        });
+
+        // Check trace update event
+        expect(traceUpdate.type).toBe("trace-create");
+        expect(traceUpdate.body).toMatchObject({
+          output: {
+            role: "assistant",
+          },
+        });
+
+        // Check IDs match between events
+        const traceId = (traceCreate.body as any).id;
+        expect((generationCreate.body as any).traceId).toBe(traceId);
+        expect((generationUpdate.body as any).traceId).toBe(traceId);
+        expect((traceUpdate.body as any).id).toBe(traceId);
+      } finally {
+        process.env.LANGFUSE_SDK_ADMIN_ENABLED = undefined;
+      }
+    });
   });
 });
 

langfuse-core/src/index.ts

@@ -181,6 +181,8 @@ abstract class LangfuseCoreStateless {
   private release: string | undefined;
   private sdkIntegration: string;
   private enabled: boolean;
+  protected adminEnabled: boolean;
+  private adminIngestionEvents: SingleIngestionEvent[] = [];
   private mask: MaskFunction | undefined;
 
   // internal
@@ -217,6 +219,7 @@ abstract class LangfuseCoreStateless {
     this.requestTimeout = options?.requestTimeout ?? 10000; // 10 seconds
 
     this.sdkIntegration = options?.sdkIntegration ?? "DEFAULT";
+    this.adminEnabled = getEnv("LANGFUSE_SDK_ADMIN_ENABLED") === "true";
   }
 
   getSdkIntegration(): string {
@@ -940,6 +943,14 @@ abstract class LangfuseCoreStateless {
       this._events.emit("flush", items);
     };
 
+    // If admin mode is enabled, we don't send the events to the server, but instead store them in the adminIngestionEvents array
+    if (this.adminEnabled) {
+      this.adminIngestionEvents.push(...items);
+
+      done();
+      return;
+    }
+
     const payload = JSON.stringify({
       batch: items,
       metadata: {
@@ -1123,6 +1134,21 @@ abstract class LangfuseCoreStateless {
     }
   }
 
+  async _shutdownAdmin(): Promise<SingleIngestionEvent[]> {
+    if (this.adminEnabled) {
+      clearTimeout(this._flushTimer);
+      await this.flushAsync();
+
+      const events = [...this.adminIngestionEvents];
+      this.adminIngestionEvents = [];
+
+      return events;
+    } else {
+      this._events.emit("error", "LANGFUSE_SDK_ADMIN_ENABLED is false, but _shutdownAdmin() was called.");
+      return [];
+    }
+  }
+
   shutdown(): void {
     console.warn(
       "shutdown() is deprecated. It does not wait for all events to be processed. Please use shutdownAsync() instead."
@@ -1172,7 +1198,9 @@ export abstract class LangfuseCore extends LangfuseCoreStateless {
     const { publicKey, secretKey, enabled } = params;
     let isObservabilityEnabled = enabled === false ? false : true;
 
-    if (!isObservabilityEnabled) {
+    if (getEnv("LANGFUSE_SDK_ADMIN_ENABLED") === "true") {
+      isObservabilityEnabled = true;
+    } else if (!isObservabilityEnabled) {
       console.warn("Langfuse is disabled. No observability data will be sent to Langfuse.");
     } else if (!secretKey) {
       isObservabilityEnabled = false;

langfuse-core/test/langfuse.flush.spec.ts

@@ -239,5 +239,29 @@ describe("Langfuse Core", () => {
 
       expect(mocks.fetch).toHaveBeenCalledTimes(1);
     });
+
+    it("should not send events in admin mode", async () => {
+      process.env.LANGFUSE_SDK_ADMIN_ENABLED = "true";
+      try {
+        [langfuse, mocks] = createTestClient({
+          publicKey: "pk-lf-111",
+          secretKey: "sk-lf-111",
+          flushAt: 5,
+          flushInterval: 200,
+        });
+
+        // Create multiple traces
+        const traces = ["test-trace-1", "test-trace-2", "test-trace-3"];
+        traces.forEach((name) => langfuse.trace({ name }));
+
+        expect(mocks.fetch).not.toHaveBeenCalled();
+
+        await jest.runAllTimersAsync();
+
+        expect(mocks.fetch).not.toHaveBeenCalled();
+      } finally {
+        process.env.LANGFUSE_SDK_ADMIN_ENABLED = undefined;
+      }
+    });
   });
 });