Merge pull request #14 from get-convex/ian/threads-no-order

`v0.0.5`

Author: ianmacartney

example/convex/example.ts

@@ -270,20 +270,23 @@ export const t = action({
           },
         }),
       },
-      maxSteps: 5,
+      maxSteps: 20,
     });
 
-    const { threadId, thread } = await fastAgent.createThread(ctx, {});
-    // eslint-disable-next-line @typescript-eslint/no-floating-promises
-    const s = await thread.streamText({
-      prompt:
-        "Do something four times, then do something else, then do something else again",
-    });
-    console.log("s", s);
-
-    for await (const chunk of s.textStream) {
-      console.log(chunk);
-    }
+    // // eslint-disable-next-line @typescript-eslint/no-floating-promises
+    await Promise.all(
+      Array.from({ length: 10 }).map(async (i) => {
+        const { threadId, thread } = await fastAgent.createThread(ctx, {
+          userId: "123",
+        });
+        const s = await thread.streamText({
+          prompt: "Do something twice",
+        });
+        console.log("agent", i);
+        await s.response;
+      }),
+    );
+    console.log("done");
     // return result.text;
   },
 });

example/convex/http.ts

@@ -1,12 +1,11 @@
 import { httpRouter } from "convex/server";
-import { httpAction } from "./_generated/server";
 import { streamHttpAction } from "./example";
 
 const http = httpRouter();
 
 http.route({
   path: "/streamText",
-  method: "GET",
+  method: "POST",
   handler: streamHttpAction,
 });
 

package-lock.json

@@ -7,7 +7,7 @@
     "email": "support@convex.dev",
     "url": "https://github.com/get-convex/agent/issues"
   },
-  "version": "0.0.4",
+  "version": "0.0.5-alpha.1",
   "license": "Apache-2.0",
   "keywords": [
     "convex",

package.json

@@ -14,7 +14,6 @@ import type {
   ToolChoice,
   ToolExecutionOptions,
   ToolSet,
-  Message as UIMessage,
 } from "ai";
 import {
   generateObject,
@@ -38,6 +37,7 @@ import {
   VectorDimension,
 } from "../component/vector/tables";
 import {
+  AIMessageWithoutId,
   promptOrMessagesToCoreMessages,
   serializeMessageWithId,
   serializeNewMessagesInStep,
@@ -288,17 +288,14 @@ export class Agent<AgentTools extends ToolSet> {
       included = new Set(searchMessages.map((m) => m._id));
       contextMessages.push(...searchMessages.map((m) => m.message!));
     }
-    if (args.threadId) {
+    if (args.threadId && opts.recentMessages !== 0) {
       const { page } = await ctx.runQuery(
         this.component.messages.getThreadMessages,
         {
           threadId: args.threadId,
           isTool: args.includeToolCalls ?? false,
           paginationOpts: {
-            numItems:
-              args.recentMessages ??
-              this.options.contextOptions?.recentMessages ??
-              DEFAULT_RECENT_MESSAGES,
+            numItems: opts.recentMessages ?? DEFAULT_RECENT_MESSAGES,
             cursor: null,
           },
           parentMessageId: args.parentMessageId,
@@ -555,7 +552,7 @@ export class Agent<AgentTools extends ToolSet> {
   async saveMessagesAndFetchContext<
     T extends {
       prompt?: string;
-      messages?: CoreMessage[] | Omit<UIMessage, "id">[];
+      messages?: CoreMessage[] | AIMessageWithoutId[];
       system?: string;
     },
   >(
@@ -834,7 +831,7 @@ export class Agent<AgentTools extends ToolSet> {
             ...rest,
             schema: jsonSchema(schema),
           } as unknown as OurStreamObjectArgs<unknown>);
-          for await (const chunk of value.fullStream) {
+          for await (const _ of value.fullStream) {
             // no-op, just consume the stream
           }
           return value.object;
@@ -859,23 +856,28 @@ export class Agent<AgentTools extends ToolSet> {
     args: Validator<unknown, "required", string>;
     contextOptions?: ContextOptions;
     maxSteps?: number;
+    provideMessageHistory?: boolean;
   }) {
     return createTool({
       ...spec,
-      handler: async (ctx, args) => {
+      handler: async (ctx, args, options) => {
         const maxSteps = spec.maxSteps ?? this.options.maxSteps;
         const contextOptions =
           spec.contextOptions && this.mergedContextOptions(spec.contextOptions);
-        const value = await this.generateText(
-          ctx,
-          { userId: ctx.userId, threadId: ctx.threadId },
-          {
-            prompt: JSON.stringify(args),
-            parentMessageId: ctx.messageId,
-            maxSteps,
-            ...contextOptions,
-          }
-        );
+        const { thread } = await this.createThread(ctx, {
+          parentThreadIds: ctx.threadId ? [ctx.threadId] : undefined,
+          userId: ctx.userId,
+        });
+        const messages = spec.provideMessageHistory ? options.messages : [];
+        messages.push({
+          role: "user",
+          content: JSON.stringify(args),
+        });
+        const value = await thread.generateText({
+          messages,
+          maxSteps,
+          ...contextOptions,
+        });
         return value.text;
       },
     });

src/client/index.ts

@@ -8,7 +8,6 @@
  * @module
  */
 
-import type * as lib from "../lib.js";
 import type * as messages from "../messages.js";
 import type * as vector_index from "../vector/index.js";
 import type * as vector_tables from "../vector/tables.js";
@@ -28,7 +27,6 @@ import type {
  * ```
  */
 declare const fullApi: ApiFromModules<{
-  lib: typeof lib;
   messages: typeof messages;
   "vector/index": typeof vector_index;
   "vector/tables": typeof vector_tables;

src/component/_generated/api.d.ts

@@ -55,7 +55,7 @@ export const getThreadsByUserId = query({
   handler: async (ctx, args) => {
     const threads = await paginator(ctx.db, schema)
       .query("threads")
-      .withIndex("userId_order", (q) => q.eq("userId", args.userId))
+      .withIndex("userId", (q) => q.eq("userId", args.userId))
       .order(args.order ?? "desc")
       .paginate(args.paginationOpts ?? { cursor: null, numItems: 100 });
     return threads;
@@ -68,15 +68,8 @@ const vThread = schema.tables.threads.validator;
 export const createThread = mutation({
   args: omit(vThread.fields, ["order", "status"]),
   handler: async (ctx, args) => {
-    const latestThread = await ctx.db
-      .query("threads")
-      .withIndex("userId_order", (q) => q.eq("userId", args.userId))
-      .order("desc")
-      .first();
-    const order = (latestThread?.order ?? -1) + 1;
     const threadId = await ctx.db.insert("threads", {
       ...args,
-      order,
       status: "active",
     });
     return (await ctx.db.get(threadId))!;
@@ -207,7 +200,7 @@ async function deletePageForUserId(
 ): Promise<DeleteAllReturns> {
   const threads = await paginator(ctx.db, schema)
     .query("threads")
-    .withIndex("userId_order", (q) => q.eq("userId", args.userId))
+    .withIndex("userId", (q) => q.eq("userId", args.userId))
     .order("desc")
     .paginate({
       numItems: 100,

src/component/lib.ts

@@ -7,7 +7,6 @@ import vectorTables, { vVectorId } from "./vector/tables";
 export const schema = defineSchema({
   threads: defineTable({
     userId: v.optional(v.string()), // Unset for anonymous
-    order: v.optional(v.number()), // within a domain
     // TODO: is this bubbling up in continue?
     defaultSystemPrompt: v.optional(v.string()),
     title: v.optional(v.string()),
@@ -17,7 +16,9 @@ export const schema = defineSchema({
     // the parent thread(s). There are multiple if the thread is a merging of
     // multiple threads.
     parentThreadIds: v.optional(v.array(v.id("threads"))),
-  }).index("userId_order", ["userId", "order"]),
+    // DEPRECATED
+    order: /*DEPRECATED*/ v.optional(v.number()),
+  }).index("userId", ["userId"]),
   // TODO: text search on title/ summary
   messages: defineTable({
     id: v.optional(v.string()), // external id, e.g. from Vercel AI SDK