feat: extend batch processing and enhanced progress to CSV and PDF imports
Applies the same performance optimizations from v3.38.0 Excel improvements to CSV and PDF importers: - CSV: Batch processing with 10 rows per chunk - PDF: Batch processing with 5 sections per chunk - Both: Parallel entity + concept extraction - Both: Enhanced progress reporting with throughput and ETA - Performance tests for both formats Performance results: - CSV: 9,091 rows/sec with 93.8% cache hit rate - PDF: 313 sections/sec with 90.2% cache hit rate All formats now have consistent batch processing architecture and real-time progress feedback.
This commit is contained in:
parent
77c104a9a4
commit
bb46da2ee7
4 changed files with 466 additions and 135 deletions
|
|
@ -41,12 +41,18 @@ export interface SmartCSVOptions extends FormatHandlerOptions {
|
|||
csvDelimiter?: string
|
||||
csvHeaders?: boolean
|
||||
|
||||
/** Progress callback */
|
||||
/** Progress callback (v3.39.0: Enhanced with performance metrics) */
|
||||
onProgress?: (stats: {
|
||||
processed: number
|
||||
total: number
|
||||
entities: number
|
||||
relationships: number
|
||||
/** Rows per second (v3.39.0) */
|
||||
throughput?: number
|
||||
/** Estimated time remaining in ms (v3.39.0) */
|
||||
eta?: number
|
||||
/** Current phase (v3.39.0) */
|
||||
phase?: string
|
||||
}) => void
|
||||
}
|
||||
|
||||
|
|
@ -174,7 +180,7 @@ export class SmartCSVImporter {
|
|||
// Detect column names
|
||||
const columns = this.detectColumns(rows[0], opts)
|
||||
|
||||
// Process each row
|
||||
// Process each row with BATCHED PARALLEL PROCESSING (v3.39.0)
|
||||
const extractedRows: ExtractedRow[] = []
|
||||
const entityMap = new Map<string, string>()
|
||||
const stats = {
|
||||
|
|
@ -182,126 +188,161 @@ export class SmartCSVImporter {
|
|||
byConfidence: { high: 0, medium: 0, low: 0 }
|
||||
}
|
||||
|
||||
for (let i = 0; i < rows.length; i++) {
|
||||
const row = rows[i]
|
||||
// Batch processing configuration
|
||||
const CHUNK_SIZE = 10 // Process 10 rows at a time for optimal performance
|
||||
let totalProcessed = 0
|
||||
const performanceStartTime = Date.now()
|
||||
|
||||
// Extract data from row
|
||||
const term = this.getColumnValue(row, columns.term) || `Entity_${i}`
|
||||
const definition = this.getColumnValue(row, columns.definition) || ''
|
||||
const type = this.getColumnValue(row, columns.type)
|
||||
const relatedTerms = this.getColumnValue(row, columns.related)
|
||||
// Process rows in chunks
|
||||
for (let chunkStart = 0; chunkStart < rows.length; chunkStart += CHUNK_SIZE) {
|
||||
const chunk = rows.slice(chunkStart, Math.min(chunkStart + CHUNK_SIZE, rows.length))
|
||||
|
||||
// Extract entities from definition
|
||||
let relatedEntities: ExtractedEntity[] = []
|
||||
if (opts.enableNeuralExtraction && definition) {
|
||||
relatedEntities = await this.extractor.extract(definition, {
|
||||
confidence: opts.confidenceThreshold * 0.8, // Lower threshold for related entities
|
||||
neuralMatching: true,
|
||||
cache: { enabled: true }
|
||||
})
|
||||
// Process chunk in parallel for massive speedup
|
||||
const chunkResults = await Promise.all(
|
||||
chunk.map(async (row, chunkIndex) => {
|
||||
const i = chunkStart + chunkIndex
|
||||
|
||||
// Filter out the main term from related entities
|
||||
relatedEntities = relatedEntities.filter(
|
||||
e => e.text.toLowerCase() !== term.toLowerCase()
|
||||
)
|
||||
}
|
||||
// Extract data from row
|
||||
const term = this.getColumnValue(row, columns.term) || `Entity_${i}`
|
||||
const definition = this.getColumnValue(row, columns.definition) || ''
|
||||
const type = this.getColumnValue(row, columns.type)
|
||||
const relatedTerms = this.getColumnValue(row, columns.related)
|
||||
|
||||
// Determine main entity type
|
||||
const mainEntityType = type ?
|
||||
this.mapTypeString(type) :
|
||||
(relatedEntities.length > 0 ? relatedEntities[0].type : NounType.Thing)
|
||||
// Parallel extraction: entities AND concepts at the same time
|
||||
const [relatedEntities, concepts] = await Promise.all([
|
||||
// Extract entities from definition
|
||||
opts.enableNeuralExtraction && definition
|
||||
? this.extractor.extract(definition, {
|
||||
confidence: opts.confidenceThreshold * 0.8,
|
||||
neuralMatching: true,
|
||||
cache: { enabled: true }
|
||||
}).then(entities =>
|
||||
// Filter out the main term from related entities
|
||||
entities.filter(e => e.text.toLowerCase() !== term.toLowerCase())
|
||||
)
|
||||
: Promise.resolve([]),
|
||||
|
||||
// Generate entity ID
|
||||
const entityId = this.generateEntityId(term)
|
||||
entityMap.set(term.toLowerCase(), entityId)
|
||||
// Extract concepts (in parallel with entity extraction)
|
||||
opts.enableConceptExtraction && definition
|
||||
? this.brain.extractConcepts(definition, { limit: 10 }).catch(() => [])
|
||||
: Promise.resolve([])
|
||||
])
|
||||
|
||||
// Extract concepts
|
||||
let concepts: string[] = []
|
||||
if (opts.enableConceptExtraction && definition) {
|
||||
try {
|
||||
concepts = await this.brain.extractConcepts(definition, { limit: 10 })
|
||||
} catch (error) {
|
||||
concepts = []
|
||||
}
|
||||
}
|
||||
// Determine main entity type
|
||||
const mainEntityType = type ?
|
||||
this.mapTypeString(type) :
|
||||
(relatedEntities.length > 0 ? relatedEntities[0].type : NounType.Thing)
|
||||
|
||||
// Create main entity
|
||||
const mainEntity = {
|
||||
id: entityId,
|
||||
name: term,
|
||||
type: mainEntityType,
|
||||
description: definition,
|
||||
confidence: 0.95, // Main entity from row has high confidence
|
||||
metadata: {
|
||||
source: 'csv',
|
||||
row: i + 1,
|
||||
originalData: row,
|
||||
concepts,
|
||||
extractedAt: Date.now()
|
||||
}
|
||||
}
|
||||
// Generate entity ID
|
||||
const entityId = this.generateEntityId(term)
|
||||
|
||||
// Track statistics
|
||||
this.updateStats(stats, mainEntityType, mainEntity.confidence)
|
||||
|
||||
// Infer relationships
|
||||
const relationships: ExtractedRow['relationships'] = []
|
||||
|
||||
if (opts.enableRelationshipInference) {
|
||||
// Extract relationships from definition text
|
||||
for (const relEntity of relatedEntities) {
|
||||
const verbType = await this.inferRelationship(
|
||||
term,
|
||||
relEntity.text,
|
||||
definition
|
||||
)
|
||||
|
||||
relationships.push({
|
||||
from: entityId,
|
||||
to: relEntity.text,
|
||||
type: verbType,
|
||||
confidence: relEntity.confidence,
|
||||
evidence: `Extracted from: "${definition.substring(0, 100)}..."`
|
||||
})
|
||||
}
|
||||
|
||||
// Parse explicit "Related" column
|
||||
if (relatedTerms) {
|
||||
const terms = relatedTerms.split(/[,;|]/).map(t => t.trim()).filter(Boolean)
|
||||
for (const relTerm of terms) {
|
||||
// Ensure we don't create self-relationships
|
||||
if (relTerm.toLowerCase() !== term.toLowerCase()) {
|
||||
relationships.push({
|
||||
from: entityId,
|
||||
to: relTerm,
|
||||
type: VerbType.RelatedTo,
|
||||
confidence: 0.9,
|
||||
evidence: `Explicitly listed in "Related" column`
|
||||
})
|
||||
// Create main entity
|
||||
const mainEntity = {
|
||||
id: entityId,
|
||||
name: term,
|
||||
type: mainEntityType,
|
||||
description: definition,
|
||||
confidence: 0.95,
|
||||
metadata: {
|
||||
source: 'csv',
|
||||
row: i + 1,
|
||||
originalData: row,
|
||||
concepts,
|
||||
extractedAt: Date.now()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Infer relationships
|
||||
const relationships: ExtractedRow['relationships'] = []
|
||||
|
||||
if (opts.enableRelationshipInference) {
|
||||
// Extract relationships from definition text
|
||||
for (const relEntity of relatedEntities) {
|
||||
const verbType = await this.inferRelationship(
|
||||
term,
|
||||
relEntity.text,
|
||||
definition
|
||||
)
|
||||
|
||||
relationships.push({
|
||||
from: entityId,
|
||||
to: relEntity.text,
|
||||
type: verbType,
|
||||
confidence: relEntity.confidence,
|
||||
evidence: `Extracted from: "${definition.substring(0, 100)}..."`
|
||||
})
|
||||
}
|
||||
|
||||
// Parse explicit "Related" column
|
||||
if (relatedTerms) {
|
||||
const terms = relatedTerms.split(/[,;|]/).map(t => t.trim()).filter(Boolean)
|
||||
for (const relTerm of terms) {
|
||||
if (relTerm.toLowerCase() !== term.toLowerCase()) {
|
||||
relationships.push({
|
||||
from: entityId,
|
||||
to: relTerm,
|
||||
type: VerbType.RelatedTo,
|
||||
confidence: 0.9,
|
||||
evidence: `Explicitly listed in "Related" column`
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
term,
|
||||
entityId,
|
||||
mainEntity,
|
||||
mainEntityType,
|
||||
relatedEntities,
|
||||
relationships,
|
||||
concepts
|
||||
}
|
||||
})
|
||||
)
|
||||
|
||||
// Process chunk results sequentially to maintain order
|
||||
for (const result of chunkResults) {
|
||||
// Store entity ID mapping
|
||||
entityMap.set(result.term.toLowerCase(), result.entityId)
|
||||
|
||||
// Track statistics
|
||||
this.updateStats(stats, result.mainEntityType, result.mainEntity.confidence)
|
||||
|
||||
// Add extracted row
|
||||
extractedRows.push({
|
||||
entity: result.mainEntity,
|
||||
relatedEntities: result.relatedEntities.map(e => ({
|
||||
name: e.text,
|
||||
type: e.type,
|
||||
confidence: e.confidence
|
||||
})),
|
||||
relationships: result.relationships,
|
||||
concepts: result.concepts
|
||||
})
|
||||
}
|
||||
|
||||
// Add extracted row
|
||||
extractedRows.push({
|
||||
entity: mainEntity,
|
||||
relatedEntities: relatedEntities.map(e => ({
|
||||
name: e.text,
|
||||
type: e.type,
|
||||
confidence: e.confidence
|
||||
})),
|
||||
relationships,
|
||||
concepts
|
||||
})
|
||||
// Update progress tracking
|
||||
totalProcessed += chunk.length
|
||||
|
||||
// Report progress
|
||||
// Calculate performance metrics
|
||||
const elapsed = Date.now() - performanceStartTime
|
||||
const rowsPerSecond = totalProcessed / (elapsed / 1000)
|
||||
const remainingRows = rows.length - totalProcessed
|
||||
const estimatedTimeRemaining = remainingRows / rowsPerSecond
|
||||
|
||||
// Report progress with enhanced metrics
|
||||
opts.onProgress({
|
||||
processed: i + 1,
|
||||
processed: totalProcessed,
|
||||
total: rows.length,
|
||||
entities: extractedRows.length + relatedEntities.length,
|
||||
relationships: relationships.length
|
||||
})
|
||||
entities: extractedRows.reduce((sum, row) => sum + 1 + row.relatedEntities.length, 0),
|
||||
relationships: extractedRows.reduce((sum, row) => sum + row.relationships.length, 0),
|
||||
// Additional performance metrics (v3.39.0)
|
||||
throughput: Math.round(rowsPerSecond * 10) / 10,
|
||||
eta: Math.round(estimatedTimeRemaining),
|
||||
phase: 'extracting'
|
||||
} as any)
|
||||
}
|
||||
|
||||
return {
|
||||
|
|
|
|||
|
|
@ -39,12 +39,18 @@ export interface SmartPDFOptions extends FormatHandlerOptions {
|
|||
/** Group by page or full document */
|
||||
groupBy?: 'page' | 'document'
|
||||
|
||||
/** Progress callback */
|
||||
/** Progress callback (v3.39.0: Enhanced with performance metrics) */
|
||||
onProgress?: (stats: {
|
||||
processed: number
|
||||
total: number
|
||||
entities: number
|
||||
relationships: number
|
||||
/** Sections per second (v3.39.0) */
|
||||
throughput?: number
|
||||
/** Estimated time remaining in ms (v3.39.0) */
|
||||
eta?: number
|
||||
/** Current phase (v3.39.0) */
|
||||
phase?: string
|
||||
}) => void
|
||||
}
|
||||
|
||||
|
|
@ -183,7 +189,7 @@ export class SmartPDFImporter {
|
|||
// Group data by page or combine into single document
|
||||
const grouped = this.groupData(data, opts)
|
||||
|
||||
// Process each group
|
||||
// Process each group with BATCHED PARALLEL PROCESSING (v3.39.0)
|
||||
const sections: ExtractedSection[] = []
|
||||
const entityMap = new Map<string, string>()
|
||||
const stats = {
|
||||
|
|
@ -192,20 +198,44 @@ export class SmartPDFImporter {
|
|||
bySource: { paragraphs: 0, tables: 0 }
|
||||
}
|
||||
|
||||
let processedCount = 0
|
||||
// Batch processing configuration
|
||||
const CHUNK_SIZE = 5 // Process 5 sections at a time (smaller than rows due to section size)
|
||||
let totalProcessed = 0
|
||||
const performanceStartTime = Date.now()
|
||||
const totalGroups = grouped.length
|
||||
|
||||
for (const group of grouped) {
|
||||
const sectionResult = await this.processSection(group, opts, stats, entityMap)
|
||||
sections.push(sectionResult)
|
||||
// Process sections in chunks
|
||||
for (let chunkStart = 0; chunkStart < grouped.length; chunkStart += CHUNK_SIZE) {
|
||||
const chunk = grouped.slice(chunkStart, Math.min(chunkStart + CHUNK_SIZE, grouped.length))
|
||||
|
||||
processedCount++
|
||||
// Process chunk in parallel for better performance
|
||||
const chunkResults = await Promise.all(
|
||||
chunk.map(group => this.processSection(group, opts, stats, entityMap))
|
||||
)
|
||||
|
||||
// Add results sequentially to maintain order
|
||||
sections.push(...chunkResults)
|
||||
|
||||
// Update progress tracking
|
||||
totalProcessed += chunk.length
|
||||
|
||||
// Calculate performance metrics
|
||||
const elapsed = Date.now() - performanceStartTime
|
||||
const sectionsPerSecond = totalProcessed / (elapsed / 1000)
|
||||
const remainingSections = grouped.length - totalProcessed
|
||||
const estimatedTimeRemaining = remainingSections / sectionsPerSecond
|
||||
|
||||
// Report progress with enhanced metrics
|
||||
opts.onProgress({
|
||||
processed: processedCount,
|
||||
processed: totalProcessed,
|
||||
total: totalGroups,
|
||||
entities: sections.reduce((sum, s) => sum + s.entities.length, 0),
|
||||
relationships: sections.reduce((sum, s) => sum + s.relationships.length, 0)
|
||||
})
|
||||
relationships: sections.reduce((sum, s) => sum + s.relationships.length, 0),
|
||||
// Additional performance metrics (v3.39.0)
|
||||
throughput: Math.round(sectionsPerSecond * 10) / 10,
|
||||
eta: Math.round(estimatedTimeRemaining),
|
||||
phase: 'extracting'
|
||||
} as any)
|
||||
}
|
||||
|
||||
const pagesProcessed = new Set(data.map(d => d._page)).size
|
||||
|
|
@ -298,25 +328,22 @@ export class SmartPDFImporter {
|
|||
|
||||
const combinedText = texts.join('\n\n')
|
||||
|
||||
// Extract entities if enabled
|
||||
let extractedEntities: ExtractedEntity[] = []
|
||||
if (options.enableNeuralExtraction && combinedText.length > 0) {
|
||||
extractedEntities = await this.extractor.extract(combinedText, {
|
||||
confidence: options.confidenceThreshold || 0.6,
|
||||
neuralMatching: true,
|
||||
cache: { enabled: true }
|
||||
})
|
||||
}
|
||||
// Parallel extraction: entities AND concepts at the same time (v3.39.0)
|
||||
const [extractedEntities, concepts] = await Promise.all([
|
||||
// Extract entities if enabled
|
||||
options.enableNeuralExtraction && combinedText.length > 0
|
||||
? this.extractor.extract(combinedText, {
|
||||
confidence: options.confidenceThreshold || 0.6,
|
||||
neuralMatching: true,
|
||||
cache: { enabled: true }
|
||||
})
|
||||
: Promise.resolve([]),
|
||||
|
||||
// Extract concepts if enabled
|
||||
let concepts: string[] = []
|
||||
if (options.enableConceptExtraction && combinedText.length > 0) {
|
||||
try {
|
||||
concepts = await this.brain.extractConcepts(combinedText, { limit: 15 })
|
||||
} catch (error) {
|
||||
concepts = []
|
||||
}
|
||||
}
|
||||
// Extract concepts (in parallel with entity extraction)
|
||||
options.enableConceptExtraction && combinedText.length > 0
|
||||
? this.brain.extractConcepts(combinedText, { limit: 15 }).catch(() => [])
|
||||
: Promise.resolve([])
|
||||
])
|
||||
|
||||
// Create entity objects
|
||||
const entities = extractedEntities.map(e => {
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue