乐于分享
好东西不私藏
当前位置：夜雨聆风 > 技术教程 > 软件教程 > OpenClaw 源码学习 | 会话管理和会话压缩

OpenClaw 源码学习 | 会话管理和会话压缩

当前时间： 2026-03-18 11:05:14 分类：软件教程 评论(0)

OpenClaw 源码学习 | 会话管理和会话压缩

会话文件

OpenClaw新建一个sessionManager负责会话的读写，它会链接到一个jsonl格式的本地会话文件。你可以通过~/.openclaw/agents/main/sessions目录来查看OpenClaw保存在本地的会话文件。这个文件将同一个会话下的消息记录下来。下图展示了一个真实的会话文件以及其中一条记录详情。

历史会话如何处理并传给大模型

不过，并不是所有的历史会话都需要一股脑塞给LLM。我们来看看OpenClaw如何读取这些信息并经过处理后传给大模型。

createAgentSession方法接受上面的sessionManager作为输入，并返回一个AgentSession对象来管理当前turn的会话。

具体地，createAgentSession内部会调用sessionManager.buildSessionContext()方法来构建输入给LLM的消息列表。

// packages/coding-agent/src/core/sdk.ts
exportasyncfunctioncreateAgentSession(...,sessionManager,settingManager,...){
// Check if session has existing data to restore
const existingSession = sessionManager.buildSessionContext();
}

sessionManager的buildSessionContext函数，负责从会话文件全部消息列表中提取发给LLM的消息列表，模型信息和 thinking level 信息。对于消息列表，它有两个主要处理逻辑：

  • 如果没有压缩点：直接按消息中的parentId往前回溯把所有可参与 LLM 的条目追加到消息列表
  • 如果存在压缩点：先把压缩摘要加入消息列表，再把被保留的关键消息（firstKeptEntryId 起）和压缩之后的新消息加入消息列表。 没有压缩点的处理代码片段如下：
// packages/coding-agent/src/core/session-manager.ts 
// Walk from leaf to root, collecting path
const path: SessionEntry[] = [];
let current: SessionEntry | undefined = leaf;
while (current) {
 path.unshift(current);
 current = current.parentId ? byId.get(current.parentId) : undefined;
}

const messages: AgentMessage[] = [];
const appendMessage = (entry: SessionEntry) => {
if (entry.type === "message") {
   messages.push(entry.message);
 } elseif (entry.type === "custom_message") {
   messages.push(createCustomMessage(entry.customType,   entry.content, entry.display, entry.details, entry.timestamp),
 );

 } elseif (entry.type === "branch_summary" && entry.summary) {
   messages.push(createBranchSummaryMessage(entry.summary, entry.fromId, entry.timestamp));
 }
};

if (compaction) {
// 有压缩点的处理逻辑
}else { 
// 无压缩点的处理逻辑 emit all messages, handle branch summaries and custom messages
for (const entry of path) {
  appendMessage(entry);
 }
}

如果完整消息列表中存在压缩点，则按照以下逻辑处理：

  • 将压缩点处（索引变量为compactionIdx）的摘要信息加入消息列表：
if (compaction) {
// 1、将压缩的摘要信息加入消息列表
messages.push(createCompactionSummaryMessage(compaction.summary, compaction.tokensBefore, compaction.timestamp));
//
}
  • 根据压缩点记录的压缩前消息索引firstKeptEntryId，将path完整列表从firstKeptEntryId到compactionIdx-1条关键信息加入消息列表messages
if (compaction) {
// 1、将压缩点的摘要信息加入消息列表
 ...
// 2、将压缩点前关键消息加入消息列表
let foundFirstKept = false;
for (let i = 0; i < compactionIdx; i++) {
const entry = path[i];
if (entry.id === compaction.firstKeptEntryId) {
   foundFirstKept = true;
  }
if (foundFirstKept) {
   appendMessage(entry);
  }
 }
}else {
// 无压缩点的处理逻辑 
}
  • 将压缩点以后的消息加入消息列表
