如何使用多智能体检索增强生成系统MARS构建企业AI应用

去年，我写过关于先进的检索增强生成（RAG）技术如何进入企业，以及多智能体软件工程（MASE）的出现。这两大生成AI在软件开发中的支柱自那时起不断发展，以一种承诺重新定义企业AI应用程序的方式相互交织。

AI代理的本质

首先，让我们探讨什么构成一个AI代理？一些开发者将代理描述为能够推理、行动和记忆的自主实体。举个例子，ReACT代理或能够推理和行动的代理。那么，是什么使ReACT代理与众不同呢？我认为它们由三个基本元素组成：

1. 智能：代理对大型语言模型（LLMs）的访问。2. 知识：其在特定领域或主题上的结构化和非结构化数据的存储库。3. 感知器和效应器：使代理能够感知其环境并执行任务的工具和API。这里是一个典型代理的简单图示，它具有用于智能的Anthropic LLM，作为工具的exa-search和连接到SingleStore的检索器（用于访问）以及通过检索器工具访问知识。

RAG在企业中的演变

在生成AI的世界里，去年感觉就像是一个时代，当时开发者们开始探索将自定义数据输入LLMs以增强上下文理解的想法。就我个人而言，我在更广为人知的术语——检索增强生成（RAG）流行之前，称之为“搜索和检索”，而这一术语不仅被接受，还在软件开发界引起了轰动。

然而，这种初步的方法，通常被称为“幼稚RAG”，对于大型企业复杂的分层需求来说是不够的。这些组织通常拥有大量的结构化和非结构化数据，而构建企业级AI应用程序的竞争被与RAG相关的三个巨大的挑战所阻碍——准确性、相关性和延迟。

事实证明，这就是为什么大多数企业去年没有看到生产就绪应用程序激增的首要原因。

但事情发展迅速，现在我们终于看到了几个原则的结合，这为企业级软件开发进入一个全新时代提供了一个新的视角。

首先，让我们看看在企业环境中使RAG应用程序准备好投入生产所需的关键要求：

1. 准确性： 在企业中，对幻觉的容忍度为零。这在金融和医疗等关键行业尤其如此，因为错误的余地是不存在的。2. 相关性： 在大规模下，资源是昂贵的，因此信息检索必须精确，仅查询必要的数据以确保效率。这也直接影响到下一点。3. 延迟： RAG管道必须以思维的迅速运作，完成任务的时间必须少于一秒，以保持无缝的用户体验，假设主模型的推理在网络上仍需大约一秒的响应时间。

精准与速度的炼金术

我们如何实现这种精准与速度的炼金术？让我们看看这些类别及一些最佳实践。

1. 准确性： 这里出现了一些新事物——微调的嵌入模型和匹配LLMs、有效的评估、结构化数据的函数调用，以及与人类反馈强化学习（RLHF）结合的实时可观察性和反馈循环。2. 相关性： 大多数公司都有数据湖、数据仓库或数据湖屋。所有这些数据对于相关性都是有价值的，将其与实时数据混合现在几乎是AI应用程序的基本要求，因为如果响应仅包含某一时间点的数据，响应不仅会变得无关紧要，而且也会变得危险地不准确。3. 延迟： 架构必须优化以确保亚秒级的响应时间。如果时间更长，整体体验至少会变得不理想，并失去产品的采用。

现在，为了构建这个堆栈，每一层都有几种选择，但我列出了在与一些公司交谈时发现的最常见的选择。

嵌入创建 — 在这里，Nvidia推理微服务（NIMs）在众多竞争解决方案中脱颖而出。使用NIMs，您可以调用Nvidia API，或者租用/购买自己的H100并部署模型。这可能不是每家公司都合适的选择，但如果您在寻找速度，Nvidia显然是这一领域的领导者。

Nemo真正有用之处在于，您可以将其作为服务器部署，提供一个可以从代理调用的端点。

您可以在这里查看一些Nemo的示例和文档。

语义缓存 — 在这里，您可以使用许多数据存储和框架，但根据我的经验，我始终建议使用一个平台来满足所有数据需求，该平台可以以速度和规模管理所有数据用例。对我来说，这个选择是SingleStore，尽管为了完全透明，我的观点可能会有偏见，因为我在SingleStore工作。

检索 — 继续上述主题，检索包括结构化和非结构化数据，这意味着在JSON、SQL、键值等中进行语义搜索和关键字搜索。在这里，您可以通过SingleStore 在单个查询中运行所有这些，响应时间在数PB数据中都能保持亚秒级。

安全与安全性 — 在这里，Nvidia的Nemo保护措施是显而易见的。自去年以来，这个开源库取得了长足的进步，现在可以进行输入和输出验证、掩盖个人身份信息（PII）等，具有高度的灵活性。

评估 — 我去年谈到过这个，仍然推荐这个开源库——RAGAs 。这与我们选择的框架很好地结合在一起。

单体与模块化

