Jaspers Hub

jaspers / agents

pluginOfficialReviewed★ 0↓ 1

Research Agents: Lets users define research agents that answer chat questions from SEC filings, uploaded documents, and the web, with every figure cited.

5.7 KB
Raw
1import type { BackendContext } from '@jaspers-ai/sdk'
2import { research, type JobStatus, type ServerStatus, type StoredCompany } from './research.ts'
3import { companyOf, isCurrent, readsText, tickerOf, type Category, type Reads } from './sec.ts'
4
5// Reading a company's filings into the store: the store does it, as a job on its worker, for
6// everyone. add_company names the kinds of filing and the years; the job lists them at SEC, reads
7// what the store lacks, and the embedding queue follows. This side starts the job and waits on it,
8// saying how far it is: first until it is read, then until nothing of the company is left to embed.
9
10/** How long between looks at a job or a company's embedding. */
11const POLL_MS = 2_000
12const OPEN = new Set(['queued', 'listing', 'reading'])
13
14/** Waits a poll, or throws once the answer is stopped. */
15async function pause(signal: AbortSignal, ms: number): Promise<void> {
16 if (signal.aborted) throw signal.reason ?? new Error('Stopped.')
17 await new Promise((resolve) => setTimeout(resolve, ms))
18}
19
20/** The named companies as the store holds them. */
21export async function companies(
22 ctx: BackendContext,
23 tickers: string[],
24 signal?: AbortSignal,
25): Promise<StoredCompany[]> {
26 if (tickers.length === 0) return []
27 const status = await research<ServerStatus>(ctx, 'status', { tickers }, signal)
28 return status.companies
29}
30
31/** Waits while a job is open, saying how far it is; the job as it ended. Throws with the store's words when it failed. */
32async function untilRead(
33 ctx: BackendContext,
34 job: JobStatus,
35 ticker: string,
36 signal: AbortSignal,
37 said: (step: string) => Promise<void>,
38 pollMs: number,
39): Promise<JobStatus> {
40 let current = job
41 while (OPEN.has(current.state)) {
42 await said(
43 current.state === 'reading'
44 ? `Reading ${ticker} filings: ${current.done.toLocaleString('en-US')} of ${current.total.toLocaleString('en-US')}`
45 : current.state === 'listing'
46 ? `Listing ${ticker} filings`
47 : `Waiting to read ${ticker}`,
48 )
49 await pause(signal, pollMs)
50 current = await research<JobStatus>(ctx, 'job', { id: job.id }, signal)
51 }
52 if (current.state === 'failed') throw new Error(current.error ?? `${ticker} could not be read.`)
53 if (current.state === 'unknown') throw new Error(`The store lost the job reading ${ticker}.`)
54 return current
55}
56
57/** Waits until none of a company's chunks wait for an embedding, saying how many are left; the company as it stands. */
58async function untilEmbedded(
59 ctx: BackendContext,
60 ticker: string,
61 signal: AbortSignal,
62 said: (step: string) => Promise<void>,
63 pollMs: number,
64): Promise<StoredCompany | null> {
65 for (;;) {
66 const [held] = await companies(ctx, [ticker], signal)
67 if (!held || held.unembedded === 0) return held ?? null
68 await said(`Embedding ${ticker}: ${held.unembedded.toLocaleString('en-US')} chunks to go`)
69 await pause(signal, pollMs)
70 }
71}
72
73/**
74 * What an agent's add_company does: has the store read the kinds of a company's filings for the
75 * years, and answers once they are read and embedded, so a search finds them by meaning as well as
76 * by their words. A company another account read in is already there: the job ends at once.
77 */
78export async function readCompany(
79 ctx: BackendContext,
80 rawTicker: string,
81 years: number[],
82 categories: Category[],
83 signal: AbortSignal,
84 said: (step: string) => Promise<void>,
85 pollMs: number = POLL_MS,
86): Promise<string> {
87 const { ticker, cik } = companyOf(rawTicker)
88 const reads: Reads = Object.fromEntries(categories.map((c) => [c, years]))
89 const started = await research<{ job: JobStatus }>(ctx, 'add_company', { ticker, cik, reads }, signal)
90 const job = await untilRead(ctx, started.job, ticker, signal, said, pollMs)
91 const held = await untilEmbedded(ctx, ticker, signal, said, pollMs)
92 const failed = job.failed ? `, and ${job.failed} could not be read` : ''
93 return `${held?.company ?? ticker}: ${job.done} of the ${job.total} filings, ${readsText(reads)}, are in the store and embedded${failed}. Search them now.`
94}
95
96/**
97 * When this session last checked each company, ISO. The store's recently_as_of is the record, but a
98 * check that fails (SEC out of reach) keeps no time there: without this every search would check again.
99 */
100const checked = new Map<string, string>()
101
102/**
103 * Before a conversation searches a company the store holds: when SEC last listed its filings more
104 * than CURRENT_FOR_MS ago, has the store list them again and read what is new in the kinds it keeps
105 * current, and waits until that is embedded. A company the store lacks is left alone: the agent adds it.
106 */
107export async function keepCurrent(
108 ctx: BackendContext,
109 rawTicker: string,
110 signal: AbortSignal,
111 said: (step: string) => Promise<void>,
112 pollMs: number = POLL_MS,
113): Promise<void> {
114 const { ticker } = companyOf(rawTicker)
115 const [held] = await companies(ctx, [ticker], signal)
116 const now = new Date()
117 if (!held || isCurrent(held.recently_as_of, now) || isCurrent(checked.get(ticker) ?? null, now)) return
118 checked.set(ticker, now.toISOString())
119 if (held.job && OPEN.has(held.job.state)) {
120 await untilRead(ctx, held.job, ticker, signal, said, pollMs)
121 await untilEmbedded(ctx, ticker, signal, said, pollMs)
122 return
123 }
124 await said(`Checking ${ticker} for new filings`)
125 const started = await research<{ job: JobStatus }>(ctx, 'refresh_company', { ticker }, signal)
126 await untilRead(ctx, started.job, ticker, signal, said, pollMs)
127 await untilEmbedded(ctx, ticker, signal, said, pollMs)
128}
129
130/** The chip's Refresh: what the company has filed since the store last listed it, in the kinds it keeps current. */
131export async function refresh(ctx: BackendContext, rawTicker: string): Promise<string> {
132 const ticker = tickerOf(rawTicker)
133 await research<{ job: JobStatus }>(ctx, 'refresh_company', { ticker })
134 return `Checking ${ticker}.`
135}