feat(server-utils): Rewrite LangChain integration to orchestrion

logaretm · logaretm · commit fcebfb9c3ace · 2026-07-16T11:37:13.000-04:00
Rewrites the LangChain integration to a diagnostics-channel listener
instead of `InstrumentationBase`, with orchestrion injecting the channels.

Chat models are hooked once on `@langchain/core`'s `BaseChatModel`
(`invoke` + `_streamIterator`), which every provider class inherits, so a
single module covers all providers. The listener injects the Sentry
callback handler into the call options, and LangChain's own callback
dispatch creates the spans exactly as before, so the span output is
identical. Embeddings have no shared base method, so they're hooked per
provider (`@langchain/openai` for now) and get their span via
`bindTracingChannelToSpan`, reusing the same core span builder as the
OTel path.

The OTel path stays as the fallback when orchestrion isn't injected. The
existing node-integration suite runs against both paths under
`INJECT_ORCHESTRION`, proving parity for langchain v0.3 and v1.
diff --git a/packages/core/src/shared-exports.ts b/packages/core/src/shared-exports.ts
@@ -216,6 +216,7 @@ export { instrumentStream as instrumentGoogleGenAIStream } from './tracing/googl
 export { GOOGLE_GENAI_INTEGRATION_NAME } from './tracing/google-genai/constants';
 export type { GoogleGenAIResponse } from './tracing/google-genai/types';
 export { createLangChainCallbackHandler, instrumentLangChainEmbeddings } from './tracing/langchain';
+export { _INTERNAL_getLangChainEmbeddingsSpanOptions } from './tracing/langchain/embeddings';
 export { _INTERNAL_mergeLangChainCallbackHandler } from './tracing/langchain/utils';
 export { LANGCHAIN_INTEGRATION_NAME } from './tracing/langchain/constants';
 export type { LangChainOptions, LangChainIntegration } from './tracing/langchain/types';
diff --git a/packages/core/src/tracing/langchain/embeddings.ts b/packages/core/src/tracing/langchain/embeddings.ts
@@ -55,6 +55,32 @@ function extractEmbeddingAttributes(instance: unknown): Record<string, unknown>
   return attributes;
 }
 
