nandi.uk / papers
This repository has no description
0
Fork 0

Configure Feed

Select the types of activity you want to include in your feed.

Parallelize enrichment: 5 concurrent Letta calls per batch

Was sequential (26 records × 30s = 14min). Now batches of 5
concurrent calls — should be ~3min for 26 records.

Applies to poll endpoint, cron handler, and /api/enrich.

👾 Generated with [Letta Code](https://letta.com)

Co-Authored-By: Letta Code <noreply@letta.com>

author
Nandi co-author
Letta Code date (May 17, 2026, 10:23 PM -0700) commit c2cf676b parent 00a7f7b9
+97 -47
+97 -47
src/worker.ts
··· 12 12 import { lexicons } from "../lexicons/generated/index.js"; 13 13 import { config } from "./contrail.config.js"; 14 14 import { pollNewPapers, pollFollowedAccounts, type PaperPost } from "./feed-watcher.js"; 15 - import { buildSummary, enrichWithLetta } from "./paper-summarizer.js"; 15 + import { buildSummary, enrichWithLetta, type PaperSummary } from "./paper-summarizer.js"; 16 16 import { writeSummaryToPds } from "./pds-writer.js"; 17 17 import { LANDING_PAGE } from "./landing-page.js"; 18 18 import { SUMMARY_PAGE } from "./summary-page.js"; ··· 212 212 const apiKey = process.env.LETTA_API_KEY; 213 213 const agentId = process.env.LETTA_AGENT_ID; 214 214 215 - for (const post of newPosts) { 215 + // Build heuristic summaries in parallel 216 + const heuristicResults = await Promise.all(newPosts.map(async (post) => { 216 217 try { 217 - let summary = await buildSummary(post); 218 + return { post, summary: await buildSummary(post) }; 219 + } catch (err: any) { 220 + return { post, error: err?.message ?? err }; 221 + } 222 + })); 218 223 219 - // Enrich with LLM if available 220 - if (apiKey && agentId) { 221 - try { 222 - const enrichment = await enrichWithLetta( 223 - { title: summary.title, paperUrl: summary.paperUrl, abstract: summary.abstract, domains: summary.domains, venue: summary.venue }, 224 - apiKey, 225 - agentId, 226 - ); 227 - if (enrichment.title) summary.title = enrichment.title; 228 - summary.summary = enrichment.summary; 229 - summary.domains = enrichment.domains; 230 - if (enrichment.takeaway) (summary as any).takeaway = enrichment.takeaway; 231 - if (enrichment.year) summary.year = enrichment.year; 232 - } catch (err: any) { 233 - console.error(`Enrichment failed for ${post.paperUrl}: ${err?.message ?? err}, using heuristic`); 234 - } 224 + // Enrich in parallel (batch of 5 to avoid rate limits) 225 + const enrichedResults: Array<{ post: PaperPost; summary: PaperSummary; error?: string }> = []; 226 + for (let i = 0; i < heuristicResults.length; i += 5) { 227 + const batch = heuristicResults.slice(i, i + 5); 228 + const batchResults = await Promise.all(batch.map(async ({ post, summary, error }) => { 229 + if (error || !summary) return { post, summary: summary!, error }; 230 + if (!apiKey || !agentId) return { post, summary }; 231 + try { 232 + const enrichment = await enrichWithLetta( 233 + { title: summary.title, paperUrl: summary.paperUrl, abstract: summary.abstract, domains: summary.domains, venue: summary.venue }, 234 + apiKey, 235 + agentId, 236 + ); 237 + if (enrichment.title) summary.title = enrichment.title; 238 + summary.summary = enrichment.summary; 239 + summary.domains = enrichment.domains; 240 + if (enrichment.takeaway) (summary as any).takeaway = enrichment.takeaway; 241 + if (enrichment.year) summary.year = enrichment.year; 242 + return { post, summary }; 243 + } catch (err: any) { 244 + console.error(`Enrichment failed for ${post.paperUrl}: ${err?.message ?? err}, using heuristic`); 245 + return { post, summary }; 235 246 } 247 + })); 248 + enrichedResults.push(...batchResults); 249 + } 236 250 251 + // Write all to PDS and D1 252 + for (const { post, summary, error } of enrichedResults) { 253 + if (error || !summary) { 254 + const msg = `${post.paperUrl}: ${error}`; 255 + console.error(`Failed to build summary for ${msg}`); 256 + if (errors < 3) errorDetails.push(msg.substring(0, 200)); 257 + errors++; 258 + continue; 259 + } 260 + try { 237 261 const result = await writeSummaryToPds(post, summary); 238 262 239 - // Index the new record into D1 directly 240 263 const rkey = result.uri.split("/").pop() || ""; 241 264 await db 242 265 .prepare("INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)") ··· 326 349 let enriched = 0; 327 350 let errors = 0; 328 351 329 - for (const row of rows.results) { 330 - try { 352 + // Enrich in parallel batches of 5 353 + for (let i = 0; i < rows.results.length; i += 5) { 354 + const batch = rows.results.slice(i, i + 5); 355 + const results = await Promise.allSettled(batch.map(async (row) => { 331 356 const record = JSON.parse(row.record) as { 332 357 title: string; 333 358 paperUrl: string; ··· 336 361 venue: string; 337 362 summary: string; 338 363 takeaway?: string; 364 + year?: number; 339 365 }; 340 366 341 367 const result = await enrichWithLetta( ··· 350 376 agentId, 351 377 ); 352 378 353 - // Update the record in D1 354 379 record.summary = result.summary; 355 380 record.domains = result.domains; 356 381 if (result.takeaway) record.takeaway = result.takeaway; ··· 362 387 .bind(JSON.stringify(record), row.rkey) 363 388 .run(); 364 389 365 - enriched++; 366 - console.log(`Enriched: ${record.title?.substring(0, 50)}...`); 367 - } catch (err: any) { 368 - console.error(`Enrichment failed for ${row.rkey}: ${err?.message ?? err}`); 369 - errors++; 390 + return record.title; 391 + })); 392 + 393 + for (const r of results) { 394 + if (r.status === "fulfilled") { 395 + enriched++; 396 + console.log(`Enriched: ${r.value?.substring(0, 50)}...`); 397 + } else { 398 + console.error(`Enrichment failed: ${r.reason}`); 399 + errors++; 400 + } 370 401 } 371 402 } 372 403 ··· 496 527 const apiKey = process.env.LETTA_API_KEY; 497 528 const agentId = process.env.LETTA_AGENT_ID; 498 529 499 - for (const post of newPosts) { 530 + // Build heuristic summaries in parallel 531 + const heuristicResults = await Promise.all(newPosts.map(async (post) => { 500 532 try { 501 - let summary = await buildSummary(post); 533 + return { post, summary: await buildSummary(post) }; 534 + } catch (err: any) { 535 + return { post, error: err?.message ?? err }; 536 + } 537 + })); 502 538 503 - // Enrich with LLM if available 504 - if (apiKey && agentId) { 505 - try { 506 - const enrichment = await enrichWithLetta( 507 - { title: summary.title, paperUrl: summary.paperUrl, abstract: summary.abstract, domains: summary.domains, venue: summary.venue }, 508 - apiKey, 509 - agentId, 510 - ); 511 - if (enrichment.title) summary.title = enrichment.title; 512 - summary.summary = enrichment.summary; 513 - summary.domains = enrichment.domains; 514 - if (enrichment.takeaway) (summary as any).takeaway = enrichment.takeaway; 515 - if (enrichment.year) summary.year = enrichment.year; 516 - } catch { 517 - // Fall back to heuristic summary 518 - } 539 + // Enrich in parallel (batch of 5) 540 + const enrichedResults: Array<{ post: PaperPost; summary: PaperSummary; error?: string }> = []; 541 + for (let i = 0; i < heuristicResults.length; i += 5) { 542 + const batch = heuristicResults.slice(i, i + 5); 543 + const batchResults = await Promise.all(batch.map(async ({ post, summary, error }) => { 544 + if (error || !summary) return { post, summary: summary!, error }; 545 + if (!apiKey || !agentId) return { post, summary }; 546 + try { 547 + const enrichment = await enrichWithLetta( 548 + { title: summary.title, paperUrl: summary.paperUrl, abstract: summary.abstract, domains: summary.domains, venue: summary.venue }, 549 + apiKey, 550 + agentId, 551 + ); 552 + if (enrichment.title) summary.title = enrichment.title; 553 + summary.summary = enrichment.summary; 554 + summary.domains = enrichment.domains; 555 + if (enrichment.takeaway) (summary as any).takeaway = enrichment.takeaway; 556 + if (enrichment.year) summary.year = enrichment.year; 557 + return { post, summary }; 558 + } catch { 559 + return { post, summary }; 519 560 } 561 + })); 562 + enrichedResults.push(...batchResults); 563 + } 520 564 565 + // Write all to PDS and D1 566 + for (const { post, summary, error } of enrichedResults) { 567 + if (error || !summary) { 568 + console.error(`Cron: failed to build summary for ${post.paperUrl}: ${error}`); 569 + continue; 570 + } 571 + try { 521 572 const result = await writeSummaryToPds(post, summary); 522 573 523 - // Index into D1 524 574 const rkey = result.uri.split("/").pop() || ""; 525 575 await db 526 576 .prepare("INSERT OR REPLACE INTO records_summary (uri, cid, did, rkey, record, time_us, indexed_at) VALUES (?, ?, ?, ?, ?, ?, ?)")