个人 AI 助手架构全揭秘：Monorepo + 插件系统 + 多通道网关设计，20 万行代码如何管理 6 个 IM 平台？

上一篇文章带大家体验了一个开源个人 AI 助手平台。很多人问：它到底是怎么设计的？为什么能同时接微信、企业 IM、社交平台，还能保持代码不失控？

答案藏在一个优雅的架构里——今天我们来拆解它。

一、项目整体概览

1.1 代码规模

src/
├── packages/ # 所有功能包
│ ├── core/ # 核心引擎
│ ├── gateway/ # 消息网关
│ ├── channels/ # 通道适配器
│ │ ├── wechat/
│ │ ├── dingtalk/
│ │ ├── telegram/
│ │ ├── slack/
│ │ ├── email/
│ │ └── sms/
│ ├── ai/ # AI 引擎
│ ├── plugins/ # 插件系统
│ ├── storage/ # 存储层
│ ├── scheduler/ # 定时任务
│ └── web/ # 管理后台
├── apps/ # 可部署的应用
│ ├── server/ # 主服务
│ └── cli/ # 命令行工具
├── config/ # 统一配置
├── scripts/ # 构建/部署脚本
└── pnpm-workspace.yaml # Monorepo 管理

总代码量约 20 万行 TypeScript，65 个 npm 包通过 pnpm workspace 联动。

1.2 Monorepo 设计思路

为什么选 Monorepo 而不是多仓库？

多仓库模式：
gateway-repo ──→ npm: @demo/gateway@1.2.0
ai-repo ──────→ npm: @demo/ai@0.9.0
wechat-repo ──→ npm: @demo/channel-wechat@0.5.0
…

── 问题：改一个接口要发 4 个包、更新 4 个仓库

Monorepo 模式：
demo-ai-assistant/
├── packages/gateway/ ←── 直接 import from "../ai"
├── packages/ai/ ←── 直接 import from "../core"
├── packages/channels/
└── packages/core/

── 改一个接口，所有引用方同时更新，一次 CI 验证

二、核心架构图

┌──────────────────────────────────────────────────────────────┐
│ 管理后台 (Web UI) │
│ http://localhost:3000/dashboard │
└────────────────────────────┬─────────────────────────────────┘
│ REST API
┌────────────────────────────┴─────────────────────────────────┐
│ 核心调度层 (Core Engine) │
│ │
│ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │
│ │ MessageBus │ │ PluginMgr │ │ ScheduleMgr │ │
│ │ │ │ │ │ │ │
│ │ 消息路由分发 │ │ 插件生命周期 │ │ 定时任务 │ │
│ │ pub/sub │ │ 热加载/卸载 │ │ Cron 调度 │ │
│ └───────┬───────┘ └───────┬───────┘ └───────┬───────┘ │
│ │ │ │ │
│ ┌───────┴──────────────────┴───────────────────┴───────┐ │
│ │ Pipeline Engine │ │
│ │ │ │
│ │ Message → [PreProcess] → [Route] → [AI/Plugin] │ │
│ │ → [PostProcess] → [Reply] │ │
│ └───────────────────────────────────────────────────────┘ │
└───────────────────────────────────────────────────────────────┘
│ │ │
┌─────┴────┐ ┌─────┴─────┐ ┌──────┴──────┐
│ Channels │ │ AI Engine │ │ Storage │
│ 适配层 │ │ │ │ │
│ │ │ LLM 调用 │ │ SQLite/ │
│ 微信 │ │ 上下文 │ │ Redis │
│ 钉钉 │ │ RAG │ │ VectorDB │
│ Telegram │ │ 多模型 │ │ │
└──────────┘ └───────────┘ └─────────────┘

三、消息网关——多通道统一接入

这是整个平台最精妙的部分。不同 IM 的 SDK/API 千差万别，怎么把它们抽象成统一接口？

3.1 通道适配器接口

// packages/gateway/src/channel-adapter.ts

/**
* 所有消息通道必须实现的接口
* 每种 IM 平台只要实现这个接口，就能被网关识别
*/
export interface ChannelAdapter {
// === 生命周期 ===
/** 通道名称，全局唯一 */
readonly name: string;

/** 初始化（登录、建连） */
connect(config: ChannelConfig): Promise<void>;

/** 断开连接 */
disconnect(): Promise<void>;

/** 是否在线 */
isOnline(): boolean;

// === 消息收发 ===
/** 发送消息（统一出入口） */
send(target: MessageTarget, content: MessageContent): Promise<SendResult>;

/** 接收消息的回调注册 */
onMessage(handler: MessageHandler): void;

// === 通道特有 ===
/** 通道支持的消息类型 */
supportedTypes(): MessageType[];

/** 通道元信息 */
getMetadata(): ChannelMetadata;
}

