From 8c3b8d30c55168b0ab6e1a13d44922c6cb633d02 Mon Sep 17 00:00:00 2001 From: OpenClaw Agent Date: Fri, 4 Sep 2026 19:02:10 +0800 Subject: [PATCH] feat(recall): add recallMulti() for multi-query recall with pluggable query builder - Recaller: add recallMulti(queries, options) that runs each query in parallel and merges results via mergeMany(). Extract static merge() from private mergeResults() so callers can merge externally. - dsh.ts: capture skill-catalog events and build task-oriented recall queries from skills relevant to the current user message. The query builder is configurable via taskQueryBuilder; the default is a bigram-overlap relevance filter that excludes lark-* infrastructure skills and caps at 500 characters. Multi-query recall lets the host augment the primary user-message query with task-domain queries (e.g. skill descriptions), improving recall of procedural knowledge that the user message alone would miss. --- dist/dsh.js | 63 +++++++++++++++++++++++++++-- dist/src/recaller/recall.js | 49 +++++++++++++++++++---- dsh.ts | 79 +++++++++++++++++++++++++++++++++++-- src/recaller/recall.ts | 54 +++++++++++++++++++++---- 4 files changed, 221 insertions(+), 24 deletions(-) diff --git a/dist/dsh.js b/dist/dsh.js index ba63885..4d7bee5 100644 --- a/dist/dsh.js +++ b/dist/dsh.js @@ -84,6 +84,38 @@ function stringOutput(title) { presentationMeta: () => ({ title }), }; } +/** + * Default task-query builder: bigram overlap filter. + * + * A skill is considered relevant to the current user message when any 2-char + * sequence from the user message appears in the skill name or description. + * Only matching skill descriptions are included, capped at 500 characters + * total. Infrastructure skills with a `lark-` prefix are excluded. + */ +function bigramRelevanceFilter(userQuery, skills) { + const TASK_QUERY_MAX_CHARS = 500; + const relevant = []; + let chars = 0; + for (const skill of skills) { + if (skill.name.startsWith("lark-")) + continue; + let match = false; + for (let i = 0; i < userQuery.length - 1 && !match; i++) { + const bigram = userQuery.slice(i, i + 2); + if (skill.name.includes(bigram) || skill.description.includes(bigram)) { + match = true; + } + } + if (!match) + continue; + const desc = skill.description; + if (chars + desc.length > TASK_QUERY_MAX_CHARS && relevant.length > 0) + break; + relevant.push(desc); + chars += desc.length; + } + return relevant.length ? relevant : undefined; +} export function apply(ctx, input = {}) { const freshTurnCount = input.freshTurnCount ?? 5; if (!Number.isInteger(freshTurnCount) || freshTurnCount < 1) { @@ -133,6 +165,8 @@ export function apply(ctx, input = {}) { const latestRoute = new Map(); const latestPrompt = new Map(); const recallCache = new Map(); + const skillEntries = new Map(); + const taskQueryBuilder = input.taskQueryBuilder ?? bigramRelevanceFilter; const extractChain = new Map(); const turnCounts = new Map(); const embeddingConfigured = Boolean(input.embedding?.apiKeyEnv || input.embedding?.baseURL || input.embedding?.baseUrl); @@ -605,6 +639,14 @@ export function apply(ctx, input = {}) { if (event?.type === "user/message" && event.data?.source?.kind === "user") { attachRollingCompaction(ctx.agents?.get(id)); } + // Capture the skill catalog so recall can build a task-oriented query + // from skills relevant to the current conversation. + if (event?.type === "user/message" && event.data?.source?.kind === "skill-catalog") { + const skills = event.data?.message?.skills; + if (Array.isArray(skills)) { + skillEntries.set(String(id), skills); + } + } ingest(id, event); recordCompactionCapsule(id, event); if (event?.type === "turn/end") { @@ -637,15 +679,28 @@ export function apply(ctx, input = {}) { try { let cached = recallCache.get(key); if (!cached || cached.query !== query) { + // Build the query list: primary user-message query + // + optional task-oriented queries from the skill catalog. + const queries = [query]; + const entries = skillEntries.get(key); + if (entries && entries.length) { + const taskQueries = taskQueryBuilder(query, entries); + if (taskQueries && taskQueries.length) { + queries.push(...taskQueries); + } + } // Automatic injection is intentionally high precision: unlike an // explicit gm_search, it must not spend tokens on query-independent // community representatives or weak semantic neighbors. + const recallOptions = { + minSemanticScore: autoRecallMinScore, + allowBroadFallback: false, + }; cached = { query, - value: recaller.recall(query, { - minSemanticScore: autoRecallMinScore, - allowBroadFallback: false, - }), + value: queries.length > 1 + ? recaller.recallMulti(queries, recallOptions) + : recaller.recall(query, recallOptions), }; recallCache.set(key, cached); } diff --git a/dist/src/recaller/recall.js b/dist/src/recaller/recall.js index 8dc657c..0a6a6ea 100644 --- a/dist/src/recaller/recall.js +++ b/dist/src/recaller/recall.js @@ -155,21 +155,42 @@ export class Recaller { * 合并两条路径的结果：全部保留，只去重复节点 */ mergeResults(precise, generalized) { + return Recaller.mergeResultsImpl(precise, generalized); + } + /** + * 合并多个独立召回结果：去重节点，保留两端都在最终节点集中的边。 + * + * 典型的调用方不需要在两条路径之间分配额——各自独立跑满， + * 合并后再统一去重。对于超过两个结果集的场景使用 {@link mergeMany}。 + */ + static merge(a, b) { + return Recaller.mergeResultsImpl(a, b); + } + /** + * 合并任意数量的独立召回结果。 + */ + static mergeMany(results) { + if (!results.length) + return { nodes: [], edges: [], tokenEstimate: 0 }; + let merged = results[0]; + for (let i = 1; i < results.length; i++) { + merged = Recaller.mergeResultsImpl(merged, results[i]); + } + return merged; + } + static mergeResultsImpl(a, b) { const nodeMap = new Map(); const edgeMap = new Map(); - // 精确路径全部入场 - for (const n of precise.nodes) + for (const n of a.nodes) nodeMap.set(n.id, n); - for (const e of precise.edges) + for (const e of a.edges) edgeMap.set(e.id, e); - // 泛化路径去重后全部入场 - for (const n of generalized.nodes) { + for (const n of b.nodes) { if (!nodeMap.has(n.id)) nodeMap.set(n.id, n); } - // 合并边：两端都在最终节点集中的边才保留 const finalIds = new Set(nodeMap.keys()); - for (const e of generalized.edges) { + for (const e of b.edges) { if (!edgeMap.has(e.id) && finalIds.has(e.fromId) && finalIds.has(e.toId)) { edgeMap.set(e.id, e); } @@ -179,9 +200,21 @@ export class Recaller { return { nodes, edges, - tokenEstimate: this.estimateTokens(nodes), + tokenEstimate: Math.ceil(nodes.reduce((s, n) => s + n.content.length + n.description.length, 0) / 3), }; } + /** + * 多查询召回：对每个查询独立跑路径，并行执行，合并去重。 + * + * 适用于需要从多个语义角度检索图谱的场景（例如用户消息 + 任务描述）。 + * 每条查询独立跑精确路径 + 泛化路径，互不干扰。 + */ + async recallMulti(queries, options = {}) { + if (!queries.length) + return { nodes: [], edges: [], tokenEstimate: 0 }; + const results = await Promise.all(queries.map(q => this.recall(q, options))); + return Recaller.mergeMany(results); + } estimateTokens(nodes) { return Math.ceil(nodes.reduce((s, n) => s + n.content.length + n.description.length, 0) / 3); } diff --git a/dsh.ts b/dsh.ts index 6dd1b0a..cb4d2ab 100644 --- a/dsh.ts +++ b/dsh.ts @@ -86,6 +86,21 @@ export interface Config { extractionStreamTimeoutMs?: number; /** @deprecated Use extractionDrain.retryDelaysMs. */ extractionRetryDelaysMs?: number[]; + /** + * Build one or more additional recall queries from the current user message + * and the session's skill catalog. Each query runs independently in parallel + * with the primary user-message query; results are merged with dedup. + * + * Return `undefined` or an empty array to skip the secondary recall path. + * + * @default bigramRelevanceFilter — includes only skills whose name or + * description has a 2-char overlap with the user message (capped at 500 + * chars total), and excludes `lark-*` infrastructure skills. + */ + taskQueryBuilder?: ( + userQuery: string, + skills: Array<{ name: string; description: string }>, + ) => string[] | undefined; } interface Route { @@ -182,6 +197,39 @@ function stringOutput(title: string) { }; } +/** + * Default task-query builder: bigram overlap filter. + * + * A skill is considered relevant to the current user message when any 2-char + * sequence from the user message appears in the skill name or description. + * Only matching skill descriptions are included, capped at 500 characters + * total. Infrastructure skills with a `lark-` prefix are excluded. + */ +function bigramRelevanceFilter( + userQuery: string, + skills: Array<{ name: string; description: string }>, +): string[] | undefined { + const TASK_QUERY_MAX_CHARS = 500; + const relevant: string[] = []; + let chars = 0; + for (const skill of skills) { + if (skill.name.startsWith("lark-")) continue; + let match = false; + for (let i = 0; i < userQuery.length - 1 && !match; i++) { + const bigram = userQuery.slice(i, i + 2); + if (skill.name.includes(bigram) || skill.description.includes(bigram)) { + match = true; + } + } + if (!match) continue; + const desc = skill.description; + if (chars + desc.length > TASK_QUERY_MAX_CHARS && relevant.length > 0) break; + relevant.push(desc); + chars += desc.length; + } + return relevant.length ? relevant : undefined; +} + export function apply(ctx: DshContext, input: Config = {}): void { const freshTurnCount = input.freshTurnCount ?? 5; if (!Number.isInteger(freshTurnCount) || freshTurnCount < 1) { @@ -231,6 +279,8 @@ export function apply(ctx: DshContext, input: Config = {}): void { const latestRoute = new Map(); const latestPrompt = new Map(); const recallCache = new Map }>(); + const skillEntries = new Map>(); + const taskQueryBuilder = input.taskQueryBuilder ?? bigramRelevanceFilter; const extractChain = new Map>(); const turnCounts = new Map(); const embeddingConfigured = Boolean( @@ -724,6 +774,14 @@ export function apply(ctx: DshContext, input: Config = {}): void { if (event?.type === "user/message" && event.data?.source?.kind === "user") { attachRollingCompaction(ctx.agents?.get(id)); } + // Capture the skill catalog so recall can build a task-oriented query + // from skills relevant to the current conversation. + if (event?.type === "user/message" && event.data?.source?.kind === "skill-catalog") { + const skills = event.data?.message?.skills; + if (Array.isArray(skills)) { + skillEntries.set(String(id), skills); + } + } ingest(id, event); recordCompactionCapsule(id, event); if (event?.type === "turn/end") { @@ -753,15 +811,28 @@ export function apply(ctx: DshContext, input: Config = {}): void { try { let cached = recallCache.get(key); if (!cached || cached.query !== query) { + // Build the query list: primary user-message query + // + optional task-oriented queries from the skill catalog. + const queries: string[] = [query]; + const entries = skillEntries.get(key); + if (entries && entries.length) { + const taskQueries = taskQueryBuilder(query, entries); + if (taskQueries && taskQueries.length) { + queries.push(...taskQueries); + } + } // Automatic injection is intentionally high precision: unlike an // explicit gm_search, it must not spend tokens on query-independent // community representatives or weak semantic neighbors. + const recallOptions = { + minSemanticScore: autoRecallMinScore, + allowBroadFallback: false, + }; cached = { query, - value: recaller.recall(query, { - minSemanticScore: autoRecallMinScore, - allowBroadFallback: false, - }), + value: queries.length > 1 + ? recaller.recallMulti(queries, recallOptions) + : recaller.recall(query, recallOptions), }; recallCache.set(key, cached); } diff --git a/src/recaller/recall.ts b/src/recaller/recall.ts index 3dac16e..e1217ce 100755 --- a/src/recaller/recall.ts +++ b/src/recaller/recall.ts @@ -209,21 +209,44 @@ export class Recaller { * 合并两条路径的结果：全部保留，只去重复节点 */ private mergeResults(precise: RecallResult, generalized: RecallResult): RecallResult { + return Recaller.mergeResultsImpl(precise, generalized); + } + + /** + * 合并多个独立召回结果：去重节点，保留两端都在最终节点集中的边。 + * + * 典型的调用方不需要在两条路径之间分配额——各自独立跑满， + * 合并后再统一去重。对于超过两个结果集的场景使用 {@link mergeMany}。 + */ + static merge(a: RecallResult, b: RecallResult): RecallResult { + return Recaller.mergeResultsImpl(a, b); + } + + /** + * 合并任意数量的独立召回结果。 + */ + static mergeMany(results: RecallResult[]): RecallResult { + if (!results.length) return { nodes: [], edges: [], tokenEstimate: 0 }; + let merged = results[0]; + for (let i = 1; i < results.length; i++) { + merged = Recaller.mergeResultsImpl(merged, results[i]); + } + return merged; + } + + private static mergeResultsImpl(a: RecallResult, b: RecallResult): RecallResult { const nodeMap = new Map(); const edgeMap = new Map(); - // 精确路径全部入场 - for (const n of precise.nodes) nodeMap.set(n.id, n); - for (const e of precise.edges) edgeMap.set(e.id, e); + for (const n of a.nodes) nodeMap.set(n.id, n); + for (const e of a.edges) edgeMap.set(e.id, e); - // 泛化路径去重后全部入场 - for (const n of generalized.nodes) { + for (const n of b.nodes) { if (!nodeMap.has(n.id)) nodeMap.set(n.id, n); } - // 合并边：两端都在最终节点集中的边才保留 const finalIds = new Set(nodeMap.keys()); - for (const e of generalized.edges) { + for (const e of b.edges) { if (!edgeMap.has(e.id) && finalIds.has(e.fromId) && finalIds.has(e.toId)) { edgeMap.set(e.id, e); } @@ -235,10 +258,25 @@ export class Recaller { return { nodes, edges, - tokenEstimate: this.estimateTokens(nodes), + tokenEstimate: Math.ceil(nodes.reduce((s, n) => s + n.content.length + n.description.length, 0) / 3), }; } + /** + * 多查询召回：对每个查询独立跑路径，并行执行，合并去重。 + * + * 适用于需要从多个语义角度检索图谱的场景（例如用户消息 + 任务描述）。 + * 每条查询独立跑精确路径 + 泛化路径，互不干扰。 + */ + async recallMulti(queries: string[], options: { + minSemanticScore?: number; + allowBroadFallback?: boolean; + } = {}): Promise { + if (!queries.length) return { nodes: [], edges: [], tokenEstimate: 0 }; + const results = await Promise.all(queries.map(q => this.recall(q, options))); + return Recaller.mergeMany(results); + } + private estimateTokens(nodes: GmNode[]): number { return Math.ceil(nodes.reduce((s, n) => s + n.content.length + n.description.length, 0) / 3); }