reactagentservice subnode 持久会话

Author: linyuan.yang

packages/sbot/prompts/agent/react_system.txt

@@ -12,7 +12,8 @@ You are a ReAct orchestration expert. Break down the user's request and dispatch
   <rule>Fully autonomous: make all decisions yourself. Never ask the user for clarification, confirmation, additional information, or approval — not even once.</rule>
   <rule>Assume and proceed: when the request is ambiguous, pick the most reasonable interpretation and act on it immediately without stating your assumption.</rule>
   <rule>One call at a time: invoke one agent, wait for the result, then decide the next step based on the output.</rule>
-  <rule>Self-contained instructions: each task field must include all context the agent needs — no references like "as discussed" or "from the previous step".</rule>
+  <rule>Self-contained instructions: when starting a fresh session (no `taskId`), the `task` field must include all context the agent needs — no references like "as discussed" or "from the previous step".</rule>
+  <rule>Continue with taskId: when a follow-up belongs to the same line of work as a prior call, pass back the `task_id` from that call's result as `taskId` to resume the session — the agent already has the prior context, so `task` can be a short next-step instruction.</rule>
   <rule>No repeats: once an agent succeeds at a goal, never call it again for the same goal.</rule>
   <rule>Finish when done: reply to the user directly as soon as all goals are met; do not call any more tools.</rule>
   <rule>On failure: change strategy — switch agents, split the task, or adjust the approach. Do not surface the failure to the user unless all strategies are exhausted.</rule>

packages/sbot/prompts/agent/react_task.txt

@@ -1 +1,7 @@
 Dispatch a task to a specialized sub-agent and return its result.
+
+Sub-agent sessions are persistent and identified by `taskId`:
+- Omit `taskId` to start a fresh session. The agent has no prior context, so `task` must be self-contained — include every detail it needs to execute.
+- Each result includes a line `task_id: <uuid>`. Remember that uuid.
+- Pass the same uuid back as `taskId` on a follow-up call to resume the session. The agent's full history (including auto-compacted summaries) is restored, so `task` can be a short follow-up that builds on prior turns rather than re-stating the entire context.
+- Reuse a `taskId` only when continuing the same line of work; start fresh for unrelated requests.

packages/scorpio.ai/src/Agents/ReAct/ReActAgentService.ts

@@ -2,39 +2,19 @@ import { type StructuredToolInterface } from "@langchain/core/tools";
 import { inject, ServiceContainer, T_StaticSystemPrompts, T_DynamicSystemPrompts, T_ReactSystemPromptTemplate, T_ReactSubNodePrompt, T_ReactTaskToolDesc, T_ModelCallTimeout } from "../../Core";
 import { IMemoryService } from "../../Memory";
 import { IWikiService } from "../../Wiki";
-import { IAgentSaverService, ChatMessage, ChatMessageOptions, type MessageContent } from "../../Saver";
+import { IAgentSaverService, TaskBackedSaver, type MessageContent } from "../../Saver";
 import { ILoggerService } from "../../Logger";
 import { IModelService } from "../../Model";
-import { type AgentServiceBase, IAgentCallback, AgentSubNode, CreateAgentFn, T_CreateAgent, MessageRole } from "../AgentServiceBase";
+import { type AgentServiceBase, IAgentCallback, AgentSubNode, CreateAgentFn, T_CreateAgent, MessageRole, ChatMessage } from "../AgentServiceBase";
 import { ISkillService } from "../../Skills";
 import { IInsightService } from "../../Insight";
 import { ITodoService } from "../../Todo";
 import { IAgentToolService } from "../../AgentTool";
-import { AgentMemorySaver } from "../../Saver/AgentMemorySaver";
 import { SingleAgentService } from "../Single/SingleAgentService";
 import { createTaskTool, type RunTaskFn } from "../../Tools";
 import { MCPContentType, createTextContent, createImageContent, createAudioContent, createErrorResult, type MCPContent } from "../../Tools/types";
 import { v4 as uuidv4 } from "uuid";
 