if (compaction) {
// 1、将压缩点的摘要信息加入消息列表
 ...
// 2、将压缩点前关键消息加入消息列表
  ...
// 3、将压缩点后的消息加入消息列表
for (let i = compactionIdx + 1; i < path.length; i++)  {
const entry = path[i];
  appendMessage(entry);
 }
}else {
// 无压缩点的处理逻辑 
}

得到处理后的信息后，createAgentSession函数内部会将处理好的消息传递给session内部的agent的消息列表

// packages/coding-agent/src/core/sdk.ts 
const existingSession = sessionManager.buildSessionContext();
const hasExistingSession = existingSession.messages.length > 0;
agent = new Agent({...});
// 将已加载并根据压缩点处理后的消息列表传给 agent 
if (hasExistingSession) {
 agent.replaceMessages(existingSession.messages);
 ...
}
const session = new AgentSession({
 agent,
 sessionManager,
 settingsManager,
 ...
 })
...

下面是Agent的replaceMessages方法，会将当前的消息列表保存在_state变量内。

// packages/agent/src/agent.ts
 replaceMessages(ms: AgentMessage[]) {
this._state.messages = ms.slice();
 }

openclaw源代码文件src/agents/pi-embedded-runner/run/attempt.ts中的runEmbeddedAttempt函数内部调用session的prompt方法发送用户消息。

// src/agents/pi-embedded-runner/run/attempt.ts
exportasyncfunctionrunEmbeddedAttempt(...){
let sessionManager: ReturnType<typeof guardSessionManager> | undefined;
 sessionManager = guardSessionManager(SessionManager.open(params.sessionFile), {...});
await prepareSessionManagerForRun({...})
 ...
 ({ session } = await createAgentSession({
  model: params.model,
  tools: builtInTools,
  customTools: allCustomTools,
  sessionManager,
  settingsManager,
 }));
//将构造好的系统提示词注入 pi-coding-agent 会话上下文
 applySystemPromptOverrideToSession(session, systemPromptText);
 ...
const activeSession = session;
await abortable(activeSession.prompt(effectivePrompt));//开始对话，
 ...

}

在调用activeSession.prompt时，实际请求时委托给session.agent.prompt方法。这个方法又被委托给agent的_runLoop。

// packages/agent/src/agent.ts
async prompt(input: string | AgentMessage | AgentMessage[], images?: ImageContent[]) {
awaitthis._runLoop(msgs);
}

根据我们之前对会话消息的处理分析，_runLoop会将我们记录在agent._state.messages的会话消息，封装好的系统提示词和工具列表往下传给LLM。

// packages/agent/src/agent.ts
/**
* Run the agent loop.
* If messages are provided, starts a new conversation turn with those messages.
* Otherwise, continues from existing context.
*/
privateasync _runLoop(messages?: AgentMessage[], options?: { skipInitialSteeringPoll?: boolean }) {
const context: AgentContext = {
 systemPrompt: this._state.systemPrompt,//系统提示词
 messages: this._state.messages.slice(),//传入的已处理的消息列表
 tools: this._state.tools,//工具列表
 };
}

上面的context会经过agentLoop和agentLoopContinue两个分支逻辑传递给runLoop。

// packages/agent/src/agent-loop.ts
// Agent Loop 逻辑的实现：该函数实现了协调 LLM 调用、工具执行、steering 与 follow-up 的内外循环逻辑。
asyncfunctionrunLoop(
 currentContext: AgentContext,// 这里就是传进来的上下文（历会话消息、工具列表、系统提示词）
 newMessages: AgentMessage[],
 config: AgentLoopConfig,
 signal: AbortSignal | undefined,
 stream: EventStream<AgentEvent, AgentMessage[]>,
 streamFn?: StreamFn,
): Promise<void> {
// 外循环，当前turn未结束，或有插入的消息就继续，否则停止
while (true) {
// 内循环：处理大模型问答，工具调用和插入消息
while (hasMoreToolCalls || pendingMessages.length > 0) {
// 调用大模型，可见 currentContext 已经传入
const message = await streamAssistantResponse(currentContext, config, signal, stream, streamFn);
const toolCalls = message.content.filter((c) => c.type === "toolCall");
   hasMoreToolCalls = toolCalls.length > 0;
// 以下为处理工具调用逻辑
if (hasMoreToolCalls) {
// 逐个执行大模型返回的工具调用并返回结果
   }
   }
 }
}

