Jaspers Hub

jaspers / research

pluginOfficialReviewed★ 0↓ 28

Jaspers Research: Research rooms where configurable analysts research companies in parallel over SEC filings data, answering with checked citations.

18 KB
Raw
1import type { Analyst } from './analysts'
2import {
3 buildRepairMessage,
4 citedPanel,
5 claimsFor,
6 numberCitations,
7 readShowCitations,
8 resolveCitations,
9 resolveViewCitations,
10 uncitedSentences,
11 type Citation,
12 type NumberedCitation,
13 type ServerRef,
14} from './citations.ts'
15import { errorText } from './errors.ts'
16import { evictResults, resultSummary, stepFor, type AnalystMessage } from './room.ts'
17
18// One analyst answering one message: call the model with the thread and the analyst's tools, run
19// whatever tools it asks for, keep going until it answers without asking, and write the message as
20// it grows. Beside the connections' tools it may read back a stubbed result, and read what the views
21// on the user's workspace show, a view's whole text included. When the answer is in, its citations
22// go to the connection's show_citations tool, which finds each quote in its source and links it, and
23// its quotes from views' texts are checked here against the text; a quote that is not found, or a
24// sentence stating a fact with no tag, sends the draft back to the model once. The model, the
25// tools, and the app's state arrive through the context, the files through `io`, so the loop is
26// tested with a scripted model. What the context needs is spelled out here, in the shapes the app's
27// llm, tools, and state capabilities have.
28
29export interface ToolDefinition {
30 name: string
31 description: string
32 parameters: Record<string, unknown>
33}
34
35export interface ToolCall {
36 id: string
37 name: string
38 input: Record<string, unknown>
39}
40
41export interface ToolResult {
42 callId: string
43 output: string
44 isError: boolean
45}
46
47export type Turn =
48 | { role: 'user'; text: string }
49 | { role: 'assistant'; text: string; toolCalls: ToolCall[]; raw: unknown[] | null }
50 | { role: 'tool'; results: ToolResult[] }
51
52export interface Completion {
53 text: string
54 toolCalls: ToolCall[]
55 stopReason: string
56 usage: { input: number; output: number }
57 raw: unknown[] | null
58}
59
60export interface ToolEntry {
61 id: string
62 connection: string
63 name: string
64 description: string
65 parameters: Record<string, unknown>
66}
67
68export interface RunnerContext {
69 llm: {
70 complete(req: { system: string; turns: Turn[]; tools?: ToolDefinition[]; model?: string; maxTokens?: number; signal?: AbortSignal }): Promise<Completion>
71 }
72 tools: {
73 list(filter?: { connections?: string[] }): Promise<ToolEntry[]>
74 call(id: string, args: Record<string, unknown>, opts?: { signal?: AbortSignal }): Promise<{ text: string; isError: boolean }>
75 }
76 state: {
77 get(path: string): Promise<unknown>
78 }
79}
80
81export interface RunIo {
82 /** Writes the message as it stands; `final` is its last write. */
83 save(message: AnalystMessage, final: boolean): Promise<void>
84 saveResult(ref: string, text: string): Promise<void>
85 readResult(ref: string): Promise<string>
86 /** What came into the room since the analyst last looked, as a turn for the model; null when nothing did. */
87 news(): Promise<string | null>
88 /** Milliseconds, for spacing the writes. */
89 now(): number
90 /** An ISO timestamp. */
91 iso(): string
92}
93
94export interface RunJob {
95 message: AnalystMessage
96 analyst: Analyst
97 system: string
98 /** The first user turn: the thread so far and the message for this analyst. */
99 firstTurn: string
100 /** What the room is called, for the caption the server puts on the citations. */
101 title: string
102 /** The workspace the user asked from, whose views read_workspace reads; null offers no read_workspace. */
103 workspace: string | null
104}
105
106/** The kept tool results a model is shown before the oldest large ones become stubs. */
107const RESULTS_BUDGET = 120_000
108const KEEP_NEWEST = 4
109const EVICT_MIN = 2000
110const FETCH_PAGE = 20_000
111/** What read_workspace shows of a view's text at once. */
112const TEXT_PAGE = 20_000
113const SAVE_EVERY_MS = 500
114/** The connection tool that verifies quotes and links them. The room calls it; the analyst never sees it. */
115const RESOLVER = 'show_citations'
116const CLAIMS_PER_CALL = 60
117/** Rounds the repair of an answer's citations may take, tool calls included. */
118const REPAIR_ROUNDS = 4
119
120const FETCH_RESULT: ToolDefinition = {
121 name: 'fetch_result',
122 description: 'Read back a tool result that was replaced by a stub, 20,000 characters at a time, from offset.',
123 parameters: {
124 type: 'object',
125 properties: { ref: { type: 'string', description: 'The ref in the stub, like r7.' }, offset: { type: 'integer', minimum: 0 } },
126 required: ['ref'],
127 },
128}
129
130const READ_WORKSPACE: ToolDefinition = {
131 name: 'read_workspace',
132 description:
133 "Read what the views on the user's workspace show. panels lists them: element id, view, a line on what each shows, and textLength when it has text. panels/<id>/output is what one is showing now, panels/<id>/state what it is set to, panels/<id>/summary its line; a key path after output or state reads one part, like panels/e1/output/tickers. panels/<id>/text is the whole of what a view shows as text, like a call transcript, 20,000 characters at a time from offset; quote it with <cite panel=\"<id>\" quote=\"…\">.",
134 parameters: {
135 type: 'object',
136 properties: {
137 path: { type: 'string', description: 'panels, or panels/<id>/output, /state, /summary, or /text, like panels/e1/output.' },
138 offset: { type: 'integer', minimum: 0, description: 'For panels/<id>/text: the character to read from, as the last part said.' },
139 },
140 required: ['path'],
141 },
142}
143
144export async function runAnalyst(ctx: RunnerContext, job: RunJob, io: RunIo, signal: AbortSignal): Promise<AnalystMessage> {
145 const { analyst } = job
146 const message: AnalystMessage = { ...job.message, status: 'working', startedAt: io.iso(), error: null, citations: [], audit: null }
147 let lastSave = Number.NEGATIVE_INFINITY
148 const save = async (final: boolean): Promise<void> => {
149 const now = io.now()
150 if (!final && now - lastSave < SAVE_EVERY_MS) return
151 lastSave = now
152 await io.save(message, final)
153 }
154 const end = async (status: AnalystMessage['status'], error: string | null = null): Promise<AnalystMessage> => {
155 message.status = status
156 message.error = error
157 message.endedAt = io.iso()
158 await save(true)
159 return message
160 }
161
162 let tools = new Map<string, { name: string; entry: ToolEntry }>()
163 let resolverId: string | null = null
164 let turns: Turn[] = [{ role: 'user', text: job.firstTurn }]
165 const refs = new Map<string, { ref: string; tool: string; args: string }>()
166 /** Views' texts as the analyst read them, by the id it named the panel by: what its quotes are checked against. */
167 const read = new Map<string, string>()
168 let nextRef = 1
169 /** Where the answer proper starts in the message's text: what the audit reads and a repair replaces. */
170 let draftStart = 0
171
172 /** One model call over the turns as they stand, results past the budget stubbed first. */
173 async function complete(definitions: ToolDefinition[]): Promise<Completion> {
174 turns = evictResults(turns, refs, { budget: RESULTS_BUDGET, keepNewest: KEEP_NEWEST, minSize: EVICT_MIN })
175 const reply = await ctx.llm.complete({ system: job.system, turns, tools: definitions, model: analyst.model ?? undefined, signal })
176 message.usage = { input: message.usage.input + reply.usage.input, output: message.usage.output + reply.usage.output }
177 message.turns += 1
178 turns.push({ role: 'assistant', text: reply.text, toolCalls: reply.toolCalls, raw: reply.raw })
179 return reply
180 }
181
182 /** The calls of one reply, run together; each becomes a step at `at` and a saved result. */
183 async function runCalls(calls: ToolCall[], at: number): Promise<void> {
184 const steps = calls.map((call) => {
185 const step = { ...stepFor(message.text, call.name, call.input), at }
186 message.steps.push(step)
187 return step
188 })
189 await save(false)
190 const results = await Promise.all(
191 calls.map(async (call, i): Promise<ToolResult> => {
192 const step = steps[i]!
193 const { text, isError } = await runCall(call)
194 step.isError = isError
195 step.summary = resultSummary(text, isError)
196 if (call.name !== FETCH_RESULT.name && !isError) {
197 const ref = `r${nextRef++}`
198 step.ref = ref
199 refs.set(call.id, { ref, tool: call.name, args: step.args })
200 await io.saveResult(ref, text)
201 }
202 return { callId: call.id, output: text, isError }
203 }),
204 )
205 turns.push({ role: 'tool', results })
206 }
207
208 /** Every citation in the text, a view's checked here and a filing's by the server. */
209 async function resolve(numbered: NumberedCitation[]): Promise<Citation[]> {
210 const views = numbered.filter((c) => citedPanel(c.attrs) !== null)
211 const filings = numbered.filter((c) => citedPanel(c.attrs) === null)
212 return [...(await resolveViews(views)), ...(await resolveFilings(filings))].sort((a, b) => a.n - b.n)
213 }
214
215 /**
216 * Quotes from views' texts, found in the text as the analyst read it, which stands even if the user
217 * has since opened something else there; a panel it never read is checked against its text now.
218 */
219 async function resolveViews(numbered: NumberedCitation[]): Promise<Citation[]> {
220 const texts = new Map(read)
221 for (const panel of new Set(numbered.map((c) => citedPanel(c.attrs)!))) {
222 if (texts.has(panel) || job.workspace === null) continue
223 try {
224 const value = await ctx.state.get(`workspaces/${job.workspace}/panels/${panel}/text`)
225 if (typeof value === 'string') texts.set(panel, value)
226 } catch {
227 // A panel that is gone, or never was, has no text to find the quote in.
228 }
229 }
230 return resolveViewCitations(numbered, texts)
231 }
232
233 /**
234 * Citations of filings, with the server's verdict on each when a server can give one. The server
235 * answers once per source, so two quotes from one chunk go in separate calls.
236 */
237 async function resolveFilings(numbered: NumberedCitation[]): Promise<Citation[]> {
238 if (numbered.length === 0) return []
239 if (!resolverId) return resolveCitations(numbered, [])
240 const batches: NumberedCitation[][] = []
241 for (const citation of numbered) {
242 let batch = batches.find((b) => b.length < CLAIMS_PER_CALL && !b.some((c) => c.key === citation.key))
243 if (!batch) batches.push((batch = []))
244 batch.push(citation)
245 }
246 const resolved: Citation[] = []
247 for (const batch of batches) {
248 let found: ServerRef[] = []
249 try {
250 const result = await ctx.tools.call(resolverId, { claims: claimsFor(batch), title: job.title }, { signal })
251 if (!result.isError) found = readShowCitations(result.text)
252 } catch {
253 // A server that cannot answer now leaves these unchecked rather than losing the answer.
254 }
255 resolved.push(...resolveCitations(batch, found))
256 }
257 return resolved
258 }
259
260 /** The citations checked and the uncited sentences counted, as the message carries them. */
261 async function audit(): Promise<{ uncited: string[]; unverified: Citation[] }> {
262 message.citations = await resolve(numberCitations(message.text))
263 const uncited = uncitedSentences(message.text.slice(draftStart))
264 const unverified = message.citations.filter((c) => c.verified === false)
265 message.audit = { uncited: uncited.length, unverified: unverified.length, repaired: message.audit?.repaired ?? false }
266 return { uncited, unverified }
267 }
268
269 /**
270 * The answer's citations checked and its uncited sentences found; when either turns something
271 * up and a repair is allowed, the draft goes back to the model once and is checked again.
272 */
273 async function finalize(definitions: ToolDefinition[], allowRepair: boolean): Promise<void> {
274 const first = await audit()
275 const repair = buildRepairMessage({
276 uncited: first.uncited,
277 unverified: first.unverified.map((c) => ({
278 n: c.n,
279 tag: c.panel
280 ? `<cite panel="${c.panel}">`
281 : c.sectionKey !== null
282 ? `<dbcitation document_id="${c.documentId}" section_key="${c.sectionKey}">`
283 : `<cite document_id="${c.documentId}" chunk_id="${c.chunkId ?? 'N/A'}">`,
284 quote: c.quote,
285 reason: c.panel ? `quote not found in the text of ${c.panel}` : 'quote not found in the source',
286 })),
287 })
288 if (!allowRepair || repair === null || signal.aborted) return
289 await save(false)
290 message.audit = { ...message.audit!, repaired: true }
291 turns.push({ role: 'user', text: repair })
292 let replaced = false
293 for (let round = 1; round <= REPAIR_ROUNDS && !signal.aborted; round++) {
294 let reply: Completion
295 try {
296 reply = await complete(definitions)
297 } catch {
298 break // the draft stands as it was
299 }
300 if (reply.toolCalls.length === 0) {
301 if (reply.text.trim()) {
302 message.text = message.text.slice(0, draftStart) + reply.text.trim()
303 replaced = true
304 }
305 break
306 }
307 // A source re-read while repairing sits with the answer, not after it.
308 await runCalls(reply.toolCalls, draftStart)
309 }
310 if (replaced) await audit()
311 }
312
313 try {
314 const entries = await ctx.tools.list(analyst.connections.length > 0 ? { connections: analyst.connections } : undefined)
315 tools = named(entries, analyst)
316 resolverId = entries.find((e) => e.name === RESOLVER && (analyst.connections.length === 0 || analyst.connections.includes(e.connection)))?.id ?? null
317 const definitions = [
318 ...[...tools.values()].map(({ name, entry }) => ({ name, description: entry.description, parameters: entry.parameters })),
319 FETCH_RESULT,
320 ...(job.workspace !== null ? [READ_WORKSPACE] : []),
321 ]
322 await save(true)
323
324 for (let round = 1; round <= analyst.maxTurns; round++) {
325 if (signal.aborted) return end('stopped', reasonOf(signal))
326 // What the other analysts finished and the user said since the last call, before this one.
327 const news = await io.news()
328 if (news) turns.push({ role: 'user', text: news })
329 let reply: Completion
330 try {
331 reply = await complete(definitions)
332 } catch (err) {
333 if (signal.aborted) return end('stopped', reasonOf(signal))
334 return end('error', messageOf(err))
335 }
336 if (reply.text.trim()) {
337 draftStart = message.text ? message.text.length + 2 : 0
338 message.text = message.text ? `${message.text}\n\n${reply.text.trim()}` : reply.text.trim()
339 }
340 if (reply.toolCalls.length === 0) {
341 await finalize(definitions, true)
342 return end('done')
343 }
344 await runCalls(reply.toolCalls, message.text.length)
345 if (signal.aborted) return end('stopped', reasonOf(signal))
346 await save(false)
347 }
348 // The last text the model wrote is what there is to check; no repair, the turns are spent.
349 await finalize(definitions, false)
350 message.text = `${message.text ? `${message.text}\n\n` : ''}Stopped at the ${analyst.maxTurns}-turn limit. Ask a follow-up to carry on.`
351 return end('done')
352 } catch (err) {
353 if (signal.aborted) return end('stopped', reasonOf(signal))
354 return end('error', messageOf(err))
355 }
356
357 /** One call: a connection tool, a stubbed result read back, or a read of the workspace. Failures go to the model as errors. */
358 async function runCall(call: ToolCall): Promise<{ text: string; isError: boolean }> {
359 try {
360 if (call.name === READ_WORKSPACE.name && job.workspace !== null) return await readWorkspace(job.workspace, call.input)
361 if (call.name === FETCH_RESULT.name) {
362 const ref = typeof call.input['ref'] === 'string' ? call.input['ref'] : ''
363 const offset = typeof call.input['offset'] === 'number' && call.input['offset'] > 0 ? Math.floor(call.input['offset']) : 0
364 const saved = await io.readResult(ref)
365 if (!saved) return { text: `No saved result ${ref}.`, isError: true }
366 return { text: saved.slice(offset, offset + FETCH_PAGE) || `Nothing past offset ${offset}; the result is ${saved.length} characters.`, isError: false }
367 }
368 const tool = tools.get(call.name)
369 if (!tool) return { text: `Unknown tool ${call.name}.`, isError: true }
370 return await ctx.tools.call(tool.entry.id, call.input, { signal })
371 } catch (err) {
372 return { text: messageOf(err), isError: true }
373 }
374 }
375
376 /**
377 * One read on the workspace the user asked from, kept to its panels. A key with nothing in it is an
378 * answer, not an error. A view's text comes a part at a time, as it reads rather than as JSON, so a
379 * quote copies out of it as written; the whole of it is kept for checking those quotes.
380 */
381 async function readWorkspace(workspace: string, input: Record<string, unknown>): Promise<{ text: string; isError: boolean }> {
382 const path = typeof input['path'] === 'string' ? input['path'].trim().replace(/^\/+|\/+$/g, '') : ''
383 if (path !== 'panels' && !path.startsWith('panels/')) {
384 return { text: `read_workspace reads panels, or panels/<id>/output, /state, /summary, or /text, like panels/e1/output; ${path ? `not ${path}` : 'give it a path'}.`, isError: true }
385 }
386 const value = await ctx.state.get(`workspaces/${workspace}/${path}`)
387 const panel = /^panels\/([^/]+)\/text$/.exec(path)?.[1]
388 if (panel && typeof value === 'string') {
389 read.set(panel, value)
390 const offset = Number(input['offset'] ?? 0)
391 return { text: textPart(path, value, Number.isFinite(offset) && offset > 0 ? Math.floor(offset) : 0), isError: false }
392 }
393 return { text: value === null || value === undefined ? `Nothing at ${path}.` : JSON.stringify(value), isError: false }
394 }
395}
396
397/** One part of a view's text, under a line saying where it sits in the whole and how to read on. */
398function textPart(path: string, text: string, offset: number): string {
399 const from = Math.min(offset, text.length)
400 const to = Math.min(from + TEXT_PAGE, text.length)
401 const count = (n: number): string => n.toLocaleString('en-US')
402 const where = `${path}, characters ${count(from)} to ${count(to)} of ${count(text.length)}`
403 return `${where}${to < text.length ? `; read on with offset ${to}.` : ', the end.'}\n\n${text.slice(from, to)}`
404}
405
406/**
407 * The analyst's tools by the name the model calls them: the tool's own name, or the connection and
408 * the name when two connections offer the same one, the connection's slash written as `__` too.
409 * Limited to its connections and its tool list. The citation resolver is not among them: the room
410 * calls that one.
411 */
412function named(entries: ToolEntry[], analyst: Analyst): Map<string, { name: string; entry: ToolEntry }> {
413 const allowed = entries.filter(
414 (e) =>
415 e.name !== RESOLVER &&
416 (analyst.connections.length === 0 || analyst.connections.includes(e.connection)) &&
417 (analyst.tools.length === 0 || analyst.tools.includes(e.name)),
418 )
419 const count = new Map<string, number>()
420 for (const e of allowed) count.set(e.name, (count.get(e.name) ?? 0) + 1)
421 const tools = new Map<string, { name: string; entry: ToolEntry }>()
422 for (const entry of allowed) {
423 const name = (count.get(entry.name) ?? 0) > 1 ? `${entry.connection.replace(/\//g, '__')}__${entry.name}` : entry.name
424 tools.set(name, { name, entry })
425 }
426 return tools
427}
428
429function reasonOf(signal: AbortSignal): string {
430 return errorText(signal.reason, 'stopped')
431}
432
433function messageOf(err: unknown): string {
434 return errorText(err, 'failed')
435}