feat(component): add langfuse support

Author: akshatdubeysf

README.md

@@ -492,6 +492,118 @@ export class AddTool implements IGraphTool {
 }
 ```
 
+# Observability
+
+## With Langsmith
+
+You can enable langsmith observability by simply adding the Langsmith env variables. Refer [this](https://docs.langchain.com/langsmith/observability-quickstart) for more details.
+
+## With Langfuse
+
+You can enable Langfuse based tracing with the following steps -
+
+- install the following packages -
+
+```sh
+npm i @langfuse/core @langfuse/langchain @langfuse/otel
+```
+
+- adding the following component in your LB4 application -
+
+```ts
+import {LangfuseComponent} from 'lb4-llm-chat-component/langfuse';
+...
+
+this.component(LangfuseComponent);
+```
+
+- set up langfuseSpanProcessor -
+
+```ts
+import {LangfuseSpanProcessor} from '@langfuse/otel';
+import {OTLPTraceExporter} from '@opentelemetry/exporter-trace-otlp-http';
+import {DnsInstrumentation} from '@opentelemetry/instrumentation-dns';
+import {HttpInstrumentation} from '@opentelemetry/instrumentation-http';
+import {PgInstrumentation} from '@opentelemetry/instrumentation-pg';
+import {
+  defaultResource,
+  resourceFromAttributes,
+} from '@opentelemetry/resources';
+import {NodeSDK} from '@opentelemetry/sdk-node';
+import {BatchSpanProcessor, SpanProcessor} from '@opentelemetry/sdk-trace-base';
+import {
+  ATTR_SERVICE_NAME,
+  ATTR_SERVICE_VERSION,
+} from '@opentelemetry/semantic-conventions';
+import * as dotenv from 'dotenv';
+import * as dotenvExt from 'dotenv-extended';
+
+dotenv.config();
+dotenvExt.load({
+  schema: '.env.example',
+  errorOnMissing: true,
+  includeProcessEnv: true,
+});
+
+if (!!+(process.env.ENABLE_TRACING ?? 0)) {
+  const resource = defaultResource().merge(
+    resourceFromAttributes({
+      [ATTR_SERVICE_NAME]: process.env.SERVICE_NAME ?? 'reporting-service',
+      [ATTR_SERVICE_VERSION]: process.env.SERVICE_VERSION ?? '1.0.0',
+    }),
+  );
+
+  const spanProcessors: SpanProcessor[] = [];
+  const instrumentations = [];
+
+  // Add OTLP exporter if Jaeger endpoint is configured
+  if (process.env.OPENTELEMETRY_HOST && process.env.OPENTELEMETRY_PORT) {
+    const otlpExporter = new OTLPTraceExporter({
+      url: `http://${process.env.OPENTELEMETRY_HOST}:${process.env.OPENTELEMETRY_PORT}/v1/traces`,
+    });
+    spanProcessors.push(new BatchSpanProcessor(otlpExporter));
+    instrumentations.push(
+      new HttpInstrumentation(),
+      new DnsInstrumentation(),
+      new PgInstrumentation(),
+    );
+  }
+
+  if (process.env.LANGFUSE_BASE_URL) {
+    console.log('Langfuse tracing enabled');
+    spanProcessors.push(new LangfuseSpanProcessor());
+  }
+
+  const sdk = new NodeSDK({
+    resource: resource,
+    spanProcessors: spanProcessors,
+    instrumentations,
+  });
+
+  // Initialize the SDK
+  console.log('Starting OpenTelemetry SDK');
+  sdk.start();
+
+  // Graceful shutdown
+  process.on('SIGTERM', () => {
+    sdk
+      .shutdown()
+      .then(() => console.log('Tracing terminated'))
+      .catch(error => console.log('Error terminating tracing', error))
+      .finally(() => process.exit(0));
+  });
+}
+```
+
+- setup env -
+
+```sh
+export ENABLE_TRACING=1
+export LANGFUSE_SECRET_KEY="sk-lf-..."
+export LANGFUSE_PUBLIC_KEY="pk-lf-..."
+export LANGFUSE_BASE_URL="https://cloud.langfuse.com"
+```
+
 # Testing
 
 ## Generation Acceptance Builder

package-lock.json

@@ -48,6 +48,10 @@
     "./db-query/testing": {
       "type": "./dist/components/db-query/testing/index.d.ts",
       "default": "./dist/components/db-query/testing/index.js"
+    },
+    "./langfuse": {
+      "type": "./dist/sub-modules/obf/langfuse/index.d.ts",
+      "default": "./dist/sub-modules/obf/langfuse/index.js"
     }
   },
   "typesVersions": {
@@ -78,6 +82,9 @@
       ],
       "db-query/testing": [
         "dist/components/db-query/testing/index.d.ts"
+      ],
+      "langfuse": [
+        "dist/sub-modules/obf/langfuse/index.d.ts"
       ]
     }
   },
@@ -144,6 +151,8 @@
     "@langchain/groq": "^0.2.3",
     "@langchain/ollama": "^0.2.3",
     "@langchain/openai": "^0.6.11",
+    "@langfuse/core": "^4.4.2",
+    "@langfuse/langchain": "^4.4.2",
     "@loopback/build": "^11.0.9",
     "@loopback/eslint-config": "^15.0.5",
     "@loopback/testlab": "^7.0.9",

package.json

@@ -7,6 +7,7 @@ import {ToolStore} from '../../types';
 import {BaseGraph} from '../base.graph';
 import {ChatGraphAnnotation, ChatState} from '../state';
 import {ChatNodes} from './nodes.enum';
+import {AnyObject} from '@loopback/repository';
 
 @injectable({scope: BindingScope.REQUEST})
 export class ChatGraph extends BaseGraph<ChatState> {
@@ -15,6 +16,8 @@ export class ChatGraph extends BaseGraph<ChatState> {
     private readonly tools: ToolStore,
     @inject('services.TokenCounter')
     private readonly tokenCounter: TokenCounter,
+    @inject(AiIntegrationBindings.ObfHandler, {optional: true})
+    protected readonly obfHandler?: AnyObject[string],
   ) {
     super();
   }
@@ -71,6 +74,7 @@ export class ChatGraph extends BaseGraph<ChatState> {
             this.tokenCounter.handleLlmEnd(runId, output);
           },
         },
+        this.obfHandler ? this.obfHandler : {},
       ],
     });
   }

src/graphs/chat/chat.graph.ts

@@ -43,6 +43,9 @@ export namespace AiIntegrationBindings {
   export const LimitStrategy = BindingKey.create<ILimitStrategy>(
     'services.ai-reporting.limit-strategy',
   );
+  export const ObfHandler = BindingKey.create<Function>(
+    'services.ai-reporting.obf-handler',
+  );
 }
 export const WriterDB = 'writerdb';
 export const ReaderDB = 'readerdb';

src/keys.ts

@@ -0,0 +1,2 @@
+export * from './langfuse.component';
+export * from './langfuse.provider';

src/sub-modules/obf/langfuse/index.ts

@@ -0,0 +1,12 @@
+import {Component, ProviderMap} from '@loopback/core';
+import {AiIntegrationBindings} from '../../../keys';
+import {LangfuseObfProvider} from './langfuse.provider';
+
+export class LangfuseComponent implements Component {
+  providers?: ProviderMap | undefined;
+  constructor() {
+    this.providers = {
+      [AiIntegrationBindings.ObfHandler.key]: LangfuseObfProvider,
+    };
+  }
+}

src/sub-modules/obf/langfuse/langfuse.component.ts

@@ -0,0 +1,8 @@
+import {Provider, ValueOrPromise} from '@loopback/core';
+import {CallbackHandler} from '@langfuse/langchain';
+
+export class LangfuseObfProvider implements Provider<CallbackHandler> {
+  value(): ValueOrPromise<CallbackHandler> {
+    return new CallbackHandler();
+  }
+}