diff --git a/services/telegram-bot/.env b/services/telegram-bot/.env index 939733484..b91bba627 100644 --- a/services/telegram-bot/.env +++ b/services/telegram-bot/.env @@ -4,6 +4,7 @@ TELEGRAM_BOT_TOKEN='' LLM_API_BASE_URL='' LLM_API_KEY='' LLM_MODEL='' +LLM_RESPONSE_LANGUAGE='' LLM_VISION_API_BASE_URL='' LLM_VISION_API_KEY='' diff --git a/services/telegram-bot/src/bots/telegram/index.test.ts b/services/telegram-bot/src/bots/telegram/index.test.ts deleted file mode 100644 index 5d9949761..000000000 --- a/services/telegram-bot/src/bots/telegram/index.test.ts +++ /dev/null @@ -1,142 +0,0 @@ -import type { SQL } from 'drizzle-orm' - -import { env } from 'node:process' -import { embed } from '@xsai/embed' -import { cosineDistance, desc, sql } from 'drizzle-orm' -import { beforeAll, describe, expect, it } from 'vitest' - -import { initDb, useDrizzle } from '../../db' -import { chatMessagesTable } from '../../db/schema' -import { chatMessageToOneLine } from '../../models' - -beforeAll(async () => { - await initDb() -}) - -describe.todo('telegram bot', { timeout: 30000 }, async () => { - it('should be able to run', async () => { - const db = useDrizzle() - const contextWindowSize = 5 // Number of messages to include before and after - - const embedding = await embed({ - baseURL: env.EMBEDDING_API_BASE_URL!, - apiKey: env.EMBEDDING_API_KEY!, - model: env.EMBEDDING_MODEL!, - input: '测试一下行不行', - }) - .then(res => res) - .catch((err) => { - console.error(err, err.cause) - return { embedding: [] } - }) - if (embedding.embedding.length === 0) { - throw new Error('Failed to embed the input') - } - - const relevantChatMessages = await Promise.all([embedding].map(async (embedding) => { - let similarity: SQL - - switch (env.EMBEDDING_DIMENSION) { - case '1536': - similarity = sql`(1 - (${cosineDistance(chatMessagesTable.content_vector_1536, embedding.embedding)}))` - break - case '1024': - similarity = sql`(1 - (${cosineDistance(chatMessagesTable.content_vector_1024, embedding.embedding)}))` - break - case '768': - similarity = sql`(1 - (${cosineDistance(chatMessagesTable.content_vector_768, embedding.embedding)}))` - break - default: - throw new Error(`Unsupported embedding dimension: ${env.EMBEDDING_DIMENSION}`) - } - - const timeRelevance = sql`(1 - (CEIL(EXTRACT(EPOCH FROM NOW()) * 1000)::bigint - ${chatMessagesTable.created_at}) / 86400 / 30)` - const combinedScore = sql`((1.2 * ${similarity}) + (0.2 * ${timeRelevance}))` - - // Get top messages with similarity above threshold - const relevantMessages = await db - .select({ - id: chatMessagesTable.id, - platform: chatMessagesTable.platform, - from_id: chatMessagesTable.from_id, - from_name: chatMessagesTable.from_name, - in_chat_id: chatMessagesTable.in_chat_id, - content: chatMessagesTable.content, - is_reply: chatMessagesTable.is_reply, - reply_to_name: chatMessagesTable.reply_to_name, - created_at: chatMessagesTable.created_at, - updated_at: chatMessagesTable.updated_at, - similarity: sql`${similarity} AS "similarity"`, - time_relevance: sql`${timeRelevance} AS "time_relevance"`, - combined_score: sql`${combinedScore} AS "combined_score"`, - }) - .from(chatMessagesTable) - .where(sql`${similarity} > '0.5'`) - .orderBy(desc(sql`combined_score`)) - .limit(3) - - // Now fetch the context for each message - return await Promise.all( - relevantMessages.map(async (message) => { - // Get N messages before the target message - const messagesBefore = await db - .select({ - id: chatMessagesTable.id, - platform: chatMessagesTable.platform, - from_id: chatMessagesTable.from_id, - from_name: chatMessagesTable.from_name, - in_chat_id: chatMessagesTable.in_chat_id, - content: chatMessagesTable.content, - is_reply: chatMessagesTable.is_reply, - reply_to_name: chatMessagesTable.reply_to_name, - created_at: chatMessagesTable.created_at, - updated_at: chatMessagesTable.updated_at, - }) - .from(chatMessagesTable) - .where(sql`${chatMessagesTable.in_chat_id} = ${message.in_chat_id} AND - ${chatMessagesTable.created_at} < ${message.created_at}`) - .orderBy(desc(chatMessagesTable.created_at)) - .limit(contextWindowSize) - - // Get N messages after the target message - const messagesAfter = await db - .select({ - id: chatMessagesTable.id, - platform: chatMessagesTable.platform, - from_id: chatMessagesTable.from_id, - from_name: chatMessagesTable.from_name, - in_chat_id: chatMessagesTable.in_chat_id, - content: chatMessagesTable.content, - is_reply: chatMessagesTable.is_reply, - reply_to_name: chatMessagesTable.reply_to_name, - created_at: chatMessagesTable.created_at, - updated_at: chatMessagesTable.updated_at, - }) - .from(chatMessagesTable) - .where(sql`${chatMessagesTable.in_chat_id} = ${message.in_chat_id} AND - ${chatMessagesTable.created_at} > ${message.created_at}`) - .orderBy(chatMessagesTable.created_at) - .limit(contextWindowSize) - - // Combine all messages in chronological order - const contextMessages = [ - ...messagesBefore.reverse(), // Reverse to get chronological order - message, - ...messagesAfter, - ] - - // eslint-disable-next-line no-console - console.log(contextMessages) - - const contextMessagesOneliner = (await Promise.all(contextMessages.map(m => chatMessageToOneLine(m)))) - return `One of the relevant message along with the context:\n${contextMessagesOneliner}` - }), - ) - })) - - // eslint-disable-next-line no-console - console.log(relevantChatMessages) - - expect(relevantChatMessages.length).toBe(1) - }) -}) diff --git a/services/telegram-bot/src/bots/telegram/index.ts b/services/telegram-bot/src/bots/telegram/index.ts index cabf1c4a6..540695873 100644 --- a/services/telegram-bot/src/bots/telegram/index.ts +++ b/services/telegram-bot/src/bots/telegram/index.ts @@ -14,6 +14,7 @@ import { interpretSticker } from '../../llm/sticker' import { recordMessage } from '../../models' import { listJoinedChats, recordJoinedChat } from '../../models/chats' import { readMessage } from './loop/read-message' +import { shouldInterruptProcessing } from './utils/interruption' import { sendMayStructuredMessage } from './utils/message' async function isChatIdBotAdmin(chatId: number) { @@ -21,15 +22,26 @@ async function isChatIdBotAdmin(chatId: number) { return admins.includes(chatId.toString()) } -async function handleLoop(state: BotSelf, msgs?: LLMMessage[], forGroupId?: string) { +async function handleLoop(state: BotSelf, msgs?: LLMMessage[], chatId?: string) { state.logger.log('handleLoop') + // Set the start time when beginning new processing + state.currentProcessingStartTime = Date.now() + // Create a new abort controller for this loop execution if (state.currentAbortController) { state.currentAbortController.abort() } state.currentAbortController = new AbortController() - const currentController = state.currentAbortController // Store reference to current controller + const currentController = state.currentAbortController + + // Track message processing state + if (chatId && !state.lastInteractedNChatIds.includes(chatId)) { + state.lastInteractedNChatIds.push(chatId) + } + if (state.lastInteractedNChatIds.length > 5) { + state.lastInteractedNChatIds = state.lastInteractedNChatIds.slice(-5) + } if (msgs == null) { msgs = [] @@ -37,36 +49,63 @@ async function handleLoop(state: BotSelf, msgs?: LLMMessage[], forGroupId?: stri try { try { - const action = await imagineAnAction(state.unreadMessages, currentController, msgs) + const action = await imagineAnAction(state.bot.botInfo.id.toString(), state.unreadMessages, currentController, msgs, state.lastInteractedNChatIds) switch (action.action) { case 'readMessages': - // eslint-disable-next-line no-case-declarations - let unreadMessagesForThisChat: Message[] | undefined = state.unreadMessages[action.groupId] - - if (forGroupId && forGroupId === action.groupId.toString() - && unreadMessagesForThisChat - && unreadMessagesForThisChat.length > 0) { - state.logger.log(`Interrupting message processing for group ${action.groupId} - new messages arrived`) - return handleLoop(state) - } if (Object.keys(state.unreadMessages).length === 0) { state.logger.log('No unread messages - deleting all unread messages') state.unreadMessages = {} break } - if (action.groupId == null) { + if (action.chatId == null) { state.logger.log('No group ID - deleting all unread messages') state.unreadMessages = {} break } + + // eslint-disable-next-line no-case-declarations + let unreadMessagesForThisChat: Message[] | undefined = state.unreadMessages[action.chatId] + + // Modified interruption logic + if (chatId && chatId === action.chatId + && unreadMessagesForThisChat + && unreadMessagesForThisChat.length > 0) { + const processingTime = state.currentProcessingStartTime + ? Date.now() - state.currentProcessingStartTime + : 0 + + const messageCount = unreadMessagesForThisChat.length + + // Factors to consider for interruption: + // + // 1. How long we've been processing (longer = more likely to finish) + // 2. Number of new messages (more = higher chance to interrupt) + // 3. Message content importance (could be determined by LLM) + + const shouldInterrupt = await shouldInterruptProcessing({ + processingTime, + messageCount, + currentMessages: msgs, + newMessages: unreadMessagesForThisChat, + chatId: action.chatId, + }) + + if (shouldInterrupt) { + state.logger.log(`Interrupting message processing for chat ${action.chatId} - new messages deemed more important`) + return handleLoop(state) + } + else { + state.logger.log(`Continuing current processing despite new messages in chat ${action.chatId}`) + } + } if (!Array.isArray(unreadMessagesForThisChat)) { - state.logger.log(`Unread messages for group ${action.groupId} is not an array - converting to array`) + state.logger.log(`Unread messages for group ${action.chatId} is not an array - converting to array`) unreadMessagesForThisChat = [] } if (unreadMessagesForThisChat.length === 0) { - state.logger.log(`No unread messages for group ${action.groupId} - deleting`) - delete state.unreadMessages[action.groupId] + state.logger.log(`No unread messages for group ${action.chatId} - deleting`) + delete state.unreadMessages[action.chatId] break } @@ -80,7 +119,7 @@ async function handleLoop(state: BotSelf, msgs?: LLMMessage[], forGroupId?: stri // return { break: true } // } - await readMessage(state, action, unreadMessagesForThisChat, currentController) + await readMessage(state, state.bot.botInfo.id.toString(), chatId, action, unreadMessagesForThisChat, currentController) break case 'listChats': msgs.push(message.user(`List of chats:${(await listJoinedChats()).map(chat => `ID:${chat.chat_id}, Name:${chat.chat_name}`).join('\n')}`)) @@ -108,9 +147,10 @@ async function handleLoop(state: BotSelf, msgs?: LLMMessage[], forGroupId?: stri state.logger.withError(err).log('Error occurred') } finally { - // Only clean up if this is still the current controller + // Clean up timing when done if (state.currentAbortController === currentController) { state.currentAbortController = null + state.currentProcessingStartTime = null } } } @@ -140,6 +180,8 @@ function newBotSelf(bot: Bot, logger: Logg): BotSelf { logger, processing: false, attentionHandler: undefined, + lastInteractedNChatIds: [], + currentProcessingStartTime: null, } // botSelf.attentionHandler = createAttentionHandler(botSelf, { @@ -199,14 +241,16 @@ async function processMessageQueue(state: BotSelf) { unreadMessagesForThisChat.push(nextMsg.message) - if (unreadMessagesForThisChat.length > 20) { - unreadMessagesForThisChat = unreadMessagesForThisChat.slice(-20) + if (unreadMessagesForThisChat.length > 100) { + unreadMessagesForThisChat = unreadMessagesForThisChat.slice(-100) } state.unreadMessages[nextMsg.message.chat.id] = unreadMessagesForThisChat + state.logger.withField('chatId', nextMsg.message.chat.id).log('message queue processed, triggering immediate reaction') + // Trigger immediate processing when messages are ready - handleLoop(state, [], nextMsg.message.chat.id.toString()) + await handleLoop(state, [], nextMsg.message.chat.id.toString()) state.messageQueue.shift() } } diff --git a/services/telegram-bot/src/bots/telegram/loop/read-message.ts b/services/telegram-bot/src/bots/telegram/loop/read-message.ts index df36bad6a..317db4c0e 100644 --- a/services/telegram-bot/src/bots/telegram/loop/read-message.ts +++ b/services/telegram-bot/src/bots/telegram/loop/read-message.ts @@ -14,6 +14,8 @@ import { sendMayStructuredMessage } from '../utils/message' export async function readMessage( state: BotSelf, + botId: string, + chatId: string, action: ReadMessagesAction, unreadMessages: Message[], abortController: AbortController, @@ -23,8 +25,8 @@ export async function readMessage( }> { const logger = useLogg('readMessage').useGlobalConfig() - const lastNMessages = await findLastNMessages(action.groupId, 30) - const lastNMessagesOneliner = lastNMessages.map(msg => chatMessageToOneLine(state.bot, msg)).join('\n') + const lastNMessages = await findLastNMessages(action.chatId, 30) + const lastNMessagesOneliner = lastNMessages.map(msg => chatMessageToOneLine(botId, msg)).join('\n') logger.withField('number_of_last_n_messages', lastNMessages.length).log('Successfully found last N messages') @@ -39,21 +41,18 @@ export async function readMessage( abortSignal: abortController.signal, }) - return { - embedding: embeddingResult.embedding, - message: msg, - } + return embeddingResult }) const unreadHistoryMessagesEmbedding = await Promise.all(unreadMessagesEmbeddingPromises) logger.withField('number_of_tasks', unreadMessagesEmbeddingPromises.length).log('Successfully embedded unread history messages') - const unreadHistoryMessages = await Promise.all(state.unreadMessages[action.groupId].map(msg => telegramMessageToOneLine(state.bot, msg))) + const unreadHistoryMessages = await Promise.all(state.unreadMessages[action.chatId].map(msg => telegramMessageToOneLine(botId, msg))) const unreadHistoryMessageOneliner = unreadHistoryMessages.join('\n') - state.unreadMessages[action.groupId] = [] + state.unreadMessages[action.chatId] = [] - const relevantChatMessages = await findRelevantMessages(state.bot, unreadHistoryMessagesEmbedding) + const relevantChatMessages = await findRelevantMessages(botId, chatId, unreadHistoryMessagesEmbedding) const relevantChatMessagesOneliner = (await Promise.all(relevantChatMessages.map(async msgs => msgs.join('\n')))).join('\n') logger.withField('number_of_relevant_chat_messages', relevantChatMessages.length).log('Successfully composed relevant chat messages') @@ -73,10 +72,11 @@ export async function readMessage( + 'All the messages you requested to read:\n' + `${unreadHistoryMessageOneliner || 'No messages'}` + '\n' - + 'Based on your personalities, imaging you have your own choice and interest over the world, ' + + 'Based on your personalities, imaging you have your own choice and interest over different, ' + 'giving the above context and chat history, would you like to participate in the conversation ' + 'about the topic? Or will you aggressively diss or piss off about the opinions of others?\n' + 'Feel free to ignore by just sending an empty array (i.e. []).' + + 'If you would like to participate, send me an array of messages you would like to send without telling you willing to participate.' + '\n' + 'Choose your action.', ), @@ -102,6 +102,6 @@ export async function readMessage( logger.withField('response', JSON.stringify(response.text)).log('Successfully generated response') - await sendMayStructuredMessage(state, response.text, action.groupId.toString()) + await sendMayStructuredMessage(state, response.text, action.chatId.toString()) return { break: true } } diff --git a/services/telegram-bot/src/bots/telegram/utils/interruption.ts b/services/telegram-bot/src/bots/telegram/utils/interruption.ts new file mode 100644 index 000000000..780d76f44 --- /dev/null +++ b/services/telegram-bot/src/bots/telegram/utils/interruption.ts @@ -0,0 +1,37 @@ +import type { Message as LLMMessage } from '@xsai/shared-chat' +import type { Message } from 'grammy/types' + +interface InterruptionParams { + processingTime: number + messageCount: number + currentMessages: LLMMessage[] + newMessages: Message[] + chatId: string +} + +export async function shouldInterruptProcessing(params: InterruptionParams): Promise { + // Base cases + if (params.processingTime < 1000) { + // Always allow very short processes to complete + return false + } + + if (params.processingTime > 30000) { + // Interrupt very long processes + return true + } + + // Consider message volume + const messageRatio = params.messageCount / 5 // normalize against baseline + + // Consider processing time + const timeRatio = Math.min(params.processingTime / 10000, 1) // normalize against 10s baseline + + // Calculate interruption probability + // More messages = higher chance to interrupt + // Longer processing = lower chance to interrupt + const interruptProbability = messageRatio * (1 - timeRatio) + + // Add some randomness to prevent predictable behavior + return Math.random() < interruptProbability +} diff --git a/services/telegram-bot/src/llm/actions.ts b/services/telegram-bot/src/llm/actions.ts index 6da406af6..ef533ec7b 100644 --- a/services/telegram-bot/src/llm/actions.ts +++ b/services/telegram-bot/src/llm/actions.ts @@ -4,19 +4,63 @@ import type { Action } from '../types' import { env } from 'node:process' import { useLogg } from '@guiiai/logg' +import { embed } from '@xsai/embed' import { generateText } from '@xsai/generate-text' import { message } from '@xsai/utils-chat' import { parse } from 'best-effort-json-parser' +import { chatMessageToOneLine, findLastNMessages, findRelevantMessages } from '../models' import { systemPrompt } from '../prompts/system-v1' +import { div, ul } from '../prompts/utils' -export async function imagineAnAction(unreadMessages: Record, currentAbortController: AbortController, agentMessages: LLMMessage[]) { +export async function imagineAnAction( + botId: string, + unreadMessages: Record, + currentAbortController: AbortController, + agentMessages: LLMMessage[], + lastInteractedNChatIds: string[], +) { const logger = useLogg('imagineAnAction').useGlobalConfig() if (agentMessages == null) { agentMessages = [] } + lastInteractedNChatIds.map(async (chatId) => { + const lastNMessages = await findLastNMessages(chatId, 15) + const lastNMessagesOneliner = lastNMessages.map(msg => chatMessageToOneLine(botId, msg)) + + const lastNMessagesEmbeddingPromises = lastNMessages + .map(async (msg) => { + const embeddingResult = await embed({ + baseURL: env.EMBEDDING_API_BASE_URL!, + apiKey: env.EMBEDDING_API_KEY!, + model: env.EMBEDDING_MODEL!, + input: chatMessageToOneLine(botId, msg), + abortSignal: currentAbortController.signal, + }) + + return { + embedding: embeddingResult.embedding, + message: msg, + } + }) + + const lastNMessagesEmbedding = await Promise.all(lastNMessagesEmbeddingPromises) + const relevantChatMessages = await findRelevantMessages(botId, chatId, lastNMessagesEmbedding) + const relevantChatMessagesOneliner = await Promise.all(relevantChatMessages.map(async msgs => msgs.join('\n'))) + + return message.user( + div( + 'Here are a brief list of the recent interacted chat ids\' message review', + `Last 15 messages in chat ${chatId}:`, + ul(...lastNMessagesOneliner), + `Relevant messages in chat ${chatId}:`, + ul(...relevantChatMessagesOneliner), + ), + ) + }) + agentMessages.push( message.system('' + `${systemPrompt()}` @@ -37,12 +81,12 @@ export async function imagineAnAction(unreadMessages: Record, example: { action: 'listChats' }, }, { - description: 'Send a message to a specific chat group. If you want to express anything to anyone or your friends in group, you can use this action.', - example: { action: 'sendMessage', content: '', groupId: 'id of chat to send to' }, + description: `Send a message to a specific chat group. If you want to express anything to anyone or your friends in group, you can use this action.${!!env.LLM_RESPONSE_LANGUAGE}` ? `The language of the sending message should be in ${env.LLM_RESPONSE_LANGUAGE}.` : '', + example: { action: 'sendMessage', content: '', chatId: 'id of chat to send to' }, }, { description: 'Read unread messages from a specific chat group. If you want to read the unread messages from a specific chat group, you can use this action.', - example: { action: 'readMessages', groupId: 'id of chat to send to' }, + example: { action: 'readMessages', chatId: 'id of chat to send to' }, }, { description: 'Continue the current task, which means to keep your current state unchanged, I\'ll ask you again in next tick.', @@ -83,7 +127,9 @@ export async function imagineAnAction(unreadMessages: Record, + `${Object.entries(unreadMessages).map(([key, value]) => `ID:${key}, Unread message count:${value.length}`).join('\n')}` + '', ), - message.user('What do you want to do? Respond with the action and parameters you choose in JSON only, without any explanation and markups'), + message.user('' + + 'What do you want to do? Respond with the action and parameters you choose in JSON only, without any explanation and markups', + ), ) const res = await generateText({ diff --git a/services/telegram-bot/src/llm/photo.ts b/services/telegram-bot/src/llm/photo.ts index a289747e0..2a08913ac 100644 --- a/services/telegram-bot/src/llm/photo.ts +++ b/services/telegram-bot/src/llm/photo.ts @@ -62,7 +62,7 @@ export async function interpretPhotos(state: BotSelf, msg: Message, photos: Phot input: 'Hello, world!', }) - await recordPhoto(base64, msg.sticker.file_id, files[index].file_path, res.text) + await recordPhoto(base64, msg.photo[index].file_id, files[index].file_path, res.text) state.logger.withField('photo', res.text).log('Interpreted photo') })) } diff --git a/services/telegram-bot/src/models/chat-message.ts b/services/telegram-bot/src/models/chat-message.ts index 7b6b55c08..6c15ecbd5 100644 --- a/services/telegram-bot/src/models/chat-message.ts +++ b/services/telegram-bot/src/models/chat-message.ts @@ -1,12 +1,11 @@ import type { EmbedResult } from '@xsai/embed' import type { SQL } from 'drizzle-orm' -import type { Bot } from 'grammy' import type { Message, UserFromGetMe } from 'grammy/types' import { env } from 'node:process' import { useLogg } from '@guiiai/logg' import { embed } from '@xsai/embed' -import { cosineDistance, desc, eq, sql } from 'drizzle-orm' +import { and, cosineDistance, desc, eq, gt, lt, sql } from 'drizzle-orm' import { useDrizzle } from '../db' import { chatMessagesTable } from '../db/schema' @@ -82,10 +81,10 @@ export async function findLastNMessages(chatId: string, n: number) { return res.reverse() } -export async function findRelevantMessages(bot: Bot, unreadHistoryMessagesEmbedding: { embedding: number[], message: Message }[]) { +export async function findRelevantMessages(botId: string, chatId: string, unreadHistoryMessagesEmbedding: { embedding: number[] }[]) { const db = useDrizzle() const contextWindowSize = 5 // Number of messages to include before and after - const logger = useLogg('findRelevantMessages').useGlobalConfig() + const logger = useLogg('findRelevantMessages').useGlobalConfig().withField('chatId', chatId) logger.withField('context_window_size', contextWindowSize).log('Querying relevant chat messages...') @@ -127,14 +126,18 @@ export async function findRelevantMessages(bot: Bot, unreadHistoryMessagesEmbedd combined_score: sql`${combinedScore} AS "combined_score"`, }) .from(chatMessagesTable) - .where(sql`${similarity} > '0.5' AND ${chatMessagesTable.in_chat_id} = ${embedding.message.chat.id} AND ${chatMessagesTable.platform} = 'telegram'`) + .where(and( + eq(chatMessagesTable.platform, 'telegram'), + eq(chatMessagesTable.in_chat_id, chatId), + gt(similarity, 0.5), + )) .orderBy(desc(sql`combined_score`)) .limit(3) logger.withField('number_of_relevant_messages', relevantMessages.length).log('Successfully found relevant chat messages') // Now fetch the context for each message - return await Promise.all( + const relevantMessageOneliner = await Promise.all( relevantMessages.map(async (message) => { // Get N messages before the target message const messagesBefore = await db @@ -151,7 +154,11 @@ export async function findRelevantMessages(bot: Bot, unreadHistoryMessagesEmbedd updated_at: chatMessagesTable.updated_at, }) .from(chatMessagesTable) - .where(sql`${chatMessagesTable.in_chat_id} = ${message.in_chat_id} AND ${chatMessagesTable.created_at} < ${message.created_at} AND ${chatMessagesTable.platform} = 'telegram'`) + .where(and( + eq(chatMessagesTable.platform, 'telegram'), + eq(chatMessagesTable.in_chat_id, message.in_chat_id), + lt(chatMessagesTable.created_at, message.created_at), + )) .orderBy(desc(chatMessagesTable.created_at)) .limit(contextWindowSize) @@ -170,7 +177,11 @@ export async function findRelevantMessages(bot: Bot, unreadHistoryMessagesEmbedd updated_at: chatMessagesTable.updated_at, }) .from(chatMessagesTable) - .where(sql`${chatMessagesTable.in_chat_id} = ${message.in_chat_id} AND ${chatMessagesTable.created_at} > ${message.created_at} AND ${chatMessagesTable.platform} = 'telegram'`) + .where(and( + eq(chatMessagesTable.platform, 'telegram'), + eq(chatMessagesTable.in_chat_id, message.in_chat_id), + gt(chatMessagesTable.created_at, message.created_at), + )) .orderBy(chatMessagesTable.created_at) .limit(contextWindowSize) @@ -181,11 +192,13 @@ export async function findRelevantMessages(bot: Bot, unreadHistoryMessagesEmbedd ...messagesAfter, ] - logger.withField('number_of_context_messages', contextMessages.length).log('Combined context messages') - - const contextMessagesOneliner = (await Promise.all(contextMessages.map(m => chatMessageToOneLine(bot, m)))) + const contextMessagesOneliner = (await Promise.all(contextMessages.map(m => chatMessageToOneLine(botId, m)))) return `One of the relevant message along with the context:\n${contextMessagesOneliner}` }), ) + + logger.withField('number_of_relevant_messages', relevantMessages.length).log('processed relevant chat messages with contextual messages') + + return relevantMessageOneliner })) } diff --git a/services/telegram-bot/src/models/common.ts b/services/telegram-bot/src/models/common.ts index bcdbecbb1..ae79d4ec9 100644 --- a/services/telegram-bot/src/models/common.ts +++ b/services/telegram-bot/src/models/common.ts @@ -1,14 +1,13 @@ -import type { Bot } from 'grammy' import type { Message } from 'grammy/types' import type { chatMessagesTable } from '../db/schema' import { findPhotoDescription } from './photos' import { findStickerDescription } from './stickers' -export function chatMessageToOneLine(bot: Bot, message: Omit) { +export function chatMessageToOneLine(botId: string, message: Omit) { let userDisplayName = `User [${message.from_name}]` - if (bot.botInfo.id.toString() === message.from_id) { + if (botId === message.from_id) { userDisplayName = 'Yourself' } @@ -19,13 +18,13 @@ export function chatMessageToOneLine(bot: Bot, message: Omit boolean), string][]) { + let exp + + for (let i = 0; i < args.length; i++) { + // eslint-disable-next-line no-cond-assign + if (typeof (exp = args[i][0]) === 'function' ? exp() : exp) { + return args[i][1] + } + } + + return '' +} + +export function span(...args: string[]) { + return args + .map(arg => arg.trim()) + .map(arg => arg.replaceAll(/\n\s+/g, '')) + .map(arg => arg.replaceAll(/\r\s+/g, ' ')) + .join(' ') +} + +export function div(...args: string[]) { + return args.join('\n\n') +} + +// ul + li +export function ul(...args: string[]) { + return args.map((arg) => { + return `- ${arg}` + }).join('\n') +} diff --git a/services/telegram-bot/src/types.ts b/services/telegram-bot/src/types.ts index ce683dab3..96b86a937 100644 --- a/services/telegram-bot/src/types.ts +++ b/services/telegram-bot/src/types.ts @@ -26,6 +26,8 @@ export interface BotSelf { logger: Logg processing: boolean attentionHandler: ReturnType + lastInteractedNChatIds: string[] + currentProcessingStartTime: number | null } export interface ContinueAction { @@ -95,7 +97,7 @@ export interface SearchGoogleAction { export interface ReadMessagesAction { action: 'readMessages' - groupId: string + chatId: string } export type Action = ContinueAction | BreakAction | SleepAction | LookupShortTermMemoryAction | LookupLongTermMemoryAction | MemorizeShortMemoryAction | MemorizeLongMemoryAction | ForgetShortTermMemoryAction | ForgetLongTermMemoryAction | ListChatsAction | SendMessageAction | SearchGoogleAction | ReadMessagesAction