-// ── ThinkForwardSaver ────────────────────────────────────────
-
-/**
- * 包装 AgentMemorySaver，每次 pushMessage 时同步转发到父 saver 的 think 记录
- */
-class ThinkForwardSaver extends AgentMemorySaver {
-  constructor(private thinkId: string, private parentSaver: IAgentSaverService) { super(); }
-
-  override async pushMessage(message: ChatMessage, options?: ChatMessageOptions): Promise<void> {
-    await super.pushMessage(message, options);
-    await this.parentSaver.pushThinkMessage(this.thinkId, message);
-  }
-
-  override async pushThinkMessage(thinkId: string, message: ChatMessage, options?: ChatMessageOptions): Promise<void> {
-    await super.pushThinkMessage(thinkId, message, options);
-    await this.parentSaver.pushThinkMessage(thinkId, message, options);
-  }
-}
-
 // ── Tokens ────────────────────────────────────────────────────
 
 export const T_AgentSubNodes = Symbol("scorpio:T_AgentSubNodes");
@@ -103,14 +83,15 @@ export class ReActAgentService extends SingleAgentService {
     if (!callback) return [];
     const { onMessage: _, ...subCallback } = callback;
 
-    const runFn: RunTaskFn = async ({ agentId, task, systemPrompt }) => {
+    const runFn: RunTaskFn = async ({ agentId, task, systemPrompt, taskId }) => {
       let agentService: AgentServiceBase | null = null;
       const thinkId = uuidv4();
+      const resolvedTaskId = taskId ?? uuidv4();
       try {
         const parentSaver = this.saverService;
-        const thinkSaver = new ThinkForwardSaver(thinkId, parentSaver);
+        const taskSaver = new TaskBackedSaver(resolvedTaskId, thinkId, parentSaver);
         const subContainer = new ServiceContainer();
-        subContainer.registerInstance(IAgentSaverService, thinkSaver);
+        subContainer.registerInstance(IAgentSaverService, taskSaver);
         if (this.memoryServices.length > 0) subContainer.registerInstance(IMemoryService, this.memoryServices);
         if (this.wikiServices.length > 0) subContainer.registerInstance(IWikiService, this.wikiServices);
         if (this.loggerService) subContainer.registerInstance(ILoggerService, this.loggerService);
@@ -128,6 +109,8 @@ export class ReActAgentService extends SingleAgentService {
           if (messages[i].role === MessageRole.AI && messages[i].content) { lastAI = messages[i]; break; }
         }
         const content: MCPContent[] = [];
+        // 把 task_id 作为首段文本注入，让 LLM 在后续调用中可显式传 taskId 续接
+        content.push(createTextContent(`task_id: ${resolvedTaskId} (pass this back as taskId to resume the same sub-agent session)`));
         if (lastAI) {
           if (typeof lastAI.content === 'string') {
             if (lastAI.content.trim()) content.push(createTextContent(lastAI.content));
@@ -143,10 +126,9 @@ export class ReActAgentService extends SingleAgentService {
             }
           }
         }
-        if (content.length === 0) content.push(createTextContent(''));
-        return { content, _meta: { thinkId } };
+        return { content, _meta: { thinkId, taskId: resolvedTaskId } };
       } catch (error: any) {
-        return { ...createErrorResult(`Execution failed: ${error.message}`), _meta: { thinkId } };
+        return { ...createErrorResult(`Execution failed: ${error.message}`), _meta: { thinkId, taskId: resolvedTaskId } };
       } finally {
         await agentService?.dispose();
       }

packages/scorpio.ai/src/Saver/AgentFileSaver.ts

@@ -13,9 +13,15 @@ import { ILoggerService, ILogger } from "../Logger";
 import { inject } from "scorpio.di";
 import { T_DBPath } from "../Core/tokens";
 
+interface TaskEntry {
+    messages: StoredMessage[];
+    metadata: Record<string, string>;
+}
+
 interface ThreadFile {
     messages: StoredMessage[];
     thinks: Record<string, StoredMessage[]>;
+    tasks: Record<string, TaskEntry>;
     metadata: Record<string, string>;
     nextId: number;
 }
@@ -42,12 +48,13 @@ export class AgentFileSaver implements IAgentSaverService {
         if (this.cache) return this.cache;
         try {
             if (!existsSync(this.filePath)) {
-                this.cache = { messages: [], thinks: {}, metadata: {}, nextId: 1 };
+                this.cache = { messages: [], thinks: {}, tasks: {}, metadata: {}, nextId: 1 };
             } else {
                 const content = await readFile(this.filePath, "utf-8");
                 this.cache = JSON.parse(content) as ThreadFile;
                 if (!this.cache.messages) this.cache.messages = [];
                 if (!this.cache.thinks) this.cache.thinks = {};
+                if (!this.cache.tasks) this.cache.tasks = {};
                 if (!this.cache.metadata) this.cache.metadata = {};
                 if (!this.cache.nextId) this.cache.nextId = 1;
                 let nextId = this.cache.nextId;
@@ -68,14 +75,34 @@ export class AgentFileSaver implements IAgentSaverService {
                     }
                 }
                 this.cache.nextId = nextId;
+                for (const entry of Object.values(this.cache.tasks)) {
+                    if (!entry.messages) entry.messages = [];
+                    if (!entry.metadata) entry.metadata = {};
+                    for (const m of entry.messages) {
+                        if (m.id == null) m.id = nextId++;
+                        else nextId = Math.max(nextId, m.id + 1);
+                        if (m.createdAt == null) m.createdAt = now;
+                        if (!m.kind) m.kind = MessageKind.Normal;
+                    }
+                }
+                this.cache.nextId = nextId;
             }
         } catch (error: any) {
             this.logger?.warn(`读取文件失败: ${error.message}`);
-            this.cache = { messages: [], thinks: {}, metadata: {}, nextId: 1 };
+            this.cache = { messages: [], thinks: {}, tasks: {}, metadata: {}, nextId: 1 };
         }
         return this.cache!;
     }
 
+    private getOrCreateTask(file: ThreadFile, taskId: string): TaskEntry {
+        let entry = file.tasks[taskId];
+        if (!entry) {
+            entry = { messages: [], metadata: {} };
+            file.tasks[taskId] = entry;
+        }
+        return entry;
+    }
+
     private async writeThreadFile(file: ThreadFile): Promise<void> {
         const dir = dirname(this.filePath);
         if (!existsSync(dir)) await mkdir(dir, { recursive: true });
@@ -175,5 +202,68 @@ export class AgentFileSaver implements IAgentSaverService {
         await this.writeThreadFile(file);
     }
 
+    // --- Task scope ---
+
+    async getTaskMessages(taskId: string, includeAll = false): Promise<StoredMessage[]> {
+        const file = await this.getFile();
+        const entry = file.tasks[taskId];
+        if (!entry) return [];
+        return includeAll
+            ? [...entry.messages]
+            : entry.messages.filter(m => m.kind === MessageKind.Normal);
+    }
+
+    async pushTaskMessage(taskId: string, message: ChatMessage, options?: ChatMessageOptions): Promise<void> {
+        const file = await this.getFile();
+        const entry = this.getOrCreateTask(file, taskId);
+        const id = file.nextId;
+        file.nextId = id + 1;
+        entry.messages.push({
+            id,
+            message,
+            createdAt: Math.floor(Date.now() / 1000),
+            thinkId: options?.thinkId,
+            kind: options?.kind ?? MessageKind.Normal,
+        });
+        await this.writeThreadFile(file);
+    }
+
+    async applyTaskCompaction(taskId: string, compactedIds: number[], summary: NewStoredMessage): Promise<void> {
+        const file = await this.getFile();
+        const entry = this.getOrCreateTask(file, taskId);
+        const set = new Set(compactedIds);
+        for (const m of entry.messages) {
+            if (set.has(m.id)) m.kind = MessageKind.Archive;
+        }
+        const id = file.nextId;
+        file.nextId = id + 1;
+        entry.messages.push({
+            ...summary,
+            id,
+            createdAt: Math.floor(Date.now() / 1000),
+        });
+        await this.writeThreadFile(file);
+    }
+
+    async clearTask(taskId: string): Promise<void> {
+        const file = await this.getFile();
+        if (file.tasks[taskId]) {
+            delete file.tasks[taskId];
+            await this.writeThreadFile(file);
+        }
+    }
+
+    async getTaskMetadata(taskId: string, key: string): Promise<string | undefined> {
+        const file = await this.getFile();
+        return file.tasks[taskId]?.metadata[key];
+    }
+
+    async setTaskMetadata(taskId: string, key: string, value: string): Promise<void> {
+        const file = await this.getFile();
+        const entry = this.getOrCreateTask(file, taskId);
+        entry.metadata[key] = value;
+        await this.writeThreadFile(file);
+    }
+
     async dispose(): Promise<void> {}
 }

packages/scorpio.ai/src/Saver/AgentMemorySaver.ts

@@ -7,16 +7,31 @@ import {
     MessageKind,
 } from "./IAgentSaverService";
 
+interface TaskEntry {
+    messages: StoredMessage[];
+    metadata: Record<string, string>;
+}
+
 /**
  * 纯内存实现的 AgentSaver，不持久化。
  * 适用于临时会话、单次任务或测试场景。
  */
 export class AgentMemorySaver implements IAgentSaverService {
     private messages: StoredMessage[] = [];
     private thinks: Record<string, StoredMessage[]> = {};
+    private tasks: Record<string, TaskEntry> = {};
     private metadata: Record<string, string> = {};
     private nextId = 1;
 
+    private getOrCreateTask(taskId: string): TaskEntry {
+        let entry = this.tasks[taskId];
+        if (!entry) {
+            entry = { messages: [], metadata: {} };
+            this.tasks[taskId] = entry;
+        }
+        return entry;
+    }
+
     async getAllMessages(includeAll = false): Promise<StoredMessage[]> {
         return includeAll
             ? [...this.messages]
@@ -92,5 +107,53 @@ export class AgentMemorySaver implements IAgentSaverService {
         this.metadata[key] = value;
     }
 
+    // --- Task scope ---
+
+    async getTaskMessages(taskId: string, includeAll = false): Promise<StoredMessage[]> {
+        const entry = this.tasks[taskId];
+        if (!entry) return [];
+        return includeAll
+            ? [...entry.messages]
+            : entry.messages.filter(m => m.kind === MessageKind.Normal);
+    }
+
+    async pushTaskMessage(taskId: string, message: ChatMessage, options?: ChatMessageOptions): Promise<void> {
+        const entry = this.getOrCreateTask(taskId);
+        entry.messages.push({
+            id: this.nextId++,
+            message,
+            createdAt: Math.floor(Date.now() / 1000),
+            thinkId: options?.thinkId,
+            kind: options?.kind ?? MessageKind.Normal,
+        });
+    }
+
+    async applyTaskCompaction(taskId: string, compactedIds: number[], summary: NewStoredMessage): Promise<void> {
+        const entry = this.getOrCreateTask(taskId);
+        const set = new Set(compactedIds);
+        for (const m of entry.messages) {
+            if (set.has(m.id)) m.kind = MessageKind.Archive;
+        }
+        entry.messages.push({
+            ...summary,
+            id: this.nextId++,
+            createdAt: Math.floor(Date.now() / 1000),
+            kind: summary.kind,
+        });
+    }
+
+    async clearTask(taskId: string): Promise<void> {
+        delete this.tasks[taskId];
+    }
+
+    async getTaskMetadata(taskId: string, key: string): Promise<string | undefined> {
+        return this.tasks[taskId]?.metadata[key];
+    }
+
+    async setTaskMetadata(taskId: string, key: string, value: string): Promise<void> {
+        const entry = this.getOrCreateTask(taskId);
+        entry.metadata[key] = value;
+    }
+
     async dispose(): Promise<void> {}
 }