你是不是也经常遇到这样的场景:早上在群里讨论方案,下午想引用其中一条结论,却只能耐着性子从几千条旧消息里翻找。聊天软件自带的搜索也许能把那条消息挖出来,但真正麻烦的是,你要的往往不是某个孤零零的关键词,而是当时前后几轮对话里沉淀下来的背景和判断。也正是出于这个原因,不少开源的AI助理项目开始把战场从独立网页搬进聊天软件本身。
我这边长期维护的一个私人AI助理项目,就是走这条路。它不打开新的客户端,不做另一个待办面板,而是以机器人的形式直接“住”在聊天群里,对话、记录、整理、提醒都在同一个会话流里完成。项目更新到2.0版之后,重点补上了两个很实际的能力:一个是针对历史对话的“会话搜索”,另一个是跨设备的“云端协作”。这篇文章就把这两个能力的实现思路、踩坑过程以及部署时需要注意的细节整理出来,给同样想自建一个“有记忆”的助理、或者想在聊天工具里嵌入私有知识库的朋友做参考。
1. 为什么“住在聊天软件里”,比做独立App更合适
1.1 聊天记录本身就是最真实的记忆库
过去我试过不少“知识管理工具”,最后的结局都很一致:坚持不下来。原因不复杂,人在工作时天然聚集在聊天软件里,讨论过程散落在各个群和私聊中。如果助理只活在另一个软件或网页里,那就意味着所有的想法要先手动搬到那边,整理到一半就断了。
所以2.0项目在设计上坚持一个原则:助理不要求用户迁徙,而是主动往用户已经聚集的地方扎。一个群里讨论完问题,你直接@助理,它就能把结论归档、建立提醒、甚至做阶段复盘。这样的使用成本几乎为零,长期积累下来,聊天会话本身就慢慢变成了一个私人知识库。
1.2 助理不应该只做“应答器”,还要做“档案员”
很多接入聊天软件的AI助理,本质上只是把大模型包装成了一句“吞消息、吐回复”的机器人。用户发一句,它回一段;换个时间重新问同样的问题,它完全不记得。
我把这种情况叫“失忆式助理”。它在单轮对话里表现不错,但无法回答“上周我们讨论过什么”“那个数据问题最后怎么处理的”这类需要依赖历史的问题。2.0版反复打磨的会话搜索,本质就是把“记忆”做成助理的基础能力,让它不只是回复当下的问题,还能在需要时把历史切片重现出来。
1.3 2.0版不是功能堆叠,而是补齐三个底层能力
老版已经能做到“在聊天软件里被召唤”,它的问题在于所有数据都保存在部署它的那一台电脑上。手机和电脑上各部署一个,两边就是知识孤岛;在公司讨论的内容,回家以后想查只能靠记忆。2.0级的目标很明确:
- 让助理记得住:对话内容能进入可检索的历史库;
- 让助理找得着:不靠模糊的名字,用自然语言也能把陈年旧事翻出来;
- 让助理跟得上:在家里电脑和公司电脑之间,知识不会分裂成两半。
这三件事看似不相关,落到实现上却需要同一个数据底座。底座如果有问题,后面的搜索和协作都会跟着抖。
2. 会话搜索的实现细节:从全文到语义
2.1 用户真正需要的搜索,不是输入一个关键词
做搜索功能前,我对用户可能的搜索意图做了一次分类,发现实际使用中大约有这么几类:
- 记得原话:“我之前说过xx句话是谁说的来着”——这类搜索适合用关键词全文匹配;
- 记得大概意思:“我们好像聊过一个关于产品定价的结论”——关键词完全对不上,需要语义匹配;
- 想找到某天或某个讨论主题下的上下文,而不是单条消息。
如果只做一种搜索方式,必然覆盖不全。关键词搜索快且准,但用户往往想不起原话;语义搜索能读懂“意思”,但无法精确处理数字、ID、邮箱之类的内容。两个方向都不能放弃。
2.2 轻量级双路索引:FTS5用于精确匹配,embedding用于语义召回
我一开始也考虑过把全部历史消息同步到专门的向量数据库里,后来发现对一个以个人或小团队为主的助理而言,完全没必要引入那么重的服务。现在2.0版默认使用一套“本地双路索引”方案:
- 消息入库时,原文写入SQLite,并用FTS5建全文索引,负责精确的关键字检索;
- 同时对消息做切片和向量化,向量也落库到本地,负责语义相似度匹配。
这样做的好处是部署极其简单,跑在一台低配云主机或NAS上即可。FTS5本身在SQLite里是内置的,向量部分也可以在轻量扩展里实现,且几乎没有额外运维负担。
2.3 FTS5全文检索的建表与查询
以下是我实际在项目中使用的建表逻辑,做了简化以保留核心思路。
-- 消息主表 CREATE TABLE IF NOT EXISTS messages ( id INTEGER PRIMARY KEY AUTOINCREMENT, room_id TEXT NOT NULL, sender_id TEXT NOT NULL, msg_time INTEGER NOT NULL, content TEXT, ref_msg_id TEXT UNIQUE ); -- FTS5 全文索引,内容按中文和英文常见场景分别处理 CREATE VIRTUAL TABLE IF NOT EXISTS messages_fts USING fts5( room_id UNINDEXED, content, tokenize = "unicode61 remove_diacritics 2" );插入消息后需要同步写入全文索引。
INSERT INTO messages_fts(rowid, room_id, content) VALUES (new_id, ?, ?);FTS5默认的unicode61分词对连续英文、数字友好,但中文是一个连续字符串,效果并不理想。我后面在“踩坑记录”一节会专门讲中文分词问题。先看查询:
SELECT m.id, m.sender_id, m.msg_time, m.content, bm25(messages_fts) AS rank FROM messages_fts f JOIN messages m ON m.id = f.rowid WHERE messages_fts MATCH ? AND room_id = ? ORDER BY rank LIMIT 20;这里的MATCH语法支持"phrase"、AND、OR,但需要用户输入的关键词和我传给它的检索表达式做一层安全转义,不能直接把用户原文塞进去,否则包含特殊字符时会报错。
2.4 语义向量的生成与召回
对于语义搜索,我保留了一个抽象接口:无论你用本地运行的嵌入模型还是外部API接口,都必须返回固定维度的向量。入库时大致是一个这样的管道流程:
# 消息入库后异步执行,不阻塞主流程 def index_for_vector(msg_id: int, text: str): vector = embedding_client.embed(text) vector_repo.save( msg_id=msg_id, text_snapshot=text[:512], vector=vector, )查询语义时,先把你输入的“口头化描述”也embedding成向量,然后做相似度检索。但如果只依赖向量检索,结果往往在精准度上不够。所以实际返回前还有一道合并排序。
简化版的混合召回流程是:
- 全文搜索命中前20条;
- 语义搜索召回前20条;
- 按“消息时间衰减 + 匹配来源权重 + 相似度得分”三者综合排序;
- top N作为最终候选,送进语言模型做摘要并输出。
时间衰减的概念很重要。5年前讨论的内容和昨天讨论的内容即便语义相似,后者的参考价值通常更高,因此在排序阶段把时间差作为权重因子一起算进去。
2.5 搜索结果的输出必须带上下文
只有一条消息的搜索结果对用户没有意义。用户想找回的是当时的场景。2.0版在定位到目标后,会从该条消息向前向后各取若干条作为上下文窗口,再把这段内容整理为引用片段。
实际输出的形如:
在 2024-11-28 的讨论中,你当时提到: > 灰度发布可以分成两阶段,第一批先放老用户群。 上下文补充: > A:我建议新老用户固定比例。 > B:不行。先小范围观察,再全量。 > 你:灰度发布可以分成两阶段... > A:可以,我下午改配置。这个能力看起来不复杂,却是聊天语义里最有价值的部分,因为能找回的不只是一句话,而是一段决策历程。
3. 云端协作的实现:多端同步,但不上传明文
3.1 要先想清楚“同步什么”和“不同步什么”
做云端协作时,最容易犯的错误是把整个本地数据库都搬到服务器上,让所有设备直接同步。搜索功能加上后,数据量会成倍增长,如果明文同步,隐私风险也完全不可控。
2.0版的同步策略定了一个明确的边界。
本地始终保留:完整历史消息、向量索引、全文索引、用户配置。
同步到云端的只是:事件操作日志、被主动标记为“需要跨设备可见”的常用会话ID、收藏片段、以及助理生成的结构化摘要和提醒。
简单说,宁可让我在另一台设备上发请求后等待几秒生成结果,也不要把所有原始消息明文交给服务端。云端协同的定位是“元数据与产物同步”,而不是同步整库。
3.2 数据同步的传输形态:事件日志 + 去重
我在项目实施时参考了事件溯源的做法:每一条需要同步的数据不直接传最终状态,而是产生一个追加事件。
事件示例:
{ "event_id": "d9a7e5f2-...", "type": "note.created", "payload": { "note_id": "note_001", "content_ref": "local://notes/note_001" }, "producer": "device_A", "created_at": 1735300000000 }每个设备启动后,从服务端拉取缺失的事件,再按照事件类型应用到本地。
这里有几个需要处理的坑:
- 事件去重:一个事件必须携带全局唯一ID,应用侧要记录已经处理过的事件ID,否则网络抖动时重复拉取会导致重复写入;
- 顺序一致性:使用
created_at排序并不能保证跨设备一致,因为设备间可能存在时钟偏差。我采用的办法是允许事件带一个client_created_at,服务端只做窗口排序,不强行统一; - 内容寻址:原始消息不在事件里传。如果同步的内容是一段文字,会把摘要放进去;如果是原始消息的引用,就只放一个本地ID,接收端按需再请求。
3.3 端到端加密的具体分级策略
云端协作上线前,我认真研究过E2E加密的实现。全量E2E实在太痛苦,因为云端搜索必然要求服务端能读内容,或者建立一套复杂的可搜索加密体系。经过取舍,最终实现为两种模式:
- 基础模式(低隐私):传输走TLS加密,云端保存的事件只含经过标记的会话语义描述与结构化数据;对明文消息不做落盘。
- 私密模式(高隐私):在本地用口令派生的密钥对关键数据做AES-GCM加密,再把密文交给云端的设备之间转发。
私密模式下,每个客户端的配置里有一段类似这样的设置。
sync: mode: e2e encryption_key_env: ASSISTANT_SYNC_KEYASSISTANT_SYNC_KEY的值不在配置文件里直接写,而是从本地环境变量或系统密钥链读取。客户端解密数据后只在内存中保留一分钟左右,不把解密结果写回磁盘,这个策略在防止服务端和本地物理磁盘双重泄露上比较稳妥。
3.4 不同设备同时修改,冲突怎么办
多个设备同时编辑同一条备注或同一个计划时,会发生同步冲突。我参考协作类软件中常用的“细粒度合并”策略,把数据拆到字段级来处理。
比如用户在同一条提醒上,一台设备修改了时间,另一台设备修改了标题,字段级合并后两者都可以保留。如果修改的是同一字段,就采用“最后写偏向”规则,但是保留旧值到历史版本表中,允许用户事后回溯。
一个团队或个人的长期数据,不怕变更,最怕的是消失。所以面向用户的所有可编辑业务对象,在本地都保留了一份轻量的分段历史版本。这样的代价是数据库里多一张表,但换回来的是很高的容错度。
3.5 多设备聊天记录如何保持一致
云端协作还有一个容易被忽略的点:与聊天软件本身的事件流不同,历史消息通常不能从聊天软件端完整拉取。如果有一个新设备加入,它的本地库空空如也,只能靠云端事件重建。
为此,同步协议里支持“设备历史引导”流程。新设备发起一次握手后,会从“现有在线且可信赖的设备”那里收到一次快照引用,然后按需拉取相关的会话摘要。这样避免了把大型原库放云端,也能让新设备在短时间内拥有一个可用的知识骨架。
4. 碰到的几个硬骨头:重复消息、中文分词、回调超时
4.1 聊天平台回调超时,导致AI被重复触发
做聊天机器人时,我第一版犯过的错误是在平台的消息回调里直接同步调用大模型接口。一次请求可能要消耗5到20秒,而绝大多数聊天平台配置的回调超时远远低于这个时间。结果就是平台发送了重试请求,模型被调用了两次甚至多次,带来了多余的时间和费用开销。
现在的实现是收到消息后只做校验、落库、投递到内部任务队列三步,然后立刻返回“已接收”。真正的大模型调用和语义识别全部在后台worker中执行。
def handle_webhook(request): event = parse_event(request) if dedup.is_dup(event["raw_msg_id"]): return {"ok": True, "skipped": "duplicated"} message_repo.save(event) task_queue.enqueue("process_message", event["raw_msg_id"]) return {"ok": True}这个改动同时消除了大量“回复重复”的问题。
4.2 同一条消息被投递两次:幂等处理不能省
很多聊天平台在回调设计上是“至少一次投递”。哪怕代码做了快速返回,依然可能在网络层发生同一条消息的重复推送。
如果不做幂等处理,助理对同一条消息会重复产生两次回答。为了在入口处拦住重复,我维护了一张已处理消息ID表。
处理流程中会给每个原始消息生成一个稳定的唯一ID,规则是“聊天平台类型 + 房间ID + 服务端消息ID”的联合值。新消息入库前先查这个ID是否存在,存在就直接丢弃。这张表的数据量比较大,但ID本身很短,可以在数据库里建立唯一索引。早期曾为了省空间只对当天消息保重,事实证明跨天的重复消息依然存在,后来改成全量保留ID,只定期清理三个月前的旧记录。
4.3 中文搜索效果差:分词器是个绕不开的问题
项目里的搜索上线初期,很多人反馈中文搜索“还不如聊天软件自带的”。原因不复杂:FTS5自带的unicode61分词对英文友好,但对中文会退化成按整句字符串匹配。中文本质上没有空格分词,需要借助额外工具。
我不想在默认部署里引入一堆分词中间层,于是采用了一套务实做法:对中文消息进行双字滑动切分(bigram)后存入索引列。
“灰度发布方案”会被切成“灰度/度发/发布/布方/方案”,这样用户搜“灰度发布”时能命中“灰度度发发布”对应的整数位置;搜索精度不比专业分词器差多少,但在小数据量场景中实现成本最低。
如果后续数据量大了,需要更合理的分词,可以再引入独立的中文分词服务。实际做社区反馈测试时,bigram方案对个人助理的搜索体验已经足够满意。
4.4 向量检索的质量问题:文本切片不能太随意
最初我把一整段长消息切成一整个向量,结果搜索时经常跑偏。比如一条长文档包含了多个主题,向量处理后会被平均成一个模棱两可的向量。
调整后的规则是:
- 短消息:整条消息作为一个向量片段;
- 长消息:优先按句子边界截断,每条不超过256字符;
- 相邻语义段之间允许少量重叠,避免把一个完整含义硬生生切断。
片段对应的原始消息ID被保留,搜索命中时能迅速回到原始会话上下文。经过调整,搜索“跟得上思路”的比例有了明显提升。
5. 动手部署一版:Docker Compose + 最小配置
5.1 运行环境与前置要求
2.0版做得比较轻,所以对硬件并不挑剔。一套常见的小型云主机配置即可稳定工作:
- CPU:1核以上;
- 内存:2GB以上,若使用本地向量模型建议4GB;
- 磁盘:20GB SSD,主要存放SQLite库和向量数据;
- 系统:Debian/Ubuntu或任何能运行Docker的Linux发行版均可。
如果你不想用Docker,也可以直接用Python运行,但Docker方式能少踩很多依赖坑。
5.2 一个可直接落地的Docker Compose示例
下面是我的部署文件目录中最精简但完整可用的版本。
version: "3.8" services: assistant: image: your-assistant-image:2.0 restart: unless-stopped env_file: - .env environment: - DATA_DIR=/app/data - IM_HOOK_TOKEN=${IM_HOOK_TOKEN} - IM_ROOM_ALLOWLIST=${IM_ROOM_ALLOWLIST} - AI_MODEL_ENDPOINT=${AI_MODEL_ENDPOINT} - AI_MODEL_NAME=${AI_MODEL_NAME} - ENABLE_LOCAL_SEARCH=true - ENABLE_SYNC_RELAY=${ENABLE_SYNC_RELAY} - SYNC_RELAY_URL=${SYNC_RELAY_URL} volumes: - ./data:/app/data ports: - "8800:8800"对应的.env文件大致如下:
# 聊天软件机器人的访问令牌 IM_HOOK_TOKEN=替换为你的实际token # 只允许特定会话唤起助理,多个用逗号分隔 IM_ROOM_ALLOWLIST=room_001,room_002 # 后端模型接口,兼容OpenAI格式的本地或远程服务均可 AI_MODEL_ENDPOINT=http://127.0.0.1:8000/v1 AI_MODEL_NAME=qwen2.5-7b-instruct # 是否开启同步中继 ENABLE_SYNC_RELAY=true SYNC_RELAY_URL=wss://your-relay.example.com这里需要提醒一个关键点:IM_ROOM_ALLOWLIST这个白名单非常推荐开启。如果助理能被任何一个会话随意唤起,某些无关人员可能会利用它消耗你配置的模型资源,甚至拷问私人知识库里的信息。私密助理必须有边界。
5.3 初始化本地目录与密钥
首次启动前,我习惯在本地先建好数据目录和权限隔离。
mkdir -p data chmod 700 data如果使用私密同步模式,还需要在初始化密钥时这样操作。
export ASSISTANT_SYNC_KEY=$(python3 -c "import secrets; print(secrets.token_hex(32))")整个密钥不会出现在shell历史中最好,要长期保存的话建议放入密码管理器,或写入只有数据目录所在用户可读的key文件。
5.4 验证助理是否正常工作
启动容器后,先看日志确认没有报错:
docker compose logs -f assistant确认没有异常后,在你的聊天会话中向机器人发送:
/status如果返回包含uptime和index_status的信息,说明基础连通性正常。接着手动发一条消息:
请记录:周一上午十点与设计团队对齐新版首页布局过几秒再问:
周日我有什么安排?助理如果能关联刚才的内容并给出答案,说明基础流程跑通。再继续测搜索功能,发送:
/search 新版首页布局两条链路都通了,说明本地搜索服务已正常挂载。
6. 扩展思路:从“个人助理”变成“小团队协作中枢”
6.1 用“收藏片段”建立轻量知识库
很多人问多设备同步到底有什么实际效果,我觉得最有价值的还是收藏片段。看到一条有价值的讨论,直接回复机器人:
收藏这条2.0版会保存当前消息及上下文,并自动打上时间标签。之后在任意设备上搜索“收藏 关于xx的讨论”,都能快速命中。这比手动复制粘贴到笔记软件省事得多,而且上下文不会丢。
6.2 自定义触发指令,按自己的工作流设计
每个用户的使用习惯不一样。有人需要每天下班前让助理整理一份当日讨论摘要;有人需要每周五复盘一次项目进度;也有人希望收到特定关键词时机器人自动建待办。
这些行为可以做成独立指令插件,放在/plugins目录下。每个插件只需定义一个函数入口。
from assistant.sdk import AssistantPlugin, action class WeeklyDigestPlugin(AssistantPlugin): @action(r"(本周|这周)工作(总结|复盘)") async def weekly_digest(self, ctx): messages = self.search.near_week(ctx.room_id) summary = await self.llm.summarize(messages) return summary然后实现类似“每一条新消息都会先经过插件匹配器,命中则触发对应action”的机制。想要做更多定制时,不需要改主程序,写一个插件文件即可。
6.3 助理入口固定在某个群,而不是所有群
在团队场景里,我强烈建议不要在所有群启用机器人,而是单独建一个类似“记忆库”的空间。平时把需要沉淀的信息转发到该空间并@助理,需要检索时也来这里搜索。
这样可以控制上下文的噪声,让私人助理始终关注真正重要的信息。搜索和知识沉淀自然形成一个闭环。
7. 最后再分享一点选型上的体会
做这个2.0版的过程里,反复出现的一个词是“边界”。搜索做过了头可能变成监控工具,云端协作做过了头可能直接变成隐私泄露通道。开源项目最大的优势,正在于每个部署者都能按自己的安全标准和信任边界去调节数据流转方式。我自己在维护时要求很朴素:它能让我两三台设备间的私人知识变得连贯,同时又不让我每天晚上担心云端服务器被人拖库。只要你在部署时把加密、白名单、数据备份这三件事做成默认习惯,它就能在“好用”和“安全”之间找到一个很舒服的平衡。