会话压缩（compact）

在前一节我们看到，读取会话文件后，会从当前叶子记录追溯后的完整消息列表中，根据是否有会话压缩点，来确定处理策略。如果有压缩点，我们并不需要将完整消息传给大模型。那么，在OpenClaw中，这个会话压缩是何时处理的？又如何处理的？

会话压缩的核心逻辑是将一段对话历史总结为一个 compaction 条目并写入记录文件，这是 pi-coding-agent 的内部功能。OpenClaw本身并没有提供一套独立的提示词来实现压缩，而是复用了 pi-coding-agent 的 compact 功能 自动压缩的时机是在agentSession的prompt将消息委托给agent.prompt之前。从agent中取出最后一条assistant类型的消息。如果有，则进行压缩检查。这么做是因为：

  • 触发条件依赖 assistant 消息里的信息，压缩判断需要使用量/token 统计、stopReason
  • 溢出检测基于模型响应（overflow error）等
  • 避免误触发，对 user、toolResult、custom 等非 LLM 输出做压缩没有意义
// packages/coding-agent/src/core/agent-session.ts
exportclass AgentSession {
async prompt(text: string, options?: PromptOptions): Promise<void> {

// 找出agent历史消息中的 最后一条 assistant 消息
const lastAssistant = this._findLastAssistantMessage();
if (lastAssistant) { // 如果有 assistant 消息，则做压缩检查和执行压缩
// _checkCompaction 需要最后一条助手消息做判断
awaitthis._checkCompaction(lastAssistant, false);
  }
  ....
awaitthis.agent.prompt(messages);
 }

/** Find the last assistant message in agent state (including aborted ones) */
private _findLastAssistantMessage(): AssistantMessage | undefined {
const messages = this.agent.state.messages;
for (let i = messages.length - 1; i >= 0; i--) {
const msg = messages[i];
if (msg.role === "assistant") {
return msg as AssistantMessage;
   }
  }
returnundefined;
}
}

我们下面来看看_checkCompaction方法，它负责做两件事：

  1. 确认当前是否需要做会话压缩。两种情况下我们需要进行压缩
  2. 如果需要，则执行压缩动作。 下面两种情况下我们需要进行压缩：
  • LLM最后一条消息（即上面的lastAssistant）返回的是上下文溢出错误。此时，需要从agent消息列表中删除这条错误信息，执行会话压缩
  • 当前上下文超出了设置的阈值
// packages/coding-agent/src/core/agent-session.ts
privateasync _checkCompaction(assistantMessage: AssistantMessage, skipAbortedCheck = true): Promise<void> {
// Case 1: Overflow - LLM returned context overflow error
if (sameModel && isContextOverflow(assistantMessage, contextWindow)) {
// Remove the error message from agent state (it IS saved to session for history, but we don't want it in context for the retry)
const messages = this.agent.state.messages;
if (messages.length > 0 && messages[messages.length - 1].role === "assistant") {
this.agent.replaceMessages(messages.slice(0, -1));
 }
awaitthis._runAutoCompaction("overflow", true);
return;
}

// Case 2: Threshold - context is getting large
// For error messages (no usage data), estimate from last successful response.
// This ensures sessions that hit persistent API errors (e.g. 529) can still compact.
let contextTokens: number;
contextTokens = ... // 估计当前上下文占用的tokens,具体逻辑忽略
// 根据当前已占用的 tokens数，模型上下文窗口大小和配置信息决定是否要压缩
if (shouldCompact(contextTokens, contextWindow, settings)) {
awaitthis._runAutoCompaction("threshold", false);
}
}

shouldCompact函数核心的逻辑是根据设置的阈值来确定是否需要压缩。

// packages/coding-agent/src/core/compaction.ts
/**
* Check if compaction should trigger based on context usage.
*/
exportfunctionshouldCompact(contextTokens: number, contextWindow: number, settings: CompactionSettings): boolean{
if (!settings.enabled) returnfalse;
return contextTokens > contextWindow - settings.reserveTokens;
}

