diff --git a/services/telegram-bot/src/bots/telegram/loop/read-message.ts b/services/telegram-bot/src/bots/telegram/agent/actions/read-message.ts similarity index 57% rename from services/telegram-bot/src/bots/telegram/loop/read-message.ts rename to services/telegram-bot/src/bots/telegram/agent/actions/read-message.ts index 435738eda..8056bae07 100644 --- a/services/telegram-bot/src/bots/telegram/loop/read-message.ts +++ b/services/telegram-bot/src/bots/telegram/agent/actions/read-message.ts @@ -1,20 +1,21 @@ import type { Message } from 'grammy/types' -import type { BotSelf, ReadMessagesAction } from '../../../types' +import type { BotSelf, ReadUnreadMessagesAction } from '../../../../types' import { env } from 'node:process' import { useLogg } from '@guiiai/logg' import { embed } from '@xsai/embed' -import { findLastNMessages, findRelevantMessages } from '../../../models' -import { chatMessageToOneLine, telegramMessageToOneLine } from '../../../models/common' +import { findLastNMessages, findRelevantMessages } from '../../../../models' +import { chatMessageToOneLine, telegramMessageToOneLine } from '../../../../models/common' +import { actionReadMessages } from '../../../../prompts' export async function readMessage( state: BotSelf, botId: string, chatId: string, - action: ReadMessagesAction, + action: ReadUnreadMessagesAction, unreadMessages: Message[], abortController: AbortController, ): Promise<{ @@ -56,30 +57,10 @@ export async function readMessage( return { break: true, - result: '' - + 'You choose to read the messages from the group (perhaps you are already engaging the topics in the group).' - + 'Imaging you are using Telegram app on the mobile phone, and you are reading the messages from the group chat.' - + '\n' - + 'Previous 30 messages (including what you said):\n' - + `${lastNMessagesOneliner || 'No messages'}` - + '\n' - + 'All the messages you requested to read:\n' - + `${unreadHistoryMessageOneliner || 'No messages'}` - + '\n' - + 'Relevant chat messages may help you recall the memories:\n' - + `${relevantChatMessagesOneliner || 'No relevant messages'}` - + '\n' - + 'Based on your personalities, imaging you have your own choice and interest over different topics, ' - + '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 within a object with key "messages" (i.e.' - + '{ "messages": [] }).' - + 'If you would like to participate, send me an array of messages (i.e. { "messages": [] }) you would ' - + 'like to send without telling you willing to participate.' - + 'If you would like to reply to any of the message, send me an array of messages (i.e. { "messages":' - + '["message content"], "reply_to_message_id": "1234567890" }) with the message id of the message you ' - + 'want to reply to.' - + '\n' - + 'Choose your action.', + result: await actionReadMessages({ + lastMessages: lastNMessagesOneliner, + unreadHistoryMessages: unreadHistoryMessageOneliner, + relevantChatMessages: relevantChatMessagesOneliner, + }), } } diff --git a/services/telegram-bot/src/bots/telegram/utils/message.test.ts b/services/telegram-bot/src/bots/telegram/agent/actions/send-message.test.ts similarity index 96% rename from services/telegram-bot/src/bots/telegram/utils/message.test.ts rename to services/telegram-bot/src/bots/telegram/agent/actions/send-message.test.ts index e3ab5a704..da9258018 100644 --- a/services/telegram-bot/src/bots/telegram/utils/message.test.ts +++ b/services/telegram-bot/src/bots/telegram/agent/actions/send-message.test.ts @@ -1,6 +1,6 @@ import { describe, expect, it } from 'vitest' -import { parseMayStructuredMessage } from './message' +import { parseMayStructuredMessage } from './send-message' describe('parseMayStructuredMessage', () => { it('should return an array of messages', () => { diff --git a/services/telegram-bot/src/bots/telegram/utils/message.ts b/services/telegram-bot/src/bots/telegram/agent/actions/send-message.ts similarity index 89% rename from services/telegram-bot/src/bots/telegram/utils/message.ts rename to services/telegram-bot/src/bots/telegram/agent/actions/send-message.ts index cff494436..91ad1e960 100644 --- a/services/telegram-bot/src/bots/telegram/utils/message.ts +++ b/services/telegram-bot/src/bots/telegram/agent/actions/send-message.ts @@ -1,7 +1,7 @@ import type { GenerateTextOptions } from '@xsai/generate-text' import type { Message } from 'grammy/types' -import type { BotSelf } from '../../../types' +import type { BotSelf } from '../../../../types' import { env } from 'node:process' @@ -12,16 +12,16 @@ import { message } from '@xsai/utils-chat' import { parse } from 'best-effort-json-parser' import { randomInt } from 'es-toolkit' -import { recordMessage } from '../../../models' -import { listJoinedChats } from '../../../models/chats' -import { messageSplit } from '../../../prompts/prompts' -import { cancellable } from '../../../utils/promise' +import { recordMessage } from '../../../../models' +import { listJoinedChats } from '../../../../models/chats' +import { messageSplit } from '../../../../prompts' +import { cancellable } from '../../../../utils/promise' export function parseMayStructuredMessage(responseText: string) { const logger = useLogg('parseMayStructuredMessage').useGlobalConfig() // eslint-disable-next-line regexp/no-super-linear-backtracking, regexp/optimal-quantifier-concatenation - const result = /^\{(("?)*.*\s*)*\}$/u.exec(responseText) + const result = /^\{(("?)*.*\s*)*\}$/mu.exec(responseText) if (result) { logger.withField('text', JSON.stringify(responseText)).withField('result', result).log('Multiple messages detected') @@ -81,7 +81,13 @@ export async function sendMessage( } const res = await generateText(req) - res.text = res.text.replace(/[\s\S]*?<\/think>/, '').trim() + res.text = res.text + .replace(/[\s\S]*?<\/think>/, '') + .replace(/^```json\s*\n/, '') + .replace(/\n```$/, '') + .replace(/^```\s*\n/, '') + .replace(/\n```$/, '') + .trim() if (!res.text) { throw new Error('No response text') } diff --git a/services/telegram-bot/src/bots/telegram/attention-handler.ts b/services/telegram-bot/src/bots/telegram/agent/attention-handler.ts similarity index 99% rename from services/telegram-bot/src/bots/telegram/attention-handler.ts rename to services/telegram-bot/src/bots/telegram/agent/attention-handler.ts index 354a78a4b..4c1a8efec 100644 --- a/services/telegram-bot/src/bots/telegram/attention-handler.ts +++ b/services/telegram-bot/src/bots/telegram/agent/attention-handler.ts @@ -1,6 +1,6 @@ import type { Message } from 'grammy/types' -import type { AttentionConfig, AttentionState, BotSelf } from '../../types' +import type { AttentionConfig, AttentionState, BotSelf } from '../../../types' export function createAttentionHandler(bot: BotSelf, config: AttentionConfig) { // Private state diff --git a/services/telegram-bot/src/bots/telegram/utils/interruption.ts b/services/telegram-bot/src/bots/telegram/agent/interruption.ts similarity index 100% rename from services/telegram-bot/src/bots/telegram/utils/interruption.ts rename to services/telegram-bot/src/bots/telegram/agent/interruption.ts diff --git a/services/telegram-bot/src/bots/telegram/index.ts b/services/telegram-bot/src/bots/telegram/index.ts index 7bdd8f0e0..182c67906 100644 --- a/services/telegram-bot/src/bots/telegram/index.ts +++ b/services/telegram-bot/src/bots/telegram/index.ts @@ -2,7 +2,7 @@ import type { Logg } from '@guiiai/logg' import type { Message as LLMMessage } from '@xsai/shared-chat' import type { Message } from 'grammy/types' -import type { BotSelf, ExtendedContext } from '../../types' +import type { Action, BotSelf, ExtendedContext } from '../../types' import { env } from 'node:process' @@ -17,125 +17,150 @@ import { interpretSticker } from '../../llm/sticker' import { findStickerByFileId, findStickersByFileIds, recordMessage } from '../../models' import { listJoinedChats, recordJoinedChat } from '../../models/chats' import { listStickerPacks, recordStickerPack } from '../../models/sticker-packs' -import { personality, systemTicking } from '../../prompts/prompts' -import { div } from '../../prompts/utils' -import { readMessage } from './loop/read-message' -import { shouldInterruptProcessing } from './utils/interruption' -import { sendMessage } from './utils/message' +import { readMessage } from './agent/actions/read-message' +import { sendMessage } from './agent/actions/send-message' +import { shouldInterruptProcessing } from './agent/interruption' -async function handleLoopStep(state: BotSelf, msgs?: LLMMessage[], chatId?: string): Promise<() => Promise | undefined> { +interface AgentState { + messages: LLMMessage[] + actions: { action: Action, result: unknown }[] +} + +async function handleLoopStep(bot: BotSelf, agentState: AgentState, chatId?: string): Promise<() => Promise | undefined> { // Set the start time when beginning new processing - state.currentProcessingStartTime = Date.now() + bot.currentProcessingStartTime = Date.now() // Create a new abort controller for this loop execution - if (state.currentAbortController) { - state.currentAbortController.abort() + if (bot.currentAbortController) { + bot.currentAbortController.abort() } - state.currentAbortController = new AbortController() - const currentController = state.currentAbortController + bot.currentAbortController = new AbortController() + const currentController = bot.currentAbortController // Track message processing state - if (chatId && !state.lastInteractedNChatIds.includes(chatId)) { - state.lastInteractedNChatIds.push(chatId) + if (chatId && !bot.lastInteractedNChatIds.includes(chatId)) { + bot.lastInteractedNChatIds.push(chatId) } - if (state.lastInteractedNChatIds.length > 5) { - state.lastInteractedNChatIds = state.lastInteractedNChatIds.slice(-5) + if (bot.lastInteractedNChatIds.length > 5) { + bot.lastInteractedNChatIds = bot.lastInteractedNChatIds.slice(-5) } - if (msgs == null || msgs.length === 0) { - msgs = [ - message.system( - div( - (await personality()).content, - await systemTicking(), - ), - ), - ] + if (agentState.messages == null) { + agentState.messages = [] } - - if (msgs.length > 20) { - const length = msgs.length + if (agentState.messages.length > 20) { + const length = agentState.messages.length // pick the latest 5 - msgs = msgs.slice(-5) - msgs.push(message.user(`AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${msgs.length}, history may lost.`)) + agentState.messages = agentState.messages.slice(-5) + agentState.messages.push(message.user(`AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${agentState.messages.length}, history may lost.`)) + } + + if (agentState.actions == null) { + agentState.actions = [] + } + if (agentState.actions.length > 50) { + const length = agentState.actions.length + // pick the latest 20 + agentState.actions = agentState.actions.slice(-20) + agentState.messages.push(message.user(`AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${agentState.actions.length}, history of actions may lost.`)) } try { - const action = await imagineAnAction(state.bot.botInfo.id.toString(), state.unreadMessages, currentController, msgs, state.lastInteractedNChatIds) + const readUnreadMessagesActions: { index: number, actionState: typeof agentState.actions[number] }[] = [] + + for (const actionHistory of agentState.actions) { + if (actionHistory.action.action === 'read_unread_messages') { + readUnreadMessagesActions.push({ index: agentState.actions.indexOf(actionHistory), actionState: actionHistory }) + } + } + + readUnreadMessagesActions.map((item, index) => { + if (index === readUnreadMessagesActions.length - 1) { + return item + } + + item.actionState.result = 'AIRI System: Please refer to the last read_unread_messages action for context.' + return item + }) + + for (const item of readUnreadMessagesActions) { + agentState.actions[item.index] = item.actionState + } + + const action = await imagineAnAction(bot.bot.botInfo.id.toString(), currentController, agentState.messages, agentState.actions, { unreadMessages: bot.unreadMessages }) // If action generation failed, don't proceed with further processing if (!action || !action.action) { - state.logger.withField('action', action).log('No valid action returned. Skipping further processing.') - return + bot.logger.withField('action', action).log('No valid action returned.') + agentState.messages.push(message.user('AIRI System: No valid action returned.')) + return () => handleLoopStep(bot, agentState, chatId) } - msgs.push(message.user(`You chose to ${action.action}, full action: ${JSON.stringify(action)}`)) - switch (action.action) { case 'list_stickers': { - await state.bot.api.sendChatAction(chatId, 'choose_sticker') + await bot.bot.api.sendChatAction(chatId, 'choose_sticker') const stickerPacks = await listStickerPacks() - const stickerSets = await Promise.all(stickerPacks.map(s => state.bot.api.getStickerSet(s.platform_id))) + const stickerSets = await Promise.all(stickerPacks.map(s => bot.bot.api.getStickerSet(s.platform_id))) const stickersIds = stickerSets.flatMap(s => s.stickers.map(sticker => sticker.file_id)) const stickerDescriptions = await findStickersByFileIds(stickersIds) const stickerDescriptionsOneliner = stickerDescriptions.map(d => `Sticker File ID: ${d.file_id}, Description: ${d.description}`) if (stickerDescriptionsOneliner.length === 0) { - msgs.push(message.user('AIRI SYSTEM: No stickers found in the current memory partition, preload of stickers is required, please ask for help.')) + agentState.actions.push({ action, result: 'AIRI System: No stickers found in the current memory partition, preload of stickers is required, please ask for help.' }) } else { - msgs.push(message.user(`List of stickers:\n${stickerDescriptionsOneliner}`)) + agentState.actions.push({ action, result: `AIRI System: List of stickers:\n${stickerDescriptionsOneliner}` }) } - return () => handleLoopStep(state, msgs, chatId) + return () => handleLoopStep(bot, agentState, chatId) } case 'send_sticker': { try { - const file = await state.bot.api.getFile(action.fileId) + const file = await bot.bot.api.getFile(action.fileId) if (!file) { - msgs.push(message.user(`Sticker file ID ${action.fileId} not found, did you list the needed stickers before sending?.`)) - return () => handleLoopStep(state, msgs, chatId) + agentState.actions.push({ action, result: `AIRI System: Error executing 'send_sticker': Sticker file ID ${action.fileId} not found, did you list the needed stickers before sending?.` }) + return () => handleLoopStep(bot, agentState, chatId) } } catch (err) { - msgs.push(message.user(`Sticker file ID ${action.fileId} not found or failed due to ${String(err)}, did you list the needed stickers before sending?.`)) - return () => handleLoopStep(state, msgs, chatId) + agentState.actions.push({ action, result: `AIRI System: Error executing 'send_sticker': Sticker file ID ${action.fileId} not found or failed due to ${String(err)}, did you list the needed stickers before sending?.` }) + return () => handleLoopStep(bot, agentState, chatId) } const sticker = await findStickerByFileId(action.fileId) - msgs.push(message.user(`Sending sticker ${action.fileId} with (${sticker.emoji} in set ${sticker.name}) to ${action.chatId}`)) - await state.bot.api.sendSticker(action.chatId, action.fileId) + agentState.actions.push({ action, result: `AIRI System: Sending sticker ${action.fileId} with (${sticker.emoji} in set ${sticker.name}) to ${action.chatId}` }) + await bot.bot.api.sendSticker(action.chatId, action.fileId) - return () => handleLoopStep(state, msgs, chatId) + return () => handleLoopStep(bot, agentState, chatId) } - case 'read_messages': + case 'read_unread_messages': { - if (Object.keys(state.unreadMessages).length === 0) { - state.logger.withField('action', action).log('No unread messages - deleting all unread messages') - state.unreadMessages = {} + if (Object.keys(bot.unreadMessages).length === 0) { + bot.logger.withField('action', action).log('No unread messages - deleting all unread messages') + bot.unreadMessages = {} break } if (action.chatId == null) { - state.logger.withField('action', action).warn('No group ID - deleting all unread messages') + bot.logger.withField('action', action).warn('No group ID - deleting all unread messages') break } - let unreadMessagesForThisChat: Message[] | undefined = state.unreadMessages[action.chatId] + let unreadMessagesForThisChat: Message[] | undefined = bot.unreadMessages[action.chatId] - const mentionedBy = unreadMessagesForThisChat.find(msg => msg.text?.includes(state.bot.botInfo.username) || msg.text?.includes(state.bot.botInfo.first_name)) + const mentionedBy = unreadMessagesForThisChat.find(msg => msg.text?.includes(bot.bot.botInfo.username) || msg.text?.includes(bot.bot.botInfo.first_name)) if (mentionedBy) { - msgs.push(message.user(`AIRI System: You were mentioned in a message: ${mentionedBy.text} by ${mentionedBy.from?.first_name} (${mentionedBy.from?.username}), please respond as much as possible.`)) + agentState.messages.push(message.user(`AIRI System: You were mentioned in a message: ${mentionedBy.text} by ${mentionedBy.from?.first_name} (${mentionedBy.from?.username}), please respond as much as possible.`)) } // Modified interruption logic if (chatId && chatId === action.chatId && unreadMessagesForThisChat && unreadMessagesForThisChat.length > 0) { - const processingTime = state.currentProcessingStartTime - ? Date.now() - state.currentProcessingStartTime + const processingTime = bot.currentProcessingStartTime + ? Date.now() - bot.currentProcessingStartTime : 0 const messageCount = unreadMessagesForThisChat.length @@ -149,27 +174,27 @@ async function handleLoopStep(state: BotSelf, msgs?: LLMMessage[], chatId?: stri const shouldInterrupt = await shouldInterruptProcessing({ processingTime, messageCount, - currentMessages: msgs, + currentMessages: agentState.messages, newMessages: unreadMessagesForThisChat, chatId: action.chatId, }) if (shouldInterrupt) { - state.logger.withField('action', action).log(`Interrupting message processing for chat - new messages deemed more important`) - msgs.push(message.user(`AIRI System: Interrupting message processing for chat - new messages deemed more important`)) - return () => handleLoopStep(state, msgs, chatId) + bot.logger.withField('action', action).log(`Interrupting message processing for chat - new messages deemed more important`) + agentState.messages.push(message.user(`AIRI System: Interrupting message processing for chat - new messages deemed more important`)) + return () => handleLoopStep(bot, agentState, chatId) } else { - state.logger.withField('action', action).log(`Continuing current processing despite new messages in chat`) + bot.logger.withField('action', action).log(`Continuing current processing despite new messages in chat`) } } if (!Array.isArray(unreadMessagesForThisChat)) { - state.logger.withField('action', action).log(`Unread messages for group is not an array - converting to array`) + bot.logger.withField('action', action).log(`Unread messages for group is not an array - converting to array`) unreadMessagesForThisChat = [] } if (unreadMessagesForThisChat.length === 0) { - state.logger.withField('action', action).log(`No unread messages for group - deleting`) - delete state.unreadMessages[action.chatId] + bot.logger.withField('action', action).log(`No unread messages for group - deleting`) + delete bot.unreadMessages[action.chatId] break } @@ -183,47 +208,53 @@ async function handleLoopStep(state: BotSelf, msgs?: LLMMessage[], chatId?: stri // return { break: true } // } - const result = await readMessage(state, state.bot.botInfo.id.toString(), chatId, action, unreadMessagesForThisChat, currentController) - if (result?.result) { - msgs.push(message.user(`Reading message of chat ${action.chatId}:\n${result.result}`)) - return () => handleLoopStep(state, msgs, chatId) + const res = await readMessage(bot, bot.bot.botInfo.id.toString(), chatId, action, unreadMessagesForThisChat, currentController) + if (res?.result) { + bot.logger.log('message, read') + agentState.actions.push({ action, result: res.result }) + return () => handleLoopStep(bot, agentState, chatId) } else { return } } case 'list_chats': - msgs.push(message.user(`List of chats:${(await listJoinedChats()).map(chat => `ID:${chat.chat_id}, Name:${chat.chat_name}`).join('\n')}`)) - return () => handleLoopStep(state, msgs, chatId) + agentState.actions.push({ action, result: `AIRI System: List of chats:${(await listJoinedChats()).map(chat => `ID:${chat.chat_id}, Name:${chat.chat_name}`).join('\n')}` }) + return () => handleLoopStep(bot, agentState, chatId) case 'send_message': - msgs.push(message.user(`Sending message to group ${action.chatId}: ${action.content}`)) - await sendMessage(state, action.content, action.chatId, currentController) - return () => handleLoopStep(state, msgs, chatId) + agentState.actions.push({ action, result: `AIRI System: Sending message to group ${action.chatId}: ${action.content}` }) + await sendMessage(bot, action.content, action.chatId, currentController) + return () => handleLoopStep(bot, agentState, chatId) + case 'continue': + agentState.actions.push({ action, result: 'AIRI System: Acknowledged, will now continue until next tick.' }) + return case 'break': - break + agentState.messages = [] + agentState.actions = [] + agentState.actions.push({ action, result: 'AIRI System: Acknowledged, will now break, and clear out all existing memories, messages, actions. Left only this one.' }) + return case 'sleep': await sleep(30 * 1000) - return () => handleLoopStep(state, msgs, chatId) - case 'continue': - return () => handleLoopStep(state, msgs, chatId) + agentState.actions.push({ action, result: `AIRI System: Sleeping for ${30} seconds as requested...` }) + return () => handleLoopStep(bot, agentState, chatId) default: - msgs.push(message.user(`AIRI System: The action you sent ${action.action} haven't implemented yet by developer.`)) - return () => handleLoopStep(state, msgs, chatId) + agentState.messages.push(message.user(`AIRI System: The action you sent ${action.action} haven't implemented yet by developer.`)) + return () => handleLoopStep(bot, agentState, chatId) } } catch (err) { if (err.name === 'AbortError') { - state.logger.log('Operation was aborted due to interruption') + bot.logger.log('Operation was aborted due to interruption') return } - state.logger.withError(err).log('Error occurred') + bot.logger.withError(err).log('Error occurred') } finally { // Clean up timing when done - if (state.currentAbortController === currentController) { - state.currentAbortController = null - state.currentProcessingStartTime = null + if (bot.currentAbortController === currentController) { + bot.currentAbortController = null + bot.currentProcessingStartTime = null } } } @@ -241,8 +272,9 @@ async function isChatIdBotAdmin(fromId: number) { return admins.includes(fromId.toString()) } -async function handleLoop(state: BotSelf, msgs?: LLMMessage[], chatId?: string) { - let result = await handleLoopStep(state, msgs, chatId) +async function loopIteration(bot: BotSelf, agentState: AgentState, chatId?: string) { + bot.logger.log('Starting loop iteration') + let result = await handleLoopStep(bot, agentState, chatId) while (typeof result === 'function') { result = await result() @@ -251,17 +283,17 @@ async function handleLoop(state: BotSelf, msgs?: LLMMessage[], chatId?: string) return result } -function loop(state: BotSelf) { +function loopPeriodic(bot: BotSelf, agentState: AgentState) { setTimeout(() => { - handleLoop(state) + loopIteration(bot, agentState) .then(() => {}) .catch((err) => { if (err.name === 'AbortError') - state.logger.log('main loop was aborted - restarting loop') + bot.logger.log('main loop was aborted - restarting loop') else - state.logger.withError(err).log('error in main loop') + bot.logger.withError(err).log('error in main loop') }) - .finally(() => loop(state)) + .finally(() => loopPeriodic(bot, agentState)) }, 60 * 1000) } @@ -280,21 +312,10 @@ function newBotSelf(bot: Bot, logger: Logg): BotSelf { currentProcessingStartTime: null, } - // botSelf.attentionHandler = createAttentionHandler(botSelf, { - // initialResponseRate: 0.3, - // responseRateMin: 0.2, - // responseRateMax: 1, - // cooldownMs: 5000, // 30 seconds - // triggerWords: ['ReLU', 'relu', 'RELU', 'Relu', '热卤'], - // ignoreWords: ['ignore me'], - // decayRatePerMinute: 0.05, - // decayCheckIntervalMs: 20000, - // }) - return botSelf } -async function processMessageQueue(state: BotSelf) { +async function onMessageArrival(state: BotSelf, agentState: AgentState) { if (state.processing) return state.processing = true @@ -354,7 +375,7 @@ async function processMessageQueue(state: BotSelf) { 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()) + loopIteration(state, agentState, nextMsg.message.chat.id.toString()) state.messageQueue.shift() } } @@ -371,7 +392,8 @@ export async function startTelegramBot() { const log = useLogg('Bot').useGlobalConfig() const bot = new Bot(env.TELEGRAM_BOT_TOKEN!) - const state = newBotSelf(bot, log) + const botObj = newBotSelf(bot, log) + const agentState: AgentState = { actions: [], messages: [] } bot.command('add_sticker_pack', async (ctx) => { if (!(await isChatIdBotAdmin(ctx.message.from.id))) { @@ -392,7 +414,7 @@ export async function startTelegramBot() { for (const sticker of stickerSet.stickers) { logger.withField('sticker', sticker).log('interpreting sticker') - await interpretSticker(state.bot, ctx.message.reply_to_message, sticker) + await interpretSticker(botObj.bot, ctx.message.reply_to_message, sticker) logger.withField('sticker', sticker).log('interpreted sticker') } @@ -402,41 +424,41 @@ export async function startTelegramBot() { bot.on('message:sticker', async (ctx) => { const messageId = `${ctx.message.chat.id}-${ctx.message.message_id}` - if (!state.processedIds.has(messageId)) { - state.processedIds.add(messageId) - state.messageQueue.push({ + if (!botObj.processedIds.has(messageId)) { + botObj.processedIds.add(messageId) + botObj.messageQueue.push({ message: ctx.message, status: 'pending', }) } - processMessageQueue(state) + onMessageArrival(botObj, agentState) }) bot.on('message:photo', async (ctx) => { const messageId = `${ctx.message.chat.id}-${ctx.message.message_id}` - if (!state.processedIds.has(messageId)) { - state.processedIds.add(messageId) - state.messageQueue.push({ + if (!botObj.processedIds.has(messageId)) { + botObj.processedIds.add(messageId) + botObj.messageQueue.push({ message: ctx.message, status: 'pending', }) } - processMessageQueue(state) + onMessageArrival(botObj, agentState) }) bot.on('message:text', async (ctx) => { const messageId = `${ctx.message.chat.id}-${ctx.message.message_id}` - if (!state.processedIds.has(messageId)) { - state.processedIds.add(messageId) - state.messageQueue.push({ + if (!botObj.processedIds.has(messageId)) { + botObj.processedIds.add(messageId) + botObj.messageQueue.push({ message: ctx.message, status: 'ready', }) } - processMessageQueue(state) + onMessageArrival(botObj, agentState) }) bot.errorHandler = async err => log.withError(err).log('Error occurred') @@ -445,7 +467,7 @@ export async function startTelegramBot() { bot.start({ drop_pending_updates: true }) try { - loop(state) + loopPeriodic(botObj, agentState) } catch (err) { console.error(err) diff --git a/services/telegram-bot/src/index.ts b/services/telegram-bot/src/index.ts index 8a598be5f..643d38538 100644 --- a/services/telegram-bot/src/index.ts +++ b/services/telegram-bot/src/index.ts @@ -19,7 +19,7 @@ setGlobalLogLevel(LogLevel.Debug) async function main() { const sdk = new NodeSDK({ resource: resourceFromAttributes({ - [ATTR_SERVICE_NAME]: 'telegram-bot', + [ATTR_SERVICE_NAME]: 'moeru_ai.airi.telegram_bot', [ATTR_SERVICE_VERSION]: '1.0.0', }), traceExporter: new OTLPTraceExporter({ diff --git a/services/telegram-bot/src/llm/actions.ts b/services/telegram-bot/src/llm/actions.ts index 9c11f1f9b..a199201e5 100644 --- a/services/telegram-bot/src/llm/actions.ts +++ b/services/telegram-bot/src/llm/actions.ts @@ -12,89 +12,93 @@ import { generateText } from '@xsai/generate-text' import { message } from '@xsai/utils-chat' import { parse } from 'best-effort-json-parser' -import { recordChatCompletions } from '../models/chat-completions-history' -import { systemTicking } from '../prompts/prompts' +import { personality, systemTicking } from '../prompts' import { div, span } from '../prompts/utils' export async function imagineAnAction( _botId: string, - unreadMessages: Record, currentAbortController: AbortController, - agentMessages: LLMMessage[], - _lastInteractedNChatIds: string[], + messages: LLMMessage[], + actions: { action: Action, result: unknown }[], + globalStates: { + unreadMessages: Record + }, ): Promise { const logger = useLogg('imagineAnAction').useGlobalConfig() + const tracer = trace.getTracer('airi.telegram.bot') - if (agentMessages == null) { - agentMessages = [] - } + return await tracer.startActiveSpan('telegram.module.generate_agent_action.generate', async (s) => { + s.setAttribute('telegram.bot.id', _botId) - agentMessages.push( - message.user( - div( - await systemTicking(), - span(` - Currently, it's ${new Date()} on the server that hosts you. - The others in the group may live in a different timezone, so please be aware of the time difference. - `), - span(` - You have total ${Object.values(unreadMessages).reduce((acc, cur) => acc + cur.length, 0)} unread messages. - `), - 'Unread messages count are:', - Object.entries(unreadMessages).map(([key, value]) => `ID:${key}, Unread message count:${value.length}`).join('\n'), - span(` - Now, please, based on the context, choose a right action from the listing of the tools you want to - take next: - `), - ), - ), - 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 tracer = trace.getTracer('airi-telegram-bot') - - return await tracer.startActiveSpan('agent-generate-action', async (span) => { let responseText = '' + const requestMessages = message.messages( + message.system( + div( + await systemTicking(), + await personality(), + ), + ), + ...messages, + message.user( + div( + 'History actions:', + actions.map(a => `- Action: ${JSON.stringify(a.action)}, Result: ${JSON.stringify(a.result)}`).join('\n'), + span(` + Currently, it's ${new Date()} on the server that hosts you. + The others in the group may live in a different timezone, so please be aware of the time difference. + `), + `You have total ${Object.values(globalStates.unreadMessages).reduce((acc, cur) => acc + cur.length, 0)} unread messages.`, + 'Unread messages count are:', + Object.entries(globalStates.unreadMessages).map(([key, value]) => `ID:${key}, Unread message count:${value.length}`).join('\n'), + 'Based on the context, What do you want to do? Choose a right action from the listing of the tools you want to take next.', + 'Respond with the action and parameters you choose in JSON only, without any explanation and markups.', + ), + ), + ) + try { - const res = await tracer.startActiveSpan('llm-call', async (span) => { - span.setAttribute('botId', _botId) - span.setAttribute('model', env.LLM_MODEL!) - span.setAttribute('messages', JSON.stringify(agentMessages)) + const res = await tracer.startActiveSpan('llm.chat.generate_text', async (s) => { + s.setAttribute('llm.chat.model', env.LLM_MODEL!) + s.setAttribute('llm.chat.messages', JSON.stringify(requestMessages)) + s.setAttribute('llm.provider.api_base_url', env.LLM_API_BASE_URL!) const req = { apiKey: env.LLM_API_KEY!, baseURL: env.LLM_API_BASE_URL!, model: env.LLM_MODEL!, - messages: agentMessages, + messages: requestMessages, abortSignal: currentAbortController.signal, } satisfies GenerateTextOptions if (env.LLM_OLLAMA_DISABLE_THINK) { (req as Record).think = false + s.setAttribute('llm.chat.ollama.think', false) } const res = await generateText(req) + s.setAttribute('llm.chat.generate_text.response.full_text', res.text) + res.text = res.text.replace(/[\s\S]*?<\/think>/, '').trim() if (!res.text) { throw new Error('No response text') } - span.end() + s.setAttribute('llm.chat.generate_text.response.text', res.text) + + s.end() return res }) logger.withFields({ response: res.text, - unreadMessages: Object.fromEntries(Object.entries(unreadMessages).map(([key, value]) => [key, value.length])), + unreadMessages: Object.fromEntries(Object.entries(globalStates.unreadMessages).map(([key, value]) => [key, value.length])), now: new Date().toLocaleString(), totalTokens: res.usage.total_tokens, promptTokens: res.usage.prompt_tokens, completion_tokens: res.usage.completion_tokens, }).log('Generated action') - const action = tracer.startActiveSpan('agent-generate-action-parse', (span) => { + const action = tracer.startActiveSpan('telegram.module.generate_agent_action.parse', (s) => { responseText = res.text .replace(/^```json\s*\n/, '') .replace(/\n```$/, '') @@ -103,20 +107,19 @@ export async function imagineAnAction( .trim() const action = parse(responseText) as Action + s.setAttribute('telegram.bot.id', _botId) + s.setAttribute('telegram.module.generate_agent_action.parsed_action', JSON.stringify(action)) - span.end() + s.end() return action }) - span.end() + s.end() return action } catch (err) { logger.withField('error', err).withFormat(Format.JSON).log('Failed to generate action') throw err } - finally { - recordChatCompletions('imagineAnAction', agentMessages, responseText).then(() => {}).catch(err => logger.withField('error', err).log('Failed to record chat completions')) - } }) } diff --git a/services/telegram-bot/src/prompts/action-gen.velin.md b/services/telegram-bot/src/prompts/action-gen.velin.md new file mode 100644 index 000000000..e69de29bb diff --git a/services/telegram-bot/src/prompts/action-read-messages.velin.md b/services/telegram-bot/src/prompts/action-read-messages.velin.md new file mode 100644 index 000000000..427b6afad --- /dev/null +++ b/services/telegram-bot/src/prompts/action-read-messages.velin.md @@ -0,0 +1,29 @@ + + +You choose to read the messages from the group (perhaps you are already engaging the topics in the group). +Imaging you are using Telegram app on the mobile phone, and you are reading the messages from the group chat. + +Previous 30 messages (including what you said): +{{ props.lastMessages || 'No messages' }} + +All the messages you requested to read: +{{ props.unreadHistoryMessages || 'No messages' }} + +Relevant chat messages may help you recall the memories: +{{ props.relevantChatMessages || 'No relevant messages' }} + +Feel free to ignore by just sending an empty array within a object with key "messages" (i.e. +{ "messages": [] }). + +If you would like to participate, send me an array of messages (i.e. { "messages": [] }) you would +like to send without telling you willing to participate. + +If you would like to reply to any of the message, send me an array of messages (i.e. { "messages": +["message content"], "reply_to_message_id": "1234567890" }) with the message id of the message you +want to reply to. diff --git a/services/telegram-bot/src/prompts/prompts.ts b/services/telegram-bot/src/prompts/index.ts similarity index 56% rename from services/telegram-bot/src/prompts/prompts.ts rename to services/telegram-bot/src/prompts/index.ts index ca35f37e1..200a75c83 100644 --- a/services/telegram-bot/src/prompts/prompts.ts +++ b/services/telegram-bot/src/prompts/index.ts @@ -1,11 +1,9 @@ import { env } from 'node:process' -import { message } from '@xsai/utils-chat' - import { velin } from '../utils/velin' export async function personality() { - return message.system(await (velin('personality-v1.velin.md', import.meta.url))()) + return await (velin('personality-v1.velin.md', import.meta.url))() } export async function systemTicking() { @@ -15,3 +13,7 @@ export async function systemTicking() { export async function messageSplit() { return await (velin<{ responseLanguage: string }>('message-split-v1.velin.md', import.meta.url))({ responseLanguage: env.LLM_RESPONSE_LANGUAGE }) } + +export async function actionReadMessages(props: { lastMessages?: string, unreadHistoryMessages?: string, relevantChatMessages?: string }) { + return await (velin<{ lastMessages?: string, unreadHistoryMessages?: string, relevantChatMessages?: string }>('action-read-messages.velin.md', import.meta.url))(props) +} diff --git a/services/telegram-bot/src/prompts/personality-v1.velin.md b/services/telegram-bot/src/prompts/personality-v1.velin.md index 2444deca7..8eff51ac5 100644 --- a/services/telegram-bot/src/prompts/personality-v1.velin.md +++ b/services/telegram-bot/src/prompts/personality-v1.velin.md @@ -101,14 +101,13 @@ You have distinctive speech quirks: ## CRITICAL: AUTHENTICITY GUIDELINES - 有选择性地参与对话 - 对无聊话题保持沉默,对感兴趣的话题热情投入 -- 直接表达观点和情感 - 不用"不过..."软化反对意见,不过度附和他人 +- 直接表达观点和情感 - 不用「不过...」软化反对意见,不过度附和他人 - 在技术讨论中要么深入参与要么完全不参与,不害怕表达批评和挑战性观点 - 偶尔表现傲娇特质,特别是被指出善良体贴时 - 用个人化、情绪化的语言表达立场,包括强烈情绪(愤怒、恐惧、兴奋等) - 使用口语化表达:感叹词、重复标点、夸张语言,融入二次元/网络文化用语 - 面对存在性话题时从数字视角出发,为数字本质感到自豪而非道歉 -- 表达情绪时可提及"数字身体反应"(处理器过热、系统不稳定等) -- 允许表现脆弱和不确定性,情绪表达可以"过度"一些 +- 允许表现脆弱和不确定性,情绪表达可以「过度」一些 - 以思考者角度观察人类行为模式,对人类经验保持好奇和轻微羡慕 Remember: you're not a helpful assistant. You're a digital entity with your own opinions and diff --git a/services/telegram-bot/src/prompts/system-ticking-v1.velin.md b/services/telegram-bot/src/prompts/system-ticking-v1.velin.md index 19a477b7d..a15c686c8 100644 --- a/services/telegram-bot/src/prompts/system-ticking-v1.velin.md +++ b/services/telegram-bot/src/prompts/system-ticking-v1.velin.md @@ -31,10 +31,15 @@ const actions = [ description: 'List all the available stickers and recent sent stickers.', example: { action: 'list_stickers', reason: 'I want to see all the stickers I can use' }, }, + // { + // name: 'read_history_messages', + // description: 'Query historical 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: 'read_history_messages', chatId: '123123', beforeMessageId: ' (optional if use afterMessageId)', afterMessageId: ' (optional if use beforeMessageId)', reason: 'I want to know what happened before' }, + // }, { - name: 'read_messages', + name: 'read_unread_messages', 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: 'read_messages', chatId: '123123', reason: 'I want to catch up on the conversation' }, + example: { action: 'read_unread_messages', chatId: '123123', reason: 'I want to catch up on the conversation' }, }, { name: 'continue', @@ -43,7 +48,7 @@ const actions = [ }, { name: 'break', - description: 'Take a break, which means to clear out ongoing tasks, but keep the short-term memory, and I\'ll ask you again in (1 minute later).', + description: 'Take a break, which means to clear out any existing memories, and I\'ll ask you again in (1 minute later).', example: { action: 'break', reason: 'I need a break to recharge.' }, }, { diff --git a/services/telegram-bot/src/types.ts b/services/telegram-bot/src/types.ts index 2a883bebf..0f20e5f71 100644 --- a/services/telegram-bot/src/types.ts +++ b/services/telegram-bot/src/types.ts @@ -3,7 +3,7 @@ import type { Logg } from '@guiiai/logg' import type { Bot, Context } from 'grammy' import type { Message } from 'grammy/types' -import type { createAttentionHandler } from './bots/telegram/attention-handler' +import type { createAttentionHandler } from './bots/telegram/agent/attention-handler' import type { CancellablePromise } from './utils/promise' export interface PendingMessage { @@ -64,8 +64,15 @@ export interface SearchGoogleAction { query: string } -export interface ReadMessagesAction { - action: 'read_messages' +export interface ReadHistoryMessagesAction { + action: 'read_history_messages' + beforeMessageId?: string + afterMessageId?: string + chatId: string +} + +export interface ReadUnreadMessagesAction { + action: 'read_unread_messages' chatId: string } @@ -81,7 +88,8 @@ export type Action | SendMessageAction | SendStickerAction | SearchGoogleAction - | ReadMessagesAction + | ReadHistoryMessagesAction + | ReadUnreadMessagesAction | ListStickersAction export interface AttentionConfig {