Jaspers Hub

jaspers / agents

pluginOfficialReviewed★ 0↓ 1

Research Agents: Agents you define that answer research chats from SEC filings, your documents, and the web, citing every passage.

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