可见，核心的逻辑是判断 contextWindow - contextTokens 是否大于 settings.reserveTokens。这个reserveTokens可以通过~/.openclaw/openclaw.json来配置。

{
  agents: {
    defaults: {
      compaction: {
// 核心设置：当上下文剩余token低于40000时，触发强制压缩
"reserveTokensFloor": 40000, 
        }
    }
   }
}

真正执行会话压缩的是_runAutoCompaction函数，这个函数会检查是否配置了外部插件来处理压缩，如果有则使用外部插件。如果没有，则使用内置的方法。压缩完成后，会往会话历史添加一条summary类型的消息，同时更新当前agent的上下文。

privateasync _runAutoCompaction(reason: "overflow" | "threshold",...){
let summary: string;
let firstKeptEntryId: string;
 ...
if (extensionCompaction) {
// 由外部压缩插件则使用
 }else{ // 没有则使用内置方法
const compactResult = await compact(...,this.model, apiKey,...)
 }
// 添加一条 summary 类型消息
this.sessionManager.appendCompaction(summary, firstKeptEntryId, tokensBefore, details, fromExtension);
const newEntries = this.sessionManager.getEntries();
// 更新当前上下文
const sessionContext = this.sessionManager.buildSessionContext();
// 更新agent消息列表
this.agent.replaceMessages(sessionContext.messages);
 ...
}

以上是自动压缩的调用逻辑，我们也可以通过/compact命令手动触发会话压缩。这个入口是AgentSession的compact函数。最后跟上面的_runAutoCompaction一样，都会去调用packages/coding-agent/src/core/compaction.ts文件的compact来做压缩。

接下来我们看看compact的核心逻辑。最开始传入的preparation是通过调用prepareCompaction函数计算的关于会话压缩的关键信息，包括：

  • firstKeptEntryId：决定真正保留的上下文边界，在这以前的内容做摘要压缩，之后的消息保留，它是通过设置信息里的keepRecentTokens来从后往回推算出来的
  • messagesToSummarize
  • turnPrefixMessages：当切点落在某个 turn 的中间（即切点不是以用户消息开始）时，turnStartIndex 是该被切分的 turn 的起始用户消息的索引。turnPrefixMessages是从turnStartIndex到firstKeptEntryId之间的信息。这部分需要做个特殊的摘要
  • tokensBefore：保留firstKeptEntryId以前部分已消耗的上下文
  • previousSummary：如果历史消息中已做过压缩，记录这些压缩信息
  • fileOps ：待压缩的信息中包括的文件操作列表

关于这个firstKeptEntryId、turnStartIndex和turnPrefixMessages，计算逻辑在prepareCompaction函数，这里我们通过简单的例子理解一下即可。keepRecentTokens设置决定我们压缩最近的多少消息要保留，所以需要从当前消息往前回溯找到这个切点。

示例 A — 切点在用户消息（非拆分 turn）

  • 假设 entries（索引: 类型）：
    • 0: session header
    • 1: user
    • 2: assistant
    • 3: user ← 这是切点
    • 4: assistant
    • 5: assistant
  • 结果：
    • firstKeptEntryIndex = 3（从索引 3 开始保留）
    • turnStartIndex = -1（因为切点本身就是用户消息，不是拆分）
    • 行为：保留 3..end；把 1..2 摘要并丢弃（或合并为历史摘要）。

示例 B — 切点落在一个正在进行的 turn（拆分 turn）

  • 假设 entries：
    • 0: user (old task)
    • 1: assistant
    • 2: user (新请求，turn 起点) ← turnStartIndex = 2
    • 3: assistant (prefix 部分)
    • 4: assistant (继续，cut 点落在这里)
    • 5: assistant (recent suffix 被保留，从这里开始) 结果：
    • firstKeptEntryIndex = 5
    • turnStartIndex = 2
    • 行为：索引 0..1（早期历史）会被汇总为历史摘要并丢弃；索引 2..4（turn 的前缀）会单独做“turn 前缀摘要”（保留摘要文本以便理解后面的 suffix）；从索引 5 开始的消息（recent suffix）整体保留，不会被丢弃，从而继续当前正在进行的工作。