+/**
+ * Builds the span options for a LangChain embedding call from the embeddings instance and input.
+ *
+ * @internal Exported so the diagnostics-channel (orchestrion) instrumentation can build the same
+ * span as the prototype-patching path below.
+ */
+export function _INTERNAL_getLangChainEmbeddingsSpanOptions(
+  instance: unknown,
+  input: unknown,
+  options: LangChainOptions = {},
+): { name: string; op: string; attributes: Record<string, SpanAttributeValue> } {
+  const { recordInputs } = resolveAIRecordingOptions(options);
+  const attributes = extractEmbeddingAttributes(instance);
+  const modelName = attributes[GEN_AI_REQUEST_MODEL_ATTRIBUTE] || 'unknown';
+
+  if (recordInputs && input != null) {
+    attributes[GEN_AI_EMBEDDINGS_INPUT_ATTRIBUTE] = typeof input === 'string' ? input : JSON.stringify(input);
+  }
+
+  return {
+    name: `embeddings ${modelName}`,
+    op: GEN_AI_EMBEDDINGS_OPERATION_ATTRIBUTE,
+    attributes: attributes as Record<string, SpanAttributeValue>,
+  };
+}
+
 /**
  * Wraps a LangChain embedding method (embedQuery or embedDocuments) to create Sentry spans.
  *
@@ -64,35 +90,16 @@ export function instrumentEmbeddingMethod(
   originalMethod: (...args: unknown[]) => Promise<unknown>,
   options: LangChainOptions = {},
 ): (...args: unknown[]) => Promise<unknown> {
-  const { recordInputs } = resolveAIRecordingOptions(options);
-
   return new Proxy(originalMethod, {
     apply(target, thisArg, args: unknown[]): Promise<unknown> {
-      const attributes = extractEmbeddingAttributes(thisArg);
-      const modelName = attributes[GEN_AI_REQUEST_MODEL_ATTRIBUTE] || 'unknown';
-
-      if (recordInputs) {
-        const input = args[0];
-        if (input != null) {
-          attributes[GEN_AI_EMBEDDINGS_INPUT_ATTRIBUTE] = typeof input === 'string' ? input : JSON.stringify(input);
-        }
-      }
-
-      return startSpan(
-        {
-          name: `embeddings ${modelName}`,
-          op: GEN_AI_EMBEDDINGS_OPERATION_ATTRIBUTE,
-          attributes: attributes as Record<string, SpanAttributeValue>,
-        },
-        () => {
-          return Reflect.apply(target, thisArg, args).then(undefined, error => {
-            captureException(error, {
-              mechanism: { handled: false, type: 'auto.ai.langchain' },
-            });
-            throw error;
+      return startSpan(_INTERNAL_getLangChainEmbeddingsSpanOptions(thisArg, args[0], options), () => {
+        return Reflect.apply(target, thisArg, args).then(undefined, error => {
+          captureException(error, {
+            mechanism: { handled: false, type: 'auto.ai.langchain' },
           });
-        },
-      );
+          throw error;
+        });
+      });
     },
   }) as (...args: unknown[]) => Promise<unknown>;
 }
diff --git a/packages/server-utils/src/integrations/tracing-channel/langchain.ts b/packages/server-utils/src/integrations/tracing-channel/langchain.ts
@@ -0,0 +1,120 @@
+import * as diagnosticsChannel from 'node:diagnostics_channel';
+import type { IntegrationFn, LangChainOptions, Span } from '@sentry/core';
+import {
+  _INTERNAL_getLangChainEmbeddingsSpanOptions,
+  _INTERNAL_mergeLangChainCallbackHandler,
+  _INTERNAL_skipAiProviderWrapping,
+  ANTHROPIC_AI_INTEGRATION_NAME,
+  createLangChainCallbackHandler,
+  debug,
+  defineIntegration,
+  GOOGLE_GENAI_INTEGRATION_NAME,
+  LANGCHAIN_INTEGRATION_NAME,
+  OPENAI_INTEGRATION_NAME,
+  startInactiveSpan,
+  waitForTracingChannelBinding,
+} from '@sentry/core';
+import { DEBUG_BUILD } from '../../debug-build';
+import { CHANNELS } from '../../orchestrion/channels';
+import { bindTracingChannelToSpan } from '../../tracing-channel';
+
+// Same name as the OTel integration by design: when enabled, the OTel 'LangChain' integration is
+// dropped from the default set (see the Node opt-in loader).
+const INTEGRATION_NAME = LANGCHAIN_INTEGRATION_NAME;
+
+// LangChain drives the underlying AI provider SDKs itself, so while it's active those providers must
+// not also instrument, or every call would produce two spans (mirrors the OTel path's skip list).
+const SKIPPED_PROVIDERS = [OPENAI_INTEGRATION_NAME, ANTHROPIC_AI_INTEGRATION_NAME, GOOGLE_GENAI_INTEGRATION_NAME];
+
+// The chat-model channels carry the live args array of `invoke(input, options)` / `_streamIterator(input, options)`.
+interface RunnableChannelContext {
+  arguments: unknown[];
+}
+
+// The embeddings channels carry the instance (`self`) and the `embedQuery(text)` / `embedDocuments(texts)` args.
+interface EmbeddingsChannelContext {
+  self?: unknown;
+  arguments: unknown[];
+}
+
+let subscribed = false;
+
+// Registered lazily on the first LangChain call (not at `setupOnce`) so a direct provider call made
+// before any LangChain call still gets its own span — matches the OTel patch-on-import timing. It
+// also stops the underlying SDK from double-instrumenting embeddings, whose `embedQuery`/
+// `embedDocuments` call the provider SDK (e.g. `openai`) internally.
+function markProvidersSkipped(): void {
+  _INTERNAL_skipAiProviderWrapping(SKIPPED_PROVIDERS);
+}
+
+const _langChainChannelIntegration = ((options: LangChainOptions = {}) => {
+  return {
+    name: INTEGRATION_NAME,
+    setupOnce() {
+      // `tracingChannel` is unavailable before Node 18.19, and a second `init()` would double-subscribe.
+      if (!diagnosticsChannel.tracingChannel || subscribed) {
+        return;
+      }
+      subscribed = true;
+
+      // One stateful handler tracks spans across the whole run tree, just like the OTel path.
+      const sentryHandler = createLangChainCallbackHandler(options);
+
+      // Chat models: inject the Sentry callback handler into the call options (arg 1). LangChain's own
+      // callback dispatch then creates the spans, exactly as in the OTel path, so no span is opened
+      // here — a `start` subscriber (which also makes orchestrion wrap the function) is enough.
+      const injectHandler = (message: unknown): void => {
+        markProvidersSkipped();
+
+        const args = (message as RunnableChannelContext).arguments;
+        if (!Array.isArray(args)) {
+          return;
+        }
+
+        let callOptions = args[1] as Record<string, unknown> | undefined;
+        if (!callOptions || typeof callOptions !== 'object' || Array.isArray(callOptions)) {
+          callOptions = {};
+          args[1] = callOptions;
+        }
+
+        callOptions.callbacks = _INTERNAL_mergeLangChainCallbackHandler(callOptions.callbacks, sentryHandler);
+      };
+
+      for (const channelName of [CHANNELS.LANGCHAIN_CHAT_MODEL_INVOKE, CHANNELS.LANGCHAIN_CHAT_MODEL_STREAM]) {
+        DEBUG_BUILD && debug.log(`[orchestrion:langchain] subscribing to channel "${channelName}"`);
+        diagnosticsChannel.tracingChannel<RunnableChannelContext>(channelName).start.subscribe(injectHandler);
+      }
+
+      // Embeddings don't use the callback system — the OTel path wraps the method in its own span, so
+      // do the same here. `bindTracingChannelToSpan` needs the async-context binding that
+      // `initOpenTelemetry()` registers after `setupOnce`, so wait for it before subscribing.
+      waitForTracingChannelBinding(() => {
+        for (const channelName of [CHANNELS.LANGCHAIN_EMBED_QUERY, CHANNELS.LANGCHAIN_EMBED_DOCUMENTS]) {
+          DEBUG_BUILD && debug.log(`[orchestrion:langchain] subscribing to channel "${channelName}"`);
+          bindTracingChannelToSpan(
+            diagnosticsChannel.tracingChannel<EmbeddingsChannelContext>(channelName),
+            data => createEmbeddingsSpan(data, options),
+            { captureError: () => ({ mechanism: { handled: false, type: 'auto.ai.langchain' } }) },
+          );
+        }
+      });
+    },
+  };
+}) satisfies IntegrationFn;
+
+function createEmbeddingsSpan(data: EmbeddingsChannelContext, options: LangChainOptions): Span {
+  // `embedQuery`/`embedDocuments` call the provider SDK internally, so skip that SDK's own
+  // instrumentation before its channel fires (the producer runs at the embeddings channel's `start`).
+  markProvidersSkipped();
+
+  const input = (data.arguments ?? [])[0];
+
+  return startInactiveSpan(_INTERNAL_getLangChainEmbeddingsSpanOptions(data.self, input, options));
+}
+
+/**
+ * EXPERIMENTAL — orchestrion-driven LangChain integration. Subscribes to the diagnostics_channels
+ * injected into `@langchain/core`'s `BaseChatModel` (to inject the Sentry callback handler) and into
+ * `@langchain/openai`'s embedding methods, so it requires the orchestrion runtime hook or bundler plugin.
+ */
+export const langChainChannelIntegration = defineIntegration(_langChainChannelIntegration);
diff --git a/packages/server-utils/src/orchestrion/config/langchain.ts b/packages/server-utils/src/orchestrion/config/langchain.ts
@@ -1,6 +1,51 @@
 import type { InstrumentationConfig } from '@apm-js-collab/code-transformer';
 