3.2 微信通道实现

// packages/channels/wechat/src/wechat-adapter.ts
import type { ChannelAdapter, Message, MessageHandler } from '@demo/gateway';

export class WeChatAdapter implements ChannelAdapter {
readonly name = 'wechat';
private bot: WeChatBot | null = null;
private messageHandlers: MessageHandler[] = [];

async connect(config: WeChatConfig): Promise<void> {
// WeChat Bot SDK 初始化
this.bot = new WeChatBot({
name: config.botName || 'demo-bot',
// 扫码登录
onLogin: (user) => {
console.log(`微信登录成功: ${user.name()}`);
},
// 接收消息的统一入口
onMessage: async (rawMsg: WeChatMessage) => {
await this.handleIncomingMessage(rawMsg);
}
});

await this.bot.start();
}

async disconnect(): Promise<void> {
if (this.bot) {
await this.bot.logout();
this.bot = null;
}
}

isOnline(): boolean {
return this.bot?.isLoggedIn ?? false;
}

// 核心方法：原始消息 → 统一格式
private async handleIncomingMessage(rawMsg: WeChatMessage): Promise<void> {
// 格式化——将微信消息转成内部标准格式
const message: StandardMessage = {
id: rawMsg.id,
channel: 'wechat',
type: this.mapMessageType(rawMsg.type()),
from: {
id: rawMsg.talker().id,
name: rawMsg.talker().name(),
type: rawMsg.room() ? 'group' : 'individual',
},
room: rawMsg.room() ? {
id: rawMsg.room().id,
topic: await rawMsg.room().topic(),
} : undefined,
content: {
text: rawMsg.text(),
// 如果有引用回复
quote: rawMsg.quotedText() ?? undefined,
},
timestamp: rawMsg.date(),
raw: rawMsg, // 保留原始引用，供高级插件使用
};

// 分发给所有注册的消息处理器
for (const handler of this.messageHandlers) {
await handler(message);
}
}

async send(target: MessageTarget, content: MessageContent): Promise<SendResult> {
if (!this.bot) throw new Error('微信通道未连接');

let contact;
if (target.type === 'individual') {
contact = this.bot.Contact.find({ id: target.id });
} else {
contact = this.bot.Room.find({ id: target.id });
}

if (!contact) throw new Error(`目标未找到: ${target.id}`);

// 根据内容类型发送
if (content.type === 'text') {
await contact.say(content.text);
} else if (content.type === 'image') {
// 微信端图片处理
const fileBox = FileBox.fromUrl(content.url!);
await contact.say(fileBox);
}

return { success: true, timestamp: Date.now() };
}

onMessage(handler: MessageHandler): void {
this.messageHandlers.push(handler);
}

// 消息类型映射——微信特有类型 → 标准类型
private mapMessageType(wechatType: number): MessageType {
const typeMap: Record<number, MessageType> = {
7: 'text', // 文本
3: 'image', // 图片
6: 'file', // 文件
0: 'unknown',
};
return typeMap[wechatType] ?? 'unknown';
}

supportedTypes(): MessageType[] {
return ['text', 'image', 'file', 'unknown'];
}

getMetadata(): ChannelMetadata {
return {
name: this.name,
displayName: '微信',
version: '1.0.0',
capabilities: ['text', 'image', 'file', 'group', 'at'],
};
}
}

四、核心管道引擎——消息处理流水线

4.1 管道模式

消息进入 → [预处理器链] → [路由决策] → [AI/插件处理] → [后处理器链] → 回复
│ │ │ │
过滤垃圾消息 判断是否要回复 生成回复内容 格式化回复
去重 选择处理器 调用模型 发给目标通道
安全检测

4.2 管道实现

// packages/core/src/pipeline/pipeline-engine.ts

export interface PipelineStage {
name: string;
process(ctx: PipelineContext): Promise<PipelineContext>;
}

