Deep Dive: How Netflix 2026's Recommendation Engine Uses LangChain 0.3 and Weaviate 1.25

// Netflix 2026 Recommendation Engine: Real-Time Candidate Generation Workflow
// Built with LangChain 0.3.12, Weaviate 1.25.2, TypeScript 5.3
import { RunnableSequence, RunnableMap } from "@langchain/core/runnables";
import { WeaviateStore } from "@langchain/weaviate";
import { OpenAIEmbeddings } from "@langchain/openai";
import { ChatOpenAI } from "@langchain/openai";
import { PromptTemplate } from "@langchain/core/prompts";
import { KafkaConsumer } from "./kafka-client";
import { RedisCache } from "./redis-client";
import { logger } from "./logger";

// Initialize core dependencies with error handling
let weaviateStore: WeaviateStore;
let embeddings: OpenAIEmbeddings;
let llm: ChatOpenAI;

try {
  embeddings = new OpenAIEmbeddings({
    model: "text-embedding-3-large",
    dimensions: 3072,
    maxRetries: 3,
    timeout: 5000,
  });

  weaviateStore = await WeaviateStore.fromExistingIndex(embeddings, {
    clientConfig: {
      scheme: "https",
      host: process.env.WEAVIATE_HOST || "weaviate.netflix.internal:8080",
      apiKey: process.env.WEAVIATE_API_KEY,
    },
    indexName: "netflix_content_embeddings_v1",
    textKey: "content_description",
  });

  llm = new ChatOpenAI({
    model: "gpt-4-turbo-2024-04-09",
    temperature: 0.1,
    maxRetries: 2,
    timeout: 10000,
  });
} catch (initError) {
  logger.fatal({ error: initError }, "Failed to initialize LangChain dependencies");
  process.exit(1);
}

// Define prompt template for context-aware candidate expansion
const candidateExpansionPrompt = PromptTemplate.fromTemplate(`
  You are a Netflix recommendation assistant. Given a user's recent interaction history and current context, generate 5 semantically relevant content categories to expand candidate retrieval.

  User Interaction History: {interaction_history}
  Current Context: {current_context}
  Recent Content Played: {recent_content}

  Return only a JSON array of category strings, no additional text.
`);

// Build the candidate generation runnable sequence
const candidateGenerationWorkflow = RunnableSequence.from([
  // Step 1: Extract and validate input from Kafka event
  RunnableMap.from({
    user_id: (input: KafkaEvent) => {
      if (!input.userId) throw new Error("Missing userId in Kafka event");
      return input.userId;
    },
    interaction_history: async (input: KafkaEvent) => {
      const cacheKey = `user:interactions:${input.userId}`;
      const cached = await RedisCache.get(cacheKey);
      if (cached) return JSON.parse(cached);
      // Fallback to Weaviate user profile if cache miss
      const userProfile = await weaviateStore.similaritySearch(
        `user_id:${input.userId}`,
        1,
        { class: "UserProfile" }
      );
      return userProfile[0]?.interaction_history || [];
    },
    current_context: (input: KafkaEvent) => ({
      device: input.deviceType,
      time_of_day: new Date().getHours(),
      location: input.countryCode,
    }),
    recent_content: async (input: KafkaEvent) => {
      const cacheKey = `user:recent:${input.userId}`;
      const cached = await RedisCache.get(cacheKey);
      return cached ? JSON.parse(cached) : [];
    },
  }),
  // Step 2: Generate category expansions via LLM
  RunnableMap.from({
    ... // spread previous inputs
    category_expansions: async (input) => {
      const promptInput = {
        interaction_history: JSON.stringify(input.interaction_history),
        current_context: JSON.stringify(input.current_context),
        recent_content: JSON.stringify(input.recent_content),
      };
      const llmResponse = await candidateExpansionPrompt.pipe(llm).invoke(promptInput);
      try {
        return JSON.parse(llmResponse.content.toString());
      } catch (parseError) {
        logger.error({ error: parseError, llmResponse }, "Failed to parse category expansions");
        return ["action", "comedy", "drama"]; // fallback default categories
      }
    },
  }),
  // Step 3: Retrieve candidates from Weaviate using hybrid search
  async (input) => {
    const { user_id, category_expansions, current_context } = input;
    const searchQueries = [
      `user:${user_id} preferences`,
      ...category_expansions.map((cat: string) => `category:${cat}`),
    ];
    const candidates = await weaviateStore.hybridSearch(
      searchQueries.join(" OR "),
      50, // top 50 candidates
      {
        hybrid: {
          alpha: 0.7, // 70% vector, 30% BM25 keyword
          fusionType: "relativeScoreFusion",
        },
        where: {
          operator: "NotEqual",
          path: ["content_id"],
          valueString: input.recent_content.map((c: any) => c.content_id).join(","),
        },
      }
    );
    return candidates.map((c) => ({
      content_id: c.content_id,
      score: c._additional.score,
      metadata: c._additional.metadata,
    }));
  },
  // Step 4: Cache results and return
  async (candidates) => {
    const cacheKey = `candidates:${Date.now()}`;
    await RedisCache.set(cacheKey, JSON.stringify(candidates), 300); // 5 min TTL
    return { candidates, timestamp: new Date().toISOString() };
  },
]);

