diff --git a/.env.example b/.env.example index c5c9434a5..6adcfe202 100644 --- a/.env.example +++ b/.env.example @@ -321,16 +321,14 @@ SAGEMAKER_ACCEPT=application/json # Response content type # ============================================================================= # CONVERSATION MEMORY CONFIGURATION (Optional) # ============================================================================= -# Enable conversation memory feature (disabled by default) -NEUROLINK_MEMORY_ENABLED=false +# Enable conversation memory feature (ENABLED BY DEFAULT) +NEUROLINK_MEMORY_ENABLED=true # Enable conversation memory (default: true) # Memory storage type (memory or redis) STORAGE_TYPE=memory # Options: memory, redis # Memory limits NEUROLINK_MEMORY_MAX_SESSIONS=50 # Maximum number of sessions to keep in memory -NEUROLINK_MEMORY_MAX_TURNS_PER_SESSION=50 # Maximum conversation turns per session - # Redis Storage Configuration (used when STORAGE_TYPE=redis) REDIS_HOST=localhost # Redis server hostname REDIS_PORT=6379 # Redis server port @@ -403,12 +401,16 @@ NEUROLINK_MAX_TOOLS_PER_PROVIDER=5 # Config File Location NEUROLINK_CONFIG_FILE=./neurolink.config.json -# Summarization (Conversation Memory) -NEUROLINK_SUMMARIZATION_ENABLED=false -NEUROLINK_SUMMARIZATION_THRESHOLD_TURNS=20 -NEUROLINK_SUMMARIZATION_TARGET_TURNS=10 -NEUROLINK_SUMMARIZATION_PROVIDER=google-ai -NEUROLINK_SUMMARIZATION_MODEL=gemini-2.5-flash +# Summarization (Conversation Memory) - TOKEN-BASED MEMORY ENABLED BY DEFAULT +# Token-based memory is enabled by default and uses 80% of each model's context window +NEUROLINK_SUMMARIZATION_ENABLED=true # Enable summarization (default: true) +NEUROLINK_TOKEN_THRESHOLD=50000 # Optional: Override token threshold (default: 80% of model context) +NEUROLINK_SUMMARIZATION_PROVIDER=vertex # Provider for summarization (default: vertex) +NEUROLINK_SUMMARIZATION_MODEL=gemini-2.5-flash # Model for summarization (default: gemini-2.5-flash) + +# Deprecated: Turn-based memory settings (use TOKEN_THRESHOLD instead) +# NEUROLINK_SUMMARIZATION_THRESHOLD_TURNS=20 # Deprecated: Use token threshold +# NEUROLINK_SUMMARIZATION_TARGET_TURNS=10 # Deprecated: Use token threshold # Default Generation Parameters NEUROLINK_DEFAULT_MAX_TOKENS=4096 diff --git a/src/cli/loop/optionsSchema.ts b/src/cli/loop/optionsSchema.ts index f7cc16d86..a4bed7d48 100644 --- a/src/cli/loop/optionsSchema.ts +++ b/src/cli/loop/optionsSchema.ts @@ -87,4 +87,9 @@ export const textGenerationOptionsSchema: Record< type: "string", description: "Context about tools/MCPs used in the interaction.", }, + enableSummarization: { + type: "boolean", + description: + "Enable or disable automatic conversation summarization for this request.", + }, }; diff --git a/src/lib/config/conversationMemory.ts b/src/lib/config/conversationMemory.ts index d0d9cfa95..ee7d2bcd5 100644 --- a/src/lib/config/conversationMemory.ts +++ b/src/lib/config/conversationMemory.ts @@ -33,6 +33,24 @@ IMPORTANT: You are continuing an ongoing conversation. The previous messages in Always reference and build upon this conversation history when relevant. If the user asks about information mentioned earlier in the conversation, refer to those previous messages to provide accurate, contextual responses.`; +/** + * Percentage of model context window to use for conversation memory threshold + * Default: 80% of model's context window + */ +export const MEMORY_THRESHOLD_PERCENTAGE = 0.8; + +/** + * Fallback token threshold if model context unknown + */ +export const DEFAULT_FALLBACK_THRESHOLD = 50000; + +/** + * Ratio of threshold to keep as recent unsummarized messages + * When summarization triggers, this percentage of tokens from the end + * are preserved as detailed messages, while older content gets summarized. + */ +export const RECENT_MESSAGES_RATIO = 0.3; + /** * Structured output instructions for JSON/structured output mode * Used to ensure AI providers output only valid JSON without conversational filler @@ -57,17 +75,23 @@ export function getConversationMemoryDefaults(): ConversationMemoryConfig { enabled: process.env.NEUROLINK_MEMORY_ENABLED === "true", maxSessions: Number(process.env.NEUROLINK_MEMORY_MAX_SESSIONS) || DEFAULT_MAX_SESSIONS, + enableSummarization: + process.env.NEUROLINK_SUMMARIZATION_ENABLED !== "false", + tokenThreshold: process.env.NEUROLINK_TOKEN_THRESHOLD + ? Number(process.env.NEUROLINK_TOKEN_THRESHOLD) + : undefined, + summarizationProvider: + process.env.NEUROLINK_SUMMARIZATION_PROVIDER || "vertex", + summarizationModel: + process.env.NEUROLINK_SUMMARIZATION_MODEL || "gemini-2.5-flash", + + // Deprecated (for backward compatibility) maxTurnsPerSession: Number(process.env.NEUROLINK_MEMORY_MAX_TURNS_PER_SESSION) || DEFAULT_MAX_TURNS_PER_SESSION, - enableSummarization: process.env.NEUROLINK_SUMMARIZATION_ENABLED === "true", summarizationThresholdTurns: Number(process.env.NEUROLINK_SUMMARIZATION_THRESHOLD_TURNS) || 20, summarizationTargetTurns: Number(process.env.NEUROLINK_SUMMARIZATION_TARGET_TURNS) || 10, - summarizationProvider: - process.env.NEUROLINK_SUMMARIZATION_PROVIDER || "vertex", - summarizationModel: - process.env.NEUROLINK_SUMMARIZATION_MODEL || "gemini-2.5-flash", }; } diff --git a/src/lib/core/conversationMemoryFactory.ts b/src/lib/core/conversationMemoryFactory.ts index 6eabb56d0..99f8ad884 100644 --- a/src/lib/core/conversationMemoryFactory.ts +++ b/src/lib/core/conversationMemoryFactory.ts @@ -27,10 +27,7 @@ export function createConversationMemoryManager( config: { enabled: config.enabled, maxSessions: config.maxSessions, - maxTurnsPerSession: config.maxTurnsPerSession, enableSummarization: config.enableSummarization, - summarizationThresholdTurns: config.summarizationThresholdTurns, - summarizationTargetTurns: config.summarizationTargetTurns, summarizationProvider: config.summarizationProvider, summarizationModel: config.summarizationModel, }, diff --git a/src/lib/core/conversationMemoryInitializer.ts b/src/lib/core/conversationMemoryInitializer.ts index 3764ae278..5f66582e0 100644 --- a/src/lib/core/conversationMemoryInitializer.ts +++ b/src/lib/core/conversationMemoryInitializer.ts @@ -113,15 +113,6 @@ export async function initializeConversationMemory(config?: { logger.info( "[conversationMemoryInitializer] Redis conversation memory manager created successfully", - { - configSource, - host: redisConfig.host || "localhost", - port: redisConfig.port || 6379, - keyPrefix: redisConfig.keyPrefix || "neurolink:conversation:", - maxSessions: memoryConfig.maxSessions, - maxTurnsPerSession: memoryConfig.maxTurnsPerSession, - managerType: redisMemoryManager?.constructor?.name, - }, ); // Perform basic validation diff --git a/src/lib/core/conversationMemoryManager.ts b/src/lib/core/conversationMemoryManager.ts index a6369400a..f7f366b26 100644 --- a/src/lib/core/conversationMemoryManager.ts +++ b/src/lib/core/conversationMemoryManager.ts @@ -8,21 +8,34 @@ import type { SessionMemory, ConversationMemoryStats, ChatMessage, + StoreConversationTurnOptions, } from "../types/conversation.js"; import { ConversationMemoryError } from "../types/conversation.js"; import { - DEFAULT_MAX_TURNS_PER_SESSION, DEFAULT_MAX_SESSIONS, + MEMORY_THRESHOLD_PERCENTAGE, + RECENT_MESSAGES_RATIO, MESSAGES_PER_TURN, } from "../config/conversationMemory.js"; import { logger } from "../utils/logger.js"; -import { NeuroLink } from "../neurolink.js"; +import { randomUUID } from "crypto"; +import { TokenUtils } from "../constants/tokens.js"; +import { + buildContextFromPointer, + getEffectiveTokenThreshold, + generateSummary, +} from "../utils/conversationMemory.js"; export class ConversationMemoryManager { private sessions: Map = new Map(); public config: ConversationMemoryConfig; private isInitialized: boolean = false; + /** + * Track sessions currently being summarized to prevent race conditions + */ + private summarizationInProgress: Set = new Set(); + constructor(config: ConversationMemoryConfig) { this.config = config; } @@ -52,60 +65,78 @@ export class ConversationMemoryManager { /** * Store a conversation turn for a session - * ULTRA-OPTIMIZED: Direct ChatMessage[] storage with zero conversion overhead + * TOKEN-BASED: Validates message size and triggers summarization based on tokens */ async storeConversationTurn( - sessionId: string, - userId: string | undefined, - userMessage: string, - aiResponse: string, - _startTimeStamp: Date | undefined, + options: StoreConversationTurnOptions, ): Promise { await this.ensureInitialized(); try { // Get or create session - let session = this.sessions.get(sessionId); + let session = this.sessions.get(options.sessionId); if (!session) { - session = this.createNewSession(sessionId, userId); - this.sessions.set(sessionId, session); + session = this.createNewSession(options.sessionId, options.userId); + this.sessions.set(options.sessionId, session); } - // ULTRA-OPTIMIZED: Direct message storage - no intermediate objects - session.messages.push( - { role: "user", content: userMessage }, - { role: "assistant", content: aiResponse }, + const tokenThreshold = options.providerDetails + ? getEffectiveTokenThreshold( + options.providerDetails.provider, + options.providerDetails.model, + this.config.tokenThreshold, + session.tokenThreshold, + ) + : this.config.tokenThreshold || 50000; + + const userMsg = await this.validateAndPrepareMessage( + options.userMessage, + "user", + tokenThreshold, + ); + const assistantMsg = await this.validateAndPrepareMessage( + options.aiResponse, + "assistant", + tokenThreshold, ); + session.messages.push(userMsg, assistantMsg); session.lastActivity = Date.now(); - if (this.config.enableSummarization) { - const userAssistantCount = session.messages.filter( - (msg) => msg.role === "user" || msg.role === "assistant", - ).length; - const currentTurnCount = Math.floor( - userAssistantCount / MESSAGES_PER_TURN, - ); - if ( - currentTurnCount >= (this.config.summarizationThresholdTurns || 20) - ) { - await this._summarizeSession(session); - } - } else { - const maxMessages = - (this.config.maxTurnsPerSession || DEFAULT_MAX_TURNS_PER_SESSION) * - MESSAGES_PER_TURN; - if (session.messages.length > maxMessages) { - session.messages = session.messages.slice(-maxMessages); + const shouldSummarize = + options.enableSummarization !== undefined + ? options.enableSummarization + : this.config.enableSummarization; + + if (shouldSummarize) { + // Only trigger summarization if not already in progress for this session + if (!this.summarizationInProgress.has(options.sessionId)) { + setImmediate(async () => { + try { + await this.checkAndSummarize(session, tokenThreshold); + } catch (error) { + logger.error("Background summarization failed", { + sessionId: session.sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } + }); + } else { + logger.debug( + "[ConversationMemoryManager] Summarization already in progress, skipping", + { + sessionId: options.sessionId, + }, + ); } } this.enforceSessionLimit(); } catch (error) { throw new ConversationMemoryError( - `Failed to store conversation turn for session ${sessionId}`, + `Failed to store conversation turn for session ${options.sessionId}`, "STORAGE_ERROR", { - sessionId, + sessionId: options.sessionId, error: error instanceof Error ? error.message : String(error), }, ); @@ -113,108 +144,220 @@ export class ConversationMemoryManager { } /** - * Build context messages for AI prompt injection (ULTRA-OPTIMIZED) - * Returns pre-stored message array with zero conversion overhead + * Validate and prepare a message before adding to session + * Truncates if message exceeds token limit + */ + private async validateAndPrepareMessage( + content: string, + role: ChatMessage["role"], + threshold: number, + ): Promise { + const id = randomUUID(); + const tokenCount = TokenUtils.estimateTokenCount(content); + + const maxMessageSize = Math.floor(threshold * MEMORY_THRESHOLD_PERCENTAGE); + if (tokenCount > maxMessageSize) { + const truncated = TokenUtils.truncateToTokenLimit( + content, + maxMessageSize, + ); + + logger.warn("Message truncated due to token limit", { + id, + role, + originalTokens: tokenCount, + threshold, + truncatedTo: maxMessageSize, + }); + + return { + id, + role, + content: truncated, + timestamp: new Date().toISOString(), + metadata: { + truncated: true, + }, + }; + } + + return { + id, + role, + content, + timestamp: new Date().toISOString(), + }; + } + + /** + * Check if summarization is needed based on token count + */ + private async checkAndSummarize( + session: SessionMemory, + threshold: number, + ): Promise { + // Acquire lock - if already in progress, skip + if (this.summarizationInProgress.has(session.sessionId)) { + logger.debug( + "[ConversationMemoryManager] Summarization already in progress, skipping", + { + sessionId: session.sessionId, + }, + ); + return; + } + + this.summarizationInProgress.add(session.sessionId); + + try { + const contextMessages = buildContextFromPointer(session); + const tokenCount = this.estimateTokens(contextMessages); + + session.lastTokenCount = tokenCount; + session.lastCountedAt = Date.now(); + + logger.debug("Token count check", { + sessionId: session.sessionId, + tokenCount, + threshold, + needsSummarization: tokenCount >= threshold, + }); + + if (tokenCount >= threshold) { + await this.summarizeSessionTokenBased(session, threshold); + } + } catch (error) { + logger.error("Token counting or summarization failed", { + sessionId: session.sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } finally { + // Release lock when done + this.summarizationInProgress.delete(session.sessionId); + } + } + + /** + * Estimate total tokens for a list of messages + */ + private estimateTokens(messages: ChatMessage[]): number { + return messages.reduce((total, msg) => { + return total + TokenUtils.estimateTokenCount(msg.content); + }, 0); + } + + /** + * Build context messages for AI prompt injection (TOKEN-BASED) + * Returns messages from pointer onwards (or all if no pointer) * Now consistently async to match Redis implementation */ async buildContextMessages(sessionId: string): Promise { const session = this.sessions.get(sessionId); - return session ? session.messages : []; + return session ? buildContextFromPointer(session) : []; } public getSession(sessionId: string): SessionMemory | undefined { return this.sessions.get(sessionId); } - public createSummarySystemMessage(content: string): ChatMessage { + public createSummarySystemMessage( + content: string, + summarizesFrom?: string, + summarizesTo?: string, + ): ChatMessage { return { + id: `summary-${randomUUID()}`, role: "system", content: `Summary of previous conversation turns:\n\n${content}`, + timestamp: new Date().toISOString(), + metadata: { + isSummary: true, + summarizesFrom, + summarizesTo, + }, }; } - private async _summarizeSession(session: SessionMemory): Promise { - logger.info( - `[ConversationMemory] Summarizing session ${session.sessionId}...`, - ); - const targetTurns = this.config.summarizationTargetTurns || 10; - const splitIndex = Math.max( - 0, - session.messages.length - targetTurns * MESSAGES_PER_TURN, + /** + * Token-based summarization (pointer-based, non-destructive) + */ + private async summarizeSessionTokenBased( + session: SessionMemory, + threshold: number, + ): Promise { + const startIndex = session.summarizedUpToMessageId + ? session.messages.findIndex( + (m) => m.id === session.summarizedUpToMessageId, + ) + 1 + : 0; + + const recentMessages = session.messages.slice(startIndex); + if (recentMessages.length === 0) { + return; + } + + const targetRecentTokens = threshold * RECENT_MESSAGES_RATIO; + const splitIndex = await this.findSplitIndexByTokens( + recentMessages, + targetRecentTokens, ); - const messagesToSummarize = session.messages.slice(0, splitIndex); - const recentMessages = session.messages.slice(splitIndex); + const messagesToSummarize = recentMessages.slice(0, splitIndex); if (messagesToSummarize.length === 0) { return; } - const summarizationPrompt = - this._createSummarizationPrompt(messagesToSummarize); - - const summarizer = new NeuroLink({ - conversationMemory: { enabled: false }, - }); - try { - const providerName = this.config.summarizationProvider; - - // Map provider names to correct format - let mappedProvider = providerName; - if (providerName === "vertex") { - mappedProvider = "googlevertex"; - } - - if (!mappedProvider) { - logger.error(`[ConversationMemory] Missing summarization provider`); - return; - } + const summary = await generateSummary( + messagesToSummarize, + this.config, + "[ConversationMemory]", + session.summarizedMessage, + ); - logger.debug( - `[ConversationMemory] Using provider: ${mappedProvider} for summarization`, + if (!summary) { + logger.warn( + `[ConversationMemory] Summary generation failed for session ${session.sessionId}`, ); + return; + } - const summaryResult = await summarizer.generate({ - input: { text: summarizationPrompt }, - provider: mappedProvider, - model: this.config.summarizationModel, - disableTools: true, - }); + const lastSummarized = messagesToSummarize[messagesToSummarize.length - 1]; + session.summarizedUpToMessageId = lastSummarized.id; + session.summarizedMessage = summary; // Store summary separately + + logger.info( + `[ConversationMemory] Summarization complete for session ${session.sessionId}`, + { + summarizedCount: messagesToSummarize.length, + totalMessages: session.messages.length, + pointer: session.summarizedUpToMessageId, + }, + ); + } - if (summaryResult.content) { - session.messages = [ - this.createSummarySystemMessage(summaryResult.content), - ...recentMessages, - ]; - logger.info( - `[ConversationMemory] Summarization complete for session ${session.sessionId}.`, - ); - } else { - logger.warn( - `[ConversationMemory] Summarization failed for session ${session.sessionId}. History not modified.`, - ); + /** + * Find split index to keep recent messages within target token count + */ + private async findSplitIndexByTokens( + messages: ChatMessage[], + targetRecentTokens: number, + ): Promise { + let recentTokens = 0; + let splitIndex = messages.length; + + for (let i = messages.length - 1; i >= 0; i--) { + const msgTokens = TokenUtils.estimateTokenCount(messages[i].content); + + if (recentTokens + msgTokens > targetRecentTokens) { + splitIndex = i + 1; + break; } - } catch (error) { - logger.error( - `[ConversationMemory] Error during summarization for session ${session.sessionId}`, - { error }, - ); + + recentTokens += msgTokens; } - } - private _createSummarizationPrompt(history: ChatMessage[]): string { - const formattedHistory = history - .map((msg) => `${msg.role}: ${msg.content}`) - .join("\n\n"); - return ` -You are a context summarization AI. Your task is to condense the following conversation history for another AI assistant. -The summary must be a concise, third-person narrative that retains all critical information, including key entities, technical details, decisions made, and any specific dates or times mentioned. -Ensure the summary flows logically and is ready to be used as context for the next turn in the conversation. - -Conversation History to Summarize: ---- -${formattedHistory} ---- -`.trim(); + // To ensure at least one message is summarized + return Math.max(1, splitIndex); } private async ensureInitialized(): Promise { diff --git a/src/lib/core/redisConversationMemoryManager.ts b/src/lib/core/redisConversationMemoryManager.ts index 0f09c138d..7bd2280a9 100644 --- a/src/lib/core/redisConversationMemoryManager.ts +++ b/src/lib/core/redisConversationMemoryManager.ts @@ -11,6 +11,8 @@ import type { RedisStorageConfig, SessionMetadata, RedisConversationObject, + SessionMemory, + StoreConversationTurnOptions, } from "../types/conversation.js"; import { ConversationMemoryError } from "../types/conversation.js"; import type { PendingToolExecution } from "../types/tools.js"; @@ -26,6 +28,12 @@ import { deserializeConversation, scanKeys, } from "../utils/redis.js"; +import { TokenUtils } from "../constants/tokens.js"; +import { + buildContextFromPointer, + getEffectiveTokenThreshold, + generateSummary, +} from "../utils/conversationMemory.js"; /** * Redis-based implementation of the ConversationMemoryManager @@ -51,6 +59,12 @@ export class RedisConversationMemoryManager { */ private titleGenerationInProgress: Set = new Set(); + /** + * Track sessions currently being summarized to prevent race conditions + * Key format: "${sessionId}:${userId}" + */ + private summarizationInProgress: Set = new Set(); + constructor( config: ConversationMemoryConfig, redisConfig: RedisStorageConfig = {}, @@ -208,16 +222,6 @@ export class RedisConversationMemoryManager { } } - /** - * Generate next message ID for a conversation - */ - private generateMessageId( - conversation: { messages?: ChatMessage[] } | null, - ): string { - const currentCount = conversation?.messages?.length || 0; - return `msg_${currentCount + 1}`; - } - /** * Generate current timestamp in ISO format */ @@ -225,13 +229,6 @@ export class RedisConversationMemoryManager { return new Date().toISOString(); } - /** - * Generate a unique conversation ID using UUID v4 - */ - private generateUniqueId(): string { - return randomUUID(); - } - /** * Store tool execution data for a session (temporarily to avoid race conditions) */ @@ -335,17 +332,11 @@ export class RedisConversationMemoryManager { * Store a conversation turn for a session */ async storeConversationTurn( - sessionId: string, - userId: string | undefined, - userMessage: string, - aiResponse: string, - startTimeStamp: Date | undefined, + options: StoreConversationTurnOptions, ): Promise { logger.debug("[RedisConversationMemoryManager] Storing conversation turn", { - sessionId, - userId, - userMessageLength: userMessage.length, - aiResponseLength: aiResponse.length, + sessionId: options.sessionId, + userId: options.userId, }); await this.ensureInitialized(); @@ -355,53 +346,35 @@ export class RedisConversationMemoryManager { throw new Error("Redis client not initialized"); } - // Generate Redis key - const redisKey = getSessionKey(this.redisConfig, sessionId, userId); - - // Get existing conversation object + const redisKey = getSessionKey( + this.redisConfig, + options.sessionId, + options.userId, + ); const conversationData = await this.redisClient.get(redisKey); let conversation = deserializeConversation(conversationData); const currentTime = new Date().toISOString(); - const normalizedUserId = userId || "randomUser"; + const normalizedUserId = options.userId || "randomUser"; - // If no existing conversation, create a new one if (!conversation) { - // Generate title asynchronously in the background (non-blocking) - const titleGenerationKey = `${sessionId}:${normalizedUserId}`; + const titleGenerationKey = `${options.sessionId}:${normalizedUserId}`; setImmediate(async () => { - // Check if title generation is already in progress for this session if (this.titleGenerationInProgress.has(titleGenerationKey)) { - logger.debug( - "[RedisConversationMemoryManager] Title generation already in progress, skipping", - { - sessionId, - userId: normalizedUserId, - titleGenerationKey, - }, - ); return; } - - // Mark title generation as in progress this.titleGenerationInProgress.add(titleGenerationKey); try { - const title = await this.generateConversationTitle(userMessage); - logger.info( - "[RedisConversationMemoryManager] Successfully generated conversation title", - { - sessionId, - userId: normalizedUserId, - title, - }, + const title = await this.generateConversationTitle( + options.userMessage, ); const updatedRedisKey = getSessionKey( this.redisConfig, - sessionId, - userId || undefined, + options.sessionId, + options.userId || undefined, ); const updatedConversationData = await this.redisClient?.get(updatedRedisKey); @@ -427,7 +400,7 @@ export class RedisConversationMemoryManager { logger.warn( "[RedisConversationMemoryManager] Failed to generate conversation title in background", { - sessionId, + sessionId: options.sessionId, userId: normalizedUserId, error: titleError instanceof Error @@ -436,130 +409,123 @@ export class RedisConversationMemoryManager { }, ); } finally { - // Always remove from tracking set when done (success or failure) this.titleGenerationInProgress.delete(titleGenerationKey); - - logger.debug( - "[RedisConversationMemoryManager] Title generation completed, removed from tracking", - { - sessionId, - userId: normalizedUserId, - titleGenerationKey, - remainingInProgress: this.titleGenerationInProgress.size, - }, - ); } }); conversation = { - id: this.generateUniqueId(), // Generate unique UUID v4 for conversation + id: randomUUID(), title: "New Conversation", // Temporary title until generated - sessionId, + sessionId: options.sessionId, userId: normalizedUserId, - createdAt: startTimeStamp?.toISOString() || currentTime, - updatedAt: startTimeStamp?.toISOString() || currentTime, + createdAt: options.startTimeStamp?.toISOString() || currentTime, + updatedAt: options.startTimeStamp?.toISOString() || currentTime, messages: [], }; } else { - // Update existing conversation timestamp conversation.updatedAt = currentTime; } - logger.info("[RedisConversationMemoryManager] Processing conversation", { - isNewConversation: !conversationData, - messageCount: conversation.messages.length, - sessionId: conversation.sessionId, - userId: conversation.userId, - }); + const tokenThreshold = options.providerDetails + ? getEffectiveTokenThreshold( + options.providerDetails.provider, + options.providerDetails.model, + this.config.tokenThreshold, + conversation.tokenThreshold, + ) + : this.config.tokenThreshold || 50000; - // Add new messages to conversation history with new format const userMsg: ChatMessage = { - id: this.generateMessageId(conversation), - timestamp: startTimeStamp?.toISOString() || this.generateTimestamp(), + id: randomUUID(), + timestamp: + options.startTimeStamp?.toISOString() || this.generateTimestamp(), role: "user", - content: userMessage, + content: options.userMessage, }; conversation.messages.push(userMsg); await this.flushPendingToolData( conversation, - sessionId, + options.sessionId, normalizedUserId, ); const assistantMsg: ChatMessage = { - id: this.generateMessageId(conversation), + id: randomUUID(), timestamp: this.generateTimestamp(), role: "assistant", - content: aiResponse, + content: options.aiResponse, }; conversation.messages.push(assistantMsg); logger.info("[RedisConversationMemoryManager] Added new messages", { - newMessageCount: conversation.messages.length, - latestMessages: [ - { - role: conversation.messages[conversation.messages.length - 2]?.role, - contentLength: - conversation.messages[conversation.messages.length - 2]?.content - .length, - }, - { - role: conversation.messages[conversation.messages.length - 1]?.role, - contentLength: - conversation.messages[conversation.messages.length - 1]?.content - .length, - }, - ], + sessionId: conversation.sessionId, + userId: conversation.userId, }); - // Save updated conversation object - const serializedData = serializeConversation(conversation); - logger.debug( - "[RedisConversationMemoryManager] Saving conversation to Redis", - { - redisKey, - messageCount: conversation.messages.length, - serializedDataLength: serializedData.length, - title: conversation.title, - }, - ); - logger.info("Storing conversation data to Redis", { - sessionId, - dataLength: serializedData.length, - messageCount: conversation.messages.length, - }); + // Use per-request enableSummarization with higher priority than instance config + const shouldSummarize = + options.enableSummarization !== undefined + ? options.enableSummarization + : this.config.enableSummarization; + + if (shouldSummarize && options.providerDetails) { + const normalizedUserId = options.userId || "randomUser"; + const summarizationKey = `${options.sessionId}:${normalizedUserId}`; + + // Only trigger summarization if not already in progress for this session + if (!this.summarizationInProgress.has(summarizationKey)) { + setImmediate(async () => { + try { + await this.checkAndSummarize( + conversation, + tokenThreshold, + options.sessionId, + options.userId, + ); + } catch (error) { + logger.error("Background summarization failed", { + sessionId: conversation.sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } + }); + } else { + logger.debug( + "[RedisConversationMemoryManager] Summarization already in progress, skipping", + { + sessionId: options.sessionId, + userId: normalizedUserId, + }, + ); + } + } + const serializedData = serializeConversation(conversation); await this.redisClient.set(redisKey, serializedData); - // Set TTL if configured if (this.redisConfig.ttl > 0) { - logger.debug("[RedisConversationMemoryManager] Setting Redis TTL", { - redisKey, - ttl: this.redisConfig.ttl, - }); await this.redisClient.expire(redisKey, this.redisConfig.ttl); } - // Add session to user's session set - if (userId) { - await this.addUserSession(userId, sessionId); + if (options.userId) { + await this.addUserSession(options.userId, options.sessionId); } logger.debug( "[RedisConversationMemoryManager] Successfully stored conversation turn", { - sessionId, + sessionId: options.sessionId, totalMessages: conversation.messages.length, title: conversation.title, }, ); } catch (error) { throw new ConversationMemoryError( - `Failed to store conversation turn in Redis for session ${sessionId}`, + `Failed to store conversation turn in Redis for session ${options.sessionId}`, "STORAGE_ERROR", { - sessionId, + sessionId: options.sessionId, error: error instanceof Error ? error.message : String(error), }, ); @@ -567,11 +533,179 @@ export class RedisConversationMemoryManager { } /** - * Build context messages for AI prompt injection + * Check if summarization is needed based on token count + */ + private async checkAndSummarize( + conversation: RedisConversationObject, + threshold: number, + sessionId: string, + userId?: string, + ): Promise { + const normalizedUserId = userId || "randomUser"; + const summarizationKey = `${sessionId}:${normalizedUserId}`; + + // Acquire lock - if already in progress, skip + if (this.summarizationInProgress.has(summarizationKey)) { + logger.debug( + "[RedisConversationMemoryManager] Summarization already in progress, skipping", + { + sessionId, + userId: normalizedUserId, + }, + ); + return; + } + + this.summarizationInProgress.add(summarizationKey); + + try { + const session: SessionMemory = { + sessionId: conversation.sessionId, + userId: conversation.userId, + messages: conversation.messages, + summarizedUpToMessageId: conversation.summarizedUpToMessageId, + summarizedMessage: conversation.summarizedMessage, + tokenThreshold: conversation.tokenThreshold, + lastTokenCount: conversation.lastTokenCount, + lastCountedAt: conversation.lastCountedAt, + createdAt: new Date(conversation.createdAt).getTime(), + lastActivity: new Date(conversation.updatedAt).getTime(), + }; + + const contextMessages = buildContextFromPointer(session); + const tokenCount = this.estimateTokens(contextMessages); + + conversation.lastTokenCount = tokenCount; + conversation.lastCountedAt = Date.now(); + + if (tokenCount >= threshold) { + await this.summarizeSessionTokenBased( + conversation, + threshold, + sessionId, + userId, + ); + } + } catch (error) { + logger.error("Token counting or summarization failed", { + sessionId: conversation.sessionId, + error: error instanceof Error ? error.message : String(error), + }); + } finally { + // Release lock when done + this.summarizationInProgress.delete(summarizationKey); + } + } + + /** + * Estimate total tokens for a list of messages + */ + private estimateTokens(messages: ChatMessage[]): number { + return messages.reduce((total, msg) => { + return total + TokenUtils.estimateTokenCount(msg.content); + }, 0); + } + + /** + * Token-based summarization (pointer-based, non-destructive) + */ + private async summarizeSessionTokenBased( + conversation: RedisConversationObject, + threshold: number, + sessionId: string, + userId?: string, + ): Promise { + const startIndex = conversation.summarizedUpToMessageId + ? conversation.messages.findIndex( + (m) => m.id === conversation.summarizedUpToMessageId, + ) + 1 + : 0; + + const recentMessages = conversation.messages.slice(startIndex); + + if (recentMessages.length === 0) { + return; + } + + // We only want to include user, assistant, and system messages in summarization + const filteredRecentMessages = recentMessages.filter( + (msg) => msg.role !== "tool_call" && msg.role !== "tool_result", + ); + + const targetRecentTokens = threshold * 0.3; + const splitIndex = await this.findSplitIndexByTokens( + filteredRecentMessages, + targetRecentTokens, + ); + + const messagesToSummarize = filteredRecentMessages.slice(0, splitIndex); + + if (messagesToSummarize.length === 0) { + return; + } + + const summary = await generateSummary( + messagesToSummarize, + this.config, + "[RedisConversationMemoryManager]", + conversation.summarizedMessage, + ); + + if (!summary) { + logger.warn( + `[RedisConversationMemoryManager] Summary generation failed for session ${conversation.sessionId}`, + ); + return; + } + + const lastSummarized = messagesToSummarize[messagesToSummarize.length - 1]; + conversation.summarizedUpToMessageId = lastSummarized.id; + conversation.summarizedMessage = summary; + if (this.redisClient) { + const redisKey = getSessionKey(this.redisConfig, sessionId, userId); + const serializedData = serializeConversation(conversation); + await this.redisClient.set(redisKey, serializedData); + + if (this.redisConfig.ttl > 0) { + await this.redisClient.expire(redisKey, this.redisConfig.ttl); + } + } + } + + /** + * Find split index to keep recent messages within target token count + */ + private async findSplitIndexByTokens( + messages: ChatMessage[], + targetRecentTokens: number, + ): Promise { + let recentTokens = 0; + let splitIndex = messages.length; + + for (let i = messages.length - 1; i >= 0; i--) { + const msgTokens = TokenUtils.estimateTokenCount(messages[i].content); + + if (recentTokens + msgTokens > targetRecentTokens) { + splitIndex = i + 1; + break; + } + + recentTokens += msgTokens; + } + + // Ensure we're summarizing at least something + return Math.max(1, splitIndex); + } + + /** + * Build context messages for AI prompt injection (TOKEN-BASED) + * Returns messages from pointer onwards (or all if no pointer) + * Filters out tool_call and tool_result messages when summarization is enabled */ async buildContextMessages( sessionId: string, userId?: string, + enableSummarization?: boolean, ): Promise { logger.info("[RedisConversationMemoryManager] Building context messages", { sessionId, @@ -579,40 +713,46 @@ export class RedisConversationMemoryManager { method: "buildContextMessages", }); - const messages = await this.getUserSessionHistory( - userId || "randomUser", - sessionId, - ); + const redisKey = getSessionKey(this.redisConfig, sessionId, userId); + const conversationData = await this.redisClient?.get(redisKey); + const conversation = deserializeConversation(conversationData || null); - if (!messages) { - logger.info( - "[RedisConversationMemoryManager] No context messages found", - { - sessionId, - userId, - }, - ); + if (!conversation) { return []; } - logger.info("[RedisConversationMemoryManager] Retrieved messages", { - messageCount: messages.length, - hasMessages: messages.length > 0, - }); + const session: SessionMemory = { + sessionId: conversation.sessionId, + userId: conversation.userId, + messages: conversation.messages, + summarizedUpToMessageId: conversation.summarizedUpToMessageId, + summarizedMessage: conversation.summarizedMessage, + tokenThreshold: conversation.tokenThreshold, + lastTokenCount: conversation.lastTokenCount, + lastCountedAt: conversation.lastCountedAt, + createdAt: new Date(conversation.createdAt).getTime(), + lastActivity: new Date(conversation.updatedAt).getTime(), + }; + + const contextMessages = buildContextFromPointer(session); + const isSummarizationEnabled = + enableSummarization !== undefined + ? enableSummarization + : this.config.enableSummarization === true; + + let finalMessages = contextMessages; + if (isSummarizationEnabled) { + finalMessages = contextMessages.filter( + (msg) => msg.role !== "tool_call" && msg.role !== "tool_result", + ); + } logger.info("[RedisConversationMemoryManager] Retrieved context messages", { sessionId, userId, - messageCount: messages.length, - messageRoles: messages.map((m) => m.role), - firstMessagePreview: messages[0]?.content?.substring(0, 50), - lastMessagePreview: messages[messages.length - 1]?.content?.substring( - 0, - 50, - ), }); - return messages; + return finalMessages; } /** @@ -934,10 +1074,21 @@ User message: "${userMessage}`; /** * Create summary system message */ - public createSummarySystemMessage(content: string): ChatMessage { + public createSummarySystemMessage( + content: string, + summarizesFrom?: string, + summarizesTo?: string, + ): ChatMessage { return { + id: `summary-${randomUUID()}`, role: "system", content: `Summary of previous conversation turns:\n\n${content}`, + timestamp: new Date().toISOString(), + metadata: { + isSummary: true, + summarizesFrom, + summarizesTo, + }, }; } @@ -1247,7 +1398,7 @@ User message: "${userMessage}`; toolCallMap.set(toolCallId, toolName); const toolCallMessage: ChatMessage = { - id: this.generateMessageId(conversation), + id: randomUUID(), timestamp: toolCall.timestamp?.toISOString() || this.generateTimestamp(), role: "tool_call", @@ -1269,7 +1420,7 @@ User message: "${userMessage}`; const toolName = toolCallMap.get(toolCallId) || "unknown"; const toolResultMessage: ChatMessage = { - id: this.generateMessageId(conversation), + id: randomUUID(), timestamp: toolResult.timestamp?.toISOString() || this.generateTimestamp(), role: "tool_result", diff --git a/src/lib/neurolink.ts b/src/lib/neurolink.ts index 3e1aa8d87..0540f5be9 100644 --- a/src/lib/neurolink.ts +++ b/src/lib/neurolink.ts @@ -110,6 +110,7 @@ import { EventEmitter } from "events"; import type { ConversationMemoryConfig, ChatMessage, + ProviderDetails, } from "./types/conversation.js"; import { ConversationMemoryManager } from "./core/conversationMemoryManager.js"; import { RedisConversationMemoryManager } from "./core/redisConversationMemoryManager.js"; @@ -144,6 +145,9 @@ import type { ObservabilityConfig } from "./types/observability.js"; import type { NeurolinkConstructorConfig } from "./types/configTypes.js"; import { initializeMem0, type Mem0Config } from "./memory/mem0Initializer.js"; +const { initializeConversationMemory } = await import( + "./core/conversationMemoryInitializer.js" +); export class NeuroLink { private mcpInitialized = false; @@ -2930,20 +2934,23 @@ Current user's request: ${currentInput}`; const userId = ( enhancedOptions.context as Record )?.userId as string; + let providerDetails: ProviderDetails | undefined = undefined; + if (enhancedOptions.model) { + providerDetails = { + provider: providerName, + model: enhancedOptions.model, + }; + } try { - await self.conversationMemory.storeConversationTurn( + await self.conversationMemory.storeConversationTurn({ sessionId, userId, - originalPrompt ?? "", - accumulatedContent, - new Date(startTime), - ); - - logger.debug("Stream conversation turn stored", { - sessionId, - userInputLength: originalPrompt?.length ?? 0, - responseLength: accumulatedContent.length, + userMessage: originalPrompt ?? "", + aiResponse: accumulatedContent, + startTimeStamp: new Date(startTime), + providerDetails, + enableSummarization: enhancedOptions.enableSummarization, }); } catch (error) { logger.warn("Failed to store stream conversation turn", { @@ -3100,8 +3107,9 @@ Current user's request: ${currentInput}`; const conversationMessages = await getConversationMessages( this.conversationMemory, { + ...options, prompt: options.input.text, - context: options.context as Record, + context: options.context, } as TextGenerationOptions, ); @@ -3264,20 +3272,23 @@ Current user's request: ${currentInput}`; )?.sessionId as string; const userId = (enhancedOptions?.context as Record) ?.userId as string; + let providerDetails: ProviderDetails | undefined = undefined; + if (options.model) { + providerDetails = { + provider: providerName, + model: options.model, + }; + } try { - await self.conversationMemory.storeConversationTurn( - sessionId || (options.context?.sessionId as string), - userId || (options.context?.userId as string), - originalPrompt ?? "", - fallbackAccumulatedContent, - new Date(startTime), - ); - - logger.debug("Fallback stream conversation turn stored", { - sessionId: sessionId || options.context?.sessionId, - userInputLength: originalPrompt?.length ?? 0, - responseLength: fallbackAccumulatedContent.length, + await self.conversationMemory.storeConversationTurn({ + sessionId: sessionId || (options.context?.sessionId as string), + userId: userId || (options.context?.userId as string), + userMessage: originalPrompt ?? "", + aiResponse: fallbackAccumulatedContent, + startTimeStamp: new Date(startTime), + providerDetails, + enableSummarization: enhancedOptions?.enableSummarization, }); } catch (error) { logger.warn("Failed to store fallback stream conversation turn", { @@ -5906,39 +5917,12 @@ Current user's request: ${currentInput}`; generateInternalHrTimeStart: bigint, ): Promise { try { - // Import the integration module - const { initializeConversationMemory } = await import( - "./core/conversationMemoryInitializer.js" - ); - - // Use the integration module to create the appropriate memory manager - const memoryManagerCreateStartTime = process.hrtime.bigint(); const memoryManager = await initializeConversationMemory( this.conversationMemoryConfig, ); // Assign to conversationMemory with proper type to handle both memory manager types this.conversationMemory = memoryManager; - const memoryManagerCreateEndTime = process.hrtime.bigint(); - const memoryManagerCreateDurationNs = - memoryManagerCreateEndTime - memoryManagerCreateStartTime; - - logger.info(`[NeuroLink] ✅ LOG_POINT_G004_MEMORY_LAZY_INIT_SUCCESS`, { - logPoint: "G004_MEMORY_LAZY_INIT_SUCCESS", - generateInternalId, - timestamp: new Date().toISOString(), - elapsedMs: Date.now() - generateInternalStartTime, - elapsedNs: ( - process.hrtime.bigint() - generateInternalHrTimeStart - ).toString(), - memoryManagerCreateDurationNs: memoryManagerCreateDurationNs.toString(), - memoryManagerCreateDurationMs: - Number(memoryManagerCreateDurationNs) / 1000000, - storageType: process.env.STORAGE_TYPE || "memory", - message: - "Lazy conversation memory initialization completed successfully", - }); - // Reset the lazy init flag since we've now initialized this.conversationMemoryNeedsInit = false; } catch (error) { diff --git a/src/lib/types/conversation.ts b/src/lib/types/conversation.ts index 8c931ca23..76e2e644f 100644 --- a/src/lib/types/conversation.ts +++ b/src/lib/types/conversation.ts @@ -15,17 +15,11 @@ export type ConversationMemoryConfig = { /** Maximum number of sessions to keep in memory (default: 50) */ maxSessions?: number; - /** Maximum number of conversation turns to keep per session (default: 20) */ - maxTurnsPerSession?: number; - /** Enable automatic summarization */ enableSummarization?: boolean; - /** Turn count to trigger summarization */ - summarizationThresholdTurns?: number; - - /** Target turn count for the summary */ - summarizationTargetTurns?: number; + /** Token threshold to trigger summarization (optional - defaults to 80% of model context) */ + tokenThreshold?: number; /** Provider to use for summarization */ summarizationProvider?: string; @@ -41,6 +35,15 @@ export type ConversationMemoryConfig = { /** Redis configuration (optional) - overrides environment variables */ redisConfig?: RedisStorageConfig; + + /** @deprecated Use tokenThreshold instead - Maximum number of conversation turns to keep per session (default: 20) */ + maxTurnsPerSession?: number; + + /** @deprecated Use tokenThreshold instead - Turn count to trigger summarization */ + summarizationThresholdTurns?: number; + + /** @deprecated Use tokenThreshold instead - Target turn count for the summary */ + summarizationTargetTurns?: number; }; /** * Complete memory for a conversation session @@ -65,6 +68,21 @@ export type SessionMemory = { /** When this session was last active */ lastActivity: number; + /** Pointer to last summarized message ID (NEW - for token-based memory) */ + summarizedUpToMessageId?: string; + + /** Stored summary message that condenses conversation history up to summarizedUpToMessageId */ + summarizedMessage?: string; + + /** Per-session token threshold override (NEW - for token-based memory) */ + tokenThreshold?: number; + + /** Cached token count for performance (NEW - for token-based memory) */ + lastTokenCount?: number; + + /** When token count was last calculated (NEW - for token-based memory) */ + lastCountedAt?: number; + /** Optional session metadata */ metadata?: { /** User role or permissions */ @@ -93,16 +111,16 @@ export type ConversationMemoryStats = { * Chat message format for conversation history */ export type ChatMessage = { + /** Unique message identifier (required for token-based memory) */ + id: string; + /** Role/type of the message */ role: "user" | "assistant" | "system" | "tool_call" | "tool_result"; /** Content of the message */ content: string; - /** Message ID (optional) - for new format */ - id?: string; - - /** Timestamp (optional) - for new format */ + /** Timestamp (ISO string) */ timestamp?: string; /** Tool name (optional) - for tool_call/tool_result messages */ @@ -119,6 +137,18 @@ export type ChatMessage = { type?: string; error?: string; }; + + /** Message metadata (NEW - for token-based memory) */ + metadata?: { + /** Is this a summary message? */ + isSummary?: boolean; + /** First message ID that this summary covers */ + summarizesFrom?: string; + /** Last message ID that this summary covers */ + summarizesTo?: string; + /** Was this message truncated due to token limits? */ + truncated?: boolean; + }; }; /** @@ -201,6 +231,19 @@ export type SessionIdentifier = { userId?: string; }; +/** + * Options for storing a conversation turn + */ +export type StoreConversationTurnOptions = { + sessionId: string; + userId?: string; + userMessage: string; + aiResponse: string; + startTimeStamp?: Date; + providerDetails?: ProviderDetails; + enableSummarization?: boolean; +}; + /** * Lightweight session metadata for efficient session listing * Contains only essential information without heavy message arrays @@ -234,6 +277,21 @@ export type ConversationBase = { /** When this conversation was last updated */ updatedAt: string; + + /** Pointer to last summarized message (token-based memory) */ + summarizedUpToMessageId?: string; + + /** Stored summary message that condenses conversation history up to summarizedUpToMessageId */ + summarizedMessage?: string; + + /** Per-session token threshold override */ + tokenThreshold?: number; + + /** Cached token count for efficiency */ + lastTokenCount?: number; + + /** Timestamp of last token count */ + lastCountedAt?: number; }; /** @@ -319,3 +377,8 @@ export type RedisStorageConfig = { [key: string]: string | number | boolean | undefined; }; }; + +export type ProviderDetails = { + provider: string; + model: string; +}; diff --git a/src/lib/types/generateTypes.ts b/src/lib/types/generateTypes.ts index 0f093d9ea..1b5f623d4 100644 --- a/src/lib/types/generateTypes.ts +++ b/src/lib/types/generateTypes.ts @@ -397,6 +397,8 @@ export type TextGenerationOptions = { formatStyle?: "raw" | "markdown" | "json"; includeHeaders?: boolean; }; + + enableSummarization?: boolean; // Enable/disable summarization for this specific request }; /** diff --git a/src/lib/types/sdkTypes.ts b/src/lib/types/sdkTypes.ts index 701ee0197..cb60aa54a 100644 --- a/src/lib/types/sdkTypes.ts +++ b/src/lib/types/sdkTypes.ts @@ -202,7 +202,6 @@ export type { export type { ConversationMemoryConfig, SessionMemory, - ConversationMemoryStats, ChatMessage, MessageContent, MultimodalChatMessage, diff --git a/src/lib/types/streamTypes.ts b/src/lib/types/streamTypes.ts index 8a9413fde..8a43c8e6b 100644 --- a/src/lib/types/streamTypes.ts +++ b/src/lib/types/streamTypes.ts @@ -355,6 +355,8 @@ export type StreamOptions = { // NEW: Middleware related config middleware?: MiddlewareFactoryOptions; + + enableSummarization?: boolean; // Enable/disable summarization for this specific request }; /** diff --git a/src/lib/utils/conversationMemory.ts b/src/lib/utils/conversationMemory.ts index 76b0dbcf7..833a2b00f 100644 --- a/src/lib/utils/conversationMemory.ts +++ b/src/lib/utils/conversationMemory.ts @@ -6,6 +6,8 @@ import type { ConversationMemoryConfig, ChatMessage, + SessionMemory, + ProviderDetails, } from "../types/conversation.js"; import type { ConversationMemoryManager } from "../core/conversationMemoryManager.js"; import type { RedisConversationMemoryManager } from "../core/redisConversationMemoryManager.js"; @@ -13,8 +15,14 @@ import type { TextGenerationOptions, TextGenerationResult, } from "../types/generateTypes.js"; -import { getConversationMemoryDefaults } from "../config/conversationMemory.js"; +import { + getConversationMemoryDefaults, + MEMORY_THRESHOLD_PERCENTAGE, + DEFAULT_FALLBACK_THRESHOLD, +} from "../config/conversationMemory.js"; +import { TokenUtils } from "../constants/tokens.js"; import { logger } from "./logger.js"; +import { NeuroLink } from "../neurolink.js"; /** * Apply conversation memory defaults to user configuration @@ -67,10 +75,11 @@ export async function getConversationMessages( | string | undefined; - // Remove duplicate summarization logic - it should be handled in ConversationMemoryManager + const enableSummarization = options.enableSummarization ?? undefined; const messages = await conversationMemory.buildContextMessages( sessionId, userId, + enableSummarization, ); logger.debug( "[conversationMemoryUtils] Conversation messages retrieved successfully", @@ -172,15 +181,25 @@ export async function storeConversationTurn( const userMessage = originalOptions.originalPrompt || originalOptions.prompt || ""; - const aiResponse = result.content; + + const aiResponse = result.content ?? ""; + let providerDetails: ProviderDetails | undefined = undefined; + if (result.provider && result.model) { + providerDetails = { + provider: result.provider, + model: result.model, + }; + } try { - await conversationMemory.storeConversationTurn( + await conversationMemory.storeConversationTurn({ sessionId, userId, userMessage, aiResponse, startTimeStamp, - ); + providerDetails, + enableSummarization: originalOptions.enableSummarization, + }); logger.debug( "[conversationMemoryUtils] Conversation turn stored successfully", @@ -202,3 +221,199 @@ export async function storeConversationTurn( }); } } + +/** + * Build context messages from pointer onwards (token-based memory) + * Returns summary message (if exists) + all messages after the summarized pointer + * @param session - Session memory with pointer + * @returns Context messages to send to LLM + */ +export function buildContextFromPointer(session: SessionMemory): ChatMessage[] { + if (!session.summarizedUpToMessageId || !session.summarizedMessage) { + return session.messages; + } + + // find a better way to wirte this + const pointerIndex = session.messages.findIndex( + (msg) => msg.id === session.summarizedUpToMessageId, + ); + + if (pointerIndex === -1) { + logger.warn("Pointer message not found, returning all messages", { + sessionId: session.sessionId, + pointer: session.summarizedUpToMessageId, + totalMessages: session.messages.length, + }); + return session.messages; + } + + const messagesAfterPointer = session.messages.slice(pointerIndex + 1); + + // Construct context: summary message + recent messages + const summaryMessage: ChatMessage = { + id: `summary-${session.summarizedUpToMessageId}`, + role: "system", + content: `Previous conversation summary: ${session.summarizedMessage}`, + timestamp: new Date().toISOString(), + metadata: { + isSummary: true, + summarizesTo: session.summarizedUpToMessageId, + }, + }; + + logger.debug("Building context with summary", { + sessionId: session.sessionId, + pointerIndex, + messagesAfterPointer: messagesAfterPointer.length, + totalMessages: session.messages.length, + summaryLength: session.summarizedMessage.length, + }); + + return [summaryMessage, ...messagesAfterPointer]; +} + +/** + * Create summarization prompt from message history + * Used by both in-memory and Redis conversation managers + * @param history - Messages to summarize + * @param previousSummary - Optional previous summary to build upon + */ +export function createSummarizationPrompt( + history: ChatMessage[], + previousSummary?: string, +): string { + const formattedHistory = history + .map((msg) => `${msg.role}: ${msg.content}`) + .join("\n\n"); + + const previousSummarySection = previousSummary + ? `Previous Summary: +--- +${previousSummary} +--- + +` + : ""; + + return ` +You are a context summarization AI. Your task is to condense the following conversation history for another AI assistant. +${previousSummary ? "Build upon the previous summary and incorporate the new conversation turns below." : ""} +The summary must be a concise, third-person narrative that retains all critical information, including key entities, technical details, decisions made, and any specific dates or times mentioned. +Ensure the summary flows logically and is ready to be used as context for the next turn in the conversation. + +${previousSummarySection}Conversation History to Summarize: +--- +${formattedHistory} +--- +`.trim(); +} + +/** + * Calculate token threshold based on model's output token limit + * Uses existing provider token limits as proxy for context window + * @param provider - AI provider name + * @param model - Model name + * @returns Token threshold (80% of model's token limit) + */ +export function calculateTokenThreshold( + provider: string, + model: string, +): number { + try { + // Get model's token limit from existing TokenUtils + const modelTokenLimit = TokenUtils.getProviderTokenLimit(provider, model); + + // Return 80% of token limit for conversation memory + // This is conservative since output limits are typically smaller than input limits + return Math.floor(modelTokenLimit * MEMORY_THRESHOLD_PERCENTAGE); + } catch (error) { + logger.warn("Failed to calculate model threshold, using fallback", { + provider, + model, + error: error instanceof Error ? error.message : String(error), + }); + return DEFAULT_FALLBACK_THRESHOLD; + } +} + +/** + * Get effective token threshold for a session + * Priority: session override > env var > model-based (80%) > fallback + * @param provider - AI provider name + * @param model - Model name + * @param envOverride - Environment variable override + * @param sessionOverride - Per-session token threshold override + * @returns Effective token threshold + */ +export function getEffectiveTokenThreshold( + provider: string, + model: string, + envOverride?: number, + sessionOverride?: number, +): number { + // Priority 1: Session-level override + if (sessionOverride && sessionOverride > 0) { + return sessionOverride; + } + + // Priority 2: Environment variable override + if (envOverride && envOverride > 0) { + return envOverride; + } + + // Priority 3: Model-based calculation (80% of context window) + try { + return calculateTokenThreshold(provider, model); + } catch (error) { + logger.warn("Failed to calculate effective threshold, using fallback", { + provider, + model, + error: error instanceof Error ? error.message : String(error), + }); + // Priority 4: Fallback for unknown models + return DEFAULT_FALLBACK_THRESHOLD; + } +} + +/** + * Generate summary using configured provider and model + * Centralized summarization logic used by both ConversationMemoryManager and RedisConversationMemoryManager + * @param messages - Messages to summarize + * @param config - Conversation memory configuration containing provider/model settings + * @param previousSummary - Optional previous summary to build upon + * @param logPrefix - Prefix for log messages (e.g., "[ConversationMemory]" or "[RedisConversationMemoryManager]") + * @returns Summary text or null if generation fails + */ +export async function generateSummary( + messages: ChatMessage[], + config: ConversationMemoryConfig, + logPrefix = "[ConversationMemory]", + previousSummary?: string, +): Promise { + const summarizationPrompt = createSummarizationPrompt( + messages, + previousSummary, + ); + const summarizer = new NeuroLink({ + conversationMemory: { enabled: false }, + }); + + try { + if (!config.summarizationProvider || !config.summarizationModel) { + logger.error(`${logPrefix} Missing summarization provider`); + return null; + } + + const summaryResult = await summarizer.generate({ + input: { text: summarizationPrompt }, + provider: config.summarizationProvider, + model: config.summarizationModel, + disableTools: true, + }); + + return summaryResult.content || null; + } catch (error) { + logger.error(`${logPrefix} Error generating summary`, { error }); + return null; + } +} diff --git a/src/lib/utils/conversationMemoryUtils.ts b/src/lib/utils/conversationMemoryUtils.ts index acacff9fd..921630b79 100644 --- a/src/lib/utils/conversationMemoryUtils.ts +++ b/src/lib/utils/conversationMemoryUtils.ts @@ -6,6 +6,7 @@ import type { ConversationMemoryConfig, ChatMessage, + ProviderDetails, } from "../types/conversation.js"; import type { ConversationMemoryManager } from "../core/conversationMemoryManager.js"; import type { @@ -88,14 +89,25 @@ export async function storeConversationTurn( return; } + let providerDetails: ProviderDetails | undefined = undefined; + if (result.provider && result.model) { + providerDetails = { + provider: result.provider, + model: result.model, + }; + } + try { - await conversationMemory.storeConversationTurn( + await conversationMemory.storeConversationTurn({ sessionId, userId, - originalOptions.originalPrompt || originalOptions.prompt || "", - result.content, + userMessage: + originalOptions.originalPrompt || originalOptions.prompt || "", + aiResponse: result.content, startTimeStamp, - ); + providerDetails, + enableSummarization: originalOptions.enableSummarization, + }); logger.debug("Conversation turn stored", { sessionId, diff --git a/src/lib/utils/redis.ts b/src/lib/utils/redis.ts index 0b47b82e1..ab4e1a6ae 100644 --- a/src/lib/utils/redis.ts +++ b/src/lib/utils/redis.ts @@ -133,18 +133,10 @@ export function deserializeConversation( data: string | null, ): RedisConversationObject | null { if (!data) { - logger.debug( - "[redisUtils] No conversation data to deserialize, returning null", - ); return null; } try { - logger.debug("[redisUtils] Deserializing conversation", { - dataLength: data.length, - dataPreview: data.substring(0, 100) + (data.length > 100 ? "..." : ""), - }); - // Parse as unknown first, then validate before casting const parsedData = JSON.parse(data) as unknown;