fix: speed up handbook sync and stop llm_calls 404 noise
Handbook sync ran files sequentially under a 5 req/min limiter with a hardcoded 60s LLM timeout, causing long syncs and AbortError timeouts on large files. Now: limiter at 20 req/min, files processed with concurrency 4, handbook extraction timeout raised to 180s, and near-empty files skip the LLM call. callLLM gains a timeoutMs option; passing a signal no longer silently disables the per-request timeout. llm_calls telemetry self-disables after the first 404 so deploys without the migration applied don't spam the console. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -261,25 +261,53 @@ const KnowledgeGraph = () => {
|
|||||||
}
|
}
|
||||||
});
|
});
|
||||||
|
|
||||||
setSyncResult({ added, modified, unchanged });
|
setSyncResult({ added, modified, unchanged, failed: [] });
|
||||||
|
|
||||||
const filesToProcess = [...added, ...modified];
|
const filesToProcess = [...added, ...modified];
|
||||||
if (filesToProcess.length > 0) {
|
if (filesToProcess.length > 0) {
|
||||||
setSyncProgress(`Processing 0 of ${filesToProcess.length} files...`);
|
const total = filesToProcess.length;
|
||||||
let count = 0;
|
let done = 0;
|
||||||
for (const file of filesToProcess) {
|
const failed = [];
|
||||||
count++;
|
setSyncProgress(`Processing 0 of ${total} files...`);
|
||||||
setSyncProgress(`Processing ${count} of ${filesToProcess.length}: ${file.name}...`);
|
|
||||||
try {
|
// Run files in parallel with bounded concurrency. The extractionLimiter
|
||||||
const rawContent = await getFileContent('respellion', 'employee-handbook', file.path);
|
// inside analyzeHandbookDelta still governs the actual API request rate;
|
||||||
// Pacing is handled centrally by extractionLimiter inside analyzeHandbookDelta.
|
// parallelism here just hides GitHub fetch latency and overlaps with the
|
||||||
await analyzeHandbookDelta(rawContent, file.path);
|
// limiter's spacing instead of waiting serially.
|
||||||
await db.updateHandbookSyncState(file.path, file.sha);
|
const CONCURRENCY = 4;
|
||||||
} catch (err) {
|
const queue = filesToProcess.slice();
|
||||||
console.error('Failed to process file:', file.path, err);
|
|
||||||
|
async function worker() {
|
||||||
|
while (queue.length > 0) {
|
||||||
|
const file = queue.shift();
|
||||||
|
if (!file) return;
|
||||||
|
try {
|
||||||
|
const rawContent = await getFileContent('respellion', 'employee-handbook', file.path);
|
||||||
|
// Skip near-empty files — they only burn an LLM call to extract nothing.
|
||||||
|
if (rawContent.trim().length < 50) {
|
||||||
|
await db.updateHandbookSyncState(file.path, file.sha);
|
||||||
|
} else {
|
||||||
|
await analyzeHandbookDelta(rawContent, file.path);
|
||||||
|
await db.updateHandbookSyncState(file.path, file.sha);
|
||||||
|
}
|
||||||
|
} catch (err) {
|
||||||
|
console.error('Failed to process file:', file.path, err);
|
||||||
|
failed.push({ path: file.path, message: err?.message || String(err) });
|
||||||
|
} finally {
|
||||||
|
done++;
|
||||||
|
setSyncProgress(`Processing ${done} of ${total} files (${failed.length} failed)...`);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
setSyncProgress('Sync Complete! Click "Analyze & Optimize Graph" above to clean up and merge.');
|
|
||||||
|
await Promise.all(Array.from({ length: Math.min(CONCURRENCY, total) }, worker));
|
||||||
|
|
||||||
|
setSyncResult({ added, modified, unchanged, failed });
|
||||||
|
setSyncProgress(
|
||||||
|
failed.length === 0
|
||||||
|
? 'Sync Complete! Click "Analyze & Optimize Graph" above to clean up and merge.'
|
||||||
|
: `Sync finished with ${failed.length} failure(s). See console for details.`
|
||||||
|
);
|
||||||
reloadKb();
|
reloadKb();
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -461,6 +489,9 @@ ${JSON.stringify({ topics: compactTopics, relations: compactRelations })}`;
|
|||||||
<p>Added files: {syncResult.added.length}</p>
|
<p>Added files: {syncResult.added.length}</p>
|
||||||
<p>Modified files: {syncResult.modified.length}</p>
|
<p>Modified files: {syncResult.modified.length}</p>
|
||||||
<p>Unchanged: {syncResult.unchanged.length}</p>
|
<p>Unchanged: {syncResult.unchanged.length}</p>
|
||||||
|
{syncResult.failed?.length > 0 && (
|
||||||
|
<p className="text-red-600">Failed: {syncResult.failed.length}</p>
|
||||||
|
)}
|
||||||
</div>
|
</div>
|
||||||
)}
|
)}
|
||||||
{syncProgress && (
|
{syncProgress && (
|
||||||
|
|||||||
@@ -67,6 +67,7 @@ export async function analyzeHandbookDelta(fileContent, filePath, { signal } = {
|
|||||||
tools: [EMIT_HANDBOOK_DELTA_TOOL],
|
tools: [EMIT_HANDBOOK_DELTA_TOOL],
|
||||||
toolChoice: { type: 'tool', name: EMIT_HANDBOOK_DELTA_TOOL.name },
|
toolChoice: { type: 'tool', name: EMIT_HANDBOOK_DELTA_TOOL.name },
|
||||||
maxTokens: 8192,
|
maxTokens: 8192,
|
||||||
|
timeoutMs: 180_000,
|
||||||
limiter: extractionLimiter,
|
limiter: extractionLimiter,
|
||||||
signal,
|
signal,
|
||||||
});
|
});
|
||||||
|
|||||||
@@ -112,9 +112,16 @@ function buildMessages({ messages, user }) {
|
|||||||
throw new Error('callLLM requires either `messages` or `user`.');
|
throw new Error('callLLM requires either `messages` or `user`.');
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Telemetry collection is optional. If the migration hasn't been applied on a
|
||||||
|
// given deploy, the first POST returns 404; we then disable further attempts
|
||||||
|
// to keep the console clean and avoid wasted round-trips.
|
||||||
|
let llmCallsDisabled = false;
|
||||||
function logLlmCall(record) {
|
function logLlmCall(record) {
|
||||||
|
if (llmCallsDisabled) return;
|
||||||
try {
|
try {
|
||||||
pb.collection('llm_calls').create(record).catch(() => {});
|
pb.collection('llm_calls').create(record).catch((err) => {
|
||||||
|
if (err?.status === 404) llmCallsDisabled = true;
|
||||||
|
});
|
||||||
} catch {
|
} catch {
|
||||||
/* collection may not exist yet — swallow */
|
/* collection may not exist yet — swallow */
|
||||||
}
|
}
|
||||||
@@ -271,6 +278,7 @@ function validateToolInputs(toolUses, task, toolSchemas) {
|
|||||||
* @property {Record<string, import('zod').ZodTypeAny>} [toolSchemas] Overrides for tool_use input validation.
|
* @property {Record<string, import('zod').ZodTypeAny>} [toolSchemas] Overrides for tool_use input validation.
|
||||||
* @property {number} [maxTokens=4096]
|
* @property {number} [maxTokens=4096]
|
||||||
* @property {number} [temperature=0]
|
* @property {number} [temperature=0]
|
||||||
|
* @property {number} [timeoutMs=60000] Per-request timeout in ms. Increase for large structured extractions.
|
||||||
* @property {AbortSignal} [signal]
|
* @property {AbortSignal} [signal]
|
||||||
* @property {{ acquire: (opts?:{signal?:AbortSignal}) => Promise<void>, pauseUntil: (untilMs:number) => void }} [limiter]
|
* @property {{ acquire: (opts?:{signal?:AbortSignal}) => Promise<void>, pauseUntil: (untilMs:number) => void }} [limiter]
|
||||||
*/
|
*/
|
||||||
@@ -291,6 +299,7 @@ export async function callLLM(options) {
|
|||||||
toolSchemas,
|
toolSchemas,
|
||||||
maxTokens = 4096,
|
maxTokens = 4096,
|
||||||
temperature = 0,
|
temperature = 0,
|
||||||
|
timeoutMs = DEFAULT_TIMEOUT_MS,
|
||||||
signal,
|
signal,
|
||||||
limiter,
|
limiter,
|
||||||
} = options;
|
} = options;
|
||||||
@@ -321,9 +330,9 @@ export async function callLLM(options) {
|
|||||||
result = await withRetry(
|
result = await withRetry(
|
||||||
async () => {
|
async () => {
|
||||||
if (limiter) await limiter.acquire({ signal });
|
if (limiter) await limiter.acquire({ signal });
|
||||||
const timeoutCtl = signal ? null : new AbortController();
|
const timeoutCtl = new AbortController();
|
||||||
const timer = timeoutCtl ? setTimeout(() => timeoutCtl.abort(new DOMException('Timeout', 'AbortError')), DEFAULT_TIMEOUT_MS) : null;
|
const timer = setTimeout(() => timeoutCtl.abort(new DOMException('Timeout', 'AbortError')), timeoutMs);
|
||||||
const fetchSignal = linkSignals(signal, timeoutCtl?.signal);
|
const fetchSignal = linkSignals(signal, timeoutCtl.signal);
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const response = await fetch(ANTHROPIC_URL, {
|
const response = await fetch(ANTHROPIC_URL, {
|
||||||
|
|||||||
@@ -160,10 +160,12 @@ export function createLimiter({ rps = 1, burst = 1 } = {}) {
|
|||||||
|
|
||||||
/**
|
/**
|
||||||
* Shared limiter for the multi-call extraction loops (source chunks,
|
* Shared limiter for the multi-call extraction loops (source chunks,
|
||||||
* handbook file sync). 5 requests/minute matches the lowest published
|
* handbook file sync). 20 req/min sits comfortably under Anthropic Tier 1
|
||||||
* Anthropic tier so we stay well clear of 429.
|
* (50/min for Sonnet) and lets the handbook sync finish in a reasonable
|
||||||
|
* time without us redesigning the rate-limit story; if a 429 still slips
|
||||||
|
* through, `withRetry` honours Retry-After.
|
||||||
*/
|
*/
|
||||||
export const extractionLimiter = createLimiter({ rps: 5 / 60, burst: 1 });
|
export const extractionLimiter = createLimiter({ rps: 20 / 60, burst: 2 });
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Run `fn(attempt)` with retry. `fn` may throw a `RetryableError` to request
|
* Run `fn(attempt)` with retry. `fn` may throw a `RetryableError` to request
|
||||||
|
|||||||
Reference in New Issue
Block a user