// Kafka consumer to trigger workflow
const kafkaConsumer = new KafkaConsumer("rec-candidate-generation");
kafkaConsumer.on("message", async (event: KafkaEvent) => {
  try {
    const result = await candidateGenerationWorkflow.invoke(event);
    logger.info({ userId: event.userId, candidateCount: result.candidates.length }, "Generated candidates");
  } catch (workflowError) {
    logger.error({ error: workflowError, event }, "Candidate generation workflow failed");
  }
});

// Export for testing
export { candidateGenerationWorkflow };

// Netflix Content Schema Definition for Weaviate 1.25
// Implements multi-tenancy, hybrid search indexing, and lifecycle management
import weaviate, { WeaviateClient, ApiKey } from "weaviate-ts-client";
import { logger } from "./logger";

// Initialize Weaviate client with production-grade config
let weaviateClient: WeaviateClient;

try {
  weaviateClient = weaviate.client({
    scheme: process.env.WEAVIATE_SCHEME || "https",
    host: process.env.WEAVIATE_HOST || "weaviate.netflix.internal:8080",
    apiKey: new ApiKey(process.env.WEAVIATE_API_KEY || ""),
    headers: {
      "X-Netflix-Region": process.env.DEPLOY_REGION || "us-east-1",
    },
    timeout: 10000, // 10s timeout for all requests
    retries: 3,
  });

  // Verify cluster health on startup
  const health = await weaviateClient.cluster.healthGetter().do();
  if (!health.isHealthy) {
    throw new Error(`Weaviate cluster unhealthy: ${JSON.stringify(health)}`);
  }
} catch (initError) {
  logger.fatal({ error: initError }, "Failed to initialize Weaviate client");
  process.exit(1);
}

// Define Weaviate 1.25 schema for Netflix content embeddings
const contentSchema = {
  class: "NetflixContent",
  description: "Stores vector embeddings and metadata for all Netflix content (movies, series, specials)",
  multiTenancyConfig: {
    enabled: true,
    autoTenantCreation: true,
    autoTenantActivation: true,
  },
  vectorIndexConfig: {
    distance: "cosine",
    vectorIndexType: "hnsw",
    hnsw: {
      efConstruction: 256,
      maxConnections: 64,
      ef: 128,
      dynamicEfMin: 64,
      dynamicEfMax: 256,
      dynamicEfFactor: 8,
    },
  },
  invertedIndexConfig: {
    bm25: {
      b: 0.75,
      k1: 1.2,
    },
    stopwords: {
      preset: "en",
    },
    indexTimestamps: true,
  },
  properties: [
    {
      name: "content_id",
      dataType: ["string"],
      description: "Unique Netflix content identifier",
      indexFilterable: true,
      indexSearchable: false,
    },
    {
      name: "title",
      dataType: ["string"],
      description: "Content title",
      indexFilterable: true,
      indexSearchable: true,
    },
    {
      name: "description",
      dataType: ["text"],
      description: "Full content description",
      indexFilterable: false,
      indexSearchable: true,
    },
    {
      name: "categories",
      dataType: ["string[]"],
      description: "Content categories (e.g., action, comedy)",
      indexFilterable: true,
      indexSearchable: true,
    },
    {
      name: "release_year",
      dataType: ["int"],
      description: "Year of content release",
      indexFilterable: true,
      indexSearchable: false,
    },
    {
      name: "average_rating",
      dataType: ["number"],
      description: "Average user rating (1-5)",
      indexFilterable: true,
      indexSearchable: false,
    },
    {
      name: "popularity_score",
      dataType: ["number"],
      description: "Global popularity score (0-100)",
      indexFilterable: true,
      indexSearchable: false,
    },
  ],
  additional: {
    userId: {
      dataType: ["string"],
      description: "Tenant ID (region code) for multi-tenancy",
    },
  },
};

