Jaspers Hub

jaspers / agents

pluginOfficialReviewed★ 0↓ 1

Research Agents: Lets you define research agents that chat and answer with citations from SEC filings, your documents, and the web, via Jaspers' research-store service.

6.9 KB
Raw
1import type { BackendContext } from '@jaspers-ai/sdk'
2import { loadAgents, publishAgents, saveAgent } from './agent-files.ts'
3import { firstAgent, type Agent } from './agents.ts'
4import { savePage } from './doc-files.ts'
5import {
6 companiesUsed,
7 firstTurn,
8 interrupted,
9 newConversationId,
10 recipients,
11 tagged,
12 type AgentMessage,
13 type Chat,
14 type UserMessage,
15} from './chat.ts'
16import { errorText } from './errors.ts'
17import { research, type Hit } from './research.ts'
18import { companies, keepCurrent, readCompany } from './sec-files.ts'
19import { runAgent, type RunnerContext } from './runner.ts'
20
21// The conversations: each kept as chats/<id>.json in the plugin's data folder and held here once read.
22// A view shows one conversation, the one its panel names, and a panel placed anew starts a new one;
23// chat/<id> is the live value it shows. A message starts one job per agent it goes to, keyed
24// <id>/<seq>, which writes the agent's answer into its conversation as it grows.
25
26const FOLDER = 'chats'
27const chats = new Map<string, Chat>()
28/** Writes go one after another, so a later state never lands under an earlier one. */
29let writes: Promise<void> = Promise.resolve()
30
31async function load(ctx: BackendContext, id: string): Promise<Chat> {
32 const held = chats.get(id)
33 if (held) return held
34 const chat = interrupted(JSON.parse(await ctx.files.read(`${FOLDER}/${id}.json`)) as Chat)
35 // Another call may have loaded it while this one read.
36 const raced = chats.get(id)
37 if (raced) return raced
38 chats.set(id, chat)
39 return chat
40}
41
42function commit(ctx: BackendContext, id: string, chat: Chat): void {
43 chats.set(id, chat)
44 ctx.live.set(`chat/${id}`, chat)
45 writes = writes
46 .then(() => ctx.files.write(`${FOLDER}/${id}.json`, JSON.stringify(chat)))
47 .catch((err: unknown) => console.warn(`conversation ${id}: ${errorText(err, 'not saved')}`))
48}
49
50function checkId(id: string): string {
51 if (!/^[\w-]+$/.test(id)) throw new Error(`No conversation ${id}.`)
52 return id
53}
54
55/** Publishes a panel's conversation and says which: the one it names while that is kept, else a new one. */
56export async function openConversation(ctx: BackendContext, wanted?: string): Promise<string> {
57 if (wanted && /^[\w-]+$/.test(wanted)) {
58 try {
59 ctx.live.set(`chat/${wanted}`, await load(ctx, wanted))
60 return wanted
61 } catch {
62 // Its file is gone: the panel starts again.
63 }
64 }
65 const now = new Date().toISOString()
66 const id = newConversationId(Date.now(), Math.random())
67 commit(ctx, id, { messages: [], lastTo: [], created: now, updated: now })
68 return id
69}
70
71/**
72 * Notes a file the user attached in a conversation, before the next message: whoever answers that
73 * message is told it was just attached, since attaching and then asking is asking about it.
74 */
75export async function attach(ctx: BackendContext, rawId: string, doc: string, name: string): Promise<void> {
76 const id = checkId(rawId)
77 const chat = await load(ctx, id)
78 const at = new Date().toISOString()
79 const before = (chat.messages.at(-1)?.seq ?? 0) + 1
80 commit(ctx, id, { ...chat, attached: [...(chat.attached ?? []), { doc, name, at, before }], updated: at })
81}
82
83export async function post(ctx: BackendContext, rawId: string, text: string): Promise<string> {
84 const id = checkId(rawId)
85 const words = text.trim()
86 if (!words) throw new Error('Write a message.')
87 const agents = await usableAgents(ctx)
88 const chat = await load(ctx, id)
89 const to = recipients(words, agents, chat.lastTo)
90 const at = new Date().toISOString()
91 let seq = (chat.messages.at(-1)?.seq ?? 0) + 1
92 const asked: UserMessage = { seq, role: 'user', text: words, to, at }
93 const answers: AgentMessage[] = to.map((agent) => ({
94 seq: ++seq,
95 role: 'agent',
96 agent,
97 inReplyTo: asked.seq,
98 status: 'working',
99 text: '',
100 searches: [],
101 usage: { input: 0, output: 0 },
102 error: null,
103 at,
104 }))
105 const named = tagged(words, agents)
106 commit(ctx, id, {
107 ...chat,
108 messages: [...chat.messages, asked, ...answers],
109 lastTo: named.length > 0 ? named : chat.lastTo,
110 updated: at,
111 })
112
113 const names = new Map(agents.map((a) => [a.id, a.name]))
114 const turn = firstTurn(chats.get(id)!.messages, asked.seq, names, chat.attached)
115 for (const answer of answers) {
116 const agent = agents.find((a) => a.id === answer.agent)!
117 ctx.jobs.start(`${id}/${answer.seq}`, async (signal) => {
118 await runAgent(runnerContext(ctx, id), agent, answer, turn, async (message) => replace(ctx, id, message), signal)
119 })
120 }
121 return `Sent to ${to.map((a) => `@${a}`).join(', ')}.`
122}
123
124/** The agents that read, one made first when there are none, so a message always has someone to answer it. */
125async function usableAgents(ctx: BackendContext): Promise<Agent[]> {
126 const all = await loadAgents(ctx)
127 const usable = all.filter((a) => a.error === null)
128 if (usable.length > 0) return usable
129 await saveAgent(ctx, firstAgent(all.map((a) => a.id)))
130 return (await publishAgents(ctx)).filter((a) => a.error === null)
131}
132
133/** Stops every answer still being written in a conversation. */
134export function stop(ctx: BackendContext, rawId: string): string {
135 const prefix = `${checkId(rawId)}/`
136 const running = ctx.jobs.running().filter((key) => key.startsWith(prefix))
137 for (const key of running) ctx.jobs.abort(key, 'Stopped.')
138 return running.length > 0 ? 'Stopped.' : 'Nothing was running.'
139}
140
141function replace(ctx: BackendContext, id: string, message: AgentMessage): void {
142 const chat = chats.get(id)
143 // A message no longer in its conversation has nowhere to go.
144 const same = (m: Chat['messages'][number]): boolean =>
145 m.seq === message.seq && m.role === 'agent' && m.agent === message.agent && m.at === message.at
146 if (!chat?.messages.some(same)) return
147 commit(ctx, id, { ...chat, messages: chat.messages.map((m) => (same(m) ? message : m)) })
148}
149
150/** The companies every conversation held here touches, each once: what the status poller asks the store about. */
151export function openTickers(): string[] {
152 const out = new Set<string>()
153 for (const chat of chats.values()) for (const ticker of companiesUsed(chat.messages)) out.add(ticker)
154 return [...out]
155}
156
157function runnerContext(ctx: BackendContext, id: string): RunnerContext {
158 return {
159 llm: ctx.llm,
160 search: (args, signal) => research<Hit[]>(ctx, 'search', args, signal),
161 fetchPage: (url, signal) => savePage(ctx, url, signal),
162 searchWeb: (query, signal) => ctx.web.search(query, { signal }),
163 // The store holds every account's companies: the agent is told of the ones this conversation touches.
164 holdings: async (signal) => {
165 const held = await companies(ctx, companiesUsed(chats.get(id)?.messages ?? []), signal)
166 return held.map((c) => ({ ticker: c.ticker, name: c.company ?? c.ticker, reads: c.reads }))
167 },
168 addCompany: (ticker, years, categories, signal, progress) =>
169 readCompany(ctx, ticker, years, categories, signal, progress),
170 keepCurrent: (ticker, signal, progress) => keepCurrent(ctx, ticker, signal, progress),
171 skills: ctx.skills,
172 }
173}