赛博肥鱼群友 - 实现小结

最近小肥鱼越来越火,不少人也都做了自己的肥鱼伙伴。我一向是对把 AI 当朋友没什么兴趣的,但是出于“大家都做了我也要试试”的心态,我也开始做起了自己的大肥鱼。

但是相比于使用现成的框架,对于这种“我的想法可以落地”的计划,我通常更喜欢自己手搓一套处理管线,也当是练练手了。

本文就是记录一下现阶段的设计历程,并为未来的改进提供存档。如果能帮到同样想自己搓一份的朋友,那就再好不过了。

1. 基本事实

在继续讨论之前,有一些基本事实应当了解。

1.1. LLM 是个无状态的黑盒模型

大语言模型(LLM)的行为非常简单:输入一段文字,它会回给你对这段文字“最应该的回答”,黑盒表示如下图:

flowchart LRA["输入文字"] --> B{{"LLM 黑盒
黑盒表示"}}B --> C["最应该的回答"]

它有一个非常关键的特点:无状态。

所谓无状态,意思就是你的每一次调用都不会改变模型状态,每一次调用都和其他调用无关。直观表现上,就是如果你第一个输入是“我喜欢吃西红柿”,而第二个输入是“我喜欢吃什么”,那么第二次输入的回答大概率不是“西红柿”。因为无状态,两个输入是完全独立、互不影响的。

因此,你的每一次请求都必须带有你和模型对话的上下文,才能让模型看起来有“记忆”,即如果你的第一次请求如下:

你:我喜欢吃西红柿,记住了
模型:我记住了

那么你的第二次请求的输入应该类似于下面这样:

你:我喜欢吃西红柿,记住了
模型:我记住了
你:我喜欢吃什么?

这种情况下模型才会几乎准确地回答出“西红柿”。

1.2. 前缀缓存

由于无状态,要让模型有聊天的“记忆”,就必须在每次输入时都携带之前与模型聊天的上下文,这就导致随着对话轮数越来越多,每次传递的对话将会越来越长,token 消耗也越来越大。例如,经过 N 轮的对话,假设每次都携带完整前文对话,则总消耗为:

[

S = Nx_1 + (N-1)x_2 + \cdots + x_N

]

(x_i) 表示第 (i) 轮新增消耗的 token,((N-i+1)) 表示第 (i) 轮内容在后续对话中被重复携带的次数。

可以看到,在这种“每次都携带完整前文对话”的条件下,在第 N 轮对话中,有相当一部分内容其实和 N-1 轮对话从头到尾都完全相同。

LLM 推理中的前缀缓存正是针对这一冗余设计的:前缀缓存是一种加速和节省算力开销的方式,对于共享相同前缀的请求,系统可以直接复用此前已计算并缓存的 KV Cache,跳过对相同前缀的重复预填充(prefill)计算。

换句话说,对于新一轮输入中未发生变化的前缀内容,其对应的注意力键值状态在上一轮已经计算完毕并驻留在缓存中,本轮只需计算新增 token 对应的增量部分即可,大大提高 prefill 速度的同时降低算力成本开销。

flowchart LRP["共享前缀 P"] --> CACHE[("前缀缓存")]CACHE --> S1["后缀 1"]CACHE --> S2["后缀 2"]CACHE --> S3["后缀 3"]S1 --> O1["回答 1"]S2 --> O2["回答 2"]S3 --> O3["回答 3"]

所谓共享相同前缀,就是两段文本从开头起完全相同的连续部分,比如:

我喜欢吃西红柿
我喜欢吃土豆

上面两句话的相同前缀就是 我喜欢吃,如果两段话被连续请求,那么在第二次请求时,我喜欢吃 这部分就会缓存命中。

模型 API 缓存命中与未命中的价格差异可能达到数十倍。因此在设计 prompt 时就必须利用前缀缓存:尽可能提高缓存命中率,从而提升模型响应速度并降低 token 成本。

前缀缓存严格要求前缀完全相同,从第一个不相同的字符处开始,此后所有字符都不再吃任何缓存命中的收益。

特别需要说明的是,DeepSeek API 的 tools 是注入到 system 与 user 之间的,也就是:system + tools + user。因此在设计 prompt 组装管线时,尽量考虑总是提供相同的 tools,以避免因为 tools 变更带来的对前缀缓存的破坏。

1.3. 被动与轮询

模型自身并不会主动发起消息,它只能在被动收到输入后再输出对应消息,因此你无法期望模型在没有任何工程辅助的情况下主动给你发送一段消息。目前让模型“主动”发起消息的方法基本就一种:轮询,在一个长循环中,定期将上下文提交给模型,并要求模型按需回复:

while True:delay(10)ask_deepseek("看看要不要找谁聊聊天")

1.4. 工具调用

这是现代 Agent 工程实践的核心。所谓工具调用,就是在请求时为模型提供一组工具,模型回复时可以选择调用这些工具,并给出调用时所需的参数。

Agent 框架所需要处理的就是将这些调用信息转化为实际调用,然后将结果返回给模型,由模型决定下一步,直到模型认为任务完成。

flowchart LRM["模型"] -->|"工具调用请求
工具名 + 参数"| A["Agent 框架"]A -->|"执行"| T["工具"]T -->|"结果"| AA -->|"结果返回"| MM -->|"完成"| R["回答"]M -.->|"未完成,继续循环"| A

这套简单的循环为语言模型装上了手脚,只要给出恰当的工具,模型就能利用工具做对应的事,包括但不限于在电脑上执行命令,编辑文件,或者将一张图发送给某个人。

1.5. 上下文长度限制与注意力稀释

能一次传递给模型的文本长度是有限制的,按中文字数估算,目前主流大语言模型的上下文上限约为 160 万个中文字(1M 上下文),超过这个长度的输入,模型便无法正常处理。

因此不可能将所有记录一股脑全部塞给模型,期望模型记住所有细节并给出准确回答。既然如此,为了让模型记住“以前的事”,就不能依赖上下文长度,必须接入外部记忆机制。

然而这一部分也是最难实现“低成本,良效果”的,远远不是一个简单的 RAG 就能解决(或者可以说 RAG 根本不能算是记忆)。效果好的成本高,成本低的效果基本都不佳。这种记忆也是人类最强的优势,模拟人类记忆的自动召回与筛选机制算是 LLM RP 最难跨过的一道门槛。

另一方面,越长的上下文越容易让模型不知道应该关注什么信息(注意力稀释,Lost in the Middle),从而导致模型给出质量更差的回答。因此即便模型真的支持 1M 上下文,也应该控制输入 prompt 的长度。如果可行的话,可以考虑控制在 16 万字以下(约 10 万 token;按 1 token 约等于 1.6 个中文字估算,大部分情况够用)。

1.6. 首因效应与近因效应

现实中的模型都加入了位置编码(绝对/相对/RoPE 等),并且是因果注意力(只能看前面的 token),所以段落在文本中的位置往往会影响模型对信息的利用:

开头:通常放系统提示、角色设定、全局规则。模型在训练中习惯了“开头是指令”的格式,因此开头内容往往有较高权重,类似“首因效应”。

末尾:因为是因果注意力,最后一个 token 能看到前面所有内容,且生成下一个 token 时,末尾内容离输出最近,影响最直接,类似“近因效应”。

中间:容易被忽略。长上下文里有个著名现象叫 “Lost in the Middle”:关键信息放在中间时,模型召回率明显低于放在开头或结尾。

基于这一效应,在设计 prompt 组装管线时,应该着重考虑某一段 prompt 放在哪里,以在最大化利用前缀缓存的同时提高模型对信息的利用效率。

2. 工作流

这一节主要讲讲我是如何设计肥鱼的工作流的。

2.1. 事实假定与指导思想

  1. 不同群之间的记忆隔离,以尽可能降低上下文污染,同时便于隔离测试。
  2. 越新的对话越应该重点保存。
  3. 简单且易于维护与从源码扩展的架构胜过复杂且全能的设计。
  4. 相信模型的智能。

2.2. 主循环

截至撰写本文时,我总共给我的肥鱼设计过三种(或者也可以说两种)处理管线。第一种大概基于消息泵 + 独立 prompt 的处理管线,但实际用下来我发现它不仅效果不行,而且代码复杂,会引入各种边界问题。

旧版的处理管线大概如下:

%%{init: {"flowchart": {"wrappingWidth": 340, "nodeSpacing": 30, "rankSpacing": 45}}}%% flowchart TDIn["群消息 / 定时事项"] --> Pump["该群串行泵
等安静窗口,攒一批"]Pump --> Gen["一次生成
模型 ⇄ 工具循环"]Gen --> Send["解析输出并逐段发送"]Send --> Progress["推进"]Progress --> Pump

确实可用,效果也不算很差,但很不符合直觉,而且会引入很多状态,导致整体处理流复杂,难以调整和维护。经过反复实践与迭代下来,目前采用的管线反而就是最简单的轮询 Agent 循环:

flowchart TDA[事件] --> B[构建上下文]B --> C[请求模型]C --> D{模型决定}D -- 调用工具 --> E[执行工具] --> CD -- 结束 --> F[本轮结束]F --> A

并让整个处理采用群内串行,群间并行的方案。通过消息队列来保证串行性,避免因为并发写导致的各种时序问题。更详细一点的处理循环伪代码见下:

chat_context: 请求的上下文,主要包括历史对话结果与当前对话
pop_events: 取新事件 #(1)新事件
peek_events: 检查是否有新事件def agent_loop(group):while True:delay(1)events = pop_events()if not events and not should_active():continuechat_context.update()while True:if peek_events():breakif chat_context.is_overflow(): #(2)上下文预算超限compact_memory_compress(chat_context)res = ask_deepseek(chat_context)if should_terminal(): #(3)结束breakif res.tool_calls:call_tools(res.called_tools)continuebreakparallel(agent_loop(group) for group in groups)

(1)新事件:主要包括新消息、重启补漏、随机激活以及定时事件。

(2)上下文预算超限:

我用了三种限制来控制上下文的长度:

  • 聊天消息上下文窗口

所谓的“聊天消息上下文”是指上下文消息中非 synthetic(程序合成的提示性信息,不代表聊天消息) 的 user 消息与肥鱼自己通过 reply 回复的消息。

窗口大小用 contextLow 与 contextHigh 定义,限制“最多有几条最近的聊天消息保留在上下文中”。采用水位线淘汰策略,当消息数量超过 contextHigh 时,通过剔除最老的一部分消息,将其压到 contextLow 范围内。高低水位线是为了避免单阈值在满载后每次剔除最旧一条消息时上下文都会发生变动,从而破坏前缀缓存。

flowchart LRsubgraph S[单阈值:每来一条删一条]direction LRS1[到达上限] --> S2[追加 1 条
删最老 1 条] --> S3[头部每步都动
前缀缓存每步失效 ❌]endsubgraph D[双水位线:批量淘汰]direction LRD1[处于 low~high 稳定区] --> D2[只追加、不删
头部不动,缓存命中 ✅] --> D3[超过 high
一次性删到 low] --> D4[仅淘汰那刻失效一次
随后重回稳定区 ✅]end
  • 总会话条目

这条属于简单约束,即限制所有消息行数量。考虑到在一问一答的情况下,一次对话会有 5 条消息 —— 一条 user 消息,两条来自 reply 工具调用的往返,两条来自 finish_turn 工具调用的往返。

 新的群聊消息
 reply ...
 工具调用结果
 finish_turn ...
 工具调用结果

因此这个值应该比消息上下文的高位窗口大 5 倍甚至更多。

  • 最大 token 数量:字面意思,控制当前消息的最大 token 消耗。更多是为了避免单次会话上下文过长导致注意力被稀释。

(3)结束:

  • 超过本轮工具最大执行次数;
  • 或模型主动发起结束;
  • 或其他预期外的非法状态(修复后继续)。

2.3. 会话压缩与归档

由于每次请求都需要传入整个对话会话上下文,因此会话压缩几乎是必然的需求。否则肥鱼一次回复中调用的工具上下文就可能撑爆 Token 预算。

下面为会话上下文预算超限时执行的压缩与归档(compact_memory_compress)的粗略流程伪代码:

def compact_memory_compress(chat_context):compact(chat_context) #(1)紧凑化if not require_archive(chat_context):returnif memory(chat_context):compress(chat_context)

compress 阶段会损失上下文信息,因此必须等归档(记忆)任务成功后再继续。

(1)紧凑化:

为了尽可能长时间地保留上下文,即便上下文超出了设定的预算,工作流也不会立刻对上下文进行压缩,而是会先尝试紧凑化处理,如果紧凑化后预算回到范围内,则跳过记忆与压缩阶段。

紧凑化指的是将上下文消息中的工具调用结果进行简化,主要处理步骤是将长文本(例如网页)与图像替换为基于 SHA256 内容寻址的“引用消息”,以达到大幅减少上下文长度以保留对话信息的目的。需要注意的是,作为代价,这一步会破坏一次前缀缓存。

对于可能返回大量内容的工具调用——例如抓取网页、读取图片(以 base64 传输)——会同时保存两种结果:一种是原始结果,另一种是存到本地后基于 SHA256 的内容寻址引用:

flowchart LRA[工具执行结果] --> B[原始结果]A --> C[SHA256 引用]B -.存储.-> D[(内容存储)]C --> E[返回给 LLM]E -.按需取回.-> D

紧凑化的具体工作就是在压缩阶段尝试将请求上下文中的工具结果替换为引用文本,例如将一段图像结果替换为 [[image:]] 这一形式。具体来说:

{"role": "tool","tool_call_id": "call_00_read_img","content": [{"type": "text","text": "找到的照片"},{"type": "image_url","image_url": {"url": "data:image/png;base64,iVBORw0KGgoAA...(几百 KB,示例截断)","detail": "high"}}]
}

会被转化为:

{"role": "tool","tool_call_id": "call_00_read_img","content": [{"type": "text","text": "找到的照片"},{"type": "text","text": "[[image:3f9a1c2d4e5f60718293a4b5c6d7e8f90123456789abcdef0123456789abcdef]] "}]
}

紧凑化会跳过对最后一次工具调用结果的处理,以避免陷入下面这种死循环:

flowchart LRA[上下文预算不足] --> B[工具结果转为引用]B --> C[LLM 读取引用原文]C --> D[下一轮上下文再次超预算]D --> B

2.3.1. 记忆任务

这一部分的详细描述见“记忆”一章,这里只做简单的流程描述。

%%{init: {"flowchart": {"wrappingWidth": 340, "nodeSpacing": 30, "rankSpacing": 45}}}%% flowchart TDIn["需要归档"] --> Readsubgraph Memory["执行记忆任务"]direction TBRead["先看已经记过什么
记忆类型 / 相关日记 / 必要时翻历史"]Read --> Judge["从这段聊天里辨认
发生了什么、谁是谁
哪些事实、偏好、认识值得留下"]Judge --> Kind{"放哪里?"}Kind -- "长期记忆" --> M["更新长期记忆
稳定认识 / 事实 / 人物 / 未完事项
按模块写回最新全文(先读后写)"]Kind -- "当天日记" --> D["补写当天日记
具体经过 / 当天细节
一天一篇(先读后写)"]Kind -- "不写" --> Skip["普通寒暄 / 一次性细节"]D --> DoneSkip --> DoneM --> Done["归档完成"]endDone --> Out["..."]

2.3.2. 重建窗口

重建窗口用于在上下文超限时压缩上下文,以将上下文长度缩减到一个可以承担后续多次对话的范围中。

窗口重建的一个重要原则是尽可能保留近期消息的完整性,为此我将最近的 5 条消息定义为受保护块,在重建的时候尽量保留这 5 条消息的完整性。

def compress(chat_context):# 最近 5 条真实 user 消息及之后完整保留# 把旧段(受保护块之前的历史行)简化(1)为一条聊天消息一条 usersimplify_old_chats(chat_context)while has_old_simply_chats(chat_context):if not chat_context.is_overflow():returndrop_oldest_chats(chat_context)# 无旧段且依然超限,则开始强制简化保护段# 从最早的一段开始,直到只剩最后一段while has_chats and full_chat_count() > 0:if not chat_context.is_overflow():returnforce_simplify_oldest_chat(chat_context)if chat_context.is_overflow():warn()

(1)简化:将本轮对话中的 assistant / tool 细节移除,转化为一行一条、表示聊天记录的 user 消息。

2.4. 定时任务与主动激活

为了让模型在没收到消息时也能够主动找机会发送消息,有两种方式:

  1. 每次轮询时,让模型检查一下近期聊天记录,并附加提示词,让其决定要不要说什么。
  2. 采用定时任务触发,到点后让模型根据近期聊天记录决定要不要说点什么。

轮询判断成本过高,因此定时任务很有必要:

flowchart LRA[(定时队列)] --> B{每轮轮询:任务到点可执行?}B -- "否" --> BB -- "是" --> C[交给模型处理]

目前定时任务的结构设计如下:

{"uuid": "3f9a1c2d-8e4f-4a6b-9c1d-2e3f4a5b6c7d","meta": {"gid": "1087126969","createdAt": 1750000000000,"creator": "self","repeat": true,"cron": "30 20 * * *"},"lastFireAt": 0,"nextFireAt": 1750123456789,"message": "平时这时候群里会约游戏,今天一直没人说话;看一眼,还没人就问一句"
}

得益于整体轮询的方案,很多事情都可以很简单地实现。在定时任务的具体实现上,我采用内存存储的方案,每次轮询时从中取出已到点(nextFireAt <= 当前时间)且没有过期(超过 1 小时)的一条任务并处理。同时在定时任务队列更新时持久化(仅用于在重启时恢复)。

定时任务有两种类型,一种是一次性的,用完即扔;另一种则是重复任务,以 cron 表达式设定时间。

主动激活:

尽管每次都让模型判断并不现实,但如果配合概率判定,结合轮询来主动激活模型也并非不可:

while True:if should_active():ask_deepseek("找点事做") 

其中 should_active 是一个基于某种概率算法的函数,每次调用时随机返回 true / false。这样只需要使用简单的概率模型就能实现“每 N 小时期望激活模型 M 次”。

原本主动激活我是想通过定时任务简单完成的,但实践下来我发现如果要以此再实现“随机激活”,会给定时任务增加一些复杂度,而这个复杂度仅仅是为了支撑这种简单的功能,毫无必要。最后我选择将其作为独立功能。

3. 记忆

尽管检索技术已经有相当成熟的工程解决方案,然而 LLM 长期记忆的核心难点并不是单纯地记住内容,或者说“记忆”远远不是“能找到记录”那么简单。其核心是以下四种能力:

能力 核心问题 难度
Formation 什么信息值得记住? 高
Consolidation 如何将碎片信息整理成稳定记忆? 很高
Retrieval 当前应该回忆哪些信息? 很高
Evolution 如何更新、修正和遗忘旧记忆? 很高

另一方面,人类的记忆有以下三种关键特点:

  1. 人类不会平等地记住所有事情。
  2. 人类会主动整理和重构记忆。
  3. 线索驱动的记忆召回。

其中第 1 点可以通过设计一个记忆优先级模型尝试改善,第 2 点也可以定期让模型自行整理来实现。而真正困难的在于第 3 点:人类并不是每次思考都遍历自己的全部记忆。当前的环境、话题、人物、情绪和目标,都会成为触发历史记忆的线索。这一困难远远不是一个嵌入模型 + 向量库能解决的。这也是人类记忆的核心特点:记忆并不是被动存储的,而是与当前认知活动紧密联系的。

提这一点并不是为了说明我解决了什么问题,恰恰相反,事实上我自己目前也没有找到比较好的解决方案,所以关于这一章的内容可能很难有什么学习价值。提这一点主要是为了说明设计一个良好的记忆系统并非易事,要做好投入大量精力的准备。

目前而言,我的整个记忆体系的设计概览如下:

flowchart TDsubgraph CHANGE["变迁层"]MEM[("模块化记忆:按内容性质划分的记忆")]endsubgraph DYNAMIC["动态层"]NOTE[("note
跨群长期笔记
按需写入 / 查询")]endsubgraph PERSIST["持久层"]HIST[("history
原始聊天记录
逐字、全量保留")]DIARY[("diary
按天记录具体经历")]endREPLY["常规任务"]ARCHIVE["归档任务"]REPLY -->|"每轮自动注入"| MEMREPLY -.->|"按需"| NOTEREPLY -.->|"按需"| HISTREPLY -.->|"按需读"| DIARYARCHIVE -->|"读窗口 / 更早记录"| HISTARCHIVE -->|"按模块更新"| MEMARCHIVE -->|"写当天经历"| DIARYARCHIVE -.->|"按需"| NOTEstyle CHANGE fill:#eef4ff,stroke:#b9cce8,stroke-width:1pxstyle DYNAMIC fill:#f0faf4,stroke:#bfe3d1,stroke-width:1pxstyle PERSIST fill:#f7f9ff,stroke:#cbd8ef,stroke-width:1px

3.1. 记忆类型

模块化记忆(memory):

实际上就是将记忆内容划分成多个维度,将记忆放在不同维度中。一方面是让记忆更清晰,另一方面则是为了尽可能减少单次记忆更新的文本量。预定义维度如下,同时允许肥鱼自己定义维度:

维度 收什么
自我认知 在这个群实际参与中逐渐形成的兴趣、关注点、看法、对相处的期待;可以变化,不急着给自己贴完整人格标签
经历 对群、对关系有持续意义的共同事件:发生了什么、结果如何、为什么以后可能再提起
事实 以后还会用到的已确认信息、群内约定、术语和黑话;保留必要时间、适用范围与来源
人物 身份与别名对应、明确表达的偏好、长期项目、交流边界,以及有依据的关系认识
表达习惯 类似场景下目前适用的说话方式:解释深度、玩笑分寸、回应顺序、某位群友喜欢的交流方式
未完事项 尚未解决的问题、正在推进的事、已经答应的事、等待的反馈;保留对象、状态和后续线索

原始聊天记录(history):

原始的聊天记录数据,经过不会导致信息丢失的修改后直接存入本地,并存储相关附件(如图片)。

所谓不会导致信息丢失的修改,例如为带图片的消息对象新增一条 sha256 字段,用于记录这条消息中该图像的哈希值,此后取消息时可以直接按内容寻址从本地中读取,避免图片链接失效等问题。

日记(diary):

以天为粒度的记录,在记忆阶段让肥鱼记录/更新当天的聊天情况。

笔记(note):

任何时候都可以自由读写的内容,主要定义如下(JSONC 表示):

{// 对外递增 id"note_id": 42,// 标签"tags": ["测试", "约定"],// 正文"content": "周五晚上准备开黑,先记一下。",// 重要性:0–100"importance": 80,// 附件在 blob 库里的 sha256;空串表示没有附件"attachment": "a1b2c3d4e5f6...",// 附件类型:image / text / file;没有附件时为空串"attachment_kind": "image",// 首次创建时间,RFC3339 UTC"created_at": "2026-10-30T03:20:00Z",// 最后一次更新时间,RFC3339 UTC"updated_at": "2026-10-30T04:35:00Z"
}

3.2. 记忆独立

目前的设计是将每个群的记录与记忆独立,一方面是通过隔离,可以方便地测试各种改动的影响;另一方面也是为了降低肥鱼取到不恰当的数据导致上下文污染的几率。当然,代价就是每个群的肥鱼表现都各不相同。

目前先观察效果,等确实有需要的时候再考虑共享记忆。

但 note 不是跨群的,这算是刻意留的口子。

3.3. 记忆安全

所谓记忆安全,就是“经历过的事在被处理后,未来是否能找回原始记录”。对于聊天来说,原始记录就是聊天记录,因此当收到一条聊天记录后,在正式进入处理管线前,需要先做以下处理:

  1. 将聊天消息的附件内容拉取到本地中。
  2. 将聊天消息的原始内容存储到数据库中。
  3. 拉取合并转发的消息。

以确保消息在被肥鱼处理前就已经能回源。

流程限制:

整个记忆更新机制依赖于一组工具,在常规处理管线中只允许运行只读性质的工具(如读取聊天记录),只有进入记忆处理节点时才会允许使用修改性质的工具(如更新记忆)。

flowchart LRR["常规回复
工具:只读查询"] --> Q{"上下文溢出?"}Q -- "否" --> RQ -- "是" --> M["记忆归档
工具:记忆 / 日记读写"]M --> R

先读后写:

为了避免肥鱼盲写记忆,我用了一个在程序上强制先读后写的机制:每次更新内容后,会随机生成一个 6 位长度的 token(令牌),在下一次写入记忆时,必须传入上一次记忆 token,而要获取该 token,就必须先将记忆读出来(首次写入除外,可以直接传入空 token)。如下图所示:

sequenceDiagramparticipant M as 模型participant P as 程序participant D as 记忆库M->>P: memory_read(type)P->>D: 读该类最新一份快照D-->>P: 正文 + tokenP-->>M: 正文 + tokenM->>P: memory_update(type, token, 新全文)P->>D: 比对传入 token 与最新快照alt token 一致P->>D: 写入新快照(随机生成新 token)P-->>M: 已写入,新 token: xxxxxxelse token 不一致P-->>M: 拒绝写入,先重新 memory_readend

追加写入:

记忆一旦写入就不原地更新,新的记忆采用当前时间戳保存全量文本。在读取记忆时,仅返回时间戳最新的那一份记忆:

flowchart LRW1["t1:写入全量 A"] --> W2["t2:写入全量 B"] --> W3["t3:写入全量 C"]W3 --> R["读取:只返回最新一份 C"]R --> Ctx["进入回复上下文的是 C"]W1 -. "保留在库里" .-> H[("历史快照")]W2 -. "保留在库里" .-> H

4. 工具

4.1. 工具主导

原本我是让模型输出固定的 JSON 格式,来决定消息怎么发,发给谁。大概就是下面这样的结构:

{// 回复的一段一段话,ref_message 表示消息引用"segments": [{ "ref_message": "", "content": ""}],// 如果模型不想回复,简短给出不回复的理由"whyEmpty": "解释为什么不回复"
}
flowchart LRsubgraph old["旧:固定 JSON 输出"]A1["模型输出 JSON
segments / whyEmpty"] --> A2{"JSON 合法?"}A2 -- 否 --> A3["兜底重试 / 修复"] --> A1A2 -- 是 --> A4["程序解析并逐条发送"]end

实践下来我发现这样有两个问题:

  1. 模型并不总是能给出正确的 JSON 响应,需要兜底重试。
  2. 最重要的是,这样做确实很别扭。

但必须保留模型“回复多段话”的能力,最终我的选择是将“回复”这一操作也作为一种工具提供,将模型真正的输出仅加入到 prompt 上下文与日志记录,如下:

flowchart LRsubgraph new["新:回复也是一个工具"]B1["模型正常输出文本"] --> B2{"要说话吗?"}B2 -- 说 --> B3["调用 reply 工具
一次调用 = 一条消息"]B3 --> B4["要多段就再调用一次 reply"]B2 -- 不说 --> B5["调用 finish_turn
说明原因"]B1 -. "正文(output)只记录到日志" .-> B6[("日志")]end

这样显然也更自然。

核心工具组:

这是我自己的定义,主要就是区分哪些工具是核心的、决定性的、应该重点维护的;哪些工具更多只是为了“好玩”或服务于某种特定的目的。核心工具缺失会导致关键功能不可用,非核心工具缺失只是在某类场景下弱化表现:

工具名 作用
reply 发送一条消息
finish_turn 明确结束本轮
read_* 内容读取工具组:按 sha256 从内容库取回内容
history_* 聊天记录工具组
memory_* 长期记忆工具组
diary_* 日记工具组
schedule_* 定时任务工具组
note_* 笔记工具组

另外给出部分非核心工具组示例:

工具名 作用
web_search / web_read 联网搜索与网页读取
qq_card_* 处理 QQ Card 相关的事项(如读 B 站分享卡片)
artifact_query / artifact_summary 自动抓取 / 下载产物的索引与概览
computer_* 肥鱼的电脑控制

动态调用:

有的工具比较小,将其作为一个独立的工具提供会多占一个工具名额,过于浪费。对于这类比较小巧的工具,更好的办法是采用动态调用。即通过一个工具(如 dynamic_call(name, args))作为入口,按照传入工具名和参数动态调用对应的工具。

sequenceDiagramparticipant M as 模型participant D as dynamic_discoverparticipant C as dynamic_callparticipant R as dynamic 注册表participant T as 小工具 A / B / CNote over M: 请求的 tools 里只有
dynamic_discover / dynamic_callM->>D: 先调用 discoverD->>R: 查询已登记的小工具R-->>D: 工具名 / 描述 / 参数签名D-->>M: 返回动态工具清单Note over M: 模型按签名选择要调用的工具M->>C: call(name, args)C->>R: 按 name 查找对应工具R->>T: 执行小工具T-->>R: 返回结果R-->>C: 返回结果C-->>M: 工具结果Note over R,T: 小工具只在注册表里
不占请求的 tools 数量

所谓的“占用工具数量”是指为模型提供的 tools 应该控制在一定数量(< 50)中,以避免工具过多影响模型判断。

5. 肥鱼的电脑

我并没有直接将本机交给肥鱼控制,我的想法是给肥鱼一台近乎完全隔离的电脑,让肥鱼无论怎么在里面折腾都不会有什么风险。且电脑应当有以下特征:

  • 内置 Python 与众多常用功能性包(如用于图像处理的 PIL,处理文档的 pandoc)作为预装工具,以使电脑在完全无网环境的情况下也能有用武之地。
  • 内置 Chrome 与 Playwright,以使浏览器可用。
  • 可限制的硬件资源(例如内存 2G),以避免因奇奇怪怪的任务导致过度占用主机资源。
  • 无持久化,销毁即清空。

我的做法是专门构建了一个 Docker 镜像作为肥鱼的电脑,以一个简单的控制层来控制容器(电脑)的启停与命令执行。大概机制如下:

flowchart LRF["肥鱼"] -->|"computer_* 工具"| C["控制层
会话 / 执行 / 回收"]C -->|"Podman socket"| P["电脑容器
每群一台"]P --> R["runtime 镜像
只读根 + /workspace + 预装工具"]

而为了让肥鱼能够尽可能自由地控制电脑,至少需要提供以下工具(最小集合):

flowchart LRA["电脑控制能力"] --> B["生命周期控制"]B --> B1["computer_create"]B --> B2["computer_dispose"]A --> C["命令执行"]C --> C1["computer_exec
前台 / 后台任务"]A --> D["文件操作"]D --> D1["computer_put"]D --> D2["computer_get"]A --> F["环境信息"]F --> F1["computer_packages"]

另一方面,还可以一些辅助性的工具,如 computer_peek 直接查看电脑中的文件,避免先 get 再 read 浪费 token;再比如 computer_exec_then_peekfile 一次性实现“执行一条命令+直接查看电脑中的指定文件”。

生命周期控制:

一台“电脑”的生命周期如下:

stateDiagram-v2state "创建中" as Creatingstate "运行中" as Runningstate "已回收" as Reapedstate "已丢失" as Lost[*] --> Creating: computer_createCreating --> Running: 容器启动成功Creating --> [*]: 创建失败Running --> Running: computer_exec / put / getRunning --> Reaped: 空闲 300 秒或总存活 1800 秒Running --> Reaped: computer_disposeRunning --> Lost: 外部删除 / 调度层重启Reaped --> [*]Lost --> [*]note right of Running/workspace 跨多次 exec 保留;后台任务或待投递通知会延后空闲回收end notenote right of Reaped容器删除,workspace 清空computer_id 失效,需要重新 createend note

命令执行:

调度层直接在容器中执行传入的任意命令,并返回 stdout / stderr。同时命令执行还需要引入后台执行选项,将任务放到后台运行,并在完成时通过通知机制通知肥鱼,以避免执行长时任务卡住肥鱼。

文件传输与命令访问:

向容器传入/传出文件也是一种必要的功能,否则肥鱼的任何产出都无法带出来,并且会随着容器销毁而清空;另一方面,如果无法向电脑传入文件,则肥鱼也不方便“把收到的放到电脑中处理”(例如给表情 P 图)。

环境信息:

用于让肥鱼获取关于这台电脑的配置信息,包括但不限于基本使用提示、使用的 Linux 发行版、已安装的包、已安装的 Python 包、预定义的各种辅助脚本等内容。以此提高肥鱼的决策效率。

5.1. 网络控制

出于安全考虑,调度层默认采用 --network none 启动容器,因此默认情况下电脑完全无网(但预制工具和 Python 包已经足够让肥鱼在无网状态下做很多事,甚至还能让肥鱼自己做游戏自己玩)。

但有时候确实有必要提供网络(例如想让肥鱼刷 B 站)。为此我用了一种一次性 token 机制,在肥鱼启动电脑(computer_create)时如果传入正确的 token,则调度层会启动带有网络连接的容器:

flowchart TDT["提供一次性 token"] --> A["肥鱼调用 computer_create(network_token)"]A --> C{"调度层校验 token"}C -->|"有效"| N["创建联网容器"]N -.->|"token 立即失效"| G["生成新 token"]G -.->|"供下次使用"| TC -->|"缺少 / 错误 / 未启用"| O["创建离线容器"]

6. Prompt 组织

按照尽可能吃到前缀缓存与利用提示词的位置效应,整个 Prompt 的构建管线如下:

flowchart TDP["persona.md"] --> SYS["system
persona + active"]ACT["active.md"] --> SYSSYS --> TOOLS["tools
全量工具定义"]TOOLS --> MEM["user
记忆(memory)"]MEM --> CHATS["messages
聊天消息与提示性信息"]CHATS --> TASK{"附加任务说明"}TASK -->|"常规"| REQ["请求模型"]TASK -->|"归档"| INS["末尾再追加简短的 instruction
归档说明"]INS --> REQclassDef note fill:#fff8e6,stroke:#e0b44c,stroke-width:1px,color:#6b4b0a;

System 中的 persona.md 为人设定义,active.md 则为任务说明(例如说明如何回复,如何更新记忆)。

即无论什么任务,system / tools / memory / chats 这几项的内容与排列顺序都必须完全相同,仅在特殊任务(如记忆任务)的末尾追加一段 instruction,以此尽可能复用前缀缓存。当然,作为代价,必须满足以下条件:

  • active.md 文件中必须写明所有可能的任务(回复、记忆等)需要遵循的要求;
  • 工具(tools)必须始终全量提供,仅在运行时做拦截(例如在常规任务中不允许用更新记忆的工具);

具体实现实测下来缓存命中率在 99.4% 左右。

7. 容错

7.1. 消息补漏

这一段先放着,只简单提一下。所谓消息补漏,就是系统离线/停机后如何补回期间的消息。主要考虑以下情况:

  • 系统离线后重新登录时如何补漏。
  • 防止因为漏掉的消息过多而导致补漏消息撑满上下文预算。
  • 补漏过程和实时消息接收发生竞态(打包处理与去重)。

补漏时机:系统启动时、运行中的定时补漏。

7.2. 工具失败与重试

这一段也先放着,简单来说就是如果工具调用出现了异常应该如何修正。主要考虑以下情况:

  • 协议形状错误(如 tool_calls 缺 id / function / name)。
  • 使用了未知工具 / 当前任务不可用的工具。
  • 工具自身抛出预期外的异常。
  • 会引起副作用的工具返回了不确定的结果(如调用会发送消息的 reply 后得到无法判断消息是否发送成功的结果)。
  • 工具批次不完整:悬空 tool_calls(存在 assistant 的 tool_calls 没有对应的 tool 结果)。
  • 轮次与预算(工具有轮次上限;接近上限时提醒模型收尾)。
  • 中断与并发:处理期间来新消息时,在当前工具批次结束后中断本轮;已执行的工具结果必须完整落地,不能留下没有结果的调用。