跳到主要内容

aibitat 内核:对话图与主循环

本章讲什么: AnythingLLM 的 agent 引擎叫 aibitat(名字源自上游早期 fork 的同名库)。要看懂工具循环,得先看懂它最底层的抽象——它不直接「调模型生成回复」,而是把对话建模成一张「节点互发消息的图」。本章讲清这个抽象、三个递归方法、以及消息历史的状态机。

1. 它要解决的小问题​

一个 agent 对话里,至少有两个参与者:用户和 agent。复杂场景下还可能有多个 agent 在一个「群」里轮流发言。怎么用一套统一的代码表达「用户→agent」「agent→用户」「群里选下一个发言人」这些不同的对话拓扑?

aibitat 的答案:把每个参与者当成图里的一个节点(node),把「A 对 B 说话」当成一条有向消息,然后用递归推进这张图。

2. 核心数据结构​

AIbitat 类(server/utils/agents/aibitat/index.js:15)持有几个 Map 和一个数组:

字段是什么
agents (Map)节点名 → agent 配置(role 系统提示 + functions 工具清单)
channels (Map)频道名 → 群成员列表(多 agent 群聊用)
functions (Map)工具名 → 工具定义(aibitat.function() 注册进来)
_chats (Array)整段对话历史,每条带 from/to/content/state

在 AnythingLLM 的实际用法里,只注册了两个 agent:USER(人)和 @agent(工作区 agent)。它们在 AgentHandler.#loadAgents 里被放进去(server/utils/agents/index.js:681),配置来自 defaults.js 的 USER_AGENT / WORKSPACE_AGENT(server/utils/agents/defaults.js:43、:53)。

消息历史是一个事件流,不是简单的 messages 数组。 每条 chat 带一个 state:

 state: "success"   ── 正常发生过的一轮对话(参与历史回放)
 state: "interrupt" ── 被中断,等人类输入(continue 时弹出)
 state: "error"     ── 出错,等 retry

getHistory({from,to})(index.js:1335)在回放历史时只取 state==='success' 的条目,所以 interrupt/error 的「占位条目」不会污染喂给模型的上下文。

3. 三个递归方法:start → chat → reply​

这是整个引擎的骨架。三者层层调用:

 start(message)
   │  newMessage(message)        把首条消息写进历史(state=success)
   │  emit "start"
   ▼
 chat({to: msg.from, from: msg.to})    ── 让对方回话
   │
   ├─ 若 from 是一个 channel(群):selectNext 选下一个发言人,递归 chat
   │
   └─ 否则(普通点对点):
        reply(route)            ── 真正生成一条回复(可能触发工具循环)
        │
        ├─ reply == "TERMINATE" 或 达到 maxRounds → terminate,结束
        ├─ reply == "INTERRUPT" 或 该 agent 配了 interrupt:ALWAYS → interrupt,挂起等人
        └─ 否则 keepAlive: 反过来 chat({to:from, from:to}) 让另一方接话

对应源码:start(index.js:562)、chat(index.js:582)、reply(index.js:857)。

关键点:chat 是自递归的。 一次 start 之后,chat 会让 USER↔@agent 来回接话,直到某一方说 TERMINATE、撞到 maxRounds、或被 interrupt。这就是「对话一直进行下去」的机制。

USER 节点为什么总是中断对话​

USER_AGENT 的定义里写死了 interrupt: "ALWAYS"(defaults.js:46-48):

const USER_AGENT = {
  name: "USER",
  getDefinition: () => ({
    interrupt: "ALWAYS",  // 轮到 USER「发言」时,永远停下来等真人
    role: "I am the human monitor and oversee this chat...",
  }),
};

效果:@agent 回完一轮后,chat 想让 USER 接话,但 shouldAgentInterrupt("USER") 返回 true(index.js:672),于是触发 interrupt——对话挂起,控制权交回真人。真人再说话时,continue()(index.js:1276)把挂起的占位条目弹出、塞入新反馈、再进 chat。这就是「agent 答完、等你下一句」的底层实现。

4. reply:一轮回复怎么组装​

reply(route)(index.js:857)是「让一个节点回话」的核心。它做四件事:

① 取历史 + 注入文档上下文。 getOrFormatNodeChatHistory 把 _chats 转成 {role, content} 的标准 messages(index.js:809)。然后如果挂了 fetchParsedFileContext 回调,会把最新的已解析文件/置顶文档追加到最后一条 user 消息里(index.js:862)——这个回调由 AgentHandler.#fetchParsedFileContext 提供(server/utils/agents/index.js:718),每轮都重取,保证上下文新鲜。

② 拼系统提示。 取该 agent 的 role 当 system 消息,放在历史前面(index.js:878)。@agent 的 role 是 Provider.systemPrompt() 算出来的——它会把 workspace 自定义提示、记忆(memories)拼进去(ai-provider.js 的 systemPrompt)。

③ 取该 agent 能调的工具,并可选地重排。 从 fromConfig.functions 里按名字查出工具定义(index.js:887)。如果开启了「智能技能选择」(ToolReranker.isEnabled()),会用 embedding 把工具按和用户问题的相关度重排、只留 top-N,显著省 token(index.js:892)。

// 这段在演示「按名字解析工具 + 可选重排」,重点看 #parseFunctionName 处理子工具命名
let functions = fromConfig.functions
  ?.map((name) => this.functions.get(this.#parseFunctionName(name)))
  .filter((a) => !!a);

if (ToolReranker.isEnabled() && functions?.length) {
  const userPrompt = this.#extractUserPrompt(messages);
  if (userPrompt) functions = await toolReranker.rerank(userPrompt, functions);
}

#parseFunctionName(index.js:768)处理三种命名:普通名;parent#child(取 child);@@custom(去掉 @@)——这套命名约定在第 03 章详解。

④ 按 provider 能力分流到工具循环。 取到 provider 实例后,看它支不支持流式 agent:

if (this.providerInstance.supportsAgentStreaming) {
  content = await this.handleAsyncExecution(messages, functions, route.from); // 流式
} else {
  content = await this.handleExecution(messages, functions, route.from);      // 同步
}

(index.js:933-947)。这两个方法就是下一章的主角——递归工具调用循环。

5. 群聊与「选下一个发言人」(边角,可略读)​

虽然 AnythingLLM 的产品里只用了 USER/@agent 两节点,但 aibitat 内核保留了多 agent 群聊能力。当 chat 的 from 是一个 channel 时,selectNext(index.js:685)会:

  • 过滤掉已达 maxRounds 的成员、以及上一轮刚发言的成员;
  • 让一个 LLM 当「导演」,读群聊历史后从可用角色里挑下一个发言人(prompt 见 index.js:736);
  • 模型没选出有效角色就随机挑一个。

这块在产品主线里不走到,但解释了为什么内核里有 channels、getGroupMembers、selectNext 这些看似多余的东西——它是从一个更通用的「多 agent 协作」框架裁剪来的。

6. 巧妙之处​

  • 对话即事件流 + 状态机。 用 state: success|interrupt|error 给历史条目打标,让「正常回放」「挂起等人」「出错重试」共用一个数组,而回放时只读 success——干净地把控制流和上下文分离(index.js:1335 getHistory)。
  • interrupt:"ALWAYS" 复用同一套递归实现「人在回路」。 不需要为「答完等用户」写特殊逻辑,只要让 USER 节点永远中断即可(defaults.js:46)。
  • 每轮重取文档上下文。 不把文档塞进一次性的系统提示,而是用 fetchParsedFileContext 回调每轮重注入到 user 消息(index.js:862),所以对话中途上传的新文件也能立刻被看到。

7. 代码地图​

主题文件符号
引擎类server/utils/agents/aibitat/index.jsAIbitat
启动server/utils/agents/aibitat/index.jsstart
递归对话server/utils/agents/aibitat/index.jschat
单轮回复server/utils/agents/aibitat/index.jsreply
历史过滤server/utils/agents/aibitat/index.jsgetHistory, getOrFormatNodeChatHistory
中断判定server/utils/agents/aibitat/index.jsshouldAgentInterrupt, interrupt, continue
工具名解析server/utils/agents/aibitat/index.js#parseFunctionName
群聊选人server/utils/agents/aibitat/index.jsselectNext, getGroupMembers
内置 agentserver/utils/agents/defaults.jsUSER_AGENT, WORKSPACE_AGENT