// Create or update schema with error handling
async function initializeContentSchema() {
  try {
    const existingSchema = await weaviateClient.schema.getter().do();
    const classExists = existingSchema.classes?.some((c) => c.class === "NetflixContent");

    if (!classExists) {
      await weaviateClient.schema.classCreator().withClass(contentSchema).do();
      logger.info("Created NetflixContent schema in Weaviate");
    } else {
      await weaviateClient.schema.classUpdater().withClass(contentSchema).do();
      logger.info("Updated NetflixContent schema in Weaviate");
    }

    // Create initial tenants for top 10 regions
    const topRegions = ["us-east-1", "eu-west-1", "ap-southeast-1", "sa-east-1"];
    for (const region of topRegions) {
      await weaviateClient.schema.tenantCreator("NetflixContent", region).do();
      logger.info({ region }, "Created Weaviate tenant for region");
    }
  } catch (schemaError) {
    logger.error({ error: schemaError }, "Failed to initialize Weaviate schema");
    throw schemaError;
  }
}

// Hybrid search implementation for content retrieval
async function hybridContentSearch(
  query: string,
  tenant: string,
  limit: number = 50,
  filters?: Record
) {
  try {
    const searchQuery = weaviateClient.graphql
      .get()
      .withClassName("NetflixContent")
      .withTenant(tenant)
      .withHybrid({
        query,
        alpha: 0.7, // 70% vector, 30% BM25
        fusionType: "relativeScoreFusion",
      })
      .withFields([
        "content_id",
        "title",
        "description",
        "categories",
        "average_rating",
        "popularity_score",
        "_additional { score }",
      ])
      .withLimit(limit);

    // Apply additional filters if provided
    if (filters) {
      const whereClause: any = { operator: "And", operands: [] };
      if (filters.categories) {
        whereClause.operands.push({
          operator: "ContainsAny",
          path: ["categories"],
          valueStringArray: filters.categories,
        });
      }
      if (filters.minRating) {
        whereClause.operands.push({
          operator: "GreaterThanEqual",
          path: ["average_rating"],
          valueNumber: filters.minRating,
        });
      }
      if (filters.minYear) {
        whereClause.operands.push({
          operator: "GreaterThanEqual",
          path: ["release_year"],
          valueInt: filters.minYear,
        });
      }
      searchQuery.withWhere(whereClause);
    }

    const result = await searchQuery.do();
    return result.data.Get.NetflixContent || [];
  } catch (searchError) {
    logger.error({ error: searchError, query, tenant }, "Hybrid search failed");
    throw searchError;
  }
}

// Initialize schema on module load
initializeContentSchema().catch((err) => {
  logger.fatal({ error: err }, "Fatal error during schema initialization");
  process.exit(1);
});

export { weaviateClient, hybridContentSearch, initializeContentSchema };

// Netflix 2026 Context-Aware Ranking Workflow
// Uses LangChain 0.3 and Weaviate 1.25 to rank 50 candidates to top 10 recommendations
import { RunnableSequence, RunnableMap } from "@langchain/core/runnables";
import { ChatOpenAI } from "@langchain/openai";
import { PromptTemplate } from "@langchain/core/prompts";
import { WeaviateStore } from "@langchain/weaviate";
import { RedisCache } from "./redis-client";
import { weaviateClient } from "./weaviate-schema";
import { logger } from "./logger";

// Initialize LLM for ranking with strict latency constraints
const rankingLLM = new ChatOpenAI({
  model: "gpt-4-turbo-2024-04-09",
  temperature: 0.0, // deterministic ranking
  maxTokens: 1024,
  timeout: 8000, // 8s max latency for ranking step
  maxRetries: 1, // only 1 retry to avoid p99 spikes
});

// Ranking prompt template with strict output formatting
const rankingPrompt = PromptTemplate.fromTemplate(`
  You are a Netflix senior ranking engineer. Rank the following 50 content candidates for a user based on their context and interaction history. Return exactly 10 content IDs in order of relevance, highest first.

  User Context: {user_context}
  Interaction History: {interaction_history}
  Content Candidates (JSON array with content_id, title, categories, score): {candidates}

  Return only a JSON array of 10 content ID strings, no additional text or formatting.
`);

