一文深入：AI 智能体系统架构设计
如何构建一个有效的主动智能系统？如何在开发过程中发现潜在问题，避免它们在生产环境中引发重大麻烦？
要回答这些问题，你需要将主动智能系统分为三个部分：工具、推理和行动。每一层都有其独特的挑战。某一层的错误可能会波及到其他层，导致意想不到的失败。例如，检索功能可能拉取不相关数据；推理不充分可能导致工作流程不完整或出现循环；行动在生产环境中可能失效。
主动智能系统只有在其最薄弱的环节足够强大时才能正常工作。本文将向你展示如何设计能够规避这些问题的系统。我们的目标是：构建在关键时刻可靠、可预测且具有韧性的智能体。
将分四个部分进行介绍。
设计主动智能系统（一）：智能体架构
架构概览
主动智能系统在三个逻辑层面上运作：工具层、推理层和行动层。每一层在使智能体能够有效检索、处理和行动信息方面都扮演着特定角色。理解这些层之间的交互对于设计功能性和可扩展的系统至关重要。
1. 工具层 ：系统的基础。此层与外部数据源和服务交互，包括 API、向量数据库、操作数据、知识库和用户交互。它负责获取系统所依赖的原始信息。设计良好的工具确保智能体高效检索相关且高质量的数据。
2. 行动层 ：有时也称作编排层。此层负责协调大型语言模型（LLM）与外部世界（工具）之间的交互。在适用情况下，它还处理与用户的交互。它接收来自 LLM 的关于下一步行动的指令，执行该行动，然后将结果提供给推理层的 LLM。
3. 推理层 ：系统智能的核心。此层使用大型语言模型（LLM）处理检索到的信息。它根据上下文、逻辑和预设目标确定智能体接下来需要做什么。推理不充分会导致错误，例如重复查询或行动不一致。
为了准确起见，主动智能系统中的推理并不总是由 LLM 执行。例如，Roomba 吸尘器是一种不使用 LLM 的主动智能系统。本文仅聚焦于基于 LLM 的主动智能系统。
主动智能工作流程
行动层 / 编排层是驱动主动智能系统行为的主要引擎。此层提供了一个主处理循环，大致如下：
智能体应用与 LLM 的首次交互定义了系统试图完成的总体目标。这可以是从生成房地产列表到撰写博客文章，再到处理用户在客户服务应用程序中的开放式请求等各种任务。
与这些指令一起的还有一个 LLM 可调用的函数列表。每个函数都有名称、描述以及它所接受参数的 JSON 模式。以下是 OpenAI 文档中的一个简单函数示例：
{"model": "gpt-4o","messages": [{"role": "user","content": "波士顿今天天气如何？"}],"tools": [{"type": "function","function": {"name": "get_current_weather","description": "获取特定地点的当前天气","parameters": {"type": "object","properties": {"location": {"type": "string","description": "城市和州，例如旧金山，加州"},"unit": {"type": "string","enum": ["摄氏度", "华氏度"]}},"required": ["location"]}}}],"tool_choice": "auto"
}由推理层的 LLM 决定下一步应调用哪个函数以更接近指定目标。
当 LLM 响应时，它将指示应调用哪个函数以及应提供给该函数的参数。
根据用例和所使用的 LLM 在推理层的能力，LLM 可能能够在下一次循环前指定要调用的一组函数（理想情况下是并行调用）。
最好提供一个退出函数，以便推理层可以指示何时已完成处理，行动层应成功退出。
设计原则
表面上看，这些似乎都很简单。然而，随着任务复杂性的增加，函数列表也会增加。随着需要覆盖的范围扩大，推理层出错的可能性也随之增加。一旦你开始添加新的 API、专用子智能体和多个数据源，你会发现要管理的内容远不止于插入一个提示词然后点击 “开始” 那么简单。
设计主动智能系统（二）：模块化
在第二部分，我们将深入探讨模块化的概念。我们将讨论为什么将主动智能系统分解为更小、更专注的子智能体有助于避免单体设计的陷阱。
每个子智能体可以处理其自身的专用领域 —— 退货、订单、产品信息 —— 这种分离使父智能体能够自由地分派任务，而不必在一个庞大的提示词中同时处理所有可能的函数。
模块化
实现复杂主动智能系统稳定性能的最佳方法之一，是将问题领域分解为更小的子智能体。这通过智能体之间的父子关系来实现。顶层的父智能体作为进入主动智能处理的入口。推理层提供的大多数函数都是交接任务，将处理工作分配给其中一个子智能体。
每个子智能体处理问题领域的一个特定部分，实际上，这将微服务架构中常用的有界上下文概念应用于人工智能智能体领域。
模块化的优势
假设你正在为一家在线零售商构建一个客户服务智能体。该智能体将处理诸如退货、订单状态、产品咨询等事务。如果你从单体架构开始，很快就会发现发送到推理层的提示变得又长又复杂，函数列表也会变得过长。随着这两种情况的发生，你的主动智能系统的准确性和性能将受到影响。
采用更模块化的方法可以简化整体架构的许多方面。有一种方法是通过功能领域来分离系统的关注点。结果可能如下图所示：
更可预测的决策流程
通过将系统分解为专门的子智能体，每个智能体承担一组明确定义的职责。随着父智能体编排工作流程，更容易看到正在发生的事情以及原因。请求可以交给最适合处理它的特定智能体，而不是试图在一个庞大的流程中同时处理多项任务。
这使得交互更加透明，数据路径更加清晰，并且任务按照逻辑逐步推进。在单体方法中，责任界限变得模糊，这使得很难确定问题出在哪里。模块化设计通过为每个智能体的领域建立明确的界限消除了这种混乱。
关注点分离
模块化允许每个智能体专注于系统的特定方面。如果你有一个专门处理退货的 ReturnsAgent 和一个仅负责订单的 OrdersAgent，你就避免了无关数据或旁枝末节的问题。这种分离使得领域知识在每个智能体中高度集中，可以大大提高准确性，不再有随机的旁支问题引入不相关信息。
在实践中，这种方法类似于微服务和领域驱动设计中使用的微服务技术，每个组件都有“有界上下文”。通过限制每个子智能体的范围，你减少了指令和依赖项的数量，使每次交互更加精确。因此，子智能体和父智能体都受益于更高的清晰度、更简单的提示和更少的离题响应。
降低复杂性
当一个系统必须处理多项任务或大量数据时，复杂性会迅速膨胀。通过将责任分配给多个子智能体，你缩小了每个智能体的问题领域。这种分散化使得每个智能体能够更有效地推理自己的任务，而不会被不相关的事务淹没。
降低复杂性也意味着更少的程序错误或逻辑错误的机会。由于每个子智能体都是独立的，一个领域的变化不会意外地影响到另一个领域。
提高可维护性和可测试性
维护一个庞大而单体的系统可能是一场噩梦。有了模块化架构，你可以改进或替换一个子智能体而不必担心导致整个系统崩溃。这不仅减少了开发时间，还降低了在与更改无关的领域引入新错误的风险。如果你的 OrdersAgent 需要一个新功能，你可以自信地在那里实现它，确信不会破坏 ReturnsAgent 或系统的其他部分。
设计主动智能系统（三）：智能体之间的交互
在第三部分，我们将深入探讨智能体之间的交互。即使有了良好的模块化，构建统一接口以使子智能体能够以一致的方式交互仍然是一个真正的挑战。我们将探索如何定义清晰、标准化的交接，以便每个智能体都能完成其工作，而不会创建一个令人困惑的调用和回调网络。你将看到为什么拥有一个一致的接口很重要，以及这如何帮助你在出现问题时进行故障排除或升级处理。
统一分发 / 回调机制
当多个智能体需要在智能体系统中协调时，你可能会面临创建大量随意调用和不匹配数据结构的风险。通过标准化每个智能体如何调用（或分发到）其他智能体以及这些智能体如何响应，可以减少混乱、错误和维护开销。一致的接口迫使每个智能体在发出请求或返回结果时使用相同的 “语言”。
采用统一接口的动机源于这样一个现实：在一个复杂的智能体系统中，一个智能体通常不足以处理用户请求的所有方面。用户可能在一次对话中询问跟踪包裹、发起退货以及检查保修状态等事宜。如果你的系统只是简单地将任务委托给大型语言模型（LLM）选择的子智能体，那么你需要一种统一的方式来传递请求数据并获取结构化的响应。通过将这些智能体之间的交接视为具有严格模式的函数调用，可以确保无论是父智能体还是子智能体，都能以可预测的方式交换信息。
如果没有这种一致性，父智能体可能期望一种数据表示方式，而子智能体却返回了完全不同的东西。或者，当你发现子智能体试图调用另一个子智能体时，可能会出现不匹配的情况。每一个小小的不一致都可能引发令人困惑的错误，尤其是在像大型语言模型（LLM）这样动态行为的系统中，这些错误很难排查。数据形状和参数名称的一致性是使大型语言模型（LLM）能够可靠地推理应该调用哪个函数以及必须提供什么数据的关键。
Python 中的示例
父智能体需要知道如何将任务正确地委托给每个子智能体。你可以通过公开负责特定领域的函数（从大型语言模型的角度来看，像是函数调用）来实现这一点。例如：
tools = [{"type": "function","function": {"name": "handoff_to_OrdersAgent","description": "处理与订单相关的问题，如跟踪或管理订单。","parameters": {"type": "object","properties": {"user_id": {"type": "string", "description": "用户的唯一 ID。"},"message": {"type": "string", "description": "用户的查询。"}},"required": ["user_id", "message"]}}},{"type": "function","function": {"name": "handoff_to_ReturnsAgent","description": "处理与退货相关的任务，如授权或跟踪退货。","parameters": {"type": "object","properties": {"user_id": {"type": "string", "description": "用户的唯一 ID。"},"message": {"type": "string", "description": "用户的查询。"}},"required": ["user_id", "message"]}}}
]当大型语言模型（LLM）决定需要处理与订单相关的问题时，它可以调用 “handoff_to_OrdersAgent” 并附上必要的参数。然后父智能体相应地分发请求：
def dispatch_request(self, function_call):fn_name = function_call["name"]arguments = json.loads(function_call["arguments"])if fn_name == "handoff_to_OrdersAgent":result = self.child_agents["OrdersAgent"].process_request(arguments)elif fn_name == "handoff_to_ReturnsAgent":result = self.child_agents["ReturnsAgent"].process_request(arguments)else:result = {"status": "error", "message": f"未知函数 {fn_name}"}return result这种方法使父智能体能够专注于路由，而每个子智能体则专注于其特定领域（订单、退货、产品问题等）。
在子智能体内部，你可以定义与其特定任务相关的函数。例如，OrdersAgent 可能会公开 “lookupOrder” 或 “searchOrders” 函数。子智能体自身的推理循环被限制在该领域内，这有助于避免混淆和巨大的提示上下文。
class OrdersAgent:def__init__(self):self.functions = [{"type": "function","function": {"name": "lookupOrder","parameters": {"type": "object","properties": {"order_id": {"type": "string", "description": "订单 ID。"}},"required": ["order_id"]}}},{"type": "function","function": {"name": "searchOrders","parameters": {"type": "object","properties": {"customer_id": {"type": "string", "description": "客户 ID。"}},"required": ["customer_id"]}}}]defprocess_request(self, payload):self.message_history.append({"role": "user", "content": payload["message"]})for _ inrange(3): # 限制递归调用次数response = self.run_llm_cycle(self.functions)if"function_call"in response:function_call = response["function_call"]result = self.handle_function_call(function_call)if result["status"] == "success":return resultelif result["status"] == "escalate":return {"status": "escalate", "message": result["message"]}else:return {"status": "success", "data": response["content"]}return {"status": "error", "message": "超出推理步骤"}defhandle_function_call(self, function_call):if function_call["name"] == "lookupOrder":return {"status": "success", "data": "找到订单详情..."}elif function_call["name"] == "searchOrders":return {"status": "success", "data": "正在搜索订单..."}else:return {"status": "escalate", "message": f"{function_call['name']} 不支持的函数"}一旦子智能体完成任务，它会以一致的格式将结果反馈给父智能体。这就是回调。父智能体可以：
• 如果一切顺利，将响应传递回给用户。
• 用另一个子智能体重试请求。
• 如果系统无法自动处理，则将问题升级给人工处理。例如：
response = orders_agent.handle_request(payload)
if response["status"] == "success":parent_agent.add_message(role="assistant", content=response["data"])
elif response["status"] == "escalate":parent_agent.add_message(role="system", content="OrdersAgent 无法完成请求。")# 可选择用另一个智能体重试在任何现实系统中，某些查询都会因不可预见的原因失败 —— API 故障、数据缺失，或者子智能体不支持的功能。当这种情况发生时，子智能体会返回一个 “升级” 状态：
def handle_function_call(self, function_call):if function_call["name"] == "不支持的函数":return {"status": "升级", "message": "不支持的函数"}父智能体可以捕获此状态，并决定是否重试、升级到另一个智能体，或者最终向用户返回错误消息。
设计主动智能系统（四）：数据检索和智能体 RAG
在第四部分，我们将研究数据检索和检索增强生成（RAG）。如果没有新鲜、相关的数据，你的语言模型能做的毕竟有限，因此我们将讨论如何连接数据库、API 和向量存储，为你的智能体提供所需的上下文。我们将涵盖从拉取现有系统中的结构化数据到对 PDF 等非结构化内容进行索引的所有内容，以便你的系统在扩展时保持快速和准确。
数据检索和智能体 RAG
有可能创建不需要数据检索的智能体系统。这是因为有些任务可以仅使用语言模型所训练的知识来完成。
例如，你可能能够创建一个有效的智能体来撰写关于历史事件（如罗马帝国或美国内战）的书籍。
对于大多数智能体系统而言，提供数据访问是系统设计的一个关键方面。因此，围绕这个功能领域，我们需要考虑几个设计原则。
检索增强生成（RAG）已 emergence 为连接大型语言模型（LLM）与其所需数据以生成响应的 facto 标准技术。顾名思义，实施 RAG 的一般模式包括三个步骤：
检索 – 从外部知识库中检索额外的上下文。在此上下文中，我使用知识库是一个 loosely 定义的术语，可以包括 API 调用、SQL 查询、向量搜索查询，或任何其他用于查找相关上下文的机制，以提供给大型语言模型（LLM）。
增强 – 使用在检索步骤中获得的相关上下文来增强用户的输入。
生成 – 大型语言模型（LLM）使用其预训练知识以及这种增强的上下文来生成更准确的响应，甚至关于它从未接受过训练的主题和内容。
因此，在问答系统中使用 RAG 将看起来像这样：
智能体系统几乎总是需要某种 RAG 实现来履行其职责。然而，在设计智能体系统时，考虑不同类型的数据如何影响整个系统的需求是很重要的。
结构化数据和 API
拥有成熟 API 计划的公司会发现他们的价值路径比那些没有的公司更容易导航。这是因为为智能体系统提供动力所需的大部分数据与今天用于非智能体系统的数据相同。
一个构建了生成保险报价 API 的保险公司可以比仍然处于客户 / 服务器时代的公司更容易地将该 API 接入智能体。
事实上，智能体系统有可能颠覆保险业务的几乎每一个方面。从报价、承保，到精算科学（保险业务中的赔率制定者），再到理赔管理，智能体 AI 很可能在接下来的 2-3 年内完全接管这些角色。但前提是智能体系统能够访问正确的数据和功能。
AI 优先的 API 管理
许多公司将 API 管理等同于拥有 API 网关。然而，在智能体系统的世界中，OpenAPI（Swagger）成为智能体系统使用现有 API 的关键推动因素。这是因为 OpenAPI 中使用的 JSON 模式可以轻松转换为许多大型语言模型（LLM）提供商使用的函数定义结构。
此外，允许智能体系统查找和发现 API 的服务发现机制为智能体系统自行创造涌现能力提供了有趣的机会。根据你的观点，这要么是一个令人兴奋的前景，要么是一个令人恐惧的前景。
在适当的安全措施下，甚至可以将 API 目录搜索公开为大型语言模型（LLM）可用的函数，以允许它查找可用于解决当前任务的额外功能。
非结构化数据和 RAG
虽然大多数公司都有成熟的实践来管理运营、分析和其他类型的结构化数据，但非结构化数据管理通常远远落后。
文档、电子邮件、PDF、图像、客户服务记录和其他自由形式的文本来源对于智能体系统来说可能非常有价值 —— 但前提是如果你能够在正确的时间以正确的格式检索正确的信息。这就是检索增强生成（RAG）特别强大的地方。
为什么非结构化数据具有挑战性
非结构化数据在范围上往往更大，格式也更加异构（PDF、图像、纯文本、HTML 等）。这种变异性可能会使 naive 的数据检索方法不堪重负。
与存储在关系表中的结构化数据不同，非结构化数据没有强制的模式。你不能简单地对 PDF 运行 SQL 查询，或在电子表格中进行直接的 “查找”。
这导致了新技术的快速采用以解决这些挑战。
向量数据库和语义搜索
现代向量数据库和语义搜索技术为大规模非结构化数据检索铺平了道路。这些系统将文本转换为高维向量嵌入，从而允许基于相似性的查找。用户或智能体的查询也被转换为向量，并检索数据库中最接近的向量（即上下文最相似的文本块）。
大型文档被分成较小的 “块”，每个块分别编制索引。这允许你的检索器只检索相关的块，而不是将整个文档塞入提示中。
许多向量数据库允许你将元数据（例如作者、日期、文档类型）附加到每个块。此元数据可以指导下游逻辑 —— 例如，只检索与用户问题相关的最新产品手册或知识库文章。
在高流量场景或数据质量至关重要的情况下（例如法律或医疗用例），你可以在语义搜索后应用额外的过滤器和排名标准，以确保只返回最高质量或领域批准的内容。
用于非结构化数据处理的 RAG 管道
RAG 管道充当非结构化数据源和向量数据库之间的连接组织。它们通常包括几个步骤，如提取、分块和嵌入，将杂乱或自由形式的文档转换为优化的搜索索引，以便为大型语言模型（LLM）提供相关上下文。
在下图中，你可以看到数据如何从各种非结构化数据源流动。这些可能包括知识库、文件系统中的文档、网页、SaaS 平台中的内容等。
在 RAG 管道内，非结构化数据经历一系列转换。这些包括提取、分块、元数据处理和嵌入（向量）生成，然后写入向量索引。这些管道提供了在大型语言模型（LLM）需要回答问题或完成任务时显示最相关信息的关键功能。
由于非结构化数据通常存在于不同的孤岛中，保持其同步和更新可能是一个挑战。在大型企业中，一些非结构化数据源在不断变化，这意味着 RAG 管道必须以最小的延迟捕获和处理这些变化。
过时的嵌入或陈旧的文本块可能导致你的 AI 系统给出不准确或误导性的答案，特别是如果用户正在寻找有关产品更新、政策变更或市场趋势的最新数据。确保你的管道能够处理近实时更新 —— 无论是通过事件驱动的触发器、计划的爬取还是流数据馈送 —— 大大提高了系统输出的质量和可信度。
优化 RAG 性能
熟悉构建数据管道的数据工程实践的开发人员往往对向量数据的非确定性感到措手不及。
在传统的数据管道中，我们对源系统和目标系统的数据表示有很好的了解。例如，PostgreSQL 中的一组关系表可能需要转换为 BigQuery 中的单个平面结构。我们可以编写测试来告诉我们源系统中的数据是否正确转换为我们目标系统中我们想要的表示。
总结
通过以上四部分介绍，我们将拥有构建可靠、可扩展主动智能系统所需的工具和模式 —— 这种系统不仅理论上听起来不错，而且能够在生产环境的真实压力下屹立不倒。