export class PipelineEngine {
private stages: PipelineStage[] = [];

constructor() {
// 注册内置处理阶段
this.stages = [
new PreProcessStage(), // 预处理
new RouteStage(), // 路由
new AIPluginStage(), // AI/插件
new PostProcessStage(), // 后处理
new ReplyStage(), // 发送回复
];
}

// 插件可以注入自定义阶段
addStage(stage: PipelineStage, before?: string): void {
const index = before
? this.stages.findIndex(s => s.name === before)
: this.stages.length;
this.stages.splice(index, 0, stage);
}

async execute(message: StandardMessage): Promise<ReplyResult> {
let ctx: PipelineContext = {
message,
metadata: {},
decisions: [],
reply: null,
errors: [],
};

for (const stage of this.stages) {
try {
const startTime = Date.now();
ctx = await stage.process(ctx);

// 如果某个阶段决定终止管道（如命中黑名单）
if (ctx.terminated) {
break;
}

ctx.metadata[`${stage.name}_duration`] = Date.now() – startTime;
} catch (error) {
ctx.errors.push({
stage: stage.name,
error: error.message,
});
// 不终止，继续处理
}
}

return {
replied: ctx.reply !== null,
reply: ctx.reply,
pipelineLog: ctx.metadata,
errors: ctx.errors,
};
}
}

4.3 路由阶段——决定谁处理

// packages/core/src/pipeline/route-stage.ts

export class RouteStage implements PipelineStage {
name = 'route';

constructor(
private pluginManager: PluginManager,
private aiEngine: AIEngine,
private config: PipelineConfig
) {}

async process(ctx: PipelineContext): Promise<PipelineContext> {
const { message } = ctx;

// === 规则 1：检查是否需要回复 ===
// 群聊中只有被 @ 或包含关键词才回复
if (message.from.type === 'group') {
const mentioned = message.raw.mentionedSelf?.() ?? false;
const hasKeyword = this.config.listenKeywords.some(
kw => message.content.text?.includes(kw)
);

if (!mentioned && !hasKeyword) {
ctx.terminated = true;
return ctx;
}
}

// === 规则 2：检查黑名单 ===
if (this.config.blacklist?.users?.includes(message.from.id)) {
ctx.terminated = true;
return ctx;
}

// === 规则 3：插件匹配优先 ===
// 先遍历插件，看是否有匹配的触发器
for (const plugin of this.pluginManager.getActivePlugins()) {
if (plugin.triggers?.some(trigger => {
if (trigger instanceof RegExp) {
return trigger.test(message.content.text ?? '');
}
return message.content.text?.includes(trigger);
})) {
ctx.decisions.push({
handler: 'plugin',
handlerName: plugin.name,
priority: plugin.priority ?? 50,
});
}
}

// === 规则 4：知识库匹配 ===
if (ctx.config.enableKnowledge) {
const knowledgeMatch = await this.aiEngine.searchKnowledge(message);
if (knowledgeMatch && knowledgeMatch.score > 0.7) {
ctx.decisions.push({
handler: 'knowledge',
handlerName: knowledgeMatch.source,
priority: 40,
data: knowledgeMatch,
});
}
}

// === 规则 5：兜底给 AI ===
if (ctx.config.enableAI) {
ctx.decisions.push({
handler: 'ai',
handlerName: 'default-ai',
priority: 10,
});
}

// === 按优先级排序，选最高的 ===
ctx.decisions.sort((a, b) => b.priority – a.priority);
ctx.metadata.selectedHandler = ctx.decisions[0];

return ctx;
}
}

五、插件系统——热插拔架构

5.1 插件接口设计

// packages/plugins/src/plugin-types.ts

export interface Plugin {
/** 唯一标识 */
name: string;

/** 显示名称 */
displayName: string;

/** 版本 */
version: string;

/** 描述 */
description?: string;

/** 触发条件——正则或字符串 */
triggers?: (string | RegExp)[];

/** 优先级，越高越先匹配 */
priority?: number;

/** 需要的配置项 */
schema?: PluginSchema;

/** 初始化 */
init?(config: PluginConfig): Promise<void>;

/** 核心执行 */
execute(message: StandardMessage, context: PluginContext): Promise<PluginResult>;

/** 销毁 */
destroy?(): Promise<void>;
}

export interface PluginContext {
/** 当前消息 */
message: StandardMessage;

/** 对话历史 */
history: Message[];

/** 用户画像 */
userProfile?: UserProfile;

/** 全局存储（跨请求） */
storage: PluginStorage;

/** 调用 AI 引擎 */
callAI(prompt: string, options?: AIOptions): Promise<string>;

/** 上一条插件结果（链式调用） */
previousResult?: PluginResult;
}

5.2 插件管理器

// packages/plugins/src/plugin-manager.ts