// packages/coding-agent/src/core/compaction.ts
exportasyncfunctioncompact(
preparation: CompactionPreparation,
model: Model<any>,
apiKey: string,
customInstructions?: string,
signal?: AbortSignal,
): Promise<CompactionResult> {
const {
  firstKeptEntryId, // 不加入压缩的首个消息，（通过设置keepRecentTokens计算）
  messagesToSummarize,  // 待压缩的历史消息
  turnPrefixMessages,  // firstKeptEntryId 如果刚好跨 turn, 需要往回找到 turn 开始，记录 turn 开始到 firstKeptEntryId 之间的消息
  isSplitTurn,
  tokensBefore,
  previousSummary, // 记录历史压缩过的信息
  fileOps, // 记录文件操作列表
  settings,
 } = preparation;  // 这个是调用 prepareCompaction 函数预先计算的关于压缩的关键信息
// Generate summaries (can be parallel if both needed) and merge into one
let summary: string;
if (isSplitTurn && turnPrefixMessages.length > 0) {
// Generate both summaries in parallel
const [historyResult, turnPrefixResult] = awaitPromise.all([
   generateSummary(messagesToSummarize,...),
   generateTurnPrefixSummary(turnPrefixMessages,...)
  ]
  summary = `${historyResult}\n\n---\n\n**Turn Context (split turn):**\n\n${turnPrefixResult}`;
 }else {
  summary = await generateSummary(messagesToSummarize,...,previousSummary)
 } 
// 摘要中还保留了文件操作的列表，供后续 turn 参考（例如如果有未完成的文件修改，下一 turn 就知道需要继续修改这个文件）
const { readFiles, modifiedFiles } = computeFileLists(fileOps);
 summary += formatFileOperations(readFiles, modifiedFiles);
return {
  summary,
  firstKeptEntryId,//本次压缩开始保留的第一条记录的 UUID，SessionManager 会以此为 parentUuid 创建新的 compaction entry
  tokensBefore,
  details: { readFiles, modifiedFiles } as CompactionDetails,
 };
}

有了上面的示例，compact核心会话压缩的逻辑也很好理解：如果需要保留的消息列表切分点刚好落在一个turn的开始，则直接对历史消息做摘要；如果切分点刚好将一个turn断开，这时我们还需要对turn的前部分单独做一个摘要。generateSummary和generateTurnPrefixSummary做的事情就是根据两个不同的压缩系统提示词模板，调用LLM返回摘要信息，这里不再深入。

下面我们来看看，模型对会话压缩后会对上下文做哪些处理工作。我们回到AgentSession的_runAutoCompaction函数。

privateasync _runAutoCompaction(reason: "overflow" | "threshold",...){
const compactResult = await compact(...,this.model,...)
// 1、添加一条 summary 类型消息
this.sessionManager.appendCompaction(summary, firstKeptEntryId, tokensBefore, details, fromExtension);
const newEntries = this.sessionManager.getEntries();
// 2、更新当前上下文
const sessionContext = this.sessionManager.buildSessionContext();
// 3、更新agent消息列表
this.agent.replaceMessages(sessionContext.messages);
 ...
}

this.sessionManager.appendCompaction函数执行后会往会话jsonl文件添加一条类型为compaction的记录。buildSessionContext则会重新从jsonl文件读取上下文消息，内部会找到刚才保存的 compaction 记录，根据其记录的firstKeptEntryId，将摘要消息以及从firstKeptEntryId以后的消息加载进sessionContext。随后通过下面一行代码将最新的agent消息列表替换

this.agent.replaceMessages(sessionContext.messages)

以上的分析仅针对内置的会话压缩，如果有压缩插件，压缩方法则依据插件执行。

AD：【资源若失效或者不可用，请留言评论或加老夜微信 al_tmen】>>点此前往<<
本站文章均为手工撰写未经允许谢绝转载：夜雨聆风 » OpenClaw 源码学习 | 会话管理和会话压缩

wang

猜你喜欢

  • 暂无文章