☰
自建AI助手实战:持久任务、沙箱执行与群聊协作
2026/10/13 7:56:14 网站建设 项目流程

一年多以前,我决定自己从零搭一个AI助手——不是那种聊两句就完事的对话玩具,而是能接持久任务、能在沙箱里安全执行代码、能被拉进群聊当团队一员的正经工具。现成的产品我试了一圈,没有一个能同时满足这三件事:会话一关就失忆,生成代码不敢让它真跑,多人协作场景更是各聊各的。于是干脆自己动手。这篇文章把整个搭建过程中的架构设计、核心模块实现、踩过的坑和最终调优方案全部写出来,适合有一定开发基础、想自己搞一个类似系统的人直接参考。

1. 先搞明白:一个"能干活的AI助手"到底需要什么

1.1 三个绕不开的硬需求

动手之前,我花了很长时间想清楚一个问题:我到底不满意现有产品的哪一点?最后归纳成三条。

第一是记忆的持续性。大多数AI产品是"会话式"的,每个请求独立处理,上下文只在当前对话窗口里有效。我要的是能连续跟踪项目的助手:周一让它调研一个技术方案,周五它还能记得当时的背景、约束和已经确认的结论,而不是每次都重新寒暄一遍。这一点直接决定了系统里必须有"持久任务"模块。

第二是执行的安全边界。助手如果只能输出文字建议,价值很有限。真正有用的是它能根据任务描述自动写代码、跑脚本、操作文件、调用接口,然后汇报结果。但这里有个核心矛盾:让它跑代码,就等于把执行权交给一个可能出错、可能被恶意提示词引导的系统。沙箱不是可选项,是安全底线。

第三是协作形态。单人对话好办,现实中很多工作是团队推进的。助手需要同时服务多个人,在群聊里被@、被指派任务、汇报进度,还得保证A群让助手做的事不会被B群的请求干扰。群聊接入不是简单接个机器人接口那么简单,背后是会话隔离和权限控制的问题。

1.2 需求到方案的对应关系

把这三条需求翻译成技术方案,对应关系非常清晰:

核心需求技术手段解决的关键问题
持续记忆任务状态机 + 上下文快照长任务不跑偏,中断可恢复
安全执行容器沙箱 + 静态扫描 + 资源限制AI代码出错不炸宿主机
多人协作消息网关 + 会话空间隔离多群多用户互不干扰

这个表格我贴在最显眼的地方,后面所有代码都是围绕这三行展开的。现在回看,最大的教训就是:需求没想清楚之前,不要急着写代码。我早期版本就是反着来的,先写了个能聊天的壳子,结果回头补记忆、补沙箱、补群聊,每一步都在重构,代价非常高。

1.3 选型思路:AI只做大脑,脏活交给成熟组件

想清楚需求后,我的路线很朴素:把传统软件工程里已经成熟的东西搬过来,AI只做"大脑",负责理解任务、拆解步骤、产出计划;脏活累活交给消息队列、任务调度器、容器沙箱和数据库。AI不直接碰系统,所有动作都通过受控的API完成。

选型上,主程序用Python,生态里处理文本、调用模型接口都方便。任务队列用Redis这类内存中间件,数据量不大、延迟低、部署简单,运维成本几乎为零。沙箱基于容器技术,用现成的容器运行时管理镜像和生命周期。数据库先用SQLite跑通流程,后期数据量上来了再迁移到PostgreSQL。模型接口走的是通用的OpenAI兼容协议,这样随时可以换供应商,不被某一家绑死。

这套选型最大的好处是每一层都能单独替换。任务队列可以换,沙箱可以换,模型可以换,连消息来源都可以换。后面你会看到,这种可替换性在踩坑调优的时候有多重要——很多问题不用改业务代码,换一个组件就解决了。

2. 整体架构:消息网关、任务调度与沙箱的协作方式

2.1 一个任务从产生到闭环的完整生命周期

在设计初期,我没急着写代码,而是先定义了一个任务从产生到结束的完整路径,这一步帮了大忙。一条典型路径是这样的:

