← 返回博客
跳至主要内容

使用 Inngest 构建 Agentic 工作流

·8 分钟阅读
Daniel Phiri
Charly Poly

Building Agentic Workflows with Inngest

AI 工作流是所有 AI 应用的基础，无论您是构建 RAG 管道、Agentic 工作流还是 AI 代理。自 2023 年以来，大多数 AI 应用都依赖于静态 AI 工作流，例如 RAG 管道。2024 年，Agentic 工作流的概念——由 LLM 规划的具有动态步骤的工作流——应运而生，以应对静态工作流的局限性，例如幻觉、缺乏相关性或灵活性。

已经宣布 2025 年将是 AI 代理的竞赛之年，拥有正确的工具来构建灵活且强大的 AI 工作流至关重要。因此，我们很高兴与 Inngest 合作，Inngest 是一种工作流引擎，已被数万开发者使用，并处理超过 1 亿次每日运行。

本文将向您展示如何使用 Inngest 和 Weaviate 构建一个 Agentic RAG 管道，该管道根据口味、过敏症和参与者人数生成完美的晚餐菜单。

完成的演示可在 GitHub 上获取，以便您可以本地按照步骤操作。

步骤 1：设置我们的晚餐生成器 Agentic 工作流架构​

我们的晚餐生成器工作流依赖于 Huggingface 上提供的三个数据集

  • Thefoodprocessor/cuisine_type：包含食谱和菜系类型的配对数据集（例如：“意大利菜”）
  • Thefoodprocessor/ingredients_alternatives：包含食谱和匹配的过敏症食材替代品数据集
  • Thefoodprocessor/wine_type：包含食谱和葡萄酒搭配的列表

一旦这些数据集加载到我们的 Weaviate 集群中，我们的工作流就会通过一个 Inngest 事件 触发，该事件具有以下属性

{
  "data": {
    "participantsCount": 4,
    "preferences": ["no nuts", "vegan"],
    "cuisine_type": "Italian"
  }
}

由上述事件触发，我们的 Inngest Agentic 工作流将经历以下步骤

An Inngest Workflow

  1. 首先，我们的工作流将从 Thefoodprocessortype Weaviate 集合中检索一些食谱。
  2. 然后，使用 OpenAI gpt-4 模型，它将尝试查看食物偏好中是否包含某些过敏症。
    2.1 如果发现某些过敏症，将执行额外的步骤，以从 Thefoodprocessoringredientsalternatives 集合中检索一些食材替代品。
  3. 然后，执行 OpenAI 调用以生成菜单（如果存在任何食材替代品）。
  4. 对 Thefoodprocessorwinetype 集合执行另一个 Weaviate 集合查询，以查找与生成的菜单匹配的葡萄酒。
  5. 进行额外的 OpenAI 调用以选择葡萄酒并更新晚餐菜单。
  6. 最后，从最终的晚餐菜单生成购物清单。

我们的晚餐生成器工作流的 Agentic 特性在于它能够在食物偏好中提到过敏症时获取其他信息的能力。

下一节将展示 Inngest 和 Weaviate 如何使构建此类 Agentic 和数据密集型工作流变得容易。

步骤 2：使用 Weaviate 矢量化器进行快速简便的嵌入​

Huggingface 数据集以 .parquet 文件形式提供，主要设计用于 Python 脚本。
因此，我们的仓库 包含翻译后的 .json 文件，以方便在 Node.js 中处理。
npm run load 脚本利用 Weaviate 批量插入 和 矢量化器 API，只需几行代码即可创建新的集合

async function processJsonFile(
  filePath: string,
  client: any,
  batchSize: number = 100
) {
  const jsonData = JSON.parse(await fs.readFile(filePath, "utf-8"));
  let count = 0;
  let records: any[] = [];

  // Create collection name from filename (without extension)
  const collectionName = path
    .basename(filePath, ".json")
    .replace(/[^a-zA-Z0-9]/g, "") // Remove special characters
    .replace(/^[0-9]/, "C$&"); // Prefix with 'C' if starts with number

  const properties = Object.keys(jsonData[0]);

  // Create collection for this file
  const collection = await client.collections.create({
    name: collectionName,
    properties: properties.map((property) => ({
      name: property,
      dataType: "text" as const,
    })),
    // Use Weaviate vectorize to create embeddings
    vectorizers: [
      weaviate.configure.vectorizer.text2VecOpenAI({
        name: "recipe_vector",
        sourceProperties: properties as any,
      }),
    ],
  });

  for (const record of jsonData) {
    const { id: _, ...cleanRecord } = record;

    records.push({
      class: collectionName,
      properties: cleanRecord,
    });

    count++;

    // we push new records in batches using Weaviate's insertMany()
    if (count % batchSize === 0) {
      console.log(
        `Processing batch from ${path.basename(filePath)}: ${
          records.length
        } records`
      );
      await collection.data.insertMany(records);
      records = [];
    }
  }

  // Process any remaining records
  if (records.length > 0) {
    console.log(
      `Processing final batch from ${path.basename(filePath)}: ${
        records.length
      } records`
    );
    await collection.data.insertMany(records);
  }

  return { count, collectionName };
}

在嵌入数千条记录时，Weaviate 的 vectorizer 和 insertMany() API 使我们免于处理 OpenAI 嵌入 API 的速率限制。

准备好我们的 Weaviate 集合后，让我们看看 Agentic 工作流的实现。

步骤 3：结合 Weaviate 和 Inngest DX 构建我们的 Agentic 工作流​

Inngest 使我们能够使用简单的 TypeScript API 创建 AI 工作流，而 Weaviate 使我们能够快速从矢量化集合中检索数据

import { openai } from "inngest";
import { inngest } from "./inngest-client";
import { weaviate } from "./weaviate-client";

export const generateMeal = inngest.createFunction(
  { id: "generate-meal", concurrency: 10 },
  { event: "meal.generate" },
  async ({ event, step }) => {
    // retrieve the input parameters from our trigger event
    const { participantsCount, preferences, cuisine_type } = event.data;

    // Query relevant recipes based on cusine type preference
    let relevantRecipes = await step.run("query-recipes", async () => {
      const collection = weaviate.collections.get("Thefoodprocessorcuisinetype");
      const query = cuisine_type;

      const result = await collection.query.nearText(query, {
        limit: 10,
      });

      return result.objects.map((r) => r.properties.recipe).join(`\n`);
    });

    const allergiesAnalysis = await step.ai.infer("Allergies analysis", {
      model: openai({ model: "gpt-4" }),
      body: {
        messages: [
          {
            role: "system",
            content:
              'Given the following list meal requirements, return a list of allergies, if not allergies, return "not allergies"',
          },
          {
            role: "user",
            content: preferences.join(", "),
          },
        ],
        temperature: 0.7,
      },
    });

    // ...
});

请参阅 仓库的 README 以获取有关设置 Inngest 的说明。
使用 Inngest 构建的 AI 工作流由使用 step.run() 或 step.ai.infer() 定义的步骤组成。有了这两个 API，我们 AI 工作流的每个步骤都将受益于自动重试、并发性和节流控制。
我们包装在 step.run() 中的 Weaviate 查询将受益于 Inngest 函数定义中配置的 concurrency，从而防止我们的集群过载。

放大 step.ai.infer()：更快开发 AI 工作流的方法

使用 step.ai.infer() 执行 LLM 调用带来了两个显著的改进

  1. LLM 请求在 Inngest DevServer（您的 Inngest 本地版本）中突出显示，使您可以检查令牌使用情况、LLM 的输入和输出，并使用新的提示重新运行给定的步骤。

    这是我们在 Inngest Dev Server 中显示的 ”过敏症分析” LLM 步骤

  2. LLM 请求被卸载到 Inngest 的基础设施上，从而更容易在Vercel 等无服务器平台上部署工作流。

Agentic 工作流步骤作为简单的条件语句

让我们回到我们的工作流，并根据我们的 ”过敏症分析” LLM 步骤的结果，向 Thefoodprocessoringredientsalternatives Weaviate 集合添加我们的动态查询。
使用 Inngest 实现这种 Agentic 模式只需要简单的条件语句

import { openai } from "inngest";
import { inngest } from "./inngest-client";
import { weaviate } from "./weaviate-client";

export const generateMeal = inngest.createFunction(
  { id: "generate-meal", concurrency: 10 },
  { event: "meal.generate" },
  async ({ event, step }) => {
    // retrieve the input parameters from our trigger event
    const { participantsCount, preferences, cuisine_type } = event.data;

    // Query relevant recipes based on cusine type preference
    let relevantRecipes = await step.run("query-recipes", async () => {
      // ...
    });

    // Look for allergies in the food preferences
    const allergiesAnalysis = await step.ai.infer("Allergies analysis", {
      // ...
    });

   let ingredientsAlternatives: any[] = [];

    // run a conditional step based on `allergiesAnalysis` result
    if (allergiesAnalysis.choices[0].message.content !== "not allergies") {
      // Query relevant recipes based on cusine type preference
      ingredientsAlternatives = await step.run(
        "query-ingredients-alternatives",
        async () => {
          const collection = client.collections.get(
            "Thefoodprocessoringredientsalternatives"
          );
          const query = allergiesAnalysis.choices[0].message.content!;

          const result = await collection.query.nearText(query, {
            limit: 10,
          });

          return result.objects;
        }
      );
    }


    // ...
});

从包含 ”不含坚果” 的食物偏好开始我们的晚餐生成器工作流，会根据 ”过敏症分析” LLM 步骤的检测触发 "query-ingredients-alternatives" Weaviate 查询步骤。

Architectural Diagram

我们的晚餐生成器工作流的后续步骤以类似的方式利用 Weaviate 的 collection.query.nearText() 和 Inngest 的 step.ai.infer() API 生成菜单、获取相关的葡萄酒搭配并生成购物清单。
现在让我们看看我们的晚餐生成器实际运行情况。

试用一下：生成一个独特的晚餐菜单​

上述代码和设置说明可在我们的 recipes 仓库中获取，并可以使用 Inngest Dev Server 在本地运行。
您可以通过尝试不同的食物偏好或菜系类型来生成多个菜单变体

Weaviate Inngest Demo

本文介绍了 Weaviate 丰富的嵌入和查询 API 如何使索引和查询大型矢量化数据集变得容易。它还演示了 Inngest 的可组合 API 如何能够使用简单的 API 编写健壮且动态的工作流。

此演示只是结合使用 Weaviate 和 Inngest 可以实现的功能的冰山一角。如果您想构建具有更多自主性的 Agentic 工作流，例如 AI 代理，您应该查看 Weaviate 生成搜索 和 Inngest 的新 AgentKit 库。或者，您可以等待我们的下一篇关于 AI 代理的文章。

准备开始构建了吗？​

请查看 快速入门教程，或使用 Weaviate Cloud (WCD) 的免费试用版构建令人惊叹的应用程序。

GitHub

论坛

Slack

X (Twitter)

不想错过另一篇博文？

注册我们的双周时事通讯以保持更新！


提交后，我同意 服务条款和 隐私政策.