diff --git a/packages/core/src/shared-exports.ts b/packages/core/src/shared-exports.ts index 271cc7bbb004..a7b3f070c2c4 100644 --- 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 index 64cda4e413c3..f6f70280e2ac 100644 --- 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 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 } { + 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, + }; +} + /** * Wraps a LangChain embedding method (embedQuery or embedDocuments) to create Sentry spans. * @@ -64,35 +90,16 @@ export function instrumentEmbeddingMethod( originalMethod: (...args: unknown[]) => Promise, options: LangChainOptions = {}, ): (...args: unknown[]) => Promise { - const { recordInputs } = resolveAIRecordingOptions(options); - return new Proxy(originalMethod, { apply(target, thisArg, args: unknown[]): Promise { - 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, - }, - () => { - 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; } diff --git a/packages/server-utils/src/integrations/tracing-channel/langchain.ts b/packages/server-utils/src/integrations/tracing-channel/langchain.ts new file mode 100644 index 000000000000..008d52638241 --- /dev/null +++ b/packages/server-utils/src/integrations/tracing-channel/langchain.ts @@ -0,0 +1,121 @@ +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 { langchainEmbeddingsChannels } from '../../orchestrion/config/langchain'; +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 | 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(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 langchainEmbeddingsChannels) { + DEBUG_BUILD && debug.log(`[orchestrion:langchain] subscribing to channel "${channelName}"`); + bindTracingChannelToSpan( + diagnosticsChannel.tracingChannel(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 index 73f8298f0a29..df6247c91d1c 100644 --- a/packages/server-utils/src/orchestrion/config/langchain.ts +++ b/packages/server-utils/src/orchestrion/config/langchain.ts @@ -1,6 +1,73 @@ import type { InstrumentationConfig } from '..'; -// 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 package. These are the provider packages the +// vendored OTel instrumentation covered that actually ship an `Embeddings` subclass: `@langchain/openai`, +// `@langchain/google-genai` and `@langchain/mistralai` define the two methods on their own class, while +// `@langchain/google-vertexai` inherits them from the shared `@langchain/google-common` base, so that base +// is the module hooked for it. (anthropic and groq are chat-only — they ship no embeddings class.) The +// `embedQuery`/`embedDocuments` channel names are per-method; orchestrion prefixes them with the module +// name, so the full channel strings stay distinct across packages. +const EMBED_QUERY = 'embedQuery'; +const EMBED_DOCUMENTS = 'embedDocuments'; + +const EMBEDDINGS_PROVIDERS = [ + { name: '@langchain/openai', versionRange: '>=0.1.0 <2.0.0', methods: [EMBED_QUERY, EMBED_DOCUMENTS] }, + { name: '@langchain/google-genai', versionRange: '>=0.1.0 <3.0.0', methods: [EMBED_QUERY, EMBED_DOCUMENTS] }, + { name: '@langchain/mistralai', versionRange: '>=0.1.0 <2.0.0', methods: [EMBED_QUERY, EMBED_DOCUMENTS] }, + // `@langchain/google-vertexai` inherits its embed methods from this shared base. The base's + // `embedQuery` delegates to `embedDocuments`, so hooking only `embedDocuments` still traces both + // entry points as a single span each, instead of emitting a nested duplicate for `embedQuery`. + { name: '@langchain/google-common', versionRange: '>=0.1.0 <3.0.0', methods: [EMBED_DOCUMENTS] }, +]; + +const embeddingsConfig = EMBEDDINGS_PROVIDERS.flatMap(({ name, versionRange, methods }) => + ['dist/embeddings.cjs', 'dist/embeddings.js'].flatMap(filePath => + methods.map(method => ({ + channelName: method, + module: { name, versionRange, filePath }, + functionQuery: { methodName: method, kind: 'Async' as const }, + })), + ), +); + +export const langchainConfig = [...chatModelConfig, ...embeddingsConfig] satisfies InstrumentationConfig[]; + +// The embeddings channel strings the subscriber binds to, derived from the provider list above so that +// adding a provider is a single edit that both instruments it and subscribes the listener to it. +export const langchainEmbeddingsChannels = EMBEDDINGS_PROVIDERS.flatMap(({ name, methods }) => + methods.map(method => `orchestrion:${name}:${method}`), +); + +export const langchainChannels = { + LANGCHAIN_CHAT_MODEL_INVOKE: 'orchestrion:@langchain/core:chatModelInvoke', + LANGCHAIN_CHAT_MODEL_STREAM: 'orchestrion:@langchain/core:chatModelStream', +} as const; diff --git a/packages/server-utils/src/orchestrion/index.ts b/packages/server-utils/src/orchestrion/index.ts index f0d6a7c3351b..eac0c5956915 100644 --- a/packages/server-utils/src/orchestrion/index.ts +++ b/packages/server-utils/src/orchestrion/index.ts @@ -13,6 +13,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 { mongooseChannelIntegration } from '../integrations/tracing-channel/mongoose'; @@ -42,6 +43,7 @@ export { ioredisChannelIntegration, kafkajsChannelIntegration, knexChannelIntegration, + langChainChannelIntegration, langGraphChannelIntegration, lruMemoizerChannelIntegration, mongooseChannelIntegration, @@ -98,6 +100,7 @@ export const channelIntegrations = { openaiIntegration: openaiChannelIntegration, anthropicIntegration: anthropicChannelIntegration, googleGenAIIntegration: googleGenAIChannelIntegration, + langChainIntegration: langChainChannelIntegration, langGraphIntegration: langGraphChannelIntegration, vercelAiIntegration: vercelAiChannelIntegration, amqplibIntegration: amqplibChannelIntegration,