-// TODO: Stub for the `langchain` orchestrion integration (ports `SentryLangChainInstrumentation`).
-export const langchainConfig: InstrumentationConfig[] = [];
+// `@langchain/*` packages ship dual CJS/ESM builds (`.cjs` for `require`, `.js` for `import`) and the
+// matcher compares `filePath` exactly, so each hook is declared once per built file.
 
-export const langchainChannels = {} as const;
+// LangChain's chat model methods live on `BaseChatModel` in `@langchain/core` and are inherited by
+// every provider class (`ChatAnthropic`, `ChatOpenAI`, …), so a single hook there covers all
+// providers. `invoke` also backs `.batch()` (which calls `invoke` per item); `_streamIterator`
+// backs `.stream()`. The vendored OTel instrumentation instead patched each provider package to
+// dodge `@langchain/core` being bundled, but orchestrion transforms its source directly regardless
+// of bundling.
+const chatModelConfig = ['dist/language_models/chat_models.cjs', 'dist/language_models/chat_models.js'].flatMap(
+  filePath => {
+    const module = { name: '@langchain/core', versionRange: '>=0.1.0 <2.0.0', filePath };
+
+    return [
+      {
+        channelName: 'chatModelInvoke',
+        module,
+        functionQuery: { className: 'BaseChatModel', methodName: 'invoke', kind: 'Async' as const },
+      },
+      {
+        channelName: 'chatModelStream',
+        module,
+        functionQuery: { className: 'BaseChatModel', methodName: '_streamIterator', kind: 'Async' as const },
+      },
+    ];
+  },
+);
+
+// Embeddings have no shared concrete method on the base class (each provider implements
+// `embedQuery`/`embedDocuments`), so they're hooked per provider. Only `@langchain/openai` is wired
+// for now; other providers follow the same shape (a class extending `Embeddings` with async
+// `embedQuery`/`embedDocuments`).
+const embeddingsConfig = ['dist/embeddings.cjs', 'dist/embeddings.js'].flatMap(filePath => {
+  const module = { name: '@langchain/openai', versionRange: '>=0.1.0 <2.0.0', filePath };
+
+  return [
+    { channelName: 'embedQuery', module, functionQuery: { methodName: 'embedQuery', kind: 'Async' as const } },
+    { channelName: 'embedDocuments', module, functionQuery: { methodName: 'embedDocuments', kind: 'Async' as const } },
+  ];
+});
+
+export const langchainConfig = [...chatModelConfig, ...embeddingsConfig] satisfies InstrumentationConfig[];
+
+export const langchainChannels = {
+  LANGCHAIN_CHAT_MODEL_INVOKE: 'orchestrion:@langchain/core:chatModelInvoke',
+  LANGCHAIN_CHAT_MODEL_STREAM: 'orchestrion:@langchain/core:chatModelStream',
+  LANGCHAIN_EMBED_QUERY: 'orchestrion:@langchain/openai:embedQuery',
+  LANGCHAIN_EMBED_DOCUMENTS: 'orchestrion:@langchain/openai:embedDocuments',
+} as const;
diff --git a/packages/server-utils/src/orchestrion/index.ts b/packages/server-utils/src/orchestrion/index.ts
@@ -12,6 +12,7 @@ import { koaChannelIntegration } from '../integrations/tracing-channel/koa';
 import { ioredisChannelIntegration } from '../integrations/tracing-channel/ioredis';
 import { kafkajsChannelIntegration } from '../integrations/tracing-channel/kafkajs';
 import { knexChannelIntegration } from '../integrations/tracing-channel/knex';
+import { langChainChannelIntegration } from '../integrations/tracing-channel/langchain';
 import { langGraphChannelIntegration } from '../integrations/tracing-channel/langgraph';
 import { lruMemoizerChannelIntegration } from '../integrations/tracing-channel/lru-memoizer';
 import { mysqlChannelIntegration } from '../integrations/tracing-channel/mysql';
@@ -38,6 +39,7 @@ export {
   ioredisChannelIntegration,
   kafkajsChannelIntegration,
   knexChannelIntegration,
+  langChainChannelIntegration,
   langGraphChannelIntegration,
   lruMemoizerChannelIntegration,
   mysqlChannelIntegration,
@@ -90,6 +92,7 @@ export const channelIntegrations = {
   openaiIntegration: openaiChannelIntegration,
   anthropicIntegration: anthropicChannelIntegration,
   googleGenAIIntegration: googleGenAIChannelIntegration,
+  langChainIntegration: langChainChannelIntegration,
   langGraphIntegration: langGraphChannelIntegration,
   vercelAiIntegration: vercelAiChannelIntegration,
   amqplibIntegration: amqplibChannelIntegration,
