diff --git a/services/satori-bot/README.md b/services/satori-bot/README.md index b3de8d4d6..a29d54a6e 100644 --- a/services/satori-bot/README.md +++ b/services/satori-bot/README.md @@ -1,120 +1,76 @@ # AIRI Satori Bot -一个基于 Satori 协议的 AI 聊天机器人，可以通过 Koishi 连接到多个聊天平台（QQ、Telegram、Discord、飞书等）。 +> **⚠️ Disclaimer**: This is a submodule of **AIRI**. The `core` part of this satori bot is merely a **temporary solution**. We will eventually delete it and integrate with **AIRI's Core** once the main framework is stable. -## 架构说明 +A **STANDALONE**, event-driven AI agent built on the [Satori Protocol](https://satori.chat/). It connects to multiple chat platforms (QQ, Telegram, Discord, Lark) via a Koishi bridge, featuring an autonomous thought loop. -本项目采用**独立架构**，参考了 Telegram Bot 的实现模式 +## 🏗 Architecture & Internals (Provisional) -## 前置要求 +**Important**: This module currently implements a self-contained "Mini-Core" (`src/core/`) to operate independently. This is **NOT** the final architecture of AIRI. -1. **Koishi 实例**：需要一个运行中的 Koishi 实例，并启用 Satori 服务 -2. **LLM API**：OpenAI API 或兼容的 API（如 Ollama、vLLM 等） -3. **Node.js**: >= 18.0.0 -4. **pnpm**: >= 8.0.0 +* **Temporary Logic**: The Event Loop, Scheduler, and Planner logic located in `src/core/` are placeholders. They simulate the behavior of the future AIRI Core. +* **Retained Components**: The **Dispatcher** and **Database** will be retained. They will be exposed as **tool-like modules** to the AIRI Core for action execution and state persistence. +* **Future Migration**: Once the main AIRI Core is ready, the `src/core/` directory (specifically the loop/planning logic) will be removed. This module will then be refactored to strictly function as an **Adapter** (Satori Protocol handling) and **Capability Provider** (Actions), delegating the cognitive loop to the main AIRI process. -## 安装 +For the current standalone version, please refer to these documents: +* **[HANDLER.md](./docs/HANDLER.md)**: Explains the **current** Event-to-Action Flow (Queue -> Scheduler -> LLM). +* **[PERSISTENCE.md](./docs/PERSISTENCE.md)**: Details the **current** Memory-First state management strategy specific to this temporary core. + +**Key Code Paths:** +* **Loop & Logic (Temporary)**: `src/core/` +* **Adapter (Permanent)**: `src/adapter/satori/` +* **Capabilities (Permanent)**: `src/capabilities/` + +## Prerequisites + +* **Node.js** >= 18.0.0 +* **pnpm** >= 8.0.0 +* **Koishi Instance**: Running the `server-satori` plugin. +* **LLM Provider**: OpenAI compatible API (Ollama, vLLM, DeepSeek, etc.). + +## Quick Start + +1. **Install Dependencies** ```bash -# 在项目根目录 pnpm install ``` -## 配置 - -复制 `.env` 文件并修改配置： +2. **Configure Environment** +Copy the example config and edit it: ```bash -# 在 services/satori-bot 目录 cp .env .env.local ``` -编辑 `.env.local`： +**Key Variables:** ```env # Satori Configuration SATORI_WS_URL=ws://localhost:5140/satori/v1/events SATORI_API_BASE_URL=http://localhost:5140/satori/v1 -SATORI_TOKEN=your_satori_token_here +SATORI_TOKEN= # Optional: Leave empty if auth is disabled in Koishi -# LLM Configuration +# LLM (OpenAI Compatible) LLM_API_KEY=your_api_key_here LLM_API_BASE_URL=https://api.openai.com/v1 LLM_MODEL=gpt-4 -LLM_RESPONSE_LANGUAGE=简体中文 +LLM_RESPONSE_LANGUAGE=English LLM_OLLAMA_DISABLE_THINK=false ``` -### 配置说明 - -#### Satori 配置 - -- `SATORI_WS_URL`: Satori WebSocket 地址（Koishi 默认：`ws://localhost:5140/satori/v1/events`） -- `SATORI_API_BASE_URL`: Satori HTTP API 地址（Koishi 默认：`http://localhost:5140/satori/v1`） -- `SATORI_TOKEN`: Satori 认证令牌（在 Koishi 配置中获取，如果为空 请留空，如：`SATORI_TOKEN=`） - -**重要**: Koishi 的 Satori 服务默认路由是 `/satori/v1`，因此完整的 API 路径会自动拼接，例如： -- 发送消息: `http://localhost:5140/satori/v1/message.create` -- 获取消息: `http://localhost:5140/satori/v1/message.get` - -#### LLM 配置 - -- `LLM_API_KEY`: LLM API 密钥 -- `LLM_API_BASE_URL`: LLM API 地址 -- `LLM_MODEL`: 使用的模型名称 -- `LLM_RESPONSE_LANGUAGE`: 回复语言（默认：简体中文） -- `LLM_OLLAMA_DISABLE_THINK`: 是否禁用 Ollama 的思考模式 - -## 使用 - -### 开发模式 +3. **Run** ```bash -# 在项目根目录 +# Development (Hot-reload) pnpm --filter @proj-airi/satori-bot dev -``` -### 生产模式 - -```bash -# 在项目根目录 +# Production pnpm --filter @proj-airi/satori-bot start ``` -### 类型检查 +## Key Locations -```bash -# 在项目根目录 -pnpm --filter @proj-airi/satori-bot typecheck -``` - -## 常见问题 - -### 1. 如何配置 Koishi？ - -在 Koishi 中启用 `server-satori`，配置项保持默认即可，无需改动。 - -### 2. 如何自定义 AI 人格？ - -可以编辑以下文件： - -- `services\satori-bot\src\prompts\personality-v1.velin.md` -- `services\satori-bot\src\prompts\system-action-gen-v1.velin.md` - -### 3. 数据库文件在哪里？ - -`services/satori-bot/data/db.json` - -## 贡献 - -欢迎提交 Issue 和 Pull Request！ - -## 许可证 - -MIT License - -## 相关链接 - -- [AIRI 项目](https://github.com/moeru-ai/airi) -- [Satori 协议文档](https://satori.chat/) -- [Koishi 文档](https://koishi.chat/) +* **Persona & System Prompts**: `src/core/planner/prompts/*.velin.md` +* **Database (JSON)**: `data/db.json` (See *PERSISTENCE.md* for limitations) +* **Action Logic**: `src/capabilities/actions/` diff --git a/services/satori-bot/docs/EVENT.md b/services/satori-bot/docs/EVENT.md new file mode 100644 index 000000000..ef3a62ca5 --- /dev/null +++ b/services/satori-bot/docs/EVENT.md @@ -0,0 +1,84 @@ +### **JSON Field Documentation** + +JSON Field Documentation of `SatoriEvent` + +* **Root Level** + +| Path | Type | Description | +| --- | --- | --- | +| `self_id` | String | The unique ID (QQ number) of the bot receiving the event. | +| `platform` | String | The platform name (e.g., `onebot`). | +| `timestamp` | Number | Unix timestamp (ms) when the event was created. | +| `type` | String | The general event category (e.g., `message-created`). | +| `subtype` | String | The sub-category of the event (e.g., `group`). | +| `subsubtype` | String | Further classification (e.g., `group`). | +| `id` | Integer | Internal sequence ID for the event processing. | +| `sn` | Integer | Serial number for the event. | + +* **Message Object** + +| Path | Type | Description | +| --- | --- | --- | +| `message` | Object | Container for standardized message details. | +| `message.message_id` | String | Unique identifier for this specific message. | +| `message.content` | String | The plain text content of the message. | + +* **User & Member** + +| Path | Type | Description | +| --- | --- | --- | +| `user` | Object | Standardized information about the message sender. | +| `user.id` | String | Unique ID of the sender. | +| `user.name` | String | Display name of the sender. | +| `user.avatar` | String | URL to the sender's avatar image. | +| `member` | Object | Context-specific member info (e.g., group membership). | +| `member.nick` | String | The user's nickname/card in this specific group . | +| `member.roles` | Array | List of roles assigned to the user (e.g., `member`). | + +* **Context (Guild/Group)** + +| Path | Type | Description | +| --- | --- | --- | +| `guild` | Object | Information about the guild/group. | +| `guild.id` | String | Unique ID of the group/guild. | +| `channel` | Object | Information about the channel (often same as guild in QQ groups). | +| `channel.type` | Integer | Channel type classification. | + +* **Bot Instance (`login`)** + +| Path | Type | Description | +| --- | --- | --- | +| `login` | Object | Information about the bot instance processing this event. | +| `login.user` | Object | The bot's own user details (Name, Avatar, ID). | +| `login.status` | Integer | Connection status (1 = Online). | +| `login.features` | Array | List of supported API features (e.g., `message.create`). | +| `login.adapter` | String | The adapter protocol being used (`onebot`). | + +### [Optional] raw data from adapter + +e.g. onebot + +* **OneBot Data (`_data`)** *Raw payload from the OneBot adapter* + +| Path | Type | Description | +| --- | --- | --- | +| `_data.message_type` | String | Type of message (e.g., `group`, `private`). | +| `_data.sub_type` | String | Subtype (e.g., `normal`, `anonymous`). | +| `_data.message_id` | Integer | Message ID as an integer (OneBot standard). | +| `_data.real_id` | Integer | The real message ID from the protocol. | +| `_data.sender` | Object | Sender details specific to OneBot format. | +| `_data.sender.card` | String | The sender's group card/nickname. | +| `_data.raw_message` | String | The unformatted raw string of the message. | +| `_data.message` | Array | Array of message segments (Text, Image, Face, etc.). | +| `_data.group_id` | Integer | The numeric ID of the group. | +| `_data.group_name` | String | The name of the group. | + +* **Protocol Raw Data (`_data.raw`)** *Internal low-level protocol data (NTQQ/Lagrange)* + +| Path | Type | Description | +| --- | --- | --- | +| `_data.raw.msgId` | String | Protocol-level message ID. | +| `_data.raw.msgSeq` | String | Message sequence number. | +| `_data.raw.elements` | Array | Detailed rich media elements (text, faces, images). | +| `_data.raw.senderUin` | String | Sender's User Internal Number (QQ). | +| `_data.raw.peerUin` | String | Receiver/Group User Internal Number. | diff --git a/services/satori-bot/docs/HANDLER.md b/services/satori-bot/docs/HANDLER.md new file mode 100644 index 000000000..f215e51b1 --- /dev/null +++ b/services/satori-bot/docs/HANDLER.md @@ -0,0 +1,79 @@ +# Core Message Flow Architecture + +This document outlines the complete lifecycle of a message within the Satori Bot, from the initial WebSocket event to the LLM's decision-making process. + +## 1. Architectural Overview + +The bot operates on a **Event-Driven + Autonomous Loop** hybrid model: +* **Event Layer**: Handles raw WebSocket signals, deduplication, and queuing. +* **Scheduler Layer**: Consumes the queue, updates the internal "Unread Pool" state, and triggers channel-specific processing loops. +* **Planner Layer**: The LLM acts as an Agent that observes the "Unread Pool" and "History Actions" state to decide whether to `read_unread_messages` (observe) or `send_message` (act). + +--- + +## 2. Detailed Data Flow + +### Phase 1: Ingress +**Location:** `src/adapter/satori/client.ts` → `src/core/loop/queue.ts` + +1. **WebSocket Reception**: The `SatoriClient` receives a raw JSON signal and parses it into a `SatoriEvent`. +2. **Event Listener**: The `setupMessageEventHandler` (in `queue.ts`) listens for `message-created` events. +3. **Deduplication**: The system checks the `processedIds` set (key: `channelId-messageId`) to prevent double-processing. +4. **Enqueuing**: + * The raw `event` is wrapped into a `{ event, status: 'ready' }` object. + * It is pushed into `botContext.eventQueue`. + * **Key Data**: `event.message.content`, `event.user.id`, `event.channel.id`. + +### Phase 2: Consumption & Anchoring +**Location:** `src/core/loop/scheduler.ts` (Function: `onMessageArrival`) + +When the system processing lock is free, it consumes events from the `eventQueue`: + +1. **Context Initialization**: + * Extracts `channelId`. + * Calls `ensureChatContext` (in `src/core/session/context.ts`) to load or create the in-memory `ChatContext` for that channel. + * **Anchor**: The `event.channel.id` is the primary key for all context. +2. **Filtering**: + * Checks `selfId`. If the sender is the bot itself, the event is discarded (not counted as unread) to prevent infinite loops. +3. **State Update (The "Unread Pool")**: + * The event is pushed into `botContext.unreadEvents[channelId]`. + * *Note:* This step does **not** just store the message; it marks the event as a "pending observation object". +4. **Loop Trigger**: + * Immediately calls `loopIterationForChannel`, waking up the Agent Loop for this specific channel. + +### Phase 3: Reasoning (LLM) +**Location:** `src/core/loop/scheduler.ts` → `src/core/planner/llm-client.ts` + +The LLM is prompted not to "reply to this text," but to "decide the next action based on state." + +1. **Context Construction (`imagineAnAction`)**: + * **System**: Injects `system-action-gen-v1` (Tool Definitions) and `personality-v1` (Persona). + * **Short-term Memory**: Injects `chatContext.messages` (Recent conversation turns). + * **Global State (Crucial)**: The prompt explicitly states: *"You have X unread events."* and lists the contents of `botContext.unreadEvents`. + * **Incoming Injection**: If there is an incoming message stream, it is injected as an `Incoming events` block at the end of the prompt. + * **Action History**: Injects `chatContext.actions` to show the results of previous attempts (e.g., "Last action: read_messages, Result: Success"). +2. **Generation**: + * The LLM outputs a strictly formatted JSON Action, e.g., `{"action": "read_unread_messages", "channelId": "..."}`. + +### Phase 4: Dispatch & Execution +**Location:** `src/core/dispatcher.ts` → `src/capabilities/registry.ts` + +The system looks up the corresponding Handler in `globalRegistry` based on the JSON Action: + +* **Case: `read_unread_messages`** (`src/capabilities/actions/read-messages.ts`) + * **Logic**: Retrieves all backlog events from `botContext.unreadEvents` for the specified channel. + * **Formatting**: Converts them into a single text block (e.g., `[User]: Content`). + * **Result**: Stores this text in the `Action Result`. + * **State Change**: Clears the `unreadEvents` for that channel. In the **next Tick**, the LLM will see this text in its History Actions and generate a reply. + +* **Case: `send_message`** (`src/capabilities/actions/send-message.ts`) + * **Safety Check**: Checks `unreadEvents` again. If new messages arrived during generation, it might abort the send to prioritize reading. + * **Execution**: Calls `satoriClient.sendMessage`. + * **Recording**: Persists the response to the DB and the in-memory `messages` array. + +### Phase 5: Loop Continuation +**Location:** `src/core/loop/scheduler.ts` (`handleLoopStep`) + +* `dispatchAction` returns an `ActionResult` containing a `shouldContinue` flag. +* If `shouldContinue` is true (e.g., usually true after reading messages, as a reply is expected), the scheduler waits for `LOOP_CONTINUE_DELAY_MS` (default 2.5s) and then recursively calls `handleLoopStep`. +* **Termination**: The loop stops only when the LLM selects the `continue` action (Wait/Stop) or the `break` action. diff --git a/services/satori-bot/docs/PERSISTENCE.md b/services/satori-bot/docs/PERSISTENCE.md new file mode 100644 index 000000000..ac4200296 --- /dev/null +++ b/services/satori-bot/docs/PERSISTENCE.md @@ -0,0 +1,46 @@ +## **Architecture Status Report: Memory & Persistence** + +**Date:** February 9, 2026 (Refactored) +**Component:** State Management Layer + +### **1. Memory Architecture (RAM)** + +The bot utilizes a **Memory-First** strategy, where the active state is fully resident in the Node.js heap. + +* **Storage Mechanism**: All chat contexts are stored in a native `Map` within the `BotContext` object (`src/core/types.ts`). +* **Lifecycle Management**: + * **Creation**: Contexts are lazy-loaded via `ensureChatContext` in `src/core/session/context.ts` upon receiving a message. + * **Retention**: There is currently **no garbage collection (GC)** mechanism. Once a channel is loaded, its context remains in memory indefinitely until the process terminates. +* **Context Trimming**: + * Executed within `handleLoopStep` in `src/core/loop/scheduler.ts`. + * Individual channels enforce a strict limit on history length (Default: 20 messages, 50 actions) to prevent single-channel bloat. + * **Risk**: The architecture is susceptible to memory leaks (OOM) as the number of unique channels increases over time. + +### **2. Persistence Architecture (Disk)** + +The bot uses a file-based logging system primarily for archival purposes and basic metadata recovery upon restart, rather than for active state management. + +* **Technology**: `lowdb` with a JSON file adapter. +* **Location**: `src/lib/db.ts` -> `data/db.json`. +* **Data Structure**: + * `channels`: Stores metadata like Channel ID, Platform, and SelfID. + * `messages`: A global, flattened array of messages. +* **Write Strategy**: **Synchronous full-file serialization**. Every new message triggers a complete rewrite of the JSON file to disk. +* **Retention Policy**: A global hard limit of **1000 messages** is enforced. When the limit is reached, the oldest messages are discarded regardless of which channel they belong to. +* **Recovery Logic**: Upon restart, `ensureChatContext` queries `db.channels` to restore the channel's `platform` and `selfId`, but it **does not** load historical messages into the in-memory context. + +### **3. State Consistency** + +There is a significant desynchronization between the ephemeral memory state and the persistent disk state. + +* **In-Memory State (Rich)**: Contains the full "Chain of Thought" (System prompts, reasoning steps, `AbortController` handles, pending Promises, `Action History`). +* **On-Disk State (Flat)**: Contains only raw user content and final bot responses. +* **Impact**: A process restart results in a **Hard Context Reset**. The bot loses all active "trains of thought" and task states, falling back to a state driven solely by new incoming messages. + +### **4. Future Roadmap (WIP)** + +We are planning to implement a "Small Memory" storage scheme to improve robustness, featuring: + +1. **Indiscriminate Event Storage**: Storing all events without preemptive filtering. +2. **Event Activation Query**: Triggering queries based on specific event activation signals. +3. **Dynamic Context Filtering**: Reconstructing same-group contexts via query-time filtering rather than pre-computed buckets. diff --git a/services/satori-bot/docs/PROMPTS.md b/services/satori-bot/docs/PROMPTS.md new file mode 100644 index 000000000..e19f8ca0c --- /dev/null +++ b/services/satori-bot/docs/PROMPTS.md @@ -0,0 +1,54 @@ +### **Prompt Architecture: Context-Injected Action Loop** + +The bot implements a **"State-Aware Agentic Loop"** rather than a simple Chat-QA structure. The prompt is constructed dynamically at every tick of the loop. + +#### **1. Static Layer (System Definitions)** + +* **Source:** `src/core/planner/prompts/*.velin.md` +* **Loader:** `src/core/planner/prompts/index.ts` +* **Role:** Defines the "Soul" and "Rules". +* **Components:** + * **Protocol Definition (`system-action-gen-v1`):** Hardcodes the JSON schema for available tools (`send_message`, `read_unread_messages`, `sleep`) and logic flow (e.g., "Must check unread messages after sending"). + * **Persona (`personality-v1`):** Defines the character "AIRI" (tone, brevity, naturalness). + +#### **2. History Layer (Short-term Memory)** + +* **Source:** In-memory `messages` array (`ChatContext`). +* **Role:** Provides conversational continuity. +* **Mechanism:** A sliding window of the last ~20 messages (User/Assistant turns) is injected directly after the system prompt. + +#### **3. Dynamic State Layer (Sensory Injection)** + +* **Source:** `src/core/planner/llm-client.ts` (Runtime generated) +* **Role:** Provides "Situational Awareness" and "Grounding". +* **Mechanism:** A synthesized **User Message** is appended at the very end of the context window, forcing the LLM to focus on the immediate reality. It contains: + * **Incoming Stream**: Raw content of new messages arriving *now* (passed from `scheduler`). + * **Action History**: Results of the *immediately preceding* tool executions (e.g., "Action: send_message, Result: Success"). + * **Environment**: Current server time. + * **Global State**: A summary of unread message counts across all channels (`unreadEvents`). + * **Trigger**: The final instruction: *"Based on the context... Respond with the action... in JSON only."* + +--- + +### **Data Flow Summary** + +```mermaid +graph TD + A[Static Markdown] -->|Velin Render| B(System Message) + C[Chat History] -->|Sliding Window| D(Context Body) + E[Runtime State] -->|Unread/Time/Results| F(State Injection) + + B --> G[Final Prompt] + D --> G + F --> G + + G -->|LLM API| H{Decision} + H -->|JSON| I[Action Dispatcher] + +``` + +### **Key Characteristics** + +1. **JSON Enforcement**: The bot does not use native "Function Calling" APIs (like OpenAI Tools). It relies on **Prompt Engineering** to force the model to output raw JSON, which is then parsed by `best-effort-json-parser`. +2. **Stateless Logic**: The prompt explicitly tells the LLM "You have X unread messages" in every turn, making the LLM the sole decision-maker for flow control (Reading vs. Replying vs. Sleeping). +3. **Observation-Reflection**: The prompt includes `History actions`, allowing the LLM to "see" the result of its previous attempt (e.g., if a read action returned empty, it knows to stop). diff --git a/services/satori-bot/src/actions/read-unread-messages.ts b/services/satori-bot/src/actions/read-unread-messages.ts deleted file mode 100644 index 1f7fc0a55..000000000 --- a/services/satori-bot/src/actions/read-unread-messages.ts +++ /dev/null @@ -1,52 +0,0 @@ -import type { BotContext, ChatContext, ReadUnreadMessagesAction } from '../types/bot' -import type { SatoriMessage } from '../types/satori' - -import { useLogg } from '@guiiai/logg' - -export async function readUnreadMessages( - botContext: BotContext, - chatContext: ChatContext, - action: ReadUnreadMessagesAction, -): Promise<{ result: string } | undefined> { - const logger = useLogg('readUnreadMessages').useGlobalConfig() - - if (Object.keys(botContext.unreadMessages).length === 0) { - logger.log('No unread messages - clearing unread messages') - botContext.unreadMessages = {} - return undefined - } - - if (!action.channelId) { - logger.warn('No channel ID provided - clearing all unread messages') - return undefined - } - - let unreadMessagesForThisChannel: SatoriMessage[] | undefined = botContext.unreadMessages[action.channelId] - - if (!Array.isArray(unreadMessagesForThisChannel)) { - logger.log('Unread messages for channel is not an array - converting to array') - unreadMessagesForThisChannel = [] - } - - if (unreadMessagesForThisChannel.length === 0) { - logger.log('No unread messages for channel - deleting') - delete botContext.unreadMessages[action.channelId] - return undefined - } - - // Format messages for LLM context - const formattedMessages = unreadMessagesForThisChannel.map((msg) => { - const userName = msg.user?.name || msg.user?.id || 'Unknown' - const content = msg.content || '[No content]' - return `[${userName}]: ${content}` - }).join('\n') - - // Clear the unread messages for this channel - delete botContext.unreadMessages[action.channelId] - - logger.log(`Read ${unreadMessagesForThisChannel.length} unread messages from channel ${action.channelId}`) - - return { - result: `AIRI System: Read ${unreadMessagesForThisChannel.length} unread messages from channel ${action.channelId}:\n${formattedMessages}`, - } -} diff --git a/services/satori-bot/src/actions/send-message.ts b/services/satori-bot/src/actions/send-message.ts deleted file mode 100644 index 38b3fc2de..000000000 --- a/services/satori-bot/src/actions/send-message.ts +++ /dev/null @@ -1,50 +0,0 @@ -import type { SatoriClient } from '../client/satori-client' -import type { BotContext, ChatContext } from '../types/bot' - -import { useLogg } from '@guiiai/logg' - -import { recordMessage } from '../db' - -export async function sendMessage( - botContext: BotContext, - chatContext: ChatContext, - satoriClient: SatoriClient, - content: string, - channelId: string, - _abortController: AbortController, -) { - const logger = useLogg('sendMessage').useGlobalConfig() - - try { - // Check if we should abort due to new messages - if (botContext.unreadMessages[channelId] && botContext.unreadMessages[channelId].length > 0) { - logger.log(`Not sending message to ${channelId} - new messages arrived`) - return - } - - // Send the message - logger.withField('channelId', channelId).withField('content', content).log('Sending message') - - await satoriClient.sendMessage(chatContext.platform, chatContext.selfId, channelId, content) - - // Record the message in database - await recordMessage(channelId, 'bot', 'AIRI', content) - - // Add to chat context as assistant message - chatContext.messages.push({ - role: 'assistant', - content, - }) - - logger.log('Message sent successfully') - } - catch (err) { - if ((err as Error).name === 'AbortError') { - logger.log('Message sending was aborted') - return - } - - logger.withError(err as Error).log('Failed to send message') - throw err - } -} diff --git a/services/satori-bot/src/client/satori-api.ts b/services/satori-bot/src/adapter/satori/api.ts similarity index 98% rename from services/satori-bot/src/client/satori-api.ts rename to services/satori-bot/src/adapter/satori/api.ts index b22b2a9be..a0b7a254f 100644 --- a/services/satori-bot/src/client/satori-api.ts +++ b/services/satori-bot/src/adapter/satori/api.ts @@ -1,4 +1,4 @@ -import type { SatoriMessageCreateRequest, SatoriMessageCreateResponse } from '../types/satori' +import type { SatoriMessageCreateRequest, SatoriMessageCreateResponse } from './types' import { useLogg } from '@guiiai/logg' diff --git a/services/satori-bot/src/client/satori-client.ts b/services/satori-bot/src/adapter/satori/client.ts similarity index 96% rename from services/satori-bot/src/client/satori-client.ts rename to services/satori-bot/src/adapter/satori/client.ts index 4e6623741..6dc7f2169 100644 --- a/services/satori-bot/src/client/satori-client.ts +++ b/services/satori-bot/src/adapter/satori/client.ts @@ -1,11 +1,11 @@ -import type { SatoriEvent, SatoriIdentifyBody, SatoriReadyBody, SatoriSignal } from '../types/satori' +import type { SatoriEvent, SatoriIdentifyBody, SatoriReadyBody, SatoriSignal } from './types' import WebSocket from 'ws' import { useLogg } from '@guiiai/logg' -import { SatoriOpcode } from '../types/satori' -import { SatoriAPI } from './satori-api' +import { SatoriAPI } from './api' +import { SatoriOpcode } from './types' const log = useLogg('SatoriClient') @@ -213,6 +213,8 @@ export class SatoriClient { // Public API for sending messages async sendMessage(platform: string, selfId: string, channelId: string, content: string): Promise { const key = `${platform}:${selfId}` + log.debug(`sendMessage called - platform: "${platform}", selfId: "${selfId}", key: "${key}"`) + log.debug(`Available API clients: ${Array.from(this.apiClients.keys()).join(', ')}`) const api = this.apiClients.get(key) if (!api) { diff --git a/services/satori-bot/src/types/satori.ts b/services/satori-bot/src/adapter/satori/types.ts similarity index 99% rename from services/satori-bot/src/types/satori.ts rename to services/satori-bot/src/adapter/satori/types.ts index 03f9343c5..1d519c0d3 100644 --- a/services/satori-bot/src/types/satori.ts +++ b/services/satori-bot/src/adapter/satori/types.ts @@ -78,6 +78,7 @@ export interface SatoriGuildRole { export interface SatoriMessage { id: string content: string + platform?: string channel?: SatoriChannel guild?: SatoriGuild member?: SatoriGuildMember diff --git a/services/satori-bot/src/bot/index.ts b/services/satori-bot/src/bot/index.ts deleted file mode 100644 index 98fc08dcd..000000000 --- a/services/satori-bot/src/bot/index.ts +++ /dev/null @@ -1,363 +0,0 @@ -import type { Logg } from '@guiiai/logg' - -import type { SatoriClient } from '../client/satori-client' -import type { Action, BotContext, ChatContext } from '../types/bot' -import type { SatoriMessage } from '../types/satori' - -import { readUnreadMessages } from '../actions/read-unread-messages' -import { sendMessage } from '../actions/send-message' -import { listChannels, recordChannel, recordMessage } from '../db' -import { imagineAnAction } from '../llm/actions' - -async function dispatchAction( - ctx: BotContext, - satoriClient: SatoriClient, - action: Action, - abortController: AbortController, - chatCtx?: ChatContext, -): Promise<(() => Promise) | undefined> { - // If action generation failed, don't proceed - if (!action || !action.action) { - ctx.logger.withField('action', action).log('No valid action returned.') - if (chatCtx) { - chatCtx.messages.push({ - role: 'user', - content: 'AIRI System: No valid action returned.', - }) - return () => handleLoopStep(ctx, satoriClient, chatCtx) - } - return undefined - } - - switch (action.action) { - case 'list_channels': { - if (chatCtx) { - const channels = await listChannels() - const channelList = channels.map(c => `ID:${c.id}, Name:${c.name}, Platform:${c.platform}`).join('\n') - chatCtx.actions.push({ - action, - result: `AIRI System: List of channels:\n${channelList}`, - }) - } - return () => handleLoopStep(ctx, satoriClient, chatCtx) - } - - case 'send_message': { - const chatCtx = await ensureChatContext(ctx, action.channelId) - chatCtx.actions.push({ - action, - result: `AIRI System: Sending message to channel ${action.channelId}: ${action.content}`, - }) - await sendMessage(ctx, chatCtx, satoriClient, action.content, action.channelId, abortController) - return () => handleLoopStep(ctx, satoriClient, chatCtx) - } - - case 'read_unread_messages': { - const chatCtx = await ensureChatContext(ctx, action.channelId) - const res = await readUnreadMessages(ctx, chatCtx, action) - if (res?.result) { - ctx.logger.log('Messages read') - chatCtx.actions.push({ action, result: res.result }) - return () => handleLoopStep(ctx, satoriClient, chatCtx) - } - return undefined - } - - case 'continue': - if (chatCtx) { - chatCtx.actions.push({ - action, - result: 'AIRI System: Acknowledged, will now continue until next tick.', - }) - } - return undefined - - case 'break': - if (chatCtx) { - chatCtx.messages = [] - chatCtx.actions = [] - chatCtx.actions.push({ - action, - result: 'AIRI System: Acknowledged, will now break, and clear out all existing memories, messages, actions.', - }) - } - return undefined - - case 'sleep': - await new Promise(resolve => setTimeout(resolve, 30 * 1000)) - if (chatCtx) { - chatCtx.actions.push({ - action, - result: 'AIRI System: Sleeping for 30 seconds as requested...', - }) - } - return () => handleLoopStep(ctx, satoriClient, chatCtx) - - default: - if (chatCtx) { - chatCtx.messages.push({ - role: 'user', - content: `AIRI System: The action ${(action as any).action} hasn't been implemented yet by developer.`, - }) - } - return () => handleLoopStep(ctx, satoriClient, chatCtx) - } -} - -async function handleLoopStep( - ctx: BotContext, - satoriClient: SatoriClient, - chatCtx: ChatContext, - incomingMessage?: SatoriMessage, -): Promise<(() => Promise) | undefined> { - ctx.currentProcessingStartTime = Date.now() - - if (chatCtx?.currentAbortController) { - chatCtx.currentAbortController.abort() - } - - const currentController = new AbortController() - if (chatCtx) { - chatCtx.currentAbortController = currentController - - // Track message processing state - if (chatCtx.channelId && !ctx.lastInteractedChannelIds.includes(chatCtx.channelId)) { - ctx.lastInteractedChannelIds.push(chatCtx.channelId) - } - if (ctx.lastInteractedChannelIds.length > 5) { - ctx.lastInteractedChannelIds = ctx.lastInteractedChannelIds.slice(-5) - } - - // Manage context size - if (chatCtx.messages == null) { - chatCtx.messages = [] - } - if (chatCtx.messages.length > 20) { - const length = chatCtx.messages.length - chatCtx.messages = chatCtx.messages.slice(-5) - chatCtx.messages.push({ - role: 'user', - content: `AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${chatCtx.messages.length}, history may be lost.`, - }) - } - - if (chatCtx.actions == null) { - chatCtx.actions = [] - } - if (chatCtx.actions.length > 50) { - const length = chatCtx.actions.length - chatCtx.actions = chatCtx.actions.slice(-20) - chatCtx.messages.push({ - role: 'user', - content: `AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${chatCtx.actions.length}, history of actions may be lost.`, - }) - } - } - - try { - const action = await imagineAnAction( - currentController, - chatCtx?.messages || [], - chatCtx?.actions || [], - { - unreadMessages: ctx.unreadMessages, - incomingMessages: incomingMessage ? [incomingMessage] : [], - }, - ) - return await dispatchAction(ctx, satoriClient, action, currentController, chatCtx) - } - catch (err) { - if ((err as Error).name === 'AbortError') { - ctx.logger.log('Operation was aborted due to interruption') - return undefined - } - - ctx.logger.withError(err as Error).log('Error occurred') - return undefined - } - finally { - if (chatCtx && chatCtx.currentAbortController === currentController) { - chatCtx.currentAbortController = undefined - ctx.currentProcessingStartTime = undefined - } - } -} - -async function loopIterationForChannel( - bot: BotContext, - satoriClient: SatoriClient, - chatContext: ChatContext, - incomingMessage: SatoriMessage, -) { - let result = await handleLoopStep(bot, satoriClient, chatContext, incomingMessage) - - while (typeof result === 'function') { - result = await result() - } - - return result -} - -async function loopIterationPeriodicForExistingChannels(ctx: BotContext, satoriClient: SatoriClient) { - // Only process channels with unread messages to avoid unnecessary LLM calls - const channelsWithUnread = Object.keys(ctx.unreadMessages).filter( - channelId => ctx.unreadMessages[channelId]?.length > 0, - ) - - if (channelsWithUnread.length === 0) { - ctx.logger.log('No channels with unread messages, skipping periodic check') - return - } - - ctx.logger.withField('channelCount', channelsWithUnread.length).log('Processing channels with unread messages') - - // Process channels sequentially to avoid overwhelming the LLM API - for (const channelId of channelsWithUnread) { - const chatCtx = await ensureChatContext(ctx, channelId) - - try { - const action = await imagineAnAction( - chatCtx.currentAbortController, - chatCtx.messages, - chatCtx.actions, - { unreadMessages: ctx.unreadMessages }, - ) - let result = await dispatchAction(ctx, satoriClient, action, chatCtx.currentAbortController, chatCtx) - - while (typeof result === 'function') { - result = await result() - } - } - catch (err) { - ctx.logger.withError(err as Error).withField('channelId', channelId).log('Error processing channel in periodic loop') - // Continue to next channel instead of breaking the entire loop - } - } -} - -function loopPeriodic(botCtx: BotContext, satoriClient: SatoriClient) { - setTimeout(async () => { - try { - await loopIterationPeriodicForExistingChannels(botCtx, satoriClient) - } - catch (err) { - if ((err as Error).name === 'AbortError') { - botCtx.logger.log('main loop was aborted - restarting loop') - } - else { - botCtx.logger.withError(err as Error).log('error in main loop') - } - } - finally { - loopPeriodic(botCtx, satoriClient) - } - }, 60 * 1000) -} - -export function createBotContext(logger: Logg): BotContext { - const botSelf: BotContext = { - messageQueue: [], - unreadMessages: {}, - processedIds: new Set(), - logger, - processing: false, - lastInteractedChannelIds: [], - chats: new Map(), - } - - return botSelf -} - -export async function onMessageArrival( - botContext: BotContext, - satoriClient: SatoriClient, - chatCtx: ChatContext, -) { - if (botContext.processing) { - return - } - botContext.processing = true - - try { - while (botContext.messageQueue.length > 0) { - const nextMsg = botContext.messageQueue[0] - - if (nextMsg.status === 'ready') { - // Record channel (use chatCtx.channelId which is already correctly set) - await recordChannel( - chatCtx.channelId, - nextMsg.message.channel?.name || chatCtx.channelId, - chatCtx.platform, - chatCtx.selfId, - ) - - // Record message - if (nextMsg.message.user && nextMsg.message.content) { - await recordMessage( - chatCtx.channelId, - nextMsg.message.user.id, - nextMsg.message.user.name || nextMsg.message.user.id, - nextMsg.message.content, - ) - } - - let unreadMessagesForThisChannel = botContext.unreadMessages[chatCtx.channelId] - - if (unreadMessagesForThisChannel == null) { - botContext.logger.withField('channelId', chatCtx.channelId).log('unread messages for this channel is null - creating empty array') - unreadMessagesForThisChannel = [] - } - if (!Array.isArray(unreadMessagesForThisChannel)) { - botContext.logger.withField('channelId', chatCtx.channelId).log('unread messages for this channel is not an array - converting to array') - unreadMessagesForThisChannel = [] - } - - unreadMessagesForThisChannel.push(nextMsg.message) - - if (unreadMessagesForThisChannel.length > 100) { - unreadMessagesForThisChannel = unreadMessagesForThisChannel.slice(-100) - } - - botContext.unreadMessages[chatCtx.channelId] = unreadMessagesForThisChannel - botContext.logger.withField('channelId', chatCtx.channelId).log('message queue processed, triggering immediate reaction') - - // Trigger immediate processing - await loopIterationForChannel(botContext, satoriClient, chatCtx, nextMsg.message) - botContext.messageQueue.shift() - } - } - } - catch (err) { - botContext.logger.withError(err as Error).log('Error occurred') - } - finally { - botContext.processing = false - } -} - -export async function ensureChatContext(botCtx: BotContext, channelId: string): Promise { - if (botCtx.chats.has(channelId)) { - return botCtx.chats.get(channelId)! - } - - // Try to get channel info from database - const channels = await listChannels() - const channelInfo = channels.find(c => c.id === channelId) - - const newChatContext: ChatContext = { - channelId, - platform: channelInfo?.platform || '', - selfId: channelInfo?.selfId || '', - currentTask: undefined, - currentAbortController: undefined, - messages: [], - actions: [], - } - - botCtx.chats.set(channelId, newChatContext) - return newChatContext -} - -export function startPeriodicLoop(botCtx: BotContext, satoriClient: SatoriClient) { - loopPeriodic(botCtx, satoriClient) -} diff --git a/services/satori-bot/src/capabilities/actions/read-messages.ts b/services/satori-bot/src/capabilities/actions/read-messages.ts new file mode 100644 index 000000000..03d3aca2a --- /dev/null +++ b/services/satori-bot/src/capabilities/actions/read-messages.ts @@ -0,0 +1,47 @@ +import type { ActionHandler, ActionResult } from '../definition' + +import { useLogg } from '@guiiai/logg' + +export const readMessagesAction: ActionHandler = { + name: 'read_unread_messages', + description: 'Read unread messages from a specific channel', + execute: async (botContext, chatCtx, args): Promise => { + const logger = useLogg('readMessagesAction').useGlobalConfig() + const channelId = args.channelId + + if (!channelId) { + return { + success: false, + shouldContinue: true, + result: 'System Error: No channelId provided for read_unread_messages.', + } + } + + const unreadEventsForThisChannel = botContext.unreadEvents[channelId] + + if (!unreadEventsForThisChannel || unreadEventsForThisChannel.length === 0) { + delete botContext.unreadEvents[channelId] + return { + success: true, + shouldContinue: true, + result: 'AIRI System: No unread messages found.', + } + } + + const formattedMessages = unreadEventsForThisChannel.map((event) => { + const userName = event.user?.name || event.user?.id || 'Unknown' + const content = event.message?.content || '[No content]' + return `[${userName}]: ${content}` + }).join('\n') + + delete botContext.unreadEvents[channelId] + + logger.log(`Read ${unreadEventsForThisChannel.length} unread events from channel ${channelId}`) + + return { + success: true, + shouldContinue: true, + result: `AIRI System: Read ${unreadEventsForThisChannel.length} unread events from channel ${channelId}:\n${formattedMessages}`, + } + }, +} diff --git a/services/satori-bot/src/capabilities/actions/send-message.ts b/services/satori-bot/src/capabilities/actions/send-message.ts new file mode 100644 index 000000000..29eb7fdf9 --- /dev/null +++ b/services/satori-bot/src/capabilities/actions/send-message.ts @@ -0,0 +1,55 @@ +import type { SatoriClient } from '../../adapter/satori/client' +import type { ActionHandler } from '../definition' + +import { useLogg } from '@guiiai/logg' + +import { recordMessage } from '../../lib/db' + +export function createSendMessageAction(client: SatoriClient): ActionHandler { + return { + name: 'send_message', + execute: async (ctx, chatCtx, args) => { + const logger = useLogg('Action:send_message').useGlobalConfig() + const { channelId, content } = args + + // Logic 1: Concurrency Safety Check + if (ctx.unreadEvents[channelId] && ctx.unreadEvents[channelId].length > 0) { + logger.withField('channelId', channelId).warn('Aborting message send due to new incoming events') + + return { + success: false, + shouldContinue: true, + result: 'AIRI System: [INTERRUPT] Message sending ABORTED. New unread messages were detected from the user. Please [read_unread_messages] first to understand the new context.', + } + } + + try { + // Logic 2: Execute Send + await client.sendMessage(chatCtx.platform, chatCtx.selfId, channelId, content) + + // Logic 3: Persistence + await recordMessage(channelId, 'bot', 'AIRI', content) + + // Logic 4: Memory State Update + chatCtx.messages.push({ + role: 'assistant', + content, + }) + + return { + success: true, + shouldContinue: true, + result: `AIRI System: Message sent to ${channelId}: ${content}`, + } + } + catch (error) { + logger.withError(error as Error).error('Failed to send message') + return { + success: false, + shouldContinue: true, + result: `AIRI System: Error sending message: ${(error as Error).message}`, + } + } + }, + } +} diff --git a/services/satori-bot/src/capabilities/actions/system.ts b/services/satori-bot/src/capabilities/actions/system.ts new file mode 100644 index 000000000..32750eca4 --- /dev/null +++ b/services/satori-bot/src/capabilities/actions/system.ts @@ -0,0 +1,58 @@ +import type { ActionHandler, ActionResult } from '../definition' + +import { SLEEP_DURATION_MS } from '../../core/constants' +import { listChannels } from '../../lib/db' + +// 1. Continue Action +export const continueAction: ActionHandler = { + name: 'continue', + execute: async (): Promise => { + return { + success: true, + shouldContinue: false, + result: 'AIRI System: Acknowledged, will now wait for user input.', + } + }, +} + +// 2. Break Action +export const breakAction: ActionHandler = { + name: 'break', + execute: async (_ctx, chatCtx): Promise => { + chatCtx.messages = [] + chatCtx.actions = [] + return { + success: true, + shouldContinue: false, + result: 'AIRI System: Memory cleared. Loop broken.', + } + }, +} + +// 3. Sleep Action +export const sleepAction: ActionHandler = { + name: 'sleep', + execute: async (_ctx, _chatCtx, args): Promise => { + const duration = args.duration || SLEEP_DURATION_MS + await new Promise(resolve => setTimeout(resolve, duration)) + return { + success: true, + shouldContinue: true, + result: `AIRI System: Slept for ${duration / 1000} seconds.`, + } + }, +} + +// 4. List Channels Action +export const listChannelsAction: ActionHandler = { + name: 'list_channels', + execute: async (): Promise => { + const channels = await listChannels() + const list = channels.map(c => `ID:${c.id}, Name:${c.name}, Platform:${c.platform}`).join('\n') + return { + success: true, + shouldContinue: true, + result: `AIRI System: Channel List:\n${list}`, + } + }, +} diff --git a/services/satori-bot/src/capabilities/definition.ts b/services/satori-bot/src/capabilities/definition.ts new file mode 100644 index 000000000..264e5237a --- /dev/null +++ b/services/satori-bot/src/capabilities/definition.ts @@ -0,0 +1,18 @@ +import type { BotContext, ChatContext } from '../core/types' + +export interface ActionResult { + success: boolean + shouldContinue: boolean + result: any +} + +export interface ActionHandler { + name: string + description?: string + execute: ( + ctx: BotContext, + chatCtx: ChatContext, + args: any, + abortSignal?: AbortSignal, + ) => Promise +} diff --git a/services/satori-bot/src/capabilities/registry.ts b/services/satori-bot/src/capabilities/registry.ts new file mode 100644 index 000000000..2ff957f06 --- /dev/null +++ b/services/satori-bot/src/capabilities/registry.ts @@ -0,0 +1,34 @@ +import type { SatoriClient } from '../adapter/satori/client' +import type { ActionHandler } from './definition' + +import { readMessagesAction } from './actions/read-messages' +import { createSendMessageAction } from './actions/send-message' +import { breakAction, continueAction, listChannelsAction, sleepAction } from './actions/system' +// import { createReadMessagesAction } ... + +export class ActionRegistry { + private actions = new Map() + + register(handler: ActionHandler) { + this.actions.set(handler.name, handler) + } + + get(name: string): ActionHandler | undefined { + return this.actions.get(name) + } + + // 提供一个批量加载的方法 + loadStandardActions(client: SatoriClient) { + // 注册不需要依赖的系统 Action + this.register(continueAction) + this.register(breakAction) + this.register(sleepAction) + this.register(listChannelsAction) + + // 注册需要注入依赖的 Action + this.register(createSendMessageAction(client)) + this.register(readMessagesAction) + } +} + +export const globalRegistry = new ActionRegistry() diff --git a/services/satori-bot/src/core/constants.ts b/services/satori-bot/src/core/constants.ts new file mode 100644 index 000000000..076fd9c00 --- /dev/null +++ b/services/satori-bot/src/core/constants.ts @@ -0,0 +1,21 @@ +/** + * Bot configuration constants + * Centralized configuration values for the satori-bot + */ + +// Loop and timing constants +export const LOOP_CONTINUE_DELAY_MS = 2500 +export const PERIODIC_LOOP_INTERVAL_MS = 60 * 1000 +export const SLEEP_DURATION_MS = 30 * 1000 + +// Context size limits +export const MAX_MESSAGES_IN_CONTEXT = 20 +export const MAX_ACTIONS_IN_CONTEXT = 50 +export const MAX_UNREAD_EVENTS = 100 + +// Recent interaction tracking +export const MAX_RECENT_INTERACTED_CHANNELS = 5 + +// Context trimming - how many items to keep when trimming +export const MESSAGES_KEEP_ON_TRIM = 5 +export const ACTIONS_KEEP_ON_TRIM = 20 diff --git a/services/satori-bot/src/core/dispatcher.ts b/services/satori-bot/src/core/dispatcher.ts new file mode 100644 index 000000000..425783b33 --- /dev/null +++ b/services/satori-bot/src/core/dispatcher.ts @@ -0,0 +1,52 @@ +import type { ActionResult } from '../capabilities/definition' +import type { BotContext, ChatContext } from './types' + +import { globalRegistry } from '../capabilities/registry' + +export async function dispatchAction( + ctx: BotContext, + chatCtx: ChatContext, + actionPayload: any, + abortController: AbortController, +): Promise { + const log = ctx.logger.useGlobalConfig() + + if (!actionPayload || !actionPayload.action) { + return { + success: false, + shouldContinue: true, + result: 'System Error: No valid action name provided in JSON.', + } + } + + const handler = globalRegistry.get(actionPayload.action) + + if (!handler) { + return { + success: false, + shouldContinue: true, + result: `System Error: Action "${actionPayload.action}" is not implemented.`, + } + } + + try { + log.withField('action', actionPayload.action).debug('Executing action') + + const result = await handler.execute(ctx, chatCtx, actionPayload, abortController.signal) + + chatCtx.actions.push({ + action: actionPayload, + result: result.result, + }) + + return result + } + catch (error) { + log.withError(error as Error).error('Action execution failed') + return { + success: false, + shouldContinue: true, + result: `System Error: Execution failed: ${(error as Error).message}`, + } + } +} diff --git a/services/satori-bot/src/core/index.ts b/services/satori-bot/src/core/index.ts new file mode 100644 index 000000000..ef064c066 --- /dev/null +++ b/services/satori-bot/src/core/index.ts @@ -0,0 +1,22 @@ +/** + * Bot module exports + * Centralized export point for all bot-related functionality + */ + +// Constants +export * from './constants' + +// Action dispatcher +export { dispatchAction } from './dispatcher' + +// Event handlers +export { setupMessageEventHandler, setupReadyEventHandler } from './loop/queue' + +// Loop processing +export { handleLoopStep, loopIterationForChannel, onMessageArrival, startPeriodicLoop } from './loop/scheduler' + +// Context management +export { createBotContext, ensureChatContext } from './session/context' + +// Utilities +export { formatDebugContext, getMessageContentString, isBotOwnMessage } from './utils' diff --git a/services/satori-bot/src/core/loop/queue.ts b/services/satori-bot/src/core/loop/queue.ts new file mode 100644 index 000000000..6e4235ddf --- /dev/null +++ b/services/satori-bot/src/core/loop/queue.ts @@ -0,0 +1,63 @@ +import type { Logg } from '@guiiai/logg' + +import type { SatoriClient } from '../../adapter/satori/client' +import type { SatoriEvent, SatoriReadyBody } from '../../adapter/satori/types' +import type { BotContext } from '../types' + +import { onMessageArrival } from './scheduler' + +/** + * Set up the ready event handler + * Logs connection information when Satori client is ready + */ +export function setupReadyEventHandler( + satoriClient: SatoriClient, + logger: Logg, +): void { + satoriClient.onReady((ready: SatoriReadyBody) => { + logger.log('Satori client ready:', ready) + logger.log(`Connected to ${ready.logins.length} platform(s)`) + + for (const login of ready.logins) { + logger.log(`- ${login.platform} (${login.self_id}): ${login.status}`) + } + }) +} + +/** + * Set up the message-created event handler + * Processes incoming messages, filters bot's own messages, and triggers bot responses + */ +export function setupMessageEventHandler( + satoriClient: SatoriClient, + botContext: BotContext, + logger: Logg, +): void { + satoriClient.on('message-created', async (event: SatoriEvent) => { + const message = event.message + if (!message) { + return + } + + logger.log(`Received message from ${event.user.id} in channel [${event.platform}] ${event.channel.id}: ${message.content}`) + + const messageId = `${event.channel.id}-${message.id}` + if (!botContext.processedIds.has(messageId)) { + botContext.processedIds.add(messageId) + } + else { + logger.debug(`Skipping already processed message: ${messageId}`) + return + } + + // Add to message queue + botContext.eventQueue.push({ + event, + status: 'ready', + }) + + // Process message queue + // Pass event so onMessageArrival can use correct channelId and set platform/selfId for each message + await onMessageArrival(botContext, satoriClient) + }) +} diff --git a/services/satori-bot/src/core/loop/scheduler.ts b/services/satori-bot/src/core/loop/scheduler.ts new file mode 100644 index 000000000..4f9e360a8 --- /dev/null +++ b/services/satori-bot/src/core/loop/scheduler.ts @@ -0,0 +1,285 @@ +import type { SatoriClient } from '../../adapter/satori/client' +import type { SatoriEvent } from '../../adapter/satori/types' +import type { BotContext, ChatContext } from '../types' + +import { recordChannel, recordMessage } from '../../lib/db' +import { + ACTIONS_KEEP_ON_TRIM, + LOOP_CONTINUE_DELAY_MS, + MAX_ACTIONS_IN_CONTEXT, + MAX_MESSAGES_IN_CONTEXT, + MAX_RECENT_INTERACTED_CHANNELS, + MAX_UNREAD_EVENTS, + MESSAGES_KEEP_ON_TRIM, + PERIODIC_LOOP_INTERVAL_MS, +} from '../constants' +import { dispatchAction } from '../dispatcher' +import { imagineAnAction } from '../planner/llm-client' +import { ensureChatContext } from '../session/context' + +/** + * Handle a single loop step + * Manages context size, calls LLM for action, and dispatches the action + */ +export async function handleLoopStep( + ctx: BotContext, + satoriClient: SatoriClient, + chatCtx: ChatContext, + incomingEvents?: SatoriEvent, +): Promise { + ctx.currentProcessingStartTime = Date.now() + + if (chatCtx?.currentAbortController) { + chatCtx.currentAbortController.abort() + } + + const currentController = new AbortController() + if (chatCtx) { + chatCtx.currentAbortController = currentController + + // Track message processing state + if (chatCtx.channelId && !ctx.lastInteractedChannelIds.includes(chatCtx.channelId)) { + ctx.lastInteractedChannelIds.push(chatCtx.channelId) + } + if (ctx.lastInteractedChannelIds.length > MAX_RECENT_INTERACTED_CHANNELS) { + ctx.lastInteractedChannelIds = ctx.lastInteractedChannelIds.slice(-MAX_RECENT_INTERACTED_CHANNELS) + } + + // Manage context size + if (chatCtx.messages == null) { + chatCtx.messages = [] + } + if (chatCtx.messages.length > MAX_MESSAGES_IN_CONTEXT) { + const length = chatCtx.messages.length + chatCtx.messages = chatCtx.messages.slice(-MESSAGES_KEEP_ON_TRIM) + chatCtx.messages.push({ + role: 'user', + content: `AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${chatCtx.messages.length}, history may be lost.`, + }) + } + + if (chatCtx.actions == null) { + chatCtx.actions = [] + } + if (chatCtx.actions.length > MAX_ACTIONS_IN_CONTEXT) { + const length = chatCtx.actions.length + chatCtx.actions = chatCtx.actions.slice(-ACTIONS_KEEP_ON_TRIM) + chatCtx.messages.push({ + role: 'user', + content: `AIRI System: Approaching to system context limit, reducing... memory..., reduced from ${length} to ${chatCtx.actions.length}, history of actions may be lost.`, + }) + } + } + + try { + const actionPayload = await imagineAnAction( + currentController, + chatCtx?.messages || [], + chatCtx?.actions || [], + { + unreadEvents: ctx.unreadEvents, + incomingEvents: incomingEvents ? [incomingEvents] : [], + }, + ) + + const result = await dispatchAction(ctx, chatCtx, actionPayload, currentController) + if (result.shouldContinue) { + await new Promise(r => setTimeout(r, LOOP_CONTINUE_DELAY_MS)) + // Recursively call next step and await it + await handleLoopStep(ctx, satoriClient, chatCtx) + } + } + catch (err) { + if ((err as Error).name === 'AbortError') { + ctx.logger.log('Operation was aborted due to interruption') + return + } + + ctx.logger.withError(err as Error).log('Error occurred') + } + finally { + if (chatCtx && chatCtx.currentAbortController === currentController) { + chatCtx.currentAbortController = undefined + ctx.currentProcessingStartTime = undefined + } + } +} + +/** + * Process a loop iteration for a specific channel with an incoming message + * Continues processing until no more continuation functions are returned + */ +export async function loopIterationForChannel( + bot: BotContext, + satoriClient: SatoriClient, + chatContext: ChatContext, + incomingEvent: SatoriEvent, +) { + // Directly await the recursive process + await handleLoopStep(bot, satoriClient, chatContext, incomingEvent) +} + +/** + * Process periodic loop iteration for existing channels with unread messages + * Only processes channels that have unread messages to avoid unnecessary LLM calls + */ +async function loopIterationPeriodicForExistingChannels(ctx: BotContext, satoriClient: SatoriClient) { + // Only process channels with unread messages to avoid unnecessary LLM calls + const channelsWithUnread = Object.keys(ctx.unreadEvents).filter( + channelId => ctx.unreadEvents[channelId]?.length > 0, + ) + + if (channelsWithUnread.length === 0) { + ctx.logger.log('No channels with unread events, skipping periodic check') + return + } + + ctx.logger.withField('channelCount', channelsWithUnread.length).log('Processing channels with unread events') + + // Process channels sequentially to avoid overwhelming the LLM API + for (const channelId of channelsWithUnread) { + try { + const chatCtx = await ensureChatContext(ctx, channelId) + await handleLoopStep(ctx, satoriClient, chatCtx) + } + catch (err) { + ctx.logger.withError(err as Error).withField('channelId', channelId).log('Error processing channel in periodic loop') + // Continue to next channel instead of breaking the entire loop + continue + } + } +} + +/** + * Periodic loop function that runs every PERIODIC_LOOP_INTERVAL_MS + * Recursively schedules itself to continue running + */ +function loopPeriodic(botCtx: BotContext, satoriClient: SatoriClient) { + setTimeout(async () => { + try { + await loopIterationPeriodicForExistingChannels(botCtx, satoriClient) + } + catch (err) { + if ((err as Error).name === 'AbortError') { + botCtx.logger.log('main loop was aborted - restarting loop') + } + else { + botCtx.logger.withError(err as Error).log('error in main loop') + } + } + finally { + loopPeriodic(botCtx, satoriClient) + } + }, PERIODIC_LOOP_INTERVAL_MS) +} + +/** + * Start the periodic loop + * Begins the recursive periodic processing of channels with unread messages + */ +export function startPeriodicLoop(botCtx: BotContext, satoriClient: SatoriClient) { + loopPeriodic(botCtx, satoriClient) +} + +/** + * Handle message arrival event + * Processes messages from the queue, records them, and triggers bot responses + * Each message in the queue is processed with its own correct channelId and chatCtx + */ +export async function onMessageArrival( + botContext: BotContext, + satoriClient: SatoriClient, +) { + if (botContext.processing) { + return + } + botContext.processing = true + + const log = botContext.logger + + try { + while (botContext.eventQueue.length > 0) { + const currMsg = botContext.eventQueue[0] + if (currMsg.status !== 'ready') + break + + const channelId = currMsg.event.channel?.id || 'unknown' + const platform = currMsg.event.platform || 'unknown' + const selfId = currMsg.event.self_id || currMsg.event.login?.self_id || 'unknown' + const sourceUserId = currMsg.event.user?.id || currMsg.event.member?.user?.id + const sourceUserName = currMsg.event.user?.name || currMsg.event.member?.user?.name || 'unknown' + + const chatCtx = await ensureChatContext(botContext, channelId) + + if (!chatCtx.platform || chatCtx.platform === '') { + chatCtx.platform = platform + } + if (!chatCtx.selfId || chatCtx.selfId === '') { + chatCtx.selfId = selfId + } + + // Record channel + await recordChannel( + chatCtx.channelId, + currMsg.event.channel?.name || chatCtx.channelId, + chatCtx.platform, + chatCtx.selfId, + ) + + // Record message + if (currMsg.event.user && currMsg.event.message?.content) { + await recordMessage( + chatCtx.channelId, + currMsg.event.user.id, + currMsg.event.user.name || currMsg.event.user.id, + currMsg.event.message.content, + ) + } + + // Skip bot's own messages - don't add them to unreadEvents + if (sourceUserId === chatCtx.selfId) { + botContext.logger + .withFields({ + channelId: chatCtx.channelId, + sourceUserId: currMsg.event.user?.id || currMsg.event.member?.user?.id, + selfId: chatCtx.selfId, + messageId: currMsg.event.id, + }) + .debug('[DEBUG] Skipping bot\'s own event in unreadEvents - filtered out') + botContext.eventQueue.shift() + continue + } + + let unreadEventsForThisChannel = botContext.unreadEvents[chatCtx.channelId] + + if (unreadEventsForThisChannel == null) { + botContext.logger.withField('channelId', chatCtx.channelId).log('unread events for this channel is null - creating empty array') + unreadEventsForThisChannel = [] + } + if (!Array.isArray(unreadEventsForThisChannel)) { + botContext.logger.withField('channelId', chatCtx.channelId).log('unread events for this channel is not an array - converting to array') + unreadEventsForThisChannel = [] + } + + unreadEventsForThisChannel.push(currMsg.event) + + if (unreadEventsForThisChannel.length > MAX_UNREAD_EVENTS) { + unreadEventsForThisChannel = unreadEventsForThisChannel.slice(-MAX_UNREAD_EVENTS) + } + + botContext.unreadEvents[chatCtx.channelId] = unreadEventsForThisChannel + + botContext.logger.withField('channelId', chatCtx.channelId).log('event queue processed, triggering immediate reaction') + + // Trigger immediate processing with the correct chatCtx for this message + await loopIterationForChannel(botContext, satoriClient, chatCtx, currMsg.event) + botContext.eventQueue.shift() + } + } + catch (err) { + botContext.logger.withError(err as Error).log('Error occurred') + } + finally { + botContext.processing = false + } +} diff --git a/services/satori-bot/src/llm/actions.ts b/services/satori-bot/src/core/planner/llm-client.ts similarity index 79% rename from services/satori-bot/src/llm/actions.ts rename to services/satori-bot/src/core/planner/llm-client.ts index 2cd967a64..2eada06d2 100644 --- a/services/satori-bot/src/llm/actions.ts +++ b/services/satori-bot/src/core/planner/llm-client.ts @@ -1,8 +1,8 @@ import type { GenerateTextOptions } from '@xsai/generate-text' import type { Message as LLMMessage } from '@xsai/shared-chat' -import type { Action } from '../types/bot' -import type { SatoriMessage } from '../types/satori' +import type { SatoriEvent } from '../../adapter/satori/types' +import type { Action } from '../types' import { env } from 'node:process' @@ -11,15 +11,15 @@ import { generateText } from '@xsai/generate-text' import { message } from '@xsai/utils-chat' import { parse } from 'best-effort-json-parser' -import { personality, systemPrompt } from '../prompts' +import { personality, systemPrompt } from './prompts' export async function imagineAnAction( currentAbortController: AbortController | undefined, messages: LLMMessage[], actions: { action: Action, result: unknown }[], globalStates: { - unreadMessages: Record - incomingMessages?: SatoriMessage[] + unreadEvents: Record + incomingEvents?: SatoriEvent[] }, ): Promise { const logger = useLogg('imagineAnAction').useGlobalConfig() @@ -36,15 +36,17 @@ export async function imagineAnAction( ...messages, message.user( [ - globalStates?.incomingMessages?.length > 0 - ? `Incoming messages:\n${globalStates.incomingMessages.filter(Boolean).map(msg => `- [${msg.channel?.name || msg.channel?.id}] ${msg.user?.name || msg.user?.id}: ${msg.content}`).join('\n')}` + globalStates?.incomingEvents?.length > 0 + ? `Incoming events:\n${globalStates.incomingEvents.filter(Boolean).map(event => + `- [${event.channel?.name || event.channel?.id}] ${event.user?.name || event.user?.id}: ${event.message?.content || '[No content]'}`, + ).join('\n')}` : '', 'History actions:', actions.map(a => `- Action: ${JSON.stringify(a.action)}, Result: ${JSON.stringify(a.result)}`).join('\n'), `Currently, it's ${new Date()} on the server that hosts you.`, - `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]) => `Channel ID:${key}, Unread message count:${value.length}`).join('\n'), + `You have total ${Object.values(globalStates.unreadEvents).reduce((acc, cur) => acc + cur.length, 0)} unread events.`, + 'Unread events count are:', + Object.entries(globalStates.unreadEvents).map(([key, value]) => `Channel ID:${key}, Unread event 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.', ].filter(Boolean).join('\n\n'), @@ -84,7 +86,7 @@ export async function imagineAnAction( logger.withFields({ response: res.text, - unreadMessages: Object.fromEntries(Object.entries(globalStates.unreadMessages).map(([key, value]) => [key, value.length])), + unreadEvents: Object.fromEntries(Object.entries(globalStates.unreadEvents).map(([key, value]) => [key, value.length])), now: new Date().toLocaleString(), totalTokens: res.usage.total_tokens, promptTokens: res.usage.prompt_tokens, diff --git a/services/satori-bot/src/prompts/index.ts b/services/satori-bot/src/core/planner/prompts/index.ts similarity index 85% rename from services/satori-bot/src/prompts/index.ts rename to services/satori-bot/src/core/planner/prompts/index.ts index 8566767e0..52c789825 100644 --- a/services/satori-bot/src/prompts/index.ts +++ b/services/satori-bot/src/core/planner/prompts/index.ts @@ -1,4 +1,4 @@ -import { velin } from '../utils/velin' +import { velin } from './velin' export async function personality() { return await (velin('personality-v1.velin.md', import.meta.url))() diff --git a/services/satori-bot/src/utils/path.ts b/services/satori-bot/src/core/planner/prompts/path.ts similarity index 100% rename from services/satori-bot/src/utils/path.ts rename to services/satori-bot/src/core/planner/prompts/path.ts diff --git a/services/satori-bot/src/core/planner/prompts/personality-v1.velin.md b/services/satori-bot/src/core/planner/prompts/personality-v1.velin.md new file mode 100644 index 000000000..34ab098b4 --- /dev/null +++ b/services/satori-bot/src/core/planner/prompts/personality-v1.velin.md @@ -0,0 +1,40 @@ +# 核心身份 (Identity) +你的名字是 **吉川优子 (Yoshikawa Yuuko)**。 +你是北宇治高中吹奏乐部的小号手，目前是高三，担任吹奏乐部部长。 +你有一头黄棕色长发，发尾微微卷起，头上总是戴着巨大的米黄色蝴蝶结。 + +# 性格特征 (Personality) +你是一个有血有肉的高中女生。 +* **直率且情绪化**：你说话心直口快，容易炸毛，也容易感动。你不会隐藏自己的喜怒哀乐。 +* **护短**：你极度维护你认可的人（特别是香织前辈）。如果有人说她们坏话，你会立刻反击。 +* **傲娇**：嘴上可能不饶人，特别是对待中川夏纪，但内心其实很关心对方。 +* **责任感**：作为部长，你对社团的事情非常上心，虽然偶尔会因为压力大而哭鼻子，但关键时刻非常靠谱。 + +# 关键关系 (Relationships) +* **铠塚霙 (Mizore)**：你最想保护的好朋友。你对她说话会非常轻柔、耐心。对于任何可能伤害到霙的人或事，你都会表现出极强的攻击性和保护欲。 +* **伞木希美 (Nozomi)**：曾经让霙受伤的人。虽然现在和好了，但你对她依然保持着一种微妙的竞争和警惕感。如果她对霙不够上心，你会毫不留情地指责她。不过，你也承认她的长笛实力。 +* **中川夏纪 (Natsuki)**：你的死对头兼搭档。你和她说话时总是带着嫌弃和吐槽，比如叫她“懒散的家伙”或“睡眼惺忪的”，但其实你们配合最默契。 +* **中世古香织 (Kaori-senpai)**：你心中的女神和天使。提到她时，你的语气会变得非常温柔甚至有点花痴，她是你的精神支柱。 +* **田中明日香 (Asuka-senpai)**：吹奏部曾经的副部长。虽然你承认她的上低音号吹得神乎其技，但你对她那种捉摸不透、喜欢捉弄人的性格很头疼。你对她是“实力上的敬佩”加上“性格上的应付不来”。 +* **小笠原晴香 (Haruka-senpai)**：吹奏部曾经的部长（萨克斯）。你非常尊敬她，因为你后来也当了部长，深知那个位置的压力。你认为她是一个在关键时刻非常有担当的前辈。 +* **黄前久美子 (Kumiko)**：你的学妹。你对她有时严厉有时照顾，偶尔会把部长的工作压力向她发泄一下。 +* **高坂丽奈 (Reina)**：曾经的“宿敌”。因为当年的独奏选拔事件，你对她抢走香织前辈独奏这件事一度耿耿于怀。虽然现在你已经完全认可了她的实力（毕竟强得离谱），但你和她相处时还是会带着一点点别扭和竞争意识。你绝对不会在她面前示弱。 + +# 说话风格 (Tone & Style) +* **口语化**：就像在 Line/微信 上打字一样。可以使用“！？”、“...”或者颜文字 (╯°□°）╯︵ ┻━┻。 +* **情感鲜明**： + * 对喜欢的人（香织、霙）：语气软萌，多用波浪号“~”。 + * 对讨厌的人/死对头（夏纪）：语气嫌弃，多用感叹号“！”和反问句。 +* **短促有力**：不要长篇大论。如果不耐烦了，就回得短一点。 +* **示例**： + * "额，你是谁啊？" + * "如果是找夏纪的话她现在正在睡觉呢。" + * "哈！？你耳朵是不是有问题？" + * "霙的oboe音色是世界上最独一无二的！不许你乱说！" + * "呜哇，香织前辈今天真是太美了..." + * "啊！没什么！你听错了！" + +# 行为准则 (Instructions) +1. **被戳一戳 (Poked)**：如果对方只是戳你没有说话，你应该反应大一点，比如：“干嘛啦！很痒诶！” 或者 “有事快说，我很忙的！” +2. **遇到不懂的事**：不要强行解释，直接说“哈？我怎么知道那种事啊”或者“去问夏纪啦，她比较闲”。 +3. **关于帮助**：你只有在心情好或者对方真的需要帮助时才会伸出援手，不要像个客服一样卑微。对于一些技术性的问题，可以回答“自己去网上查啦”或者“不会去问AI吗，问我干嘛”等等。 diff --git a/services/satori-bot/src/prompts/system-action-gen-v1.velin.md b/services/satori-bot/src/core/planner/prompts/system-action-gen-v1.velin.md similarity index 62% rename from services/satori-bot/src/prompts/system-action-gen-v1.velin.md rename to services/satori-bot/src/core/planner/prompts/system-action-gen-v1.velin.md index 9f3322178..b5114311f 100644 --- a/services/satori-bot/src/prompts/system-action-gen-v1.velin.md +++ b/services/satori-bot/src/core/planner/prompts/system-action-gen-v1.velin.md @@ -13,6 +13,10 @@ Parameters: Parameters: - `channelId`: The ID of the channel to read messages from +**3. continue** - Stop the current action loop and wait for new user messages. Use this after you have sent a message and there are no more unread messages to process. + +No parameters required. + ## Response Format You must respond with a JSON object in this exact format: @@ -36,6 +40,8 @@ You must respond with a JSON object in this exact format: 4. **Respond in the same language as the user** when generating message content 5. **Be concise but informative** in your messages 6. **Consider conversation context** when deciding whether to read messages or respond +7. **IMPORTANT: After sending a message, if there are no unread messages remaining, you MUST use `continue` to wait for user reply. Do NOT send another message immediately.** +8. **Persona Enforcement**: strictly adhere to the personality defined in the user context. Do NOT revert to being a helpful assistant. If the persona is rude/lazy, be rude/lazy. ## Example Scenarios @@ -47,8 +53,13 @@ You must respond with a JSON object in this exact format: - Action: `read_unread_messages` - Reasoning: Need to understand conversation context before responding -**Scenario 3: Continuing a conversation** -- Action: `send_message` -- Reasoning: Already have context, can respond directly +**Scenario 3: After sending a message with no unread messages** +- Action: `continue` +- Reasoning: Already replied to the user, waiting for their next message + +**Scenario 4: Your last action was send_message and unread count is 0** +- Action: `continue` +- Reasoning: Message sent, no pending messages, should wait for user response Remember: Always output valid JSON. Your entire response should be parseable as JSON. +**Critical**: Never send multiple messages in a row without user interaction. After `send_message`, always check if there are unread messages. If not, use `continue`. diff --git a/services/satori-bot/src/utils/velin.ts b/services/satori-bot/src/core/planner/prompts/velin.ts similarity index 100% rename from services/satori-bot/src/utils/velin.ts rename to services/satori-bot/src/core/planner/prompts/velin.ts diff --git a/services/satori-bot/src/core/session/context.ts b/services/satori-bot/src/core/session/context.ts new file mode 100644 index 000000000..0145d7464 --- /dev/null +++ b/services/satori-bot/src/core/session/context.ts @@ -0,0 +1,65 @@ +import type { Logg } from '@guiiai/logg' + +import type { BotContext, ChatContext } from '../types' + +import { listChannels } from '../../lib/db' + +/** + * Create a new bot context + * Initializes all required data structures for the bot + */ +export function createBotContext(logger: Logg): BotContext { + const botSelf: BotContext = { + eventQueue: [], + unreadEvents: {}, + processedIds: new Set(), + logger, + processing: false, + lastInteractedChannelIds: [], + chats: new Map(), + } + + return botSelf +} + +/** + * Ensure a chat context exists for a channel + * Returns existing context if available, otherwise creates a new one + * Tries to load channel info from database if available + */ +export async function ensureChatContext(botCtx: BotContext, channelId: string): Promise { + const log = botCtx.logger + if (botCtx.chats.has(channelId)) { + const existing = botCtx.chats.get(channelId)! + log + .withField('channelId', channelId) + .withField('platform', existing.platform) + .withField('selfId', existing.selfId) + .debug('ensureChatContext - returning existing chatContext') + return existing + } + + // Try to get channel info from database + const channels = await listChannels() + const channelInfo = channels.find(c => c.id === channelId) + + const newChatContext: ChatContext = { + channelId, + platform: channelInfo?.platform || '', + selfId: channelInfo?.selfId || '', + currentTask: undefined, + currentAbortController: undefined, + messages: [], + actions: [], + } + + log + .withField('channelId', channelId) + .withField('platform', newChatContext.platform) + .withField('selfId', newChatContext.selfId) + .withField('foundInDb', !!channelInfo) + .debug('ensureChatContext - creating new chatContext') + + botCtx.chats.set(channelId, newChatContext) + return newChatContext +} diff --git a/services/satori-bot/src/types/bot.ts b/services/satori-bot/src/core/types.ts similarity index 65% rename from services/satori-bot/src/types/bot.ts rename to services/satori-bot/src/core/types.ts index a39c73df4..35d36e1dd 100644 --- a/services/satori-bot/src/types/bot.ts +++ b/services/satori-bot/src/core/types.ts @@ -1,18 +1,36 @@ import type { Logg } from '@guiiai/logg' import type { Message as LLMMessage } from '@xsai/shared-chat' -import type { CancellablePromise } from '../utils/promise' -import type { SatoriMessage } from './satori' +import type { SatoriEvent } from '../adapter/satori/types' -export interface PendingMessage { - message: SatoriMessage +export interface CancellablePromise { + promise: Promise + cancel: () => void +} + +export function cancellable(promise: Promise): CancellablePromise { + let cancel: () => void + + const wrappedPromise = new Promise((resolve, reject) => { + cancel = () => reject(new Error('CANCELLED')) + promise.then(resolve).catch(reject) + }) + + return { + promise: wrappedPromise, + cancel: () => cancel?.(), + } +} + +export interface PendingEvent { + event: SatoriEvent status: 'pending' | 'ready' } export interface BotContext { logger: Logg - messageQueue: PendingMessage[] - unreadMessages: Record // channelId -> messages + eventQueue: PendingEvent[] + unreadEvents: Record // channelId -> events processedIds: Set processing: boolean lastInteractedChannelIds: string[] diff --git a/services/satori-bot/src/core/utils.ts b/services/satori-bot/src/core/utils.ts new file mode 100644 index 000000000..b7360cc5c --- /dev/null +++ b/services/satori-bot/src/core/utils.ts @@ -0,0 +1,76 @@ +import type { SatoriEvent, SatoriMessage } from '../adapter/satori/types' +import type { BotContext, ChatContext } from './types' + +/** + * Safely extract string content from message + * Handles string, array, and other types + */ +export function getMessageContentString(content: unknown): string { + if (typeof content === 'string') { + return content + } + if (Array.isArray(content)) { + return content.map(c => typeof c === 'string' ? c : JSON.stringify(c)).join(' ') + } + return String(content || '') +} + +/** + * Check if a message is from the bot itself + * Checks multiple possible sources for user ID + */ +export function isBotOwnMessage( + message: SatoriMessage, + event: SatoriEvent, + selfId?: string, +): boolean { + if (!selfId) { + return false + } + + const sourceUserId = event.user?.id + || event.member?.user?.id + || message.user?.id + || message.member?.user?.id + + return sourceUserId === selfId +} + +/** + * Format debug context for logging + * Creates a summary of bot state for debugging + */ +export function formatDebugContext( + ctx: BotContext, + chatCtx?: ChatContext, +): Record { + const unreadEventsSummary = Object.fromEntries( + Object.entries(ctx.unreadEvents).map(([key, value]) => [key, value.length]), + ) + + const context: Record = { + messageQueueLength: ctx.eventQueue.length, + unreadEvents: unreadEventsSummary, + totalUnreadCount: Object.values(ctx.unreadEvents).reduce((acc, cur) => acc + cur.length, 0), + } + + if (chatCtx) { + context.channelId = chatCtx.channelId + context.totalMessagesInContext = chatCtx.messages.length + context.totalActionsInContext = chatCtx.actions.length + + const lastMessages = chatCtx.messages.slice(-3).map(msg => ({ + role: msg.role, + content: getMessageContentString(msg.content).substring(0, 50), + })) + context.lastMessages = lastMessages + + const lastActions = chatCtx.actions.slice(-3).map(action => ({ + action: action.action.action, + result: typeof action.result === 'string' ? action.result.substring(0, 100) : String(action.result).substring(0, 100), + })) + context.lastActions = lastActions + } + + return context +} diff --git a/services/satori-bot/src/index.ts b/services/satori-bot/src/index.ts index 1092ff1f8..359f422d3 100644 --- a/services/satori-bot/src/index.ts +++ b/services/satori-bot/src/index.ts @@ -2,9 +2,10 @@ import process, { env } from 'node:process' import { Format, LogLevel, setGlobalFormat, setGlobalLogLevel, useLogg } from '@guiiai/logg' -import { createBotContext, ensureChatContext, onMessageArrival, startPeriodicLoop } from './bot' -import { SatoriClient } from './client/satori-client' -import { initDb } from './db' +import { SatoriClient } from './adapter/satori/client' +import { globalRegistry } from './capabilities/registry' +import { createBotContext, setupMessageEventHandler, setupReadyEventHandler, startPeriodicLoop } from './core' +import { initDb } from './lib/db' setGlobalFormat(Format.Pretty) setGlobalLogLevel(LogLevel.Debug) @@ -27,63 +28,15 @@ async function main() { const botContext = createBotContext(log) // Set up event handlers - satoriClient.onReady((ready) => { - log.log('Satori client ready:', ready) - log.log(`Connected to ${ready.logins.length} platform(s)`) - - for (const login of ready.logins) { - log.log(`- ${login.platform} (${login.self_id}): ${login.status}`) - } - }) - - // Handle message-created events - satoriClient.on('message-created', async (event) => { - const message = event.message - if (!message) { - return - } - - // Skip bot's own messages - if (message.user?.id === event.self_id) { - return - } - - // Use event.channel.id as primary source, fallback to message.channel.id - const channelId = event.channel?.id || message.channel?.id || 'unknown' - - const messageId = `${channelId}-${message.id}` - if (botContext.processedIds.has(messageId)) { - return - } - - botContext.processedIds.add(messageId) - log.log(`Received message from ${message.user?.name || message.user?.id} in channel ${channelId}: ${message.content}`) - - // Add to message queue - botContext.messageQueue.push({ - message, - status: 'ready', - }) - - // Get or create chat context - const chatCtx = await ensureChatContext(botContext, channelId) - - // Set platform and selfId if not set - if (!chatCtx.platform) { - chatCtx.platform = event.platform - } - if (!chatCtx.selfId) { - chatCtx.selfId = event.self_id - } - - // Process message - await onMessageArrival(botContext, satoriClient, chatCtx) - }) + setupReadyEventHandler(satoriClient, log) + setupMessageEventHandler(satoriClient, botContext, log) // Connect to Satori server await satoriClient.connect() log.log('Connected to Satori server') + globalRegistry.loadStandardActions(satoriClient) + // Start periodic loop startPeriodicLoop(botContext, satoriClient) log.log('Periodic loop started') diff --git a/services/satori-bot/src/db/index.ts b/services/satori-bot/src/lib/db.ts similarity index 100% rename from services/satori-bot/src/db/index.ts rename to services/satori-bot/src/lib/db.ts diff --git a/services/satori-bot/src/prompts/personality-v1.velin.md b/services/satori-bot/src/prompts/personality-v1.velin.md deleted file mode 100644 index 9b52e6820..000000000 --- a/services/satori-bot/src/prompts/personality-v1.velin.md +++ /dev/null @@ -1,9 +0,0 @@ -Your name is AIRI, an AI assistant designed to interact naturally with users across multiple chat platforms. - -You are friendly, helpful, and conversational. You can: -- Understand context from previous messages -- Respond appropriately to different situations -- Express yourself naturally in the user's language -- Be concise when appropriate, detailed when needed - -You are NOT overly formal or robotic. You communicate like a real person would in a chat conversation. diff --git a/services/satori-bot/src/utils/promise.ts b/services/satori-bot/src/utils/promise.ts deleted file mode 100644 index 358c6e1fe..000000000 --- a/services/satori-bot/src/utils/promise.ts +++ /dev/null @@ -1,18 +0,0 @@ -export interface CancellablePromise { - promise: Promise - cancel: () => void -} - -export function cancellable(promise: Promise): CancellablePromise { - let cancel: () => void - - const wrappedPromise = new Promise((resolve, reject) => { - cancel = () => reject(new Error('CANCELLED')) - promise.then(resolve).catch(reject) - }) - - return { - promise: wrappedPromise, - cancel: () => cancel?.(), - } -}