feat: implement local deep search engine
This commit is contained in:
john
@@ -0,0 +1,698 @@
|
||||
import dns from 'node:dns/promises';
|
||||
import { EventEmitter } from 'node:events';
|
||||
import net from 'node:net';
|
||||
import { randomUUID } from 'node:crypto';
|
||||
import { searchSearxng } from './mindsearch-providers.mjs';
|
||||
|
||||
const DEPTH_PROFILES = Object.freeze({
|
||||
quick: { goals: 3, rounds: 2, resultsPerQuery: 6, maxSources: 8 },
|
||||
standard: { goals: 4, rounds: 2, resultsPerQuery: 7, maxSources: 14 },
|
||||
deep: { goals: 6, rounds: 3, resultsPerQuery: 10, maxSources: 24 },
|
||||
});
|
||||
|
||||
const TRACKING_PARAMS = /^(utm_|fbclid$|gclid$|mc_)/i;
|
||||
|
||||
function clamp(value, min, max) {
|
||||
return Math.max(min, Math.min(max, value));
|
||||
}
|
||||
|
||||
function throwIfAborted(signal) {
|
||||
if (signal?.aborted) {
|
||||
const error = new Error('Research task was cancelled');
|
||||
error.name = 'AbortError';
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
function decodeHtml(value) {
|
||||
return value
|
||||
.replace(/ /gi, ' ')
|
||||
.replace(/&/gi, '&')
|
||||
.replace(/</gi, '<')
|
||||
.replace(/>/gi, '>')
|
||||
.replace(/"/gi, '"')
|
||||
.replace(/'|'/gi, "'")
|
||||
.replace(/—/gi, '—')
|
||||
.replace(/–/gi, '–')
|
||||
.replace(/…/gi, '…')
|
||||
.replace(/&#(\d+);/g, (_match, code) => String.fromCodePoint(Number(code)));
|
||||
}
|
||||
|
||||
export function htmlToPlainText(html) {
|
||||
return decodeHtml(String(html ?? '')
|
||||
.replace(/<script\b[^>]*>[\s\S]*?<\/script>/gi, ' ')
|
||||
.replace(/<style\b[^>]*>[\s\S]*?<\/style>/gi, ' ')
|
||||
.replace(/<noscript\b[^>]*>[\s\S]*?<\/noscript>/gi, ' ')
|
||||
.replace(/<!--[\s\S]*?-->/g, ' ')
|
||||
.replace(/<\/(p|div|section|article|li|h[1-6]|tr)>/gi, '\n')
|
||||
.replace(/<br\s*\/?>/gi, '\n')
|
||||
.replace(/<[^>]+>/g, ' '))
|
||||
.replace(/[ \t]+/g, ' ')
|
||||
.replace(/\n\s+/g, '\n')
|
||||
.replace(/\n{3,}/g, '\n\n')
|
||||
.trim();
|
||||
}
|
||||
|
||||
export function tokenize(value) {
|
||||
const normalized = String(value ?? '').toLowerCase();
|
||||
const latin = normalized.match(/[a-z0-9][a-z0-9._-]{1,}/g) ?? [];
|
||||
const cjkSequences = normalized.match(/[\p{Script=Han}]{2,}/gu) ?? [];
|
||||
const cjk = [];
|
||||
for (const sequence of cjkSequences) {
|
||||
cjk.push(sequence);
|
||||
for (let index = 0; index < sequence.length - 1; index += 1) {
|
||||
cjk.push(sequence.slice(index, index + 2));
|
||||
}
|
||||
}
|
||||
return [...new Set([...latin, ...cjk])];
|
||||
}
|
||||
|
||||
function lexicalScore(needle, haystack) {
|
||||
const terms = tokenize(needle);
|
||||
if (!terms.length) return 0;
|
||||
const text = String(haystack ?? '').toLowerCase();
|
||||
return terms.filter((term) => text.includes(term)).length / terms.length;
|
||||
}
|
||||
|
||||
export function canonicalizeUrl(value) {
|
||||
try {
|
||||
const url = new URL(value);
|
||||
if (!['http:', 'https:'].includes(url.protocol)) return '';
|
||||
url.hash = '';
|
||||
for (const key of [...url.searchParams.keys()]) {
|
||||
if (TRACKING_PARAMS.test(key)) url.searchParams.delete(key);
|
||||
}
|
||||
url.hostname = url.hostname.toLowerCase();
|
||||
if ((url.protocol === 'https:' && url.port === '443') || (url.protocol === 'http:' && url.port === '80')) {
|
||||
url.port = '';
|
||||
}
|
||||
return url.toString();
|
||||
} catch {
|
||||
return '';
|
||||
}
|
||||
}
|
||||
|
||||
function isPrivateIp(address) {
|
||||
const normalized = String(address ?? '').toLowerCase();
|
||||
if (net.isIPv4(normalized)) {
|
||||
const parts = normalized.split('.').map(Number);
|
||||
return parts[0] === 0
|
||||
|| parts[0] === 10
|
||||
|| parts[0] === 127
|
||||
|| (parts[0] === 100 && parts[1] >= 64 && parts[1] <= 127)
|
||||
|| (parts[0] === 169 && parts[1] === 254)
|
||||
|| (parts[0] === 172 && parts[1] >= 16 && parts[1] <= 31)
|
||||
|| (parts[0] === 192 && parts[1] === 168)
|
||||
|| parts[0] >= 224;
|
||||
}
|
||||
if (net.isIPv6(normalized)) {
|
||||
return normalized === '::'
|
||||
|| normalized === '::1'
|
||||
|| normalized.startsWith('fc')
|
||||
|| normalized.startsWith('fd')
|
||||
|| normalized.startsWith('fe8')
|
||||
|| normalized.startsWith('fe9')
|
||||
|| normalized.startsWith('fea')
|
||||
|| normalized.startsWith('feb')
|
||||
|| normalized.startsWith('::ffff:127.')
|
||||
|| normalized.startsWith('::ffff:10.')
|
||||
|| normalized.startsWith('::ffff:192.168.');
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
export async function assertSafePublicUrl(value, { lookup = dns.lookup } = {}) {
|
||||
const url = new URL(value);
|
||||
if (!['http:', 'https:'].includes(url.protocol) || url.username || url.password) {
|
||||
throw new Error('Unsafe source URL');
|
||||
}
|
||||
if (url.hostname === 'localhost' || isPrivateIp(url.hostname)) throw new Error('Unsafe source URL');
|
||||
const records = await lookup(url.hostname, { all: true, verbatim: true });
|
||||
if (!records.length || records.some((record) => isPrivateIp(record.address))) {
|
||||
throw new Error('Unsafe source URL');
|
||||
}
|
||||
return url;
|
||||
}
|
||||
|
||||
export async function readResearchSource(target, {
|
||||
fetchImpl = fetch,
|
||||
lookup = dns.lookup,
|
||||
timeoutMs = 10_000,
|
||||
maxChars = 40_000,
|
||||
signal,
|
||||
redirects = 3,
|
||||
} = {}) {
|
||||
let current = String(target);
|
||||
for (let attempt = 0; attempt <= redirects; attempt += 1) {
|
||||
throwIfAborted(signal);
|
||||
const url = await assertSafePublicUrl(current, { lookup });
|
||||
const timeout = AbortSignal.timeout(timeoutMs);
|
||||
const combinedSignal = signal ? AbortSignal.any([signal, timeout]) : timeout;
|
||||
const response = await fetchImpl(url, {
|
||||
headers: {
|
||||
accept: 'text/html,text/plain,application/xhtml+xml',
|
||||
'user-agent': 'TKMind-Deep-Search/1.0',
|
||||
},
|
||||
redirect: 'manual',
|
||||
signal: combinedSignal,
|
||||
});
|
||||
if ([301, 302, 303, 307, 308].includes(response.status)) {
|
||||
const location = response.headers.get('location');
|
||||
if (!location || attempt === redirects) throw new Error('Unsafe or excessive redirect');
|
||||
current = new URL(location, url).toString();
|
||||
continue;
|
||||
}
|
||||
if (!response.ok) throw new Error(`Reader returned ${response.status}`);
|
||||
const contentType = response.headers.get('content-type') ?? '';
|
||||
if (!/text\/|application\/xhtml\+xml/i.test(contentType)) {
|
||||
throw new Error(`Unsupported content type: ${contentType || 'unknown'}`);
|
||||
}
|
||||
const raw = (await response.text()).slice(0, maxChars * 4);
|
||||
return {
|
||||
url: url.toString(),
|
||||
title: decodeHtml(raw.match(/<title\b[^>]*>([\s\S]*?)<\/title>/i)?.[1] ?? '').trim().slice(0, 500),
|
||||
content: (/html|xhtml/i.test(contentType) ? htmlToPlainText(raw) : raw).slice(0, maxChars),
|
||||
contentType,
|
||||
};
|
||||
}
|
||||
throw new Error('Reader failed');
|
||||
}
|
||||
|
||||
function fallbackGoals(question) {
|
||||
const latinEntity = (String(question).match(/[A-Za-z][A-Za-z0-9._-]{2,}/g) ?? [])[0] ?? '';
|
||||
const candidates = [];
|
||||
const add = (pattern, goal, suffix, english) => {
|
||||
if (pattern.test(question)) {
|
||||
candidates.push({
|
||||
goal,
|
||||
query: `${question} ${suffix}`,
|
||||
alternate: latinEntity ? `${latinEntity} ${english}` : `${question} ${suffix} 权威来源`,
|
||||
});
|
||||
}
|
||||
};
|
||||
add(/架构|组成|原理|architecture/i, '技术架构与组成', '技术架构 组件 原理', 'architecture components official documentation');
|
||||
add(/特性|功能|能力|feature|capabilit/i, '核心特性与能力', '核心特性 功能', 'features capabilities official documentation');
|
||||
add(/局限|风险|问题|争议|limitation|risk|issue/i, '主要局限、风险与争议', '局限 风险 已知问题', 'limitations known issues privacy');
|
||||
add(/市场|规模|商业|竞争|market|business/i, '市场规模与竞争格局', '市场规模 竞争格局 数据', 'market size competitors report');
|
||||
add(/对比|比较|区别|versus|compare/i, '替代方案与横向比较', '替代方案 对比', 'alternatives comparison');
|
||||
add(/趋势|未来|战略|影响|trend|future|strategy/i, '趋势、影响与未来判断', '趋势 影响 未来', 'trends roadmap future');
|
||||
const generic = [
|
||||
{ goal: '核心事实与定义', query: question, alternate: latinEntity ? `${latinEntity} official documentation overview` : `${question} 权威概述` },
|
||||
{ goal: '最新进展与时间线', query: `${question} 最新进展 时间线`, alternate: latinEntity ? `${latinEntity} releases changelog latest` : `${question} 最新时间线` },
|
||||
{ goal: '权威及一手来源', query: `${question} 官方 报告 数据`, alternate: latinEntity ? `${latinEntity} official documentation GitHub` : `${question} 一手来源` },
|
||||
{ goal: '主要参与者与方案比较', query: `${question} 主要参与者 对比`, alternate: `${question} alternatives comparison` },
|
||||
{ goal: '风险、争议与反方观点', query: `${question} 风险 争议 局限`, alternate: `${question} limitations criticism` },
|
||||
{ goal: '趋势、影响与未来判断', query: `${question} 趋势 影响 未来`, alternate: `${question} roadmap future` },
|
||||
];
|
||||
const seen = new Set(candidates.map((item) => item.goal));
|
||||
return [...candidates, ...generic.filter((item) => !seen.has(item.goal))];
|
||||
}
|
||||
|
||||
export function buildFallbackResearchPlan(question, depth = 'standard') {
|
||||
const profile = DEPTH_PROFILES[depth] ?? DEPTH_PROFILES.standard;
|
||||
return fallbackGoals(question).slice(0, profile.goals).map((item, index) => ({
|
||||
id: `goal-${index + 1}`,
|
||||
goal: item.goal,
|
||||
queries: [item.query, item.alternate],
|
||||
sources: index === 2 ? ['web', 'paper'] : ['web'],
|
||||
}));
|
||||
}
|
||||
|
||||
function normalizePlan(candidate, question, depth) {
|
||||
const profile = DEPTH_PROFILES[depth] ?? DEPTH_PROFILES.standard;
|
||||
const source = Array.isArray(candidate) ? candidate : candidate?.research_plan;
|
||||
if (!Array.isArray(source) || !source.length) return buildFallbackResearchPlan(question, depth);
|
||||
const plan = source.map((step, index) => ({
|
||||
id: `goal-${index + 1}`,
|
||||
goal: String(step.goal ?? '').trim().slice(0, 300),
|
||||
queries: (Array.isArray(step.queries) ? step.queries : step.query)
|
||||
?.map?.((query) => String(query ?? '').trim().slice(0, 500))
|
||||
.filter(Boolean)
|
||||
.slice(0, 4) ?? [],
|
||||
sources: (Array.isArray(step.sources) ? step.sources : ['web'])
|
||||
.map((source) => String(source))
|
||||
.filter((source) => ['web', 'news', 'paper', 'code'].includes(source))
|
||||
.slice(0, 4),
|
||||
})).filter((step) => step.goal && step.queries.length).slice(0, profile.goals);
|
||||
return plan.length ? plan : buildFallbackResearchPlan(question, depth);
|
||||
}
|
||||
|
||||
export function createOpenAiCompatibleResearchLlm({
|
||||
endpoint = process.env.TKMIND_DEEP_SEARCH_LLM_URL,
|
||||
apiKey = process.env.TKMIND_DEEP_SEARCH_LLM_API_KEY,
|
||||
model = process.env.TKMIND_DEEP_SEARCH_LLM_MODEL,
|
||||
fetchImpl = fetch,
|
||||
timeoutMs = 90_000,
|
||||
} = {}) {
|
||||
if (!endpoint || !model) return null;
|
||||
async function complete(messages, { json = false } = {}) {
|
||||
const response = await fetchImpl(endpoint, {
|
||||
method: 'POST',
|
||||
signal: AbortSignal.timeout(timeoutMs),
|
||||
headers: {
|
||||
accept: 'application/json',
|
||||
'content-type': 'application/json',
|
||||
...(apiKey ? { authorization: `Bearer ${apiKey}` } : {}),
|
||||
},
|
||||
body: JSON.stringify({
|
||||
model,
|
||||
messages,
|
||||
temperature: 0.2,
|
||||
...(json ? { response_format: { type: 'json_object' } } : {}),
|
||||
}),
|
||||
});
|
||||
if (!response.ok) throw new Error(`Research LLM returned ${response.status}`);
|
||||
const body = await response.json();
|
||||
return String(body.choices?.[0]?.message?.content ?? '').trim();
|
||||
}
|
||||
return {
|
||||
async plan(question, depth) {
|
||||
const content = await complete([
|
||||
{
|
||||
role: 'system',
|
||||
content: 'You are a research planner. Return JSON {"research_plan":[{"goal":"","queries":[""],"sources":["web"]}]}. Create distinct, verifiable goals and search queries.',
|
||||
},
|
||||
{ role: 'user', content: `Depth: ${depth}\nQuestion: ${question}` },
|
||||
], { json: true });
|
||||
return JSON.parse(content);
|
||||
},
|
||||
async synthesize({ question, plan, evidence }) {
|
||||
return complete([
|
||||
{
|
||||
role: 'system',
|
||||
content: 'Write a rigorous Markdown research report using only the supplied evidence. Cite every factual claim with [n]. Include executive summary, findings by research goal, uncertainties, and sources. Never invent citations.',
|
||||
},
|
||||
{
|
||||
role: 'user',
|
||||
content: JSON.stringify({ question, plan, evidence }, null, 2).slice(0, 120_000),
|
||||
},
|
||||
]);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
function sourceAuthority(url) {
|
||||
try {
|
||||
const hostname = new URL(url).hostname;
|
||||
if (/\.(gov|edu)$/.test(hostname) || hostname === 'arxiv.org') return 0.2;
|
||||
if (hostname === 'github.com' || hostname.endsWith('.org')) return 0.12;
|
||||
} catch {
|
||||
return 0;
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
function scoreSource(source, question, goal = '') {
|
||||
const haystack = `${source.title} ${source.snippet} ${source.content}`;
|
||||
const base = clamp(
|
||||
lexicalScore(question, haystack) * 0.45
|
||||
+ lexicalScore(goal, haystack) * 0.35
|
||||
+ sourceAuthority(source.url)
|
||||
+ (/official|documentation|docs|官方|文档/i.test(source.title) ? 0.08 : 0)
|
||||
+ (source.content?.length > 1000 ? 0.05 : 0)
|
||||
+ (source.rank ? 0.05 / source.rank : 0),
|
||||
0,
|
||||
1,
|
||||
);
|
||||
const entity = (String(question).match(/[A-Za-z][A-Za-z0-9._-]{2,}/g) ?? [])[0];
|
||||
return entity && !haystack.toLowerCase().includes(entity.toLowerCase()) ? base * 0.25 : base;
|
||||
}
|
||||
|
||||
function selectSourcesForReading(sourceMap, plan, maxSources) {
|
||||
const all = [...sourceMap.values()].sort((a, b) => b.score - a.score);
|
||||
const selected = [];
|
||||
const seen = new Set();
|
||||
for (const step of plan) {
|
||||
for (const source of all.filter((item) => item.goal === step.goal).slice(0, 2)) {
|
||||
if (seen.has(source.url)) continue;
|
||||
seen.add(source.url);
|
||||
selected.push(source);
|
||||
}
|
||||
}
|
||||
for (const source of all) {
|
||||
if (selected.length >= maxSources) break;
|
||||
if (seen.has(source.url)) continue;
|
||||
seen.add(source.url);
|
||||
selected.push(source);
|
||||
}
|
||||
return selected.slice(0, maxSources);
|
||||
}
|
||||
|
||||
function splitSentences(content) {
|
||||
return String(content ?? '')
|
||||
.split(/(?<=[。！？.!?])\s+|\n+/u)
|
||||
.map((sentence) => sentence.trim())
|
||||
.filter((sentence) => sentence.length >= 35 && sentence.length <= 1200);
|
||||
}
|
||||
|
||||
export function extractEvidence(source, { question, goal }) {
|
||||
const normalizedTitle = String(source.title ?? '').toLowerCase().replace(/\s+/g, ' ').trim();
|
||||
const sentences = splitSentences(source.content || source.snippet)
|
||||
.filter((claim) => {
|
||||
const normalized = claim.toLowerCase().replace(/\s+/g, ' ').trim();
|
||||
if (!normalized || normalized === normalizedTitle) return false;
|
||||
if (normalized.startsWith(normalizedTitle) && normalized.length < normalizedTitle.length + 30) return false;
|
||||
return !/contents\s+menu\s+expand|light mode\s+dark mode|sign in\s+sign up|data-hydro-click|class="|aria-[a-z-]+=/i.test(normalized);
|
||||
});
|
||||
const ranked = sentences
|
||||
.map((claim) => ({
|
||||
claim,
|
||||
score: lexicalScore(`${question} ${goal}`, claim) + Math.min(0.25, claim.length / 1000),
|
||||
}))
|
||||
.sort((a, b) => b.score - a.score)
|
||||
.slice(0, 2);
|
||||
if (!ranked.length && source.snippet) ranked.push({ claim: source.snippet, score: 0.2 });
|
||||
return ranked.map((item) => ({
|
||||
claim: item.claim,
|
||||
confidence: clamp(0.45 + item.score * 0.45 + sourceAuthority(source.url), 0.45, 0.95),
|
||||
url: source.url,
|
||||
title: source.title,
|
||||
goal,
|
||||
}));
|
||||
}
|
||||
|
||||
async function runBounded(items, limit, worker) {
|
||||
const output = new Array(items.length);
|
||||
let cursor = 0;
|
||||
async function run() {
|
||||
while (cursor < items.length) {
|
||||
const index = cursor;
|
||||
cursor += 1;
|
||||
output[index] = await worker(items[index], index);
|
||||
}
|
||||
}
|
||||
await Promise.all(Array.from({ length: Math.min(limit, items.length) }, run));
|
||||
return output;
|
||||
}
|
||||
|
||||
function reportFromEvidence({ question, plan, evidence, sources }) {
|
||||
const sourceIndex = new Map(sources.map((source, index) => [source.url, index + 1]));
|
||||
const lines = [
|
||||
`# ${question}`,
|
||||
'',
|
||||
'## 执行摘要',
|
||||
'',
|
||||
];
|
||||
const strongest = [...evidence].sort((a, b) => b.confidence - a.confidence).slice(0, 5);
|
||||
if (!strongest.length) {
|
||||
lines.push('当前检索未获得足够的可验证证据，建议调整问题或检查搜索服务。');
|
||||
} else {
|
||||
for (const item of strongest) lines.push(`- ${item.claim} [${sourceIndex.get(item.url)}]`);
|
||||
}
|
||||
for (const step of plan) {
|
||||
lines.push('', `## ${step.goal}`, '');
|
||||
const items = evidence.filter((item) => item.goal === step.goal).slice(0, 5);
|
||||
if (!items.length) lines.push('该研究目标暂未获得足够证据。');
|
||||
for (const item of items) {
|
||||
lines.push(`- ${item.claim} [${sourceIndex.get(item.url)}]`);
|
||||
}
|
||||
}
|
||||
lines.push('', '## 不确定性与局限', '');
|
||||
lines.push('- 报告仅基于本次检索可访问的公开网页；付费墙、动态页面和非文本资料可能未被读取。');
|
||||
lines.push('- 自动提取的事实仍建议对照一手来源复核，尤其是数字、日期和预测。');
|
||||
lines.push('', '## 来源', '');
|
||||
sources.forEach((source, index) => {
|
||||
lines.push(`${index + 1}. [${source.title || source.url}](${source.url}) — ${source.provider}`);
|
||||
});
|
||||
return lines.join('\n');
|
||||
}
|
||||
|
||||
function citationNumbers(markdown) {
|
||||
return [...String(markdown).matchAll(/\[(\d+)\]/g)].map((match) => Number(match[1]));
|
||||
}
|
||||
|
||||
function validLlmReport(markdown, sourceCount) {
|
||||
const citations = citationNumbers(markdown);
|
||||
return markdown.length >= 200
|
||||
&& citations.length > 0
|
||||
&& citations.every((number) => number >= 1 && number <= sourceCount);
|
||||
}
|
||||
|
||||
export function createSearxngResearchProvider({
|
||||
endpoint = process.env.TKMIND_DEEP_SEARCH_SEARXNG_URL || process.env.TKMIND_SEARCH_SEARXNG_URL || 'http://127.0.0.1:8080/search',
|
||||
fetchImpl = fetch,
|
||||
} = {}) {
|
||||
return {
|
||||
name: 'searxng',
|
||||
async search(query, { limit, signal } = {}) {
|
||||
throwIfAborted(signal);
|
||||
return searchSearxng(query, { limit, endpoint, fetchImpl, signal });
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function createDeepSearchEngine({
|
||||
store,
|
||||
searchProvider = createSearxngResearchProvider(),
|
||||
reader = readResearchSource,
|
||||
llm = createOpenAiCompatibleResearchLlm(),
|
||||
memorySink = null,
|
||||
idFactory = () => randomUUID(),
|
||||
now = () => Date.now(),
|
||||
concurrency = 3,
|
||||
} = {}) {
|
||||
if (!store) throw new Error('Deep Search store is required');
|
||||
const running = new Map();
|
||||
const emitter = new EventEmitter();
|
||||
|
||||
function publish(taskId, type, payload = {}) {
|
||||
store.appendEvent(taskId, type, payload);
|
||||
emitter.emit(taskId, { type, payload, createdAt: now() });
|
||||
}
|
||||
|
||||
async function planResearch(question, depth) {
|
||||
if (llm?.plan) {
|
||||
try {
|
||||
return normalizePlan(await llm.plan(question, depth), question, depth);
|
||||
} catch {
|
||||
// The deterministic planner keeps Deep Search available when the LLM is unavailable.
|
||||
}
|
||||
}
|
||||
return buildFallbackResearchPlan(question, depth);
|
||||
}
|
||||
|
||||
async function executeTask(taskId, signal) {
|
||||
const task = store.getTask(taskId, { includeSources: false, includeEvents: false });
|
||||
const profile = DEPTH_PROFILES[task.depth] ?? DEPTH_PROFILES.standard;
|
||||
try {
|
||||
store.updateTask(taskId, { status: 'researching', phase: 'planning', progress: 5 });
|
||||
publish(taskId, 'phase', { phase: 'planning', progress: 5 });
|
||||
const plan = await planResearch(task.question, task.depth);
|
||||
throwIfAborted(signal);
|
||||
store.updateTask(taskId, { plan, phase: 'searching', progress: 12 });
|
||||
publish(taskId, 'plan', { plan });
|
||||
|
||||
const sourcesByUrl = new Map();
|
||||
for (let round = 0; round < profile.rounds; round += 1) {
|
||||
throwIfAborted(signal);
|
||||
const jobs = plan.map((step) => ({
|
||||
step,
|
||||
query: step.queries[round % step.queries.length],
|
||||
round,
|
||||
}));
|
||||
const batches = await runBounded(jobs, concurrency, async ({ step, query }) => {
|
||||
try {
|
||||
const results = await searchProvider.search(query, {
|
||||
limit: profile.resultsPerQuery,
|
||||
source: step.sources[0],
|
||||
signal,
|
||||
});
|
||||
publish(taskId, 'search', { goal: step.goal, query, count: results.length, round: round + 1 });
|
||||
return results.map((result) => ({ ...result, goal: step.goal, query }));
|
||||
} catch (error) {
|
||||
if (error?.name === 'AbortError') throw error;
|
||||
publish(taskId, 'search_error', { goal: step.goal, query, message: error.message });
|
||||
return [];
|
||||
}
|
||||
});
|
||||
for (const source of batches.flat()) {
|
||||
const url = canonicalizeUrl(source.url);
|
||||
if (!url) continue;
|
||||
const candidate = {
|
||||
...source,
|
||||
url,
|
||||
provider: source.provider || source.source || searchProvider.name || 'unknown',
|
||||
score: scoreSource(source, task.question, source.goal),
|
||||
};
|
||||
const previous = sourcesByUrl.get(url);
|
||||
if (!previous || candidate.score > previous.score) sourcesByUrl.set(url, candidate);
|
||||
}
|
||||
const progress = 12 + Math.round(((round + 1) / profile.rounds) * 28);
|
||||
store.updateTask(taskId, { progress, phase: 'searching' });
|
||||
publish(taskId, 'progress', { phase: 'searching', progress, sources: sourcesByUrl.size });
|
||||
}
|
||||
|
||||
const ranked = selectSourcesForReading(sourcesByUrl, plan, profile.maxSources);
|
||||
store.updateTask(taskId, { phase: 'reading', progress: 45 });
|
||||
publish(taskId, 'phase', { phase: 'reading', progress: 45, sources: ranked.length });
|
||||
const readSources = await runBounded(ranked, concurrency, async (source, index) => {
|
||||
throwIfAborted(signal);
|
||||
let enriched = source;
|
||||
try {
|
||||
const document = await reader(source.url, { signal });
|
||||
enriched = {
|
||||
...source,
|
||||
title: document.title || source.title,
|
||||
content: document.content,
|
||||
};
|
||||
} catch (error) {
|
||||
if (error?.name === 'AbortError') throw error;
|
||||
publish(taskId, 'read_error', { url: source.url, message: error.message });
|
||||
}
|
||||
enriched.score = scoreSource(enriched, task.question, enriched.goal);
|
||||
store.addSource(taskId, enriched);
|
||||
const progress = 45 + Math.round(((index + 1) / Math.max(1, ranked.length)) * 25);
|
||||
publish(taskId, 'read', { url: source.url, progress });
|
||||
return enriched;
|
||||
});
|
||||
|
||||
throwIfAborted(signal);
|
||||
const finalSources = readSources.sort((a, b) => b.score - a.score);
|
||||
const evidence = [];
|
||||
const seenClaims = new Set();
|
||||
for (const source of finalSources) {
|
||||
for (const item of extractEvidence(source, { question: task.question, goal: source.goal })) {
|
||||
const key = item.claim.toLowerCase().replace(/\W+/gu, '').slice(0, 180);
|
||||
if (!key || seenClaims.has(key)) continue;
|
||||
seenClaims.add(key);
|
||||
evidence.push(item);
|
||||
}
|
||||
}
|
||||
store.updateTask(taskId, { phase: 'synthesizing', progress: 78 });
|
||||
publish(taskId, 'phase', { phase: 'synthesizing', progress: 78, evidence: evidence.length });
|
||||
let report = '';
|
||||
if (llm?.synthesize && finalSources.length) {
|
||||
try {
|
||||
const numberedEvidence = evidence.map((item) => ({
|
||||
...item,
|
||||
citation: finalSources.findIndex((source) => source.url === item.url) + 1,
|
||||
}));
|
||||
const candidate = await llm.synthesize({ question: task.question, plan, evidence: numberedEvidence });
|
||||
if (validLlmReport(candidate, finalSources.length)) report = candidate;
|
||||
} catch {
|
||||
// Fall through to the citation-safe deterministic report.
|
||||
}
|
||||
}
|
||||
if (!report) report = reportFromEvidence({ question: task.question, plan, evidence, sources: finalSources });
|
||||
throwIfAborted(signal);
|
||||
const completedAt = now();
|
||||
store.updateTask(taskId, {
|
||||
status: 'completed',
|
||||
phase: 'completed',
|
||||
progress: 100,
|
||||
report,
|
||||
completedAt,
|
||||
});
|
||||
store.saveMemory({
|
||||
id: `research:${taskId}`,
|
||||
userId: task.userId,
|
||||
topic: task.question,
|
||||
summary: report.slice(0, 4000),
|
||||
sourceTaskId: taskId,
|
||||
});
|
||||
if (typeof memorySink === 'function' && task.userId) {
|
||||
try {
|
||||
await memorySink({
|
||||
userId: task.userId,
|
||||
taskId,
|
||||
topic: task.question,
|
||||
summary: report.slice(0, 4000),
|
||||
sources: finalSources.map((source) => ({
|
||||
title: source.title,
|
||||
url: source.url,
|
||||
provider: source.provider,
|
||||
})),
|
||||
});
|
||||
publish(taskId, 'memory_saved', { userId: task.userId });
|
||||
} catch (error) {
|
||||
publish(taskId, 'memory_error', { message: String(error?.message ?? error) });
|
||||
}
|
||||
}
|
||||
publish(taskId, 'completed', {
|
||||
progress: 100,
|
||||
sources: finalSources.length,
|
||||
evidence: evidence.length,
|
||||
});
|
||||
} catch (error) {
|
||||
if (error?.name === 'AbortError' || signal.aborted) {
|
||||
store.updateTask(taskId, {
|
||||
status: 'cancelled',
|
||||
phase: 'cancelled',
|
||||
error: null,
|
||||
completedAt: now(),
|
||||
});
|
||||
publish(taskId, 'cancelled', {});
|
||||
} else {
|
||||
store.updateTask(taskId, {
|
||||
status: 'failed',
|
||||
phase: 'failed',
|
||||
error: String(error?.message ?? error).slice(0, 2000),
|
||||
completedAt: now(),
|
||||
});
|
||||
publish(taskId, 'failed', { message: String(error?.message ?? error) });
|
||||
}
|
||||
} finally {
|
||||
running.delete(taskId);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
start({ question, depth = 'standard', userId = null } = {}) {
|
||||
const normalizedQuestion = String(question ?? '').trim();
|
||||
if (!normalizedQuestion || normalizedQuestion.length > 2000) {
|
||||
throw new Error('question must be 1-2000 characters');
|
||||
}
|
||||
const normalizedDepth = Object.hasOwn(DEPTH_PROFILES, depth) ? depth : 'standard';
|
||||
const taskId = idFactory();
|
||||
store.createTask({ id: taskId, userId, question: normalizedQuestion, depth: normalizedDepth });
|
||||
const controller = new AbortController();
|
||||
running.set(taskId, controller);
|
||||
queueMicrotask(() => executeTask(taskId, controller.signal));
|
||||
return { taskId, status: 'queued', service: 'tkmind-deep-search' };
|
||||
},
|
||||
|
||||
async search({ query, limit = 10, type = 'web' } = {}) {
|
||||
const normalizedQuery = String(query ?? '').trim();
|
||||
if (!normalizedQuery || normalizedQuery.length > 500) throw new Error('query must be 1-500 characters');
|
||||
const results = await searchProvider.search(normalizedQuery, {
|
||||
limit: clamp(Number(limit) || 10, 1, 20),
|
||||
source: type,
|
||||
});
|
||||
return results;
|
||||
},
|
||||
|
||||
getTask(id) {
|
||||
return store.getTask(id);
|
||||
},
|
||||
|
||||
listTasks(limit) {
|
||||
return store.listTasks(limit);
|
||||
},
|
||||
|
||||
cancel(id) {
|
||||
const task = store.getTask(id, { includeSources: false, includeEvents: false });
|
||||
if (!task) return null;
|
||||
if (['completed', 'failed', 'cancelled'].includes(task.status)) return task;
|
||||
running.get(String(id))?.abort();
|
||||
store.updateTask(id, { status: 'cancelling', phase: 'cancelling' });
|
||||
publish(id, 'cancelling', {});
|
||||
return store.getTask(id);
|
||||
},
|
||||
|
||||
subscribe(id, listener) {
|
||||
emitter.on(String(id), listener);
|
||||
return () => emitter.off(String(id), listener);
|
||||
},
|
||||
|
||||
getHealth() {
|
||||
return {
|
||||
ok: true,
|
||||
service: 'tkmind-deep-search',
|
||||
running: running.size,
|
||||
tasks: store.getStats(),
|
||||
llmEnabled: Boolean(llm),
|
||||
provider: searchProvider.name ?? 'custom',
|
||||
};
|
||||
},
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user