// Initialize Weaviate store for user profile retrieval
const userProfileStore = new WeaviateStore(
  new OpenAIEmbeddings({ model: "text-embedding-3-large" }),
  {
    client: weaviateClient,
    indexName: "UserProfile",
    textKey: "interaction_history",
  }
);

// Build the ranking workflow
const rankingWorkflow = RunnableSequence.from([
  // Step 1: Validate input and retrieve user profile
  RunnableMap.from({
    user_id: (input: { user_id: string; candidates: any[] }) => {
      if (!input.user_id) throw new Error("Missing user_id");
      if (!input.candidates || input.candidates.length === 0) {
        throw new Error("No candidates provided for ranking");
      }
      return input.user_id;
    },
    candidates: (input) => input.candidates,
    user_profile: async (input) => {
      const cacheKey = `user:profile:${input.user_id}`;
      const cached = await RedisCache.get(cacheKey);
      if (cached) return JSON.parse(cached);
      // Retrieve from Weaviate if cache miss
      const profile = await userProfileStore.similaritySearch(
        `user_id:${input.user_id}`,
        1
      );
      return profile[0] || null;
    },
  }),
  // Step 2: Enrich input with user context
  RunnableMap.from({
    ... // spread previous fields
    user_context: (input) => ({
      device: input.user_profile?.device_preference || "unknown",
      time_of_day: new Date().getHours(),
      subscription_tier: input.user_profile?.subscription_tier || "standard",
      region: input.user_profile?.region || "us-east-1",
    }),
    interaction_history: (input) => input.user_profile?.interaction_history || [],
  }),
  // Step 3: Run LLM ranking with fallback
  async (input) => {
    const promptInput = {
      user_context: JSON.stringify(input.user_context),
      interaction_history: JSON.stringify(input.interaction_history),
      candidates: JSON.stringify(input.candidates),
    };

    try {
      const llmResponse = await rankingPrompt.pipe(rankingLLM).invoke(promptInput);
      const rankedIds = JSON.parse(llmResponse.content.toString());
      // Validate output
      if (!Array.isArray(rankedIds) || rankedIds.length !== 10) {
        throw new Error(`Invalid ranking output: expected 10 IDs, got ${rankedIds.length}`);
      }
      // Map back to full candidate objects
      const rankedCandidates = rankedIds.map((id: string) => 
        input.candidates.find((c: any) => c.content_id === id)
      ).filter(Boolean);
      return rankedCandidates;
    } catch (rankingError) {
      logger.error({ error: rankingError, user_id: input.user_id }, "LLM ranking failed, falling back to score-based ranking");
      // Fallback: sort by Weaviate score + popularity
      return input.candidates
        .sort((a: any, b: any) => {
          const scoreA = a.score * 0.7 + (a.popularity_score / 100) * 0.3;
          const scoreB = b.score * 0.7 + (b.popularity_score / 100) * 0.3;
          return scoreB - scoreA;
        })
        .slice(0, 10);
    }
  },
  // Step 4: Post-process and cache recommendations
  async (rankedCandidates) => {
    const result = rankedCandidates.map((c: any) => ({
      content_id: c.content_id,
      title: c.title,
      thumbnail_url: `https://cdn.netflix.com/thumbnails/${c.content_id}`,
      score: c.score,
      categories: c.categories,
    }));
    // Cache for 15 minutes
    const cacheKey = `recs:${rankedCandidates[0]?.content_id || "empty"}`;
    await RedisCache.set(cacheKey, JSON.stringify(result), 900);
    return result;
  },
]);

// Export workflow and helper for testing
export { rankingWorkflow };

import { RunnableSequence, RunnableMap } from "@langchain/core/runnables";

const workflow = RunnableSequence.from([
  RunnableMap.from({
    user_id: (input) => input.user_id,
    context: (input) => input.context,
  }),
  // Add workflow steps here
]).withRetry({ maxRetries: 2, factor: 2 });

const contentSchema = {
  vectorIndexConfig: {
    vectorIndexType: "hnsw",
    hnsw: {
      efConstruction: 256,
      maxConnections: 64,
      dynamicEfMin: 64,
      dynamicEfMax: 256,
    },
  },
};

import { createHash } from "crypto";

const getCacheKey = (query: string, filters: any) => {
  const hash = createHash("sha256");
  hash.update(query + JSON.stringify(filters));
  return `weaviate:query:${hash.digest("hex")}`;
};