feat: phase 3 of AI pipeline hardening — extraction quality
Replace stateless one-shot extraction with a stateful, paced, cancellable pipeline. Six subtasks: - 3.1 Sentence-aware chunking with 800-char overlap (was paragraph-only at 4000 chars). Hard-split fallback for runaway sentences. - 3.2 Stateful extraction: chunks 2+ receive an "already-extracted topic IDs" hint capped at 200 IDs, so the model reuses IDs instead of inventing variants like software-developer vs software-engineer. - 3.3 Token-bucket limiter in llmRetry.js (extractionLimiter, 5 req/min). callLLM awaits the limiter before fetch; 429+Retry-After calls pauseUntil. Replaces hard setTimeout(12000) and setTimeout(15000). - 3.4 relevance_locked column on topics — admin edits to relevance are sticky across re-extraction. Migration + merge respects the flag + unlock checkbox in KnowledgeGraph edit form. - 3.5 Unify relation vocabulary — handbook prompt no longer mentions legacy "executes"; one-shot migration rewrites existing executes rows to executed_by with source/target swapped. - 3.6 Cancellation — Cancel button on UploadZone wired to an AbortController threaded into callLLM; aborted runs persist status = "cancelled" rather than "failed". Tests: 16 new unit tests for chunkText, buildKnownIdsHint, and createLimiter. All 61 tests pass, 0 lint errors, build clean. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -1,8 +1,28 @@
|
||||
import * as db from './db';
|
||||
import { callLLM } from './llm';
|
||||
import { extractionLimiter } from './llmRetry';
|
||||
import { EMIT_KNOWLEDGE_GRAPH_TOOL, EMIT_HANDBOOK_DELTA_TOOL } from './llmTools';
|
||||
import { normalizeHandbookResult } from './llmSchemas';
|
||||
|
||||
const MAX_KNOWN_IDS_HINT = 200;
|
||||
|
||||
/**
|
||||
* Build the "already-extracted topic IDs" hint that prepends every chunk
|
||||
* after the first. Capped at the most-recent `MAX_KNOWN_IDS_HINT` IDs so
|
||||
* the prompt stays a bounded size; the model uses this list to reuse IDs
|
||||
* rather than invent variants like `software-developer` for
|
||||
* `software-engineer`.
|
||||
*/
|
||||
export function buildKnownIdsHint(ids) {
|
||||
if (!ids || !ids.length) return '';
|
||||
const recent = ids.slice(-MAX_KNOWN_IDS_HINT);
|
||||
return [
|
||||
'Already-extracted topic IDs (do NOT create new IDs for these — reuse them if the same concept appears here):',
|
||||
...recent.map((id) => `- ${id}`),
|
||||
'',
|
||||
].join('\n');
|
||||
}
|
||||
|
||||
const EXTRACTION_SYSTEM_PROMPT = `You are an AI knowledge extractor for Respellion, an IT company built on radical transparency.
|
||||
You receive a source text. Extract every distinct concept, role, and process from it and emit them through the emit_knowledge_graph tool.
|
||||
|
||||
@@ -28,17 +48,17 @@ Relation types: related_to | depends_on | part_of | executed_by.
|
||||
const HANDBOOK_SYSTEM_PROMPT = `You are analysing an update to the Respellion Employee Handbook. Emit the extracted topics and relations through the emit_handbook_delta tool.
|
||||
|
||||
CRITICAL INSTRUCTIONS:
|
||||
- Every process must have a role attached (the role that executes it).
|
||||
- Every process must have a role attached. Express this as: process --executed_by--> role.
|
||||
- Every concept must connect to a process or role.
|
||||
- Mark handbook topics with metadata.source = "github_handbook".
|
||||
- Assign learning_relevance using the same scale as extraction: core | standard | peripheral | exclude.
|
||||
|
||||
Relation types: related_to | depends_on | part_of | executed_by. (Legacy "executes" relations are normalised by the client into executed_by with source/target swapped.)
|
||||
Relation types: related_to | depends_on | part_of | executed_by.
|
||||
`;
|
||||
|
||||
const cachedSystem = (text) => [{ type: 'text', text, cache_control: { type: 'ephemeral' } }];
|
||||
|
||||
export async function analyzeHandbookDelta(fileContent, filePath) {
|
||||
export async function analyzeHandbookDelta(fileContent, filePath, { signal } = {}) {
|
||||
const result = await callLLM({
|
||||
task: 'extract.handbook',
|
||||
tier: 'standard',
|
||||
@@ -47,6 +67,8 @@ export async function analyzeHandbookDelta(fileContent, filePath) {
|
||||
tools: [EMIT_HANDBOOK_DELTA_TOOL],
|
||||
toolChoice: { type: 'tool', name: EMIT_HANDBOOK_DELTA_TOOL.name },
|
||||
maxTokens: 8192,
|
||||
limiter: extractionLimiter,
|
||||
signal,
|
||||
});
|
||||
|
||||
const raw = result.toolUses[0]?.input;
|
||||
@@ -57,24 +79,79 @@ export async function analyzeHandbookDelta(fileContent, filePath) {
|
||||
return { success: true, data: extractedData };
|
||||
}
|
||||
|
||||
function chunkText(text, maxChunkSize = 4000) {
|
||||
const paragraphs = text.split(/\n+/);
|
||||
const chunks = [];
|
||||
let currentChunk = '';
|
||||
/**
|
||||
* Sentence-aware chunker with overlap.
|
||||
*
|
||||
* Targets ~2000 input tokens per chunk (`MAX_CHUNK_CHARS / 4`). Splits on
|
||||
* sentence boundaries first, then falls back to paragraph boundaries, and
|
||||
* hard-splits inside an oversized sentence as a last resort. Adjacent chunks
|
||||
* share `overlapChars` of trailing text to preserve cross-boundary context
|
||||
* for the model.
|
||||
*
|
||||
* Exported for unit tests; callers in this module use it directly.
|
||||
*
|
||||
* @param {string} text
|
||||
* @param {{ maxChars?: number, overlapChars?: number }} [opts]
|
||||
* @returns {string[]}
|
||||
*/
|
||||
export const MAX_CHUNK_CHARS = 8000;
|
||||
export const OVERLAP_CHARS = 800;
|
||||
|
||||
for (const para of paragraphs) {
|
||||
if ((currentChunk + '\n' + para).length > maxChunkSize) {
|
||||
if (currentChunk) chunks.push(currentChunk.trim());
|
||||
currentChunk = para;
|
||||
} else {
|
||||
currentChunk = currentChunk ? currentChunk + '\n' + para : para;
|
||||
export function chunkText(text, { maxChars = MAX_CHUNK_CHARS, overlapChars = OVERLAP_CHARS } = {}) {
|
||||
if (typeof text !== 'string' || !text.trim()) return [];
|
||||
const trimmed = text.trim();
|
||||
if (trimmed.length <= maxChars) return [trimmed];
|
||||
|
||||
const units = splitIntoChunkableUnits(trimmed, maxChars);
|
||||
if (units.length === 0) return [];
|
||||
|
||||
const chunks = [];
|
||||
let buf = '';
|
||||
let bufLen = 0; // length of new (non-overlap) content added since last flush
|
||||
|
||||
for (const unit of units) {
|
||||
const wouldOverflow = (buf ? buf.length + 1 + unit.length : unit.length) > maxChars;
|
||||
if (wouldOverflow && bufLen > 0) {
|
||||
chunks.push(buf.trim());
|
||||
const overlap = buf.length > overlapChars ? buf.slice(-overlapChars) : '';
|
||||
buf = overlap;
|
||||
bufLen = 0;
|
||||
}
|
||||
// If the overlap + unit still won't fit, drop the overlap so the unit fits cleanly.
|
||||
if (buf && (buf.length + 1 + unit.length) > maxChars) {
|
||||
buf = '';
|
||||
}
|
||||
buf = buf ? buf + ' ' + unit : unit;
|
||||
bufLen += unit.length + (bufLen > 0 ? 1 : 0);
|
||||
}
|
||||
if (currentChunk) chunks.push(currentChunk.trim());
|
||||
if (bufLen > 0 && buf.trim()) chunks.push(buf.trim());
|
||||
return chunks;
|
||||
}
|
||||
|
||||
export async function processSourceText(textContent, sourceName) {
|
||||
function splitIntoChunkableUnits(text, maxChars) {
|
||||
const paragraphs = text.split(/\n\s*\n+/);
|
||||
const units = [];
|
||||
for (const para of paragraphs) {
|
||||
const trimmedPara = para.trim();
|
||||
if (!trimmedPara) continue;
|
||||
const sentences = trimmedPara.split(/(?<=[.!?])\s+/);
|
||||
for (const s of sentences) {
|
||||
const sentence = s.trim();
|
||||
if (!sentence) continue;
|
||||
if (sentence.length <= maxChars) {
|
||||
units.push(sentence);
|
||||
} else {
|
||||
for (let i = 0; i < sentence.length; i += maxChars) {
|
||||
units.push(sentence.slice(i, i + maxChars));
|
||||
}
|
||||
console.warn(`[chunkText] Hard-split a sentence of ${sentence.length} chars (exceeds maxChars=${maxChars}).`);
|
||||
}
|
||||
}
|
||||
}
|
||||
return units;
|
||||
}
|
||||
|
||||
export async function processSourceText(textContent, sourceName, { signal } = {}) {
|
||||
const existing = await db.getSources();
|
||||
const alreadyDone = existing.find(
|
||||
s => s.name === sourceName && s.status === 'completed'
|
||||
@@ -87,36 +164,42 @@ export async function processSourceText(textContent, sourceName) {
|
||||
const sourceId = rec.id;
|
||||
|
||||
try {
|
||||
const chunks = chunkText(textContent, 4000);
|
||||
const chunks = chunkText(textContent);
|
||||
console.log(`[Pipeline] Split "${sourceName}" into ${chunks.length} chunks for processing.`);
|
||||
|
||||
let allExtractedTopics = [];
|
||||
let allExtractedRelations = [];
|
||||
const existingTopics = await db.getTopics();
|
||||
const knownIds = existingTopics.map((t) => t.id);
|
||||
|
||||
const allExtractedTopics = [];
|
||||
const allExtractedRelations = [];
|
||||
|
||||
for (let i = 0; i < chunks.length; i++) {
|
||||
if (i > 0) {
|
||||
console.log(`[Pipeline] Pacing delay (12s) to prevent rate limits before chunk ${i + 1}/${chunks.length}...`);
|
||||
await new Promise(r => setTimeout(r, 12000));
|
||||
}
|
||||
if (signal?.aborted) throw signal.reason ?? new DOMException('Aborted', 'AbortError');
|
||||
|
||||
console.log(`[Pipeline] Processing chunk ${i + 1}/${chunks.length} (${chunks[i].length} chars)...`);
|
||||
const hint = i > 0 ? buildKnownIdsHint(knownIds) : '';
|
||||
const result = await callLLM({
|
||||
task: 'extract.source',
|
||||
tier: 'standard',
|
||||
system: cachedSystem(EXTRACTION_SYSTEM_PROMPT),
|
||||
user: `Analyze this part of the document (${i + 1}/${chunks.length}):\n\n${chunks[i]}`,
|
||||
user: `${hint}Analyze this part of the document (${i + 1}/${chunks.length}):\n\n${chunks[i]}`,
|
||||
tools: [EMIT_KNOWLEDGE_GRAPH_TOOL],
|
||||
toolChoice: { type: 'tool', name: EMIT_KNOWLEDGE_GRAPH_TOOL.name },
|
||||
maxTokens: 8192,
|
||||
limiter: extractionLimiter,
|
||||
signal,
|
||||
});
|
||||
|
||||
const extractedData = result.toolUses[0]?.input;
|
||||
if (!extractedData) throw new Error(`Extraction did not emit a tool result for chunk ${i + 1}.`);
|
||||
|
||||
if (extractedData.topics && Array.isArray(extractedData.topics)) {
|
||||
if (Array.isArray(extractedData.topics)) {
|
||||
allExtractedTopics.push(...extractedData.topics);
|
||||
for (const t of extractedData.topics) {
|
||||
if (t?.id && !knownIds.includes(t.id)) knownIds.push(t.id);
|
||||
}
|
||||
}
|
||||
if (extractedData.relations && Array.isArray(extractedData.relations)) {
|
||||
if (Array.isArray(extractedData.relations)) {
|
||||
allExtractedRelations.push(...extractedData.relations);
|
||||
}
|
||||
}
|
||||
@@ -127,7 +210,8 @@ export async function processSourceText(textContent, sourceName) {
|
||||
return { success: true, data: { topics: allExtractedTopics, relations: allExtractedRelations } };
|
||||
|
||||
} catch (error) {
|
||||
await db.updateSourceStatus(sourceId, 'failed', error.message);
|
||||
const isAbort = error?.name === 'AbortError';
|
||||
await db.updateSourceStatus(sourceId, isAbort ? 'cancelled' : 'failed', isAbort ? 'cancelled by user' : error.message);
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
@@ -142,11 +226,16 @@ async function mergeKnowledgeGraph(newData) {
|
||||
for (const t of newData.topics) {
|
||||
if (topicsMap.has(t.id)) {
|
||||
const existing = topicsMap.get(t.id);
|
||||
topicsMap.set(t.id, {
|
||||
const merged = {
|
||||
...existing,
|
||||
...t,
|
||||
description: t.description || existing.description,
|
||||
});
|
||||
};
|
||||
if (existing.relevance_locked) {
|
||||
merged.learning_relevance = existing.learning_relevance;
|
||||
merged.relevance_locked = true;
|
||||
}
|
||||
topicsMap.set(t.id, merged);
|
||||
} else {
|
||||
topicsMap.set(t.id, t);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user