正如您所想象的，去年这些管道中的每一层或步骤都开始独立发展，开发人员开始将这些系统构建为顺序管道，但这种方法存在许多局限性：僵化、延迟，以及准确性和相关性的降低。这种单体架构让人想起旧的 C 或 C++ 应用程序，尽管性能良好，但可以说笨重且缺乏灵活性。

进入多智能体 RAG 系统 (MARS) 的时代。

现在想象一下，如果我们利用选择的工具将上述不同步骤和任务构建为专门的智能体，这些智能体在各自领域中是专家，并配备特殊工具。然后，它们可以被编排为并行运行，以获得更好的性能和可扩展性，类似于微服务。

这种替代方案提供了几个其他优势：

可维护性：专门的智能体可以处理不同的任务，允许独立更新和工具更换。2. 并行性：智能体可以同时操作，利用不同的 GPU 或计算资源。3. 优化资源：工作负载可以根据任务分配到 GPU 和/或 CPU 基于商品计算，提升效率。4. 效率和准确性：将特定任务隔离在不同的智能体中，有助于调试和通过 RLHF 进行迭代评估。

多智能体RAG系统的蓝图

现在我们已经确定了不同的任务和库的选择，在将这些内容整合在一起之前，我们需要考虑另外两件事。首先，让我们选择一个框架来创建工作流程。

在这里，可以使用像Autogen或CrewAI这样的代理框架，但我选择Langchain的LangGraph，原因有两个——易用性和与NIMs及Nemo Guardrails等工具的集成。此外，我喜欢我们可以根据企业的需求构建“图”，这些需求可能因公司而异。

其次，让我们看看其他代理的一些附加增强功能，以使我们的架构更具性能和效率。嵌入创建：我们可以考虑添加一个预处理代理，用于在创建嵌入之前进行数据清理。语义缓存：SingleStore是理想选择，但我们也可以添加一个缓存管理代理，用于失效和更新。Langchain 在其API中也提供了这一功能。检索：一旦我们掌握了基础知识，我们还可以添加一个查询理解代理，在发送检索之前增加更多上下文。安全与安全性：Nemo Guardrail在这里至关重要，但一个单独的PII检测和编辑代理也是一个非常有价值的补充，例如Protecto。评估：目前，我们暂时坚持使用RAGAs，尽管我们可以使用其他代理来使用其他框架并比较结果。

鉴于架构的模块化，我们现在还可以构建额外的代理，根据最终需求添加、修改和删除它们；

查询规划代理： 策划回答查询的最佳方法。上下文增强代理： 用额外的上下文增强检索到的信息。响应生成代理： 根据增强的信息生成最终响应。反馈集成代理： 处理用户反馈以改善系统。日志记录和监控代理： 管理全面的日志记录和实时监控。数据版本控制代理： 监督数据、嵌入和模型工件版本以确保可重复性。

编排框架

为了协调这个多代理RAG系统，如上所述，LangGraph将是我的选择。它针对多代理系统的设计、灵活的图结构、强大的状态管理、与LangChain的集成、可视化工具以及可扩展性，使其成为企业级应用的理想选择。

代码实现

第一步 — 构建代理

首先，让我们看一下使用 LangGraph 的简单代理实现。

enrichment_agent = initialize_agent(

[top_selling_products_tool],

chat_model,

agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,

verbose=True

)

第2步 — 创建工具

注意我们是如何将工具传递给代理的。这实际上是代理在需要运行查询时调用的另一种方法，在这里定义。

top_selling_products_tool = Tool(

name=”畅销产品”,

func=get_top_selling_products,

description=”获取畅销产品。您可以指定限制（默认是5）。”

)

这个工具调用一个函数 get_top_selling_products，该函数包含检索结构化数据的代码。

def get_top_selling_products(limit=5):

query = f”””

SELECT product_name, SUM(quantity_sold) as total_sold

FROM sales

GROUP BY product_name

ORDER BY total_sold DESC

LIMIT {limit}

return db.run(query)

第 3 步 — 创建一个包含图、节点和边的工作流

然后我们创建一个图，其中包含一个图（工作流）以及节点和边来将这一切组合在一起。接着，我们将节点和边添加到工作流中并运行它。

以下是通过给定一个有状态图来创建工作流的示例。AgentState 负责维护整体工作流中的状态。

workflow = StateGraph(AgentState)

最后，我们将所有这些组合在一起。

# 为代理添加节点

graph.add_node(“sql_agent”, sql_agent)

graph.add_node(“enrichment_agent”, enrichment_agent)

# 定义工作流

@graph.run

def workflow(query: str):

sql_result = graph.nodes.sql_agent(query)

enriched_result = graph.nodes.enrichment_agent(sql_result)

return enriched_result

有关完整的代理 RAG 示例，请参阅这篇文章。

结论

显然，我们在这里只是触及了表面，正如所述，我们可以选择不同的技术和选项来构建自定义的多智能体RAG系统（MAR）。希望这为您奠定了基础，如果您希望了解构建企业级AI应用程序所需的内容，以及走向一个不仅模块化而且可扩展和可控的整体框架和步骤，就像我们所熟知的微服务架构一样。

✌️

Barry's Home