中间件

Author: linyuan.yang

packages/sbot/src/Middleware/MiddlewarePipeline.ts

@@ -0,0 +1,36 @@
+export type Middleware<T> = (ctx: T, next: () => Promise<void>) => Promise<void>;
+
+export class MiddlewarePipeline<T> {
+    private stack: Middleware<T>[] = [];
+
+    use(fn: Middleware<T>): this {
+        this.stack.push(fn);
+        return this;
+    }
+
+    prepend(fn: Middleware<T>): this {
+        this.stack.unshift(fn);
+        return this;
+    }
+
+    remove(fn: Middleware<T>): boolean {
+        const idx = this.stack.indexOf(fn);
+        if (idx === -1) return false;
+        this.stack.splice(idx, 1);
+        return true;
+    }
+
+    async execute(ctx: T, final: (ctx: T) => Promise<void>): Promise<void> {
+        let index = -1;
+        const dispatch = async (i: number): Promise<void> => {
+            if (i <= index) throw new Error('next() called multiple times');
+            index = i;
+            if (i < this.stack.length) {
+                await this.stack[i](ctx, () => dispatch(i + 1));
+            } else {
+                await final(ctx);
+            }
+        };
+        await dispatch(0);
+    }
+}

packages/sbot/src/Middleware/index.ts

@@ -0,0 +1,3 @@
+export { MiddlewarePipeline, type Middleware } from './MiddlewarePipeline';
+export type { MessageContext, AIContext } from './types';
+export { intentFilterMiddleware } from './intentFilter';

packages/sbot/src/Middleware/intentFilter.ts

@@ -0,0 +1,22 @@
+import type { Middleware } from "./MiddlewarePipeline";
+import type { MessageContext } from "./types";
+import { getChannelSession } from "../Core/Database";
+import { config } from "../Core/Config";
+import { classifyIntent } from "../Processing/classifyIntent";
+
+export const intentFilterMiddleware: Middleware<MessageContext> = async (ctx, next) => {
+    if (ctx.args.mentionBot) return next();
+
+    const dbSession = await getChannelSession(ctx.args.dbSessionId);
+    const channel = ctx.args.channelId ? config.getChannel(ctx.args.channelId) : undefined;
+    const intentModel = dbSession?.intentModel != null ? dbSession.intentModel : channel?.intentModel;
+    if (!intentModel) return next();
+
+    const intentPrompt = dbSession?.intentPrompt != null ? dbSession.intentPrompt : (channel?.intentPrompt ?? null);
+    const intentThreshold = dbSession?.intentThreshold != null ? dbSession.intentThreshold : (channel?.intentThreshold ?? 0.7);
+
+    if (await classifyIntent(ctx.query, intentModel, intentPrompt, intentThreshold, ctx.threadId)) {
+        return next();
+    }
+    ctx.filtered = true;
+};

packages/sbot/src/Middleware/types.ts

@@ -0,0 +1,19 @@
+import type { MessageContent } from "scorpio.ai";
+import type { ChannelMessageArgs, ChannelSessionHandler } from "channel.base";
+import type { ChannelRouteArgs } from "../Session/SessionManager";
+import type { AgentRunOptions } from "../Agent/AgentRunner";
+
+export interface MessageContext {
+    query: MessageContent;
+    args: ChannelRouteArgs;
+    threadId: string;
+    filtered: boolean;
+}
+
+export interface AIContext {
+    query: MessageContent;
+    args: ChannelMessageArgs;
+    sessionHandler: ChannelSessionHandler;
+    runOptions: Partial<AgentRunOptions>;
+    metadata: Record<string, any>;
+}

packages/sbot/src/Session/SessionManager.ts

@@ -7,12 +7,14 @@ import { config } from "../Core/Config";
 import { ChannelSessionRow, getChannelSession } from "../Core/Database";
 import { channelManager } from "../Channel/ChannelManager";
 import { createProcessAIHandler } from "../Processing/createProcessAIHandler";
-import { classifyIntent } from "../Processing/classifyIntent";
+import { MiddlewarePipeline } from "../Middleware/MiddlewarePipeline";
+import { intentFilterMiddleware } from "../Middleware/intentFilter";
+import type { MessageContext } from "../Middleware/types";
 
 import { getBuiltInCommands } from "./BuiltInCommands";
 import { WebSocketSessionHandler } from "../Channel/web/WebSocketSessionHandler";
 
-interface ChannelRouteArgs extends ChannelMessageArgs {
+export interface ChannelRouteArgs extends ChannelMessageArgs {
     channelType: string;
     channelId: string;
     dbSessionId: number;
@@ -148,9 +150,11 @@ function mergeMessageContents(items: { query: MessageContent }[]): MessageConten
 
 export class SbotSessionManager extends SessionManager {
     private mergeBuffers = new Map<string, MergeBufferEntry>();
+    readonly messagePipeline = new MiddlewarePipeline<MessageContext>();
 
     constructor() {
         super();
+        this.messagePipeline.use(intentFilterMiddleware);
     }
 
     protected createSession(threadId: string): SessionService {
@@ -216,20 +220,12 @@ export class SbotSessionManager extends SessionManager {
     }
 
     private async dispatchToSession(threadId: string, query: MessageContent, args: ChannelRouteArgs): Promise<void> {
-        if (!await this.passIntentFilter(query, args, threadId)) return;
-        const session = this.getOrCreate(threadId);
-        await session.onReceiveMessage(query, args);
-    }
-
-    private async passIntentFilter(query: MessageContent, args: ChannelRouteArgs, threadId: string): Promise<boolean> {
-        if (args?.mentionBot) return true;
-        const dbSession = await getChannelSession(args?.dbSessionId);
-        const channel = args.channelId ? config.getChannel(args.channelId) : undefined;
-        const intentModel = dbSession?.intentModel != null ? dbSession.intentModel : channel?.intentModel;
-        if (!intentModel) return true;
-        const intentPrompt = dbSession?.intentPrompt != null ? dbSession.intentPrompt : (channel?.intentPrompt ?? null);
-        const intentThreshold = dbSession?.intentThreshold != null ? dbSession.intentThreshold : (channel?.intentThreshold ?? 0.7);
-        return classifyIntent(query, intentModel, intentPrompt, intentThreshold, threadId);
+        const ctx: MessageContext = { query, args, threadId, filtered: false };
+        await this.messagePipeline.execute(ctx, async (c) => {
+            if (c.filtered) return;
+            const session = this.getOrCreate(c.threadId);
+            await session.onReceiveMessage(c.query, c.args);
+        });
     }
 
     async onReceiveWebMessage(threadId: string, query: MessageContent, sessionId: string, dbSessionId: number): Promise<void> {