export class PluginManager {
private plugins: Map<string, Plugin> = new Map();
private activeState: Map<string, boolean> = new Map();
private eventBus: EventBus;

// 注册插件
async register(plugin: Plugin): Promise<void> {
if (this.plugins.has(plugin.name)) {
throw new Error(`插件 ${plugin.name} 已存在`);
}

this.plugins.set(plugin.name, plugin);
this.activeState.set(plugin.name, false);

this.eventBus.emit('plugin:registered', { name: plugin.name });
}

// 启用插件（热加载）
async enable(name: string, config?: PluginConfig): Promise<void> {
const plugin = this.plugins.get(name);
if (!plugin) throw new Error(`插件 ${name} 未注册`);

await plugin.init?.(config ?? {});
this.activeState.set(name, true);

this.eventBus.emit('plugin:enabled', { name });
console.log(`✅ 插件已启用: ${name}`);
}

// 禁用插件（热卸载）
async disable(name: string): Promise<void> {
const plugin = this.plugins.get(name);
if (!plugin) return;

await plugin.destroy?.();
this.activeState.set(name, false);

this.eventBus.emit('plugin:disabled', { name });
console.log(`⏸️ 插件已禁用: ${name}`);
}

// 获取所有激活的插件（按优先级排序）
getActivePlugins(): Plugin[] {
return Array.from(this.plugins.values())
.filter(p => this.activeState.get(p.name))
.sort((a, b) => (b.priority ?? 0) – (a.priority ?? 0));
}
}

5.3 一个天气预报插件（完整实现）

// packages/plugins/demo-plugins/src/weather-plugin.ts
import { Plugin, PluginContext, PluginResult, PluginSchema } from '@demo/plugins';

export class WeatherPlugin implements Plugin {
name = 'demo-weather';
displayName = '天气预报';
version = '1.0.0';
description = '查询指定城市的天气信息，支持定时播报';
priority = 60;

triggers = [
/天气/,
/(今天|明天|后天).*(冷|热|温度|穿什么)/,
/.*天气预报.*/,
];

schema: PluginSchema = {
type: 'object',
properties: {
defaultCity: {
type: 'string',
description: '默认查询城市',
default: '深圳',
},
apiKey: {
type: 'string',
description: '天气 API 密钥',
sensitive: true,
},
},
};

private config: PluginConfig = {};

async init(config: PluginConfig): Promise<void> {
this.config = config;
}

async execute(message: StandardMessage, context: PluginContext): Promise<PluginResult> {
// 从消息中提取城市名
const city = this.extractCity(message.content.text ?? '')
|| this.config.defaultCity
|| '深圳';

try {
// 调用外部天气 API
const weather = await this.fetchWeather(city, this.config.apiKey);

// 生成自然语言回复——用 AI 润色
const prompt = `
将以下天气预报转成一句友好的回答：
城市：${city}
天气：${weather.condition}
温度：${weather.temp}°C
湿度：${weather.humidity}%
建议：${weather.advice}

回复字数 80 字以内，口语化。
`;

const reply = await context.callAI(prompt, {
temperature: 0.7,
maxTokens: 120,
});

return {
reply,
metadata: {
city,
weather: {
condition: weather.condition,
temp: weather.temp,
humidity: weather.humidity,
},
},
consumed: false, // 不消费消息，允许后续继续处理
};
} catch (error) {
return {
reply: `抱歉，查询${city}天气失败: ${error.message}`,
error: error.message,
};
}
}

async destroy(): Promise<void> {
// 清理资源
}

private extractCity(text: string): string | null {
// 简单城市匹配——生产环境用 NER
const match = text.match(/(北京|上海|广州|深圳|杭州|成都|武汉|南京)/);
return match ? match[1] : null;
}

private async fetchWeather(city: string, apiKey: string): Promise<WeatherData> {
const response = await fetch(
`https://api.weather-demo.com/v1/forecast?city=${encodeURIComponent(city)}&key=${apiKey}`
);
if (!response.ok) throw new Error(`API 返回 ${response.status}`);
return response.json();
}
}

5.4 从文件系统动态加载插件

// packages/plugins/src/loader/fs-loader.ts
import { readdirSync, existsSync } from 'node:fs';
import { resolve } from 'node:path';
import { PluginManager } from '../plugin-manager';
import { Plugin } from '../plugin-types';

export class FileSystemPluginLoader {
constructor(
private pluginManager: PluginManager,
private pluginsDir: string
) {}

async loadAll(): Promise<number> {
if (!existsSync(this.pluginsDir)) {
console.warn(`插件目录不存在: ${this.pluginsDir}`);
return 0;
}

const entries = readdirSync(this.pluginsDir, { withFileTypes: true });
let loaded = 0;

for (const entry of entries) {
if (!entry.isDirectory()) continue;

const pluginPath = resolve(this.pluginsDir, entry.name);
const indexPath = resolve(pluginPath, 'index.ts'); // 或 index.js

if (!existsSync(indexPath)) {
console.warn(`插件 ${entry.name} 缺少入口文件`);
continue;
}

try {
// 动态 import
const module = await import(indexPath);
const plugin: Plugin = module.default ?? module.plugin;

if (!plugin || !plugin.name) {
console.warn(`插件 ${entry.name} 导出格式不正确`);
continue;
}

await this.pluginManager.register(plugin);

// 读取插件配置（默认启用）
const config = this.loadPluginConfig(entry.name);
if (config?.enabled !== false) {
await this.pluginManager.enable(plugin.name, config);
}

loaded++;
console.log(`📦 插件已加载: ${plugin.name} v${plugin.version}`);
} catch (error) {
console.error(`加载插件失败 ${entry.name}:`, error);
}
}

return loaded;
}

private loadPluginConfig(name: string): Record<string, any> {
const configPath = resolve(this.pluginsDir, name, 'config.yaml');
if (existsSync(configPath)) {
return parseYaml(readFileSync(configPath, 'utf-8'));
}
return {};
}
}

六、上下文管理——多轮对话如何记住历史

6.1 上下文窗口设计

// packages/ai/src/context-manager.ts

interface ContextWindow {
sessionId: string;
messages: ContextMessage[];
summary?: string; // 超过窗口时自动摘要
tokenCount: number;
maxTokens: number;
createdAt: Date;
updatedAt: Date;
}

export class ContextManager {
// === 两种缓存 ===
private activeWindows: Map<string, ContextWindow> = new Map(); // 热数据
private storage: ContextStorage; // 持久化（SQLite/Redis）

constructor(private config: ContextConfig) {}

async addMessage(sessionId: string, message: StandardMessage): Promise<void> {
let window = await this.getOrCreateWindow(sessionId);

const contextMessage: ContextMessage = {
role: 'user',
content: message.content.text ?? '',
timestamp: Date.now(),
};

window.messages.push(contextMessage);
window.tokenCount += this.estimateTokens(contextMessage.content);

// === 超过窗口大小 → 压缩成摘要 ===
if (window.tokenCount > window.maxTokens * 0.85) {
await this.compressWindow(window);
}

window.updatedAt = new Date();
await this.storage.save(sessionId, window);
}

private async compressWindow(window: ContextWindow): Promise<void> {
// 保留最近 5 条消息，其余压缩为摘要
const compressMessages = window.messages.slice(0, –5);
const recentMessages = window.messages.slice(–5);

const summary = await this.aiEngine.summarize(
compressMessages.map(m => `${m.role}: ${m.content}`).join('\\n')
);

window.summary = summary;
window.messages = [
{
role: 'system',
content: `之前的对话摘要：${summary}`,
timestamp: Date.now(),
},
…recentMessages,
];

window.tokenCount = this.estimateTokens(summary)
+ recentMessages.reduce((s, m) => s + this.estimateTokens(m.content), 0);
}

// 多用户隔离
private buildSessionId(message: StandardMessage): string {
// 同一个用户 + 同一个群的对话共享上下文
if (message.room) {
return `room:${message.room.id}:user:${message.from.id}`;
}
return `user:${message.from.id}`;
}
}

七、最后——这张架构图的几点精髓

回顾整张架构图，最值得学习的三个设计决策：

7.1 通道适配器模式

ChannelAdapter 接口 ← 微信 Adapter
← 钉钉 Adapter
← Telegram Adapter
← … 任意 IM

核心引擎只依赖接口，不关心具体实现
新加一个 IM 通道 = 实现一个 Adapter，零修改核心代码

7.2 管道 + 插件双模式

管道模式：处理"怎么处理一条消息"的固定流程
→ 预处理 → 路由 → 处理 → 后处理 → 回复

插件模式：提供"具体怎么处理"的可扩展能力
→ 天气插件 / 翻译插件 / 日程插件 / …

管道负责流程，插件负责策略。
管道不关心有几个插件，插件不关心怎么被调用的。

7.3 上下文管理的两级缓存

热数据（内存 Map） ← 毫秒级访问，高 QPS
│
│ 定期同步
▼
冷数据（SQLite/Redis） ← 持久化，可恢复

这是性能和可靠性之间的经典权衡，在高并发场景下至关重要。