feat: support human in the loop for TS

.changeset/eight-moons-perform.md

@@ -0,0 +1,5 @@
+---
+"@llamaindex/server": patch
+---
+
+feat: support human in the loop for TS

packages/server/examples/hitl/README.md

@@ -0,0 +1,6 @@
+This example demonstrates human in the loop workflow. Export OpenAI API key and start the server in dev mode.
+
+```bash
+export OPENAI_API_KEY=<your-openai-api-key>
+npx nodemon --exec tsx index.ts --ignore output/*
+```

packages/server/examples/hitl/components/cli_human_input.tsx

@@ -0,0 +1,95 @@
+import { Button } from "@/components/ui/button";
+import { Card, CardContent, CardFooter } from "@/components/ui/card";
+import { JSONValue, useChatUI } from "@llamaindex/chat-ui";
+import React, { FC, useState } from "react";
+import { z } from "zod";
+
+// This schema is equivalent to the CLICommand model defined in events.py
+const CLIInputEventSchema = z.object({
+  command: z.string(),
+});
+type CLIInputEvent = z.infer<typeof CLIInputEventSchema>;
+
+const CLIHumanInput: FC<{
+  events: JSONValue[];
+}> = ({ events }) => {
+  const inputEvent = (events || [])
+    .map((ev) => {
+      const parseResult = CLIInputEventSchema.safeParse(ev);
+      return parseResult.success ? parseResult.data : null;
+    })
+    .filter((ev): ev is CLIInputEvent => ev !== null)
+    .at(-1);
+
+  const { append } = useChatUI();
+  const [confirmedValue, setConfirmedValue] = useState<boolean | null>(null);
+  const [editableCommand, setEditableCommand] = useState<string | undefined>(
+    inputEvent?.command,
+  );
+
+  // Update editableCommand if inputEvent changes (e.g. new event comes in)
+  React.useEffect(() => {
+    setEditableCommand(inputEvent?.command);
+  }, [inputEvent?.command]);
+
+  const handleConfirm = () => {
+    append({
+      content: "Yes",
+      role: "user",
+      annotations: [
+        {
+          type: "human_response",
+          data: {
+            execute: true,
+            command: editableCommand, // Use editable command
+          },
+        },
+      ],
+    });
+    setConfirmedValue(true);
+  };
+
+  const handleCancel = () => {
+    append({
+      content: "No",
+      role: "user",
+      annotations: [
+        {
+          type: "human_response",
+          data: {
+            execute: false,
+            command: inputEvent?.command,
+          },
+        },
+      ],
+    });
+    setConfirmedValue(false);
+  };
+
+  return (
+    <Card className="my-4">
+      <CardContent className="pt-6">
+        <p className="text-sm text-gray-700">
+          Do you want to execute the following command?
+        </p>
+        <input
+          disabled
+          type="text"
+          value={editableCommand || ""}
+          onChange={(e) => setEditableCommand(e.target.value)}
+          className="my-2 w-full overflow-x-auto rounded border border-gray-300 bg-gray-100 p-3 font-mono text-xs text-gray-800"
+        />
+      </CardContent>
+      {confirmedValue === null ? (
+        <CardFooter className="flex justify-end gap-2">
+          <>
+            <Button onClick={handleConfirm}>Yes</Button>
+            <Button onClick={handleCancel}>No</Button>
+          </>
+        </CardFooter>
+      ) : null}
+    </Card>
+  );
+};
+
+export default CLIHumanInput;

packages/server/examples/hitl/index.ts

@@ -0,0 +1,20 @@
+import { OpenAI } from "@llamaindex/openai";
+import { LlamaIndexServer } from "@llamaindex/server";
+import { Settings } from "llamaindex";
+import { workflowFactory } from "./src/app/workflow";
+
+Settings.llm = new OpenAI({
+  model: "gpt-4o-mini",
+});
+
+new LlamaIndexServer({
+  workflow: workflowFactory,
+  uiConfig: {
+    starterQuestions: [
+      "Check status of git in the current directory",
+      "List all files in the current directory",
+    ],
+    componentsDir: "components",
+  },
+  port: 3000,
+}).start();

packages/server/examples/hitl/src/app/events.ts

@@ -0,0 +1,12 @@
+import { humanInputEvent, humanResponseEvent } from "@llamaindex/server";
+
+export const cliHumanInputEvent = humanInputEvent<{
+  type: "cli_human_input";
+  data: { command: string };
+  response: typeof cliHumanResponseEvent;
+}>();
+
+export const cliHumanResponseEvent = humanResponseEvent<{
+  type: "human_response";
+  data: { execute: boolean; command: string };
+}>();

packages/server/examples/hitl/src/app/tools.ts

@@ -0,0 +1,20 @@
+import { execSync } from "child_process";
+import { tool } from "llamaindex";
+import { z } from "zod";
+
+export const cliExecutor = tool({
+  name: "cli_executor",
+  description: "This tool executes a command and returns the output.",
+  parameters: z.object({ command: z.string() }),
+  execute: async ({ command }) => {
+    try {
+      const output = execSync(command, {
+        encoding: "utf-8",
+      });
+      return output;
+    } catch (error) {
+      console.error(error);
+      return "Command failed";
+    }
+  },
+});

packages/server/examples/hitl/src/app/workflow.ts

@@ -0,0 +1,106 @@
+import { OpenAI } from "@llamaindex/openai";
+import { toAgentRunEvent, writeResponseToStream } from "@llamaindex/server";
+import { chatWithTools } from "@llamaindex/tools";
+import {
+  createWorkflow,
+  getContext,
+  startAgentEvent,
+  stopAgentEvent,
+  withSnapshot,
+  workflowEvent,
+} from "@llamaindex/workflow";
+import { ChatMessage, Settings, ToolCallLLM } from "llamaindex";
+import { cliHumanInputEvent, cliHumanResponseEvent } from "./events";
+import { cliExecutor } from "./tools";
+
+Settings.llm = new OpenAI({
+  model: "gpt-4o-mini",
+});
+
+const summaryEvent = workflowEvent<string>(); // simple event to summarize the result
+
+export const workflowFactory = (body: unknown) => {
+  const llm = Settings.llm as ToolCallLLM;
+
+  if (!llm.supportToolCall) {
+    throw new Error("LLM is not a ToolCallLLM");
+  }
+
+  const { messages } = body as { messages: ChatMessage[] };
+
+  const workflow = withSnapshot(createWorkflow());
+
+  workflow.handle([startAgentEvent], async ({ data }) => {
+    const { userInput, chatHistory = [] } = data;
+    if (!userInput) {
+      throw new Error("User input is required");
+    }
+
+    // in this example, we use chatWithTools to decide should perform a tool call or not
+    // if cli executor is called, emit HumanInputEvent to ask user for permission
+    const toolCallResponse = await chatWithTools(
+      llm,
+      [cliExecutor],
+      chatHistory.concat({ role: "user", content: userInput }),
+    );
+    const cliExecutorToolCall = toolCallResponse.toolCalls.find(
+      (toolCall) => toolCall.name === cliExecutor.metadata.name,
+    );
+    const command = cliExecutorToolCall?.input?.command as string;
+    if (command) {
+      return cliHumanInputEvent.with({
+        type: "cli_human_input",
+        data: { command },
+        response: cliHumanResponseEvent,
+      });
+    }
+
+    // if no tool call, just response as normal
+    return summaryEvent.with("");
+  });
+
+  // do actions after getting response from human
+  workflow.handle([cliHumanResponseEvent], async ({ data }) => {
+    const { sendEvent } = getContext();
+    const { command, execute } = data.data;
+
+    if (!execute) {
+      // stop the workflow if user reject to execute the command
+      return summaryEvent.with(`User reject to execute the command ${command}`);
+    }
+
+    sendEvent(
+      toAgentRunEvent({
+        agent: "CLI Executor",
+        text: `Execute the command "${command}" and return the result`,
+        type: "text",
+      }),
+    );
+
+    const result = (await cliExecutor.call({ command })) as string;
+
+    return summaryEvent.with(
+      `Executed the command ${command} and got the result: ${result}`,
+    );
+  });
+
+  workflow.handle([summaryEvent], async ({ data: summaryResult }) => {
+    const { sendEvent } = getContext();
+
+    const chatHistory = messages;
+    if (summaryResult) {
+      chatHistory.push({ role: "user", content: summaryResult });
+    }
+
+    const stream = await llm.chat({
+      messages: chatHistory,
+      stream: true,
+    });
+
+    const result = await writeResponseToStream(stream, sendEvent);
+
+    return stopAgentEvent.with({ result });
+  });
+
+  return workflow;
+};

packages/server/next/app/api/chat/route.ts

@@ -1,16 +1,19 @@
-import { type AgentInputData } from "@llamaindex/workflow";
 import { type Message } from "ai";
 import { type MessageType } from "llamaindex";
 import { NextRequest, NextResponse } from "next/server";
 
 // import chat utils
 import {
+  getHumanResponsesFromMessage,
+  pauseForHumanInput,
+  processWorkflowStream,
   runWorkflow,
   sendSuggestedQuestionsEvent,
   toDataStream,
 } from "./utils";
 
 // import workflow factory and settings from local file
+import { stopAgentEvent } from "@llamaindex/workflow";
 import { initSettings } from "./app/settings";
 import { workflowFactory } from "./app/workflow";
 
@@ -21,7 +24,10 @@ export async function POST(req: NextRequest) {
     const reqBody = await req.json();
     const suggestNextQuestions = process.env.SUGGEST_NEXT_QUESTIONS === "true";
 
-    const { messages } = reqBody as { messages: Message[] };
+    const { messages, id: requestId } = reqBody as {
+      messages: Message[];
+      id?: string;
+    };
     const chatHistory = messages.map((message) => ({
       role: message.role as MessageType,
       content: message.content,
@@ -36,25 +42,31 @@ export async function POST(req: NextRequest) {
         { status: 400 },
       );
     }
-    const workflowInput: AgentInputData = {
-      userInput: lastMessage.content,
-      chatHistory,
-    };
 
     const abortController = new AbortController();
     req.signal.addEventListener("abort", () =>
       abortController.abort("Connection closed"),
     );
 
-    const workflow = await workflowFactory(reqBody);
-    const workflowEventStream = await runWorkflow(
-      workflow,
-      workflowInput,
-      abortController.signal,
+    const context = await runWorkflow({
+      workflow: await workflowFactory(reqBody),
+      input: { userInput: lastMessage.content, chatHistory },
+      human: {
+        snapshotId: requestId, // use requestId to restore snapshot
+        responses: getHumanResponsesFromMessage(lastMessage),
+      },
+    });
+
+    const stream = processWorkflowStream(context.stream).until(
+      (event) =>
+        abortController.signal.aborted || stopAgentEvent.include(event),
     );
 
-    const dataStream = toDataStream(workflowEventStream, {
+    const dataStream = toDataStream(stream, {
       callbacks: {
+        onPauseForHumanInput: async (responseEvent) => {
+          await pauseForHumanInput(context, responseEvent, requestId); // use requestId to save snapshot
+        },
         onFinal: async (completion, dataStreamWriter) => {
           chatHistory.push({
             role: "assistant" as MessageType,
@@ -66,7 +78,6 @@ export async function POST(req: NextRequest) {
         },
       },
     });
-
     return new Response(dataStream, {
       status: 200,
       headers: {

packages/server/package.json

@@ -37,7 +37,7 @@
     "prebuild": "pnpm clean",
     "build": "bunchee",
     "postbuild": "pnpm prepare:nextjs && pnpm prepare:ts-server && pnpm prepare:py-static",
-    "prepare:nextjs": "cp -r ./next ./project && cp -r ./src/utils ./project/app/api/chat && cp -r ./project-config/* ./project/",
+    "prepare:nextjs": "cp -r ./next ./project && cp -r ./src/utils/* ./project/app/api/chat && cp -r ./project-config/* ./project/",
     "prepare:ts-server": "pnpm copy:next-src && pnpm build:css && pnpm build:api",
     "prepare:py-static": "pnpm prepare:static && pnpm build:static && pnpm copy:static",
     "copy:next-src": "cp -r ./next ./server",