用户在群聊里@助手并发出一条指令;消息网关把指令转成结构化请求,交给意图解析模块;意图解析模块调用模型,把它翻译成一个任务对象,包含任务类型、参数、执行方式;任务对象进入队列,由调度器按优先级取出来;调度器创建一个沙箱实例,把任务描述和必要的数据放进去;沙箱里的工作进程调用模型生成执行代码,在受限环境中运行,不断记录进度;每完成一步,状态回写数据库;整个任务结束后,结果通过消息网关推回群里。

用户全程只看到"收到任务→执行中→完成",中间所有细节都沉淀在任务记录里。这个生命周期定义是整个系统的地基,后面所有模块都是在为这条路径上的某个环节服务。

2.2 四个核心组件各自的边界

这个架构一共四个核心组件,各自的职责必须划得非常清楚,不能有重叠:

  • 消息网关:负责对接不同的消息来源,把各种平台的消息统一转成内部格式,同时把助手的回复翻译回对应平台。
  • 任务调度器:维护任务队列,处理优先级、重试、超时,是任务生命周期的大脑。
  • 沙箱执行器:真正干活的地方,负责创建隔离环境、注入依赖、执行代码、回收资源。
  • 状态存储:记录任务状态、上下文快照、执行日志,是整个系统的"记忆"。

组件之间通过内部API通信,消息格式统一为JSON。这里有个容易被忽略但极其关键的点:任务状态的流转必须设计成幂等的。调度器可能因为重启、网络抖动重复推送同一个任务,如果处理逻辑不幂等,同一个任务被执行两次,后果可能是数据重复写入、通知发两次,在真实场景里非常致命。

我的解决办法是给每个任务分配全局唯一的task_id,数据库里加唯一索引,执行前先检查状态,已经处于completed或running的任务直接忽略。这个设计在后面处理群聊消息重复推送时也派上了大用场。

2.3 为什么没直接用现成的智能体框架

可能有人会问:现在已经有挺多现成的智能体框架,为什么不直接用?我试过,结论是:通用框架确实能快速跑通demo,但一旦涉及持久任务、沙箱和群聊这三个维度,就会遇到严重的扩展问题。很多框架把"单轮对话+工具调用"做得很好,但对任务的持久化、断点续跑、多租户隔离、多平台消息适配,要么没有抽象,要么抽象得很别扭,改造起来比从零写还费劲。

自己搭这套,代码量确实多了一些,但每一层都是自己可控的。出现问题的时候,我可以直接看队列积压情况、沙箱的资源快照、数据库里的任务状态,定位问题通常花不了几分钟。用通用框架时,出问题经常要在框架源码里翻半天,那个体验谁用谁知道。

3. 持久任务模块:把AI的短时记忆变成项目记忆

3.1 任务状态机:只允许合法路径流转

持久任务的核心,是给每个任务定义清晰的状态机。我最终使用的状态集合如下:

pending(排队中)→ running(执行中)→ paused(暂停)→ running(继续) → completed / failed / canceled

每个状态流转都有触发条件和校验逻辑。比如,只有pending和paused状态的任务可以被调度器拉取;运行中的任务不允许重复进入队列;failed状态的任务需要记录失败原因,并按照配置决定是否自动重试。代码里用一张状态流转表来控制合法性:

ALLOWED_TRANSITIONS = { "pending": {"running", "canceled"}, "running": {"paused", "completed", "failed", "pending"}, "paused": {"running", "canceled"}, } def transition(task_id, new_state): current = get_state(task_id) if new_state not in ALLOWED_TRANSITIONS[current]: raise InvalidTransition( f"task {task_id}: {current} -> {new_state} 不被允许" ) update_state(task_id, new_state)

数据库表设计上,我保留了几个关键字段:task_id(全局唯一)、task_type(任务类型)、params(参数JSON)、state(当前状态)、progress(进度百分比或描述)、context_snapshot(上下文快照)、result(最终结果)、error_msg(错误信息)、created_at和updated_at(时间戳)。其中context_snapshot是重头戏,它决定了AI能否"接上之前的话"。

3.2 上下文快照:断点续跑的关键

所谓"持久",本质上就是把大语言模型的短时记忆外置到数据库里。模型本身不记忆任何历史,所有记忆都是我们主动写进去、再在需要时取出来的。

我的做法是:每个任务维护一份结构化的上下文,包含七类信息——任务目标、已完成的步骤、当前所在阶段、关键决策及理由、待办事项、相关文件路径、历史报错摘要。每完成一个子步骤,调度器就调用模型对新增内容做一次摘要,把摘要合并进上下文快照,写回数据库。

断点续跑的逻辑很简单:任务因为超时、崩溃或手动暂停而中断后,重新恢复时不是让模型"从头想想",而是把上下文快照整体塞回提示词,再附上一条指令:"基于以上进度继续执行,不要重复已完成步骤。"实测下来,这个设计让长任务的成功率提升非常明显。早期版本没有快照机制,一个需要跑20分钟的数据处理任务,往往在第三次调用模型时就跑偏了,因为它已经忘记了最初的约束。

需要注意一个细节:上下文快照不能无限膨胀。模型上下文窗口有限,摘要合并虽然能压缩,但几十轮之后还是可能爆掉。我在快照表里加了size字段,超过阈值就触发"深度压缩",让模型把旧内容提炼成更紧凑的结构化要点,必要时丢弃过细的中间过程,只保留结论和证据。这个阈值要在任务开始前就设好,别等报错再处理。

3.3 队列可靠性:进程重启也不丢任务

持久任务最怕的就是进程挂了、任务丢了。消息队列本身能扛住一定程度的故障,但有一个坑:如果任务已经出队开始执行,执行器进程却崩了,队列里和数据库里的状态就会不一致。

我的处理策略是"数据库为准"。任务调度器把任务从队列取出后,先更新数据库状态为running,并写入执行器实例ID和心跳时间;执行器每30秒上报一次心跳。恢复程序每隔几分钟扫描一次,发现某个running状态的任务心跳超时,就把它的状态改回pending,重新入队。整个过程有一个硬约束:重试次数上限,超过上限直接标记failed,避免死循环。

CREATE TABLE tasks ( task_id TEXT PRIMARY KEY, task_type TEXT NOT NULL, params TEXT NOT NULL, state TEXT NOT NULL DEFAULT 'pending', progress TEXT, context_snapshot TEXT, result TEXT, error_msg TEXT, retry_count INTEGER DEFAULT 0, created_at TEXT DEFAULT (datetime('now')), updated_at TEXT DEFAULT (datetime('now')) ); CREATE INDEX idx_tasks_state ON tasks(state); CREATE TABLE heartbeats ( task_id TEXT PRIMARY KEY, executor_id TEXT, last_beat_at TEXT );

这个方案比依赖消息队列的ack机制更稳,因为数据库是唯一的真相来源,队列只是传输管道。实际运行中,我经历过两次宿主机重启,恢复后都有任务自动续跑,没有一个丢失,验证了这个设计是靠谱的。

4. 沙箱执行环境:在"放权"和"安全"之间划一条线

4.1 一次事故让我下定决心

让AI生成代码并执行,最担心的是两件事:模型可能写出有缺陷的代码导致资源泄漏,恶意提示词可能诱导模型执行危险操作。真实工具场景下,你无法保证每次对话都是善意的,所以执行环境的隔离不是"建议",而是"必须"。

我最早图省事,直接在宿主机上跑AI生成的代码,结果一个死循环脚本把CPU全部占满,整个服务差点卡死。那次事故之后,我彻底改成容器沙箱,再也没发生过类似问题。在安全这件事上,一次事故的教训比十篇文档都管用。

4.2 容器沙箱的核心参数配置

我用的方案是基于容器运行时的隔离,每个任务启动一个独立容器。核心配置如下:

配置项默认值说明
CPU1核,峰值2核防止死循环占满宿主
内存512MB,上限2GB内存型任务单独上调
磁盘写层1GB限制日志和临时文件膨胀
网络默认全隔离联网任务单独开白名单
超时10分钟到点直接杀容器,可配置
文件系统系统目录只读防止误改系统配置
# 创建沙箱容器的核心参数 container run \ --cpus 1 \ --memory 512m \ --memory-swap 1g \ --pids-limit 256 \ --read-only \ --tmpfs /tmp:rw,size=128m \ --network none \ --stop-timeout 30 \ --name task-xxx \ 基础镜像:版本

一个容易被忽视的配置是文件系统只读。任务的工作目录下允许读写,系统目录全部只读,可以有效防止AI误改系统配置。另一个是--pids-limit,限制进程数量能防止fork炸弹类的攻击,这个参数很多人会漏掉。

4.3 静态扫描与执行审计双保险

光靠资源限制还不够。AI生成的代码里可能出现明显危险的操作,比如删除关键目录、读取敏感文件、安装来源不明的包。我在沙箱前面加了一层静态扫描:代码生成后、执行前,先用规则库做一次危险模式匹配,命中高风险规则就不放行,并把拦截原因写进审计日志。

静态扫描防不住所有情况,所以还要加执行审计。沙箱里的所有文件改动、网络请求、进程创建都会记录到结构化日志。任务结束后,日志和任务结果一起归档。这样万一出了事,能回溯到具体是哪一步、哪段代码、哪个时间点,排查效率高很多。

两层叠加之后,安全策略才算完整:静态扫描拦在事前,资源限制挡在事中,审计日志兜底在事后。

4.4 沙箱启动性能的优化

沙箱启动是有开销的,任务很碎很频繁时,每个任务都新建容器会拖慢整体性能。我做了一层镜像精简,把基础镜像里用不到的软件包全部删掉,冷启动时间从几秒降到1秒以内。另一层是常驻预热容器池:维护几个空闲容器,任务来了直接复用。热启动几乎无感知,实测单任务从提交到开始执行的时间从平均3秒降到0.5秒以内。

容器池要注意回收策略。我设置了一个空闲时间阈值,超过5分钟未使用的容器自动销毁,避免内存被一堆僵尸容器吃掉。

5. 群聊接入:让助手成为团队里"有记忆"的成员

5.1 消息网关:一套内部格式通吃所有平台

群聊接入最忌讳的是和某个具体平台绑死。所以我先定义了一套内部消息格式,所有平台的消息都转换成统一结构:sender(发送者)、group(群标识)、content(文本内容)、mention(是否@了助手)、attachments(附件)。

消息网关包含两个方向:入站方向对接各个平台的机器人回调,把各种消息格式解析成统一结构;出站方向把助手的回复渲染成平台支持的格式。新增一个平台时,只需要实现这两个方向的适配器,其他模块完全不用动。核心处理逻辑长这样:

def handle_group_message(platform, message): parsed = gateway.parse(message) if not parsed.mention: return # 没@助手就不处理 if not permission_check(parsed.sender, parsed.group): return gateway.reply(parsed, "你没有权限执行该操作") task = create_task_from_message(parsed) scheduler.submit(task) gateway.reply(parsed, f"任务已受理:{task.task_id}")

5.2 会话空间隔离:每个群拥有独立记忆

群聊场景下的一个关键问题是:多个群、多个人共享同一个助手,但每个群的上下文必须是独立的。否则A群讨论的技术方案会被B群的请求干扰,整个体验就崩了。

我的做法是引入"会话空间"概念。每个群对应一个独立的话题存储,按群ID做隔离。助手被@后,首先根据群ID加载该群的上下文,再结合当前消息生成回复。回复内容只写入当前群的上下文,绝不会跨群污染。单人私聊也复用同一套机制,每个私聊用户就是一个独立的会话空间。

在数据库里,这就是一张简单的映射表:session_id对应一组上下文记录。隔离逻辑写清楚后,多群并发的问题就变成了普通的数据库读写问题,压力小了很多。

5.3 权限分层与并发任务的处理策略

团队协作场景还要处理权限问题。不是所有人都应该能触发所有任务。我在用户层加了角色概念:管理员、成员、访客。管理员可以触发部署、清理类高风险任务;成员只能触发查询、分析类普通任务;访客只能读取结果。权限校验在消息网关层完成,不合法请求直接拒绝并提示。

还有个实际问题:同一个群里,如果两个人同时各发了一条耗时任务,怎么处理?我的策略是:默认串行执行,同一会话空间内同一时间只允许一个任务运行,后续任务排队。这个策略牺牲了一点并行度,但避开了"两个任务同时操作同一份数据"的冲突。如果确实需要并行,任务可以显式声明互不冲突,调度器会把它们放入不同的沙箱并行执行。

串行策略还有个额外的好处:群里的执行结果按顺序展示,不会出现A的任务结果和B的任务结果交叉出现,阅读体验好很多。

6. 实战踩坑清单与调优经验

6.1 五个印象最深的坑和对应解法

第一个坑是模型输出格式不稳定。最初让模型直接返回JSON格式的任务参数,结果经常出现JSON截断、字段多一个逗号之类的问题。后面改成强制模型输出Markdown代码块包裹的JSON,并加了解析重试机制:解析失败一次,就把错误信息反馈给模型让它修复,最多三次。这个方法简单粗暴但非常有效,解析成功率从70%左右直接拉到95%以上。

第二个坑是上下文窗口溢出。早期没做摘要压缩时,一个长任务进行到后半段,提示词经常超过模型限制,直接报错。解决办法就是我前面说的深度压缩策略。关键是要在任务开始前就设好阈值,别等到报错再处理。我现在的经验是:上下文快照超过总窗口的40%就开始压缩,留足余量给后续步骤。

第三个坑是沙箱内的时区问题。容器基础镜像默认UTC时间,日志时间和宿主时间错开8小时,排查问题时一度很困惑。最终在创建容器时注入TZ环境变量解决,这个配置后来直接写进了镜像默认配置里。

第四个坑是群消息的重复推送。部分平台在机器人接收消息超时时会自动重推,导致同一条指令被处理两遍。解决方案是消息网关层面做幂等去重,按消息ID存哈希表,重复消息直接丢弃。这个坑和任务幂等设计一脉相承,系统里凡是涉及外部事件的入口,都要默认"可能重复"。

第五个坑是队列任务饿死。曾经给所有任务同等的优先级,结果大量短小任务把队列堆满,长任务迟迟得不到执行。后来改成按任务类型区分优先级,并且长任务允许抢占低优先级任务,问题才解决。任务队列的调度策略,建议一开始就按业务重要度设计,不要一刀切。

6.2 模型调用成本与沙箱资源怎么省

运行一段时间后,性能瓶颈主要出现在两个地方:模型调用次数和沙箱资源占用。

模型调用次数是成本大头。我做了一层模板化处理:很多重复性任务不需要每步都调用模型,直接走预置流程,模型只在任务偏离预期时才介入。第二层是对中间步骤结果做本地缓存,相同的子任务结果不重复请求。第三层是设置全局每日token预算,超预算后自动降级到更便宜的模型,保证服务不中断,只是效果略降。

沙箱资源占用方面,前面提到的常驻容器池帮了大忙。初始版本每个任务都新建容器,长时间运行后宿主机内存压力很大。改成池化复用后,内存占用降了大概40%。另一个容易被忽略的点是镜像版本管理:沙箱里装的依赖版本必须和代码仓库锁定的版本一致,否则会出现"本地跑得好好的,沙箱里全报错"的经典问题。

6.3 重做一遍的话,我会先做什么

整个项目从第一版能跑通到稳定运行,大概花了三周业余时间。最大的体会是:自建AI助手的难点不在于调用模型,而在于软件工程本身。记忆、隔离、并发、幂等、监控,这些传统后端领域的经典问题,在AI应用里一个都不会少,甚至因为AI的不确定性变得更难。

如果让我重新做一遍,我会在一开始就花更多时间设计状态机和数据模型,而不是急着跑通demo。前期多花一天设计,后期能省下好几天排查问题的时间。

最后分享一个很实用的小技巧:在沙箱里干活时,让AI每完成一个关键步骤就往日志里写一行"步骤名+结果摘要",这个习惯能让你在任务失败时快速定位到是哪一步出了问题,而不是对着模型的长篇输出发愁。我的所有长任务模板里都内置了这条要求,实测对排查问题帮助极大。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询