feat: add agent flows

Signed-off-by: Tomas Dvorak <toomas2d@gmail.com>

examples/flows/agent.ts

@@ -8,7 +8,7 @@ import { WikipediaTool } from "bee-agent-framework/tools/search/wikipedia";
 import { OpenMeteoTool } from "bee-agent-framework/tools/weather/openMeteo";
 import { ReadOnlyMemory } from "bee-agent-framework/memory/base";
 import { UnconstrainedMemory } from "bee-agent-framework/memory/unconstrainedMemory";
-import { Flow } from "bee-agent-framework/experimental/flows";
+import { Flow } from "bee-agent-framework/experimental/flows/flow";
 import { createConsoleReader } from "examples/helpers/io.js";
 
 const schema = z.object({

examples/flows/contentCreator.ts

@@ -1,6 +1,6 @@
 import "dotenv/config";
-import { Flow } from "bee-agent-framework/experimental/flows";
 import { z } from "zod";
+import { Flow } from "bee-agent-framework/experimental/flows/flow";
 import { BeeAgent } from "bee-agent-framework/agents/bee/agent";
 import { UnconstrainedMemory } from "bee-agent-framework/memory/unconstrainedMemory";
 import { BAMChatLLM } from "bee-agent-framework/adapters/bam/chat";

examples/flows/multiAgents.ts

@@ -0,0 +1,49 @@
+import "dotenv/config";
+import { BAMChatLLM } from "bee-agent-framework/adapters/bam/chat";
+import { UnconstrainedMemory } from "bee-agent-framework/memory/unconstrainedMemory";
+import { createConsoleReader } from "examples/helpers/io.js";
+import { OpenMeteoTool } from "bee-agent-framework/tools/weather/openMeteo";
+import { WikipediaTool } from "bee-agent-framework/tools/search/wikipedia";
+import { AgentFlow } from "bee-agent-framework/experimental/flows/agent";
+import { BaseMessage, Role } from "bee-agent-framework/llms/primitives/message";
+
+const flow = new AgentFlow();
+flow.addAgent({
+  name: "WeatherAgent",
+  instructions: "You are a weather assistant. Respond only if you can provide a useful answer.",
+  tools: [new OpenMeteoTool()],
+  llm: BAMChatLLM.fromPreset("meta-llama/llama-3-1-70b-instruct"),
+});
+flow.addAgent({
+  name: "ResearchAgent",
+  instructions: "You are a researcher assistant. Respond only if you can provide a useful answer.",
+  tools: [new WikipediaTool()],
+  llm: BAMChatLLM.fromPreset("meta-llama/llama-3-1-70b-instruct"),
+});
+flow.addAgent({
+  name: "FinalAgent",
+  instructions:
+    "You are a helpful assistant. Your task is to make a final answer from the current conversation, starting with the last user message, that provides all useful information.",
+  tools: [],
+  llm: BAMChatLLM.fromPreset("meta-llama/llama-3-1-70b-instruct"),
+});
+
+const reader = createConsoleReader();
+const memory = new UnconstrainedMemory();
+
+for await (const { prompt } of reader) {
+  await memory.add(
+    BaseMessage.of({
+      role: Role.USER,
+      text: prompt,
+    }),
+  );
+
+  const { result } = await flow.run(memory.messages).observe((emitter) => {
+    emitter.on("success", (data) => {
+      reader.write(`-> ${data.step}`, data.response?.update?.finalAnswer ?? "-");
+    });
+  });
+  await memory.addMany(result.newMessages);
+  reader.write(`Agent 🤖`, result.finalAnswer);
+}

examples/flows/nesting.ts

@@ -1,4 +1,4 @@
-import { Flow } from "bee-agent-framework/experimental/flows";
+import { Flow } from "bee-agent-framework/experimental/flows/flow";
 import { z } from "zod";
 
 const schema = z.object({

examples/flows/simple.ts

@@ -1,4 +1,4 @@
-import { Flow } from "bee-agent-framework/experimental/flows";
+import { Flow } from "bee-agent-framework/experimental/flows/flow";
 import { z } from "zod";
 
 const schema = z.object({

src/experimental/flows/agent.ts

@@ -0,0 +1,102 @@
+/**
+ * Copyright 2024 IBM Corp.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+import { BeeAgent } from "@/agents/bee/agent.js";
+import { Flow } from "@/experimental/flows/flow.js";
+import { BaseMessage } from "@/llms/primitives/message.js";
+import { AnyTool } from "@/tools/base.js";
+import { AnyChatLLM } from "@/llms/chat.js";
+import { BeeSystemPrompt } from "@/agents/bee/prompts.js";
+import { ReadOnlyMemory } from "@/memory/base.js";
+import { z } from "zod";
+import { UnconstrainedMemory } from "@/memory/unconstrainedMemory.js";
+
+export class AgentFlow {
+  protected readonly flow;
+
+  static readonly schema = z.object({
+    messages: z.array(z.instanceof(BaseMessage)).min(1),
+
+    finalAnswer: z.string().optional(),
+    newMessages: z.array(z.instanceof(BaseMessage)).default([]),
+  });
+
+  constructor(name = "AgentFlow") {
+    this.flow = new Flow({
+      name,
+      schema: AgentFlow.schema,
+      outputSchema: AgentFlow.schema.required(),
+    });
+  }
+
+  addAgent(agent: { name: string; instructions?: string; tools: AnyTool[]; llm: AnyChatLLM }) {
+    return this.addRawAgent(agent.name, (memory) => {
+      return new BeeAgent({
+        llm: agent.llm,
+        tools: agent.tools,
+        memory,
+        meta: {
+          name: agent.name,
+          description: agent.instructions ?? "",
+        },
+        templates: {
+          system: BeeSystemPrompt.fork((config) => ({
+            ...config,
+            defaults: {
+              ...config.defaults,
+              instructions: agent.instructions || config.defaults.instructions,
+            },
+          })),
+        },
+      });
+    });
+  }
+
+  delAgent(name: string) {
+    return this.flow.delStep(name);
+  }
+
+  addRawAgent(name: string, factory: (memory: ReadOnlyMemory) => BeeAgent) {
+    this.flow.addStep(name, async (state, ctx) => {
+      const memory = new UnconstrainedMemory();
+      await memory.addMany([...state.messages, ...state.newMessages]);
+
+      const agent = factory(memory.asReadOnly());
+      const { result } = await agent.run({ prompt: null }, { signal: ctx.signal });
+
+      return {
+        update: {
+          finalAnswer: result.text,
+          newMessages: state.newMessages.concat(
+            BaseMessage.of({
+              ...result,
+              text: [
+                `Assistant Name: ${agent.meta.name}`,
+                `Assistant Response: ${result.text}`,
+              ].join("\n"),
+            }),
+          ),
+        },
+      };
+    });
+  }
+
+  run(messages: BaseMessage[]) {
+    return this.flow.run({
+      messages,
+    });
+  }
+}

src/experimental/flows.ts → src/experimental/flows/flow.ts

@@ -236,14 +236,14 @@ export class Flow<
               );
             });
             const response = await step.handler(stepInput, handlers);
+            if (response.update) {
+              run.state = { ...run.state, ...response.update };
+            }
             await runContext.emitter.emit("success", {
               run: shallowCopy(run),
               response,
               step: next,
             });
-            if (response.update) {
-              run.state = { ...run.state, ...response.update };
-            }
 
             if (response.next === Flow.START) {
               next = run.steps.at(0)?.name!;
@@ -266,7 +266,7 @@ export class Flow<
           }
         }
 
-        run.result = (this.input.outputSchema ?? this.input.schema)
+        run.result = await (this.input.outputSchema ?? this.input.schema)
           .parseAsync(run.state)
           .catch((err) => {
             throw new FlowError(