feat: add multi-agents template based on workflows (#271)

---------
Co-authored-by: Thuc Pham <[email protected]>

.changeset/tall-ties-own.md

@@ -0,0 +1,5 @@
+---
+"create-llama": patch
+---
+
+Add chat agent events UI

e2e/multiagent_template.spec.ts

@@ -0,0 +1,85 @@
+/* eslint-disable turbo/no-undeclared-env-vars */
+import { expect, test } from "@playwright/test";
+import { ChildProcess } from "child_process";
+import fs from "fs";
+import path from "path";
+import type {
+  TemplateFramework,
+  TemplatePostInstallAction,
+  TemplateUI,
+} from "../helpers";
+import { createTestDir, runCreateLlama, type AppType } from "./utils";
+
+const templateFramework: TemplateFramework = "fastapi";
+const dataSource: string = "--example-file";
+const templateUI: TemplateUI = "shadcn";
+const templatePostInstallAction: TemplatePostInstallAction = "runApp";
+const appType: AppType = "--frontend";
+const userMessage = "Write a blog post about physical standards for letters";
+
+test.describe(`Test multiagent template ${templateFramework} ${dataSource} ${templateUI} ${appType} ${templatePostInstallAction}`, async () => {
+  test.skip(
+    process.platform !== "linux" ||
+      process.env.FRAMEWORK !== "fastapi" ||
+      process.env.DATASOURCE === "--no-files",
+    "The multiagent template currently only works with FastAPI and files. We also only run on Linux to speed up tests.",
+  );
+  let port: number;
+  let externalPort: number;
+  let cwd: string;
+  let name: string;
+  let appProcess: ChildProcess;
+  // Only test without using vector db for now
+  const vectorDb = "none";
+
+  test.beforeAll(async () => {
+    port = Math.floor(Math.random() * 10000) + 10000;
+    externalPort = port + 1;
+    cwd = await createTestDir();
+    const result = await runCreateLlama(
+      cwd,
+      "multiagent",
+      templateFramework,
+      dataSource,
+      vectorDb,
+      port,
+      externalPort,
+      templatePostInstallAction,
+      templateUI,
+      appType,
+    );
+    name = result.projectName;
+    appProcess = result.appProcess;
+  });
+
+  test("App folder should exist", async () => {
+    const dirExists = fs.existsSync(path.join(cwd, name));
+    expect(dirExists).toBeTruthy();
+  });
+
+  test("Frontend should have a title", async ({ page }) => {
+    await page.goto(`http://localhost:${port}`);
+    await expect(page.getByText("Built by LlamaIndex")).toBeVisible();
+  });
+
+  test("Frontend should be able to submit a message and receive the start of a streamed response", async ({
+    page,
+  }) => {
+    await page.goto(`http://localhost:${port}`);
+    await page.fill("form input", userMessage);
+
+    const responsePromise = page.waitForResponse((res) =>
+      res.url().includes("/api/chat"),
+    );
+
+    await page.click("form button[type=submit]");
+
+    const response = await responsePromise;
+    expect(response.ok()).toBeTruthy();
+  });
+
+  // clean processes
+  test.afterAll(async () => {
+    appProcess?.kill();
+  });
+});

helpers/run-app.ts

@@ -81,7 +81,7 @@ export async function runApp(
   if (template === "extractor") {
     processes.push(runReflexApp(appPath, port, externalPort));
   }
-  if (template === "streaming") {
+  if (template === "streaming" || template === "multiagent") {
     if (framework === "fastapi" || framework === "express") {
       const backendRunner = framework === "fastapi" ? runFastAPIApp : runTSApp;
       if (frontend) {

helpers/typescript.ts

@@ -33,7 +33,8 @@ export const installTSTemplate = async ({
    * Copy the template files to the target directory.
    */
   console.log("\nInitializing project with template:", template, "\n");
-  const templatePath = path.join(templatesDir, "types", template, framework);
+  const type = template === "multiagent" ? "streaming" : template; // use nextjs streaming template for multiagent
+  const templatePath = path.join(templatesDir, "types", type, framework);
   const copySource = ["**"];
 
   await copy(copySource, root, {

questions.ts

@@ -287,27 +287,25 @@ export const askQuestions = async (
           },
         ];
 
-        if (program.template !== "multiagent") {
-          const modelConfigured =
-            !program.llamapack && program.modelConfig.isConfigured();
-          // If using LlamaParse, require LlamaCloud API key
-          const llamaCloudKeyConfigured = program.useLlamaParse
-            ? program.llamaCloudKey || process.env["LLAMA_CLOUD_API_KEY"]
-            : true;
-          const hasVectorDb = program.vectorDb && program.vectorDb !== "none";
-          // Can run the app if all tools do not require configuration
-          if (
-            !hasVectorDb &&
-            modelConfigured &&
-            llamaCloudKeyConfigured &&
-            !toolsRequireConfig(program.tools)
-          ) {
-            actionChoices.push({
-              title:
-                "Generate code, install dependencies, and run the app (~2 min)",
-              value: "runApp",
-            });
-          }
+        const modelConfigured =
+          !program.llamapack && program.modelConfig.isConfigured();
+        // If using LlamaParse, require LlamaCloud API key
+        const llamaCloudKeyConfigured = program.useLlamaParse
+          ? program.llamaCloudKey || process.env["LLAMA_CLOUD_API_KEY"]
+          : true;
+        const hasVectorDb = program.vectorDb && program.vectorDb !== "none";
+        // Can run the app if all tools do not require configuration
+        if (
+          !hasVectorDb &&
+          modelConfigured &&
+          llamaCloudKeyConfigured &&
+          !toolsRequireConfig(program.tools)
+        ) {
+          actionChoices.push({
+            title:
+              "Generate code, install dependencies, and run the app (~2 min)",
+            value: "runApp",
+          });
         }
 
         const { action } = await prompts(
@@ -341,7 +339,7 @@ export const askQuestions = async (
           choices: [
             { title: "Agentic RAG (e.g. chat with docs)", value: "streaming" },
             {
-              title: "Multi-agent app (using llama-agents)",
+              title: "Multi-agent app (using workflows)",
               value: "multiagent",
             },
             { title: "Structured Extractor", value: "extractor" },
@@ -448,7 +446,7 @@ export const askQuestions = async (
 
   if (
     (program.framework === "express" || program.framework === "fastapi") &&
-    program.template === "streaming"
+    (program.template === "streaming" || program.template === "multiagent")
   ) {
     // if a backend-only framework is selected, ask whether we should create a frontend
     if (program.frontend === undefined) {

templates/types/multiagent/fastapi/README-template.md

@@ -1,4 +1,18 @@
-This is a [LlamaIndex](https://www.llamaindex.ai/) project using [FastAPI](https://fastapi.tiangolo.com/) bootstrapped with [`create-llama`](https://github.com/run-llama/LlamaIndexTS/tree/main/packages/create-llama).
+This is a [LlamaIndex](https://www.llamaindex.ai/) multi-agents project using [Workflows](https://docs.llamaindex.ai/en/stable/understanding/workflows/).
+
+## Overview
+
+This example is using three agents to generate a blog post:
+
+- a researcher that retrieves content via a RAG pipeline,
+- a writer that specializes in writing blog posts and
+- a reviewer that is reviewing the blog post.
+
+There are three different methods how the agents can interact to reach their goal:
+
+1. [Choreography](./app/examples/choreography.py) - the agents decide themselves to delegate a task to another agent
+1. [Orchestrator](./app/examples/orchestrator.py) - a central orchestrator decides which agent should execute a task
+1. [Explicit Workflow](./app/examples/workflow.py) - a pre-defined workflow specific for the task is used to execute the tasks
 
 ## Getting Started
 
@@ -8,43 +22,48 @@ First, setup the environment with poetry:
 
 ```shell
 poetry install
-poetry shell
 ```
 
 Then check the parameters that have been pre-configured in the `.env` file in this directory. (E.g. you might need to configure an `OPENAI_API_KEY` if you're using OpenAI as model provider).
 
-Second, generate the embeddings of the documents in the `./data` directory (if this folder exists - otherwise, skip this step):
+Second, generate the embeddings of the documents in the `./data` directory:
 
 ```shell
 poetry run generate
 ```
 
-Third, run all the services in one command:
+Third, run the development server:
 
 ```shell
 poetry run python main.py
 ```
 
-You can monitor and test the agent services with `llama-agents` monitor TUI:
+Per default, the example is using the explicit workflow. You can change the example by setting the `EXAMPLE_TYPE` environment variable to `choreography` or `orchestrator`.
 
-```shell
-poetry run llama-agents monitor --control-plane-url http://127.0.0.1:8001
+The example provides one streaming API endpoint `/api/chat`.
+You can test the endpoint with the following curl request:
+
+```
+curl --location 'localhost:8000/api/chat' \
+--header 'Content-Type: application/json' \
+--data '{ "messages": [{ "role": "user", "content": "Write a blog post about physical standards for letters" }] }'
 ```
 
-## Services:
+You can start editing the API by modifying `app/api/routers/chat.py` or `app/examples/workflow.py`. The API auto-updates as you save the files.
+
+Open [http://localhost:8000/docs](http://localhost:8000/docs) with your browser to see the Swagger UI of the API.
 
-- Message queue (port 8000): To exchange the message between services
-- Control plane (port 8001): A gateway to manage the tasks and services.
-- Human consumer (port 8002): To handle result when the task is completed.
-- Agent service `query_engine` (port 8003): Agent that can query information from the configured LlamaIndex index.
-- Agent service `dummy_agent` (port 8004): A dummy agent that does nothing. Good starting point to add more agents.
+The API allows CORS for all origins to simplify development. You can change this behavior by setting the `ENVIRONMENT` environment variable to `prod`:
 
-The ports listed above are set by default, but you can change them in the `.env` file.
+```
+ENVIRONMENT=prod poetry run python main.py
+```
 
 ## Learn More
 
 To learn more about LlamaIndex, take a look at the following resources:
 
 - [LlamaIndex Documentation](https://docs.llamaindex.ai) - learn about LlamaIndex.
+- [Workflows Introduction](https://docs.llamaindex.ai/en/stable/understanding/workflows/) - learn about LlamaIndex workflows.
 
 You can check out [the LlamaIndex GitHub repository](https://github.com/run-llama/llama_index) - your feedback and contributions are welcome!

templates/types/multiagent/fastapi/app/agents/multi.py

@@ -0,0 +1,83 @@
+import asyncio
+from typing import Any, List
+
+from llama_index.core.tools.types import ToolMetadata, ToolOutput
+from llama_index.core.tools.utils import create_schema_from_function
+from llama_index.core.workflow import Context, Workflow
+
+from app.agents.single import (
+    AgentRunResult,
+    ContextAwareTool,
+    FunctionCallingAgent,
+)
+from app.agents.planner import StructuredPlannerAgent
+
+
+class AgentCallTool(ContextAwareTool):
+    def __init__(self, agent: Workflow) -> None:
+        self.agent = agent
+        name = f"call_{agent.name}"
+
+        async def schema_call(input: str) -> str:
+            pass
+
+        # create the schema without the Context
+        fn_schema = create_schema_from_function(name, schema_call)
+        self._metadata = ToolMetadata(
+            name=name,
+            description=(
+                f"Use this tool to delegate a sub task to the {agent.name} agent."
+                + (f" The agent is an {agent.role}." if agent.role else "")
+            ),
+            fn_schema=fn_schema,
+        )
+
+    # overload the acall function with the ctx argument as it's needed for bubbling the events
+    async def acall(self, ctx: Context, input: str) -> ToolOutput:
+        task = asyncio.create_task(self.agent.run(input=input))
+        # bubble all events while running the agent to the calling agent
+        async for ev in self.agent.stream_events():
+            ctx.write_event_to_stream(ev)
+        ret: AgentRunResult = await task
+        response = ret.response.message.content
+        return ToolOutput(
+            content=str(response),
+            tool_name=self.metadata.name,
+            raw_input={"args": input, "kwargs": {}},
+            raw_output=response,
+        )
+
+
+class AgentCallingAgent(FunctionCallingAgent):
+    def __init__(
+        self,
+        *args: Any,
+        name: str,
+        agents: List[FunctionCallingAgent] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        agents = agents or []
+        tools = [AgentCallTool(agent=agent) for agent in agents]
+        super().__init__(*args, name=name, tools=tools, **kwargs)
+        # call add_workflows so agents will get detected by llama agents automatically
+        self.add_workflows(**{agent.name: agent for agent in agents})
+
+
+class AgentOrchestrator(StructuredPlannerAgent):
+    def __init__(
+        self,
+        *args: Any,
+        name: str = "orchestrator",
+        agents: List[FunctionCallingAgent] | None = None,
+        **kwargs: Any,
+    ) -> None:
+        agents = agents or []
+        tools = [AgentCallTool(agent=agent) for agent in agents]
+        super().__init__(
+            *args,
+            name=name,
+            tools=tools,
+            **kwargs,
+        )
+        # call add_workflows so agents will get detected by llama agents automatically
+        self.add_workflows(**{agent.name: agent for agent in agents})