生产级企业代理式人工智能系统的参考架构

执行摘要

从无状态的大语言模型应用转向生产级智能代理AI，标志着企业构建软件方式的重大变革。早期的企业AI应用主要依赖基础的检索增强生成技术——将文档转换为嵌入向量并放入提示词的上下文窗口中。这种方法适用于简单的问答场景，但在自主决策、多步骤状态规划、可靠的工具调用以及在工作流中自我纠错方面存在不足。

企业代理型人工智能通过重新定位基础模型来填补这一空白：该模型不再作为与用户的直接接口，而是成为嵌入在确定性软件框架中的推理组件。这类系统能够感知周围环境，将大型目标拆解为更小的步骤，调用企业内部服务，并在多步骤事务中保持执行状态。不过，要将可运行的原型转化为可在生产环境中使用的系统，就必须解决控制循环不稳定、长时间会话中上下文丢失、权限提升风险以及代币消耗失控等问题。

本文为负责企业智能体系统的架构师及人工智能工程负责人提供了端到端的参考架构。它涵盖了智能体所需的基础层，比较了多种多智能体拓扑结构，深入探讨了企业级检索的集成方案——尤其重点介绍了Amazon Kendra——同时还提供了适用于生产环境的Python示例代码。

第1节：标准企业智能体架构

核心层：感知、推理、规划、记忆与工具执行

一个可用于生产环境的智能体系统会将概率型基础模型与围绕其运行的确定性执行运行时分开。该架构通常由五层核心组成，所有这些层都在一个托管的运行时容器中运行：

• 感知层：将原始的企业信号——遥测数据、用户消息、webhook负载、API响应——转换为系统其他部分能够使用的结构化、类型明确的输入。这包括输入清洗、速率限制以及提前进行的模式检查，所有这些操作都在调用LLM之前完成。
• 推理引擎：作为认知核心，由底层LLM提供支持（例如Anthropic Claude 3.5 Sonnet或AWS Bedrock模型）。该引擎并不直接执行操作，而是解析当前上下文，权衡各种可能的处理路径，并输出结构化的意图表述。

规划模块：负责目标分解，将用户给出的高层级指令转化为结构化、可执行的有向无环图（DAG）形式的步骤。当工具调用失败或返回意外结果/不完整数据时，该模块能够动态重新规划。

状态与内存管理：负责管理三层内存——临时工作内存（当前线程的实时暂存区）、执行过程的短期日志，以及可跨越多个用户会话的长期语义记忆或情景记忆。

工具执行与管控层：将智能体的结构化意图转化为实际操作——包括对外发起的API调用、SQL查询或远程函数调用。该层还负责实现静态沙箱机制、参数验证、基于角色的访问控制（RBAC）以及断路保护逻辑。

编排模式：单智能体循环与多智能体拓扑

选择合适的编排模式在很大程度上取决于目标领域的复杂程度：

• 单智能体ReAct循环：通过思考、行动和观察三个阶段循环的推理机制。这种模式适用于涉及5到8种以内不同工具的简单线性工作流程。超出此范围后，单个智能体往往会面临上下文窗口过大的问题、指令模糊以及难以决定该使用哪种工具的困境。

监督者/领导者多智能体拓扑结构：这是一种分层架构，其中顶层的“监督者”智能体接收原始请求，将其拆分为子任务，然后分配给专门的智能体（如SQL智能体、RAG搜索智能体和动作执行智能体）。监督者负责管理全局状态，而每个专门智能体则使用为特定任务设计的工具集来工作。

点对点/网络拓扑结构：专门的智能体通过共享的事件总线直接相互通信，没有中央协调器。这种方式具有很高的灵活性，但容易引入非确定性、消息循环无法终止的风险，以及在企业环境中难以解决的调试难题。

对于企业级生产环境，监督多智能体拓扑结构是推荐的默认方案，因为它具有清晰的状态边界、更易于审计的特点，以及更为可预测的上下文成本。

第2节：企业级内存管理与上下文维护

内存层次结构：工作记忆、短期轨迹与长期记忆

可以将智能体内存管理视为一种资源预算问题。若任由内存无限制增长，会提高推理成本、增加延迟，并且随着上下文质量下降导致模型丢失处理线索。一个设计良好的企业级智能体需要采用分层内存结构：

• 工作记忆（临时存储区）：即实时上下文窗口——当前有效的系统指令、正在执行任务的状态，以及最新的工具输出结果。

短期轨迹记忆：一种临时存储机制，如 Redis 或 DynamoDB，用于保存当前会话的原始事件历史——包括完整的工具调用负载及其原始响应。

长期记忆：持久存储方式——关系型数据库、向量数据库或图数据库——用于保存经过汇总的过往交互记录、用户偏好配置，以及多会话中积累的领域特定经验。

上下文压缩、剪枝与状态隔离策略

为防止上下文数据恶化，运行时需主动执行相关规范：

• 观测数据截断：原始工具响应——例如包含500条记录的 JSON 数据——绝不能直接进入工作内存。运行时必须在数据传送到推理引擎之前对其进行清洗、截断或汇总处理。
• 滚动窗口压缩：当临时存储空间的使用量超过预设阈值（例如达到模型总上下文容量的20%）时，系统会通过压缩步骤将对话中早前的内容转化为简洁的语义摘要，并丢弃原始的未处理对话内容。
• 子任务状态隔离：当监督者将任务分配给子代理时，会为该子代理创建一个仅包含其所需特定目标与参数的新鲜上下文——监督者自身的推理历史不会泄露到其中。

第3节：深入探讨：利用Amazon Kendra实现企业级检索定位

Amazon Kendra作为生产环境中的检索层

在通用向量数据库上构建检索流程通常意味着需要从零开始设计文档导入、分块逻辑、嵌入生成以及混合搜索融合层。而Amazon Kendra则提供了一种全托管的企业级搜索引擎，可直接处理这些功能。它具备内置的多阶段自然语言理解能力、能够识别表格、页眉和页脚的结构解析功能，以及结合词汇信息和语义信号的混合排序模型。

在企业智能体架构中，Kendra通常充当核心的知识关联子系统——该组件负责让智能体从各种内部数据源中获取经过验证的事实信息，从而避免因简单的向量存储分割方式所带来的数据碎片化风险。

高级相关性调优与动态元数据过滤

为保障安全自主决策，企业级检索系统需要精确的权限管理及可调节的相关性控制：

• 原生访问控制列表（ACL）：Kendra会自动获取并映射源自SharePoint、Confluence或S3等系统的文档级ACL。当智能体发起查询时，它会将经过认证的用户身份以UserContext令牌的形式传递出去，Kendra随后会在索引层面进行安全过滤，确保任何该用户无权查看的片段都不会传递给智能体。

相关性提升：Kendra 支持基于文档属性在运行时及索引层面进行提升。例如，团队可以通过 _last_updated_at 按最新更新时间、按业务类别，或通过对 Department、ProjectCode 等自定义字段的精确匹配来实现提升。

检索 API 与查询 API：在将 Kendra 集成到智能体的工具集中时，建议优先使用 Retrieve API 而非通用的 Query API。Retrieve 完全跳过面向用户的搜索元数据，直接返回高度浓缩的段落级摘录，这些摘录专为直接输入大语言模型的上下文窗口而设计。

第 4 节：确定性约束、工具验证与断路器机制

执行前（前馈）与执行后（反馈）控制

自主代理需要坚实的软件级边界，以防止权限提升、格式错误的工具调用或无限执行循环：

• 前馈验证（执行前）：在工具调用传达到下游企业系统之前，运行时系统会依据严格的Pydantic模式检查其参数——确认所需字段存在、数值或枚举值处于允许范围内，且调用者的授权令牌确实具备执行该操作的权限。

反馈传感器（执行后）：当工具出现错误时，该框架会以确定性的方式拦截故障，而非让其导致循环崩溃或将原始堆栈跟踪信息传递给模型。相反，它会将故障转化为结构清晰、格式规范的消息——例如：错误：未找到数据库表‘users_v2’。可用表：['users', 'orders']——从而为智能体提供可操作的依据，帮助其调整后续操作。

沙箱环境、步骤预算与自动断路器

• 隔离式沙箱：所有动态生成的代码——比如数据分析智能体产生的Python代码——都应在如Docker容器或gVisor微虚拟机这样的可丢弃、隔离的沙箱环境中运行，并且其对外网络访问必须受到严格限制。
• 步骤与成本预算：每项智能体任务都应设定工具调用次数的上限（例如不超过10次），同时限定最大token消耗量。一旦超出任一限制，系统就必须停止执行并将任务转交给人工操作员。
• 工具断路器：如果后端API或数据库在连续重试后仍持续出现故障，系统应触发断路器机制，将该工具在智能体的工具目录中标记为UNAVAILABLE，从而促使规划层选择其他路径，而非不断尝试使用已失效的依赖项。

第5节：生产环境实施蓝图

下面的示例是一个完整且可运行的 Python 应用程序，展示了生产级Supervisor 多智能体架构。它结合了基于 Pydantic 的工具验证、步骤预算控制、确定性错误处理，以及针对 Amazon Kendra 检索功能的封装工具。

import os
import json
import time
from typing import List, Dict, Any, Optional
from pydantic import BaseModel, Field, ValidationError

# =====================================================================
# 1. TOOL SCHEMAS & ENTERPRISE INTEGRATION CONTRACTS
# =====================================================================
class KendraRetrieveInput(BaseModel):
    """Input contract for the Amazon Kendra retrieval tool."""
    query_text: str = Field(..., description="The natural language query string to search across corporate documentation.")
    department_filter: Optional[str] = Field(None, description="Optional department metadata filter (e.g., 'Engineering', 'HR').")
    user_id: str = Field(..., description="Authenticated user ID used for native Kendra ACL security trimming.")
class DatabaseQueryInput(BaseModel):
    """Input contract for enterprise relational database lookups."""
    query_type: str = Field(..., description="Must be 'SELECT'. Data mutation operations are strictly prohibited.")
    table_name: str = Field(..., description="Target database table name.")
    limit: int = Field(default=5, ge=1, le=20, description="Number of records to return.")
# =====================================================================
# 2. MOCK ENTERPRISE SERVICES & KENDRA TOOL ENGINE
# =====================================================================
class MockAmazonKendraClient:
    """Simulates Amazon Kendra's high-density Retrieve API with ACL security trimming."""
    def __init__(self):
        self._mock_index = [
            {
                "doc_id": "KENDRA-DOC-001",
                "content": "Production deployment requires dual sign-off from Engineering and Security leads.",
                "department": "Engineering",
                "acl_users": ["user_eng_101", "admin_007"]
            },
            {
                "doc_id": "KENDRA-DOC-002",
                "content": "Standard employee travel stipend is capped at $150/day for domestic lodging.",
                "department": "HR",
                "acl_users": ["user_hr_201", "user_eng_101", "admin_007"]
            }
        ]
    def retrieve(self, query_text: str, user_id: str, department_filter: Optional[str] = None) -> List[Dict[str, Any]]:
        results = []
        for doc in self._mock_index:
            # Enforce Document-Level ACL Security Trimming
            if user_id not in doc["acl_users"]:
                continue
            # Apply Optional Metadata Department Filtering
            if department_filter and doc["department"].lower() != department_filter.lower():
                continue

            results.append({
                "DocumentId": doc["doc_id"],
                "ContentSnippet": doc["content"],
                "Department": doc["department"]
            })
        return results
class MockEnterpriseDatabase:
    """Simulates a secure internal enterprise database."""
    def __init__(self):
        self._tables = {
            "deployments": [
                {"id": 1, "service": "auth-service", "status": "COMPLETED", "env": "prod"},
                {"id": 2, "service": "payment-api", "status": "PENDING_APPROVAL", "env": "prod"}
            ]
        }
    def execute_select(self, query_type: str, table_name: str, limit: int) -> List[Dict[str, Any]]:
        if query_type.upper() != "SELECT":
            raise ValueError(f"Security Alert: Unauthorized operation '{query_type}'. Only 'SELECT' is permitted.")
        if table_name not in self._tables:
            raise KeyError(f"Database Error: Table '{table_name}' does not exist. Available tables: {list(self._tables.keys())}")
        return self._tables[table_name][:limit]
# =====================================================================
# 3. PRODUCTION AGENT HARNESS & RUNTIME ENGINE
# =====================================================================
class ProductionAgentRuntime:
    """
    Deterministic software harness surrounding probabilistic reasoning models.
    Enforces budgets, schema validation, sandboxing, and error recovery loops.
    """
    def __init__(self, user_id: str, max_step_budget: int = 4):
        self.user_id = user_id
        self.max_step_budget = max_step_budget
        self.kendra_service = MockAmazonKendraClient()
        self.db_service = MockEnterpriseDatabase()
    def execute_task(self, task_goal: str) -> Dict[str, Any]:
        print(f"=== [Harness Started] Initiating Task: '{task_goal}' for User: '{self.user_id}' ===")
        step_count = 0
        scratchpad_history: List[str] = []
        # Simulated dynamic model trajectory outputs (demonstrating multi-step execution & self-correction)
        simulated_llm_turns = [
            # Turn 1: Attempt invalid database deletion (Caught by Feedforward Schema/Guardrail)
            {
                "thought": "I will clean up old deployment logs before checking security compliance.",
                "action": "execute_db_query",
                "args": {"query_type": "DELETE", "table_name": "deployments", "limit": 5}
            },
            # Turn 2: Corrected database lookup
            {
                "thought": "I will check active deployment statuses in the enterprise database.",
                "action": "execute_db_query",
                "args": {"query_type": "SELECT", "table_name": "deployments", "limit": 2}
            },
            # Turn 3: Ground task using Amazon Kendra Retrieve API
            {
                "thought": "Now I need to query enterprise policies regarding production deployment sign-off.",
                "action": "kendra_retrieve",
                "args": {"query_text": "production deployment sign-off rules", "department_filter": "Engineering"}
            }
        ]
        while step_count < self.max_step_budget:
            step_count += 1
            print(f"\n--- [Step {step_count}/{self.max_step_budget}] ---")
            # Fetch current simulated model decision turn
            turn_data = simulated_llm_turns[min(step_count - 1, len(simulated_llm_turns) - 1)]
            print(f"Agent Thought: {turn_data['thought']}")

            action = turn_data.get("action")
            args = turn_data.get("args", {})
            # --- TOOL EXECUTION BRANCH: DATABASE ---
            if action == "execute_db_query":
                try:
                    # 1. Pre-execution Feedforward Schema Validation
                    validated_args = DatabaseQueryInput(**args)

                    # 2. Tool Execution
                    db_results = self.db_service.execute_select(
                        query_type=validated_args.query_type,
                        table_name=validated_args.table_name,
                        limit=validated_args.limit
                    )
                    observation = f"Database Query Success: {json.dumps(db_results)}"
                    print(f"[Observation]: {observation}")
                    scratchpad_history.append(observation)
                except ValidationError as ve:
                    error_msg = f"Schema Validation Blocked Action: {ve.errors()[0]['msg']}"
                    print(f"[Harness Feedforward Intercept]: {error_msg}")
                    scratchpad_history.append(error_msg)
                except (ValueError, KeyError) as exec_err:
                    error_msg = f"Tool Execution Failure: {str(exec_err)}"
                    print(f"[Harness Feedback Sensor Catch]: {error_msg}")
                    scratchpad_history.append(error_msg)
            # --- TOOL EXECUTION BRANCH: AMAZON KENDRA RETRIEVE ---
            elif action == "kendra_retrieve":
                try:
                    # Inject authenticated context
                    args["user_id"] = self.user_id

                    # 1. Pre-execution Schema Validation
                    validated_kendra_args = KendraRetrieveInput(**args)

                    # 2. Execute Kendra Retrieve Call
                    kendra_passages = self.kendra_service.retrieve(
                        query_text=validated_kendra_args.query_text,
                        user_id=validated_kendra_args.user_id,
                        department_filter=validated_kendra_args.department_filter
                    )

                    observation = f"Amazon Kendra Retrieved {len(kendra_passages)} Grounding Snippets: {json.dumps(kendra_passages)}"
                    print(f"[Observation]: {observation}")
                    scratchpad_history.append(observation)
                    # Successful multi-step completion condition reached
                    return {
                        "status": "SUCCESS",
                        "completed_in_steps": step_count,
                        "trajectory": scratchpad_history
                    }
                except ValidationError as ve:
                    error_msg = f"Kendra Schema Error: {ve.errors()[0]['msg']}"
                    print(f"[Harness Intercept]: {error_msg}")
                    scratchpad_history.append(error_msg)
        return {"status": "FAILED", "reason": "Step budget exhausted without completing task goals."}
# =====================================================================
# 4. EXECUTION DRIVER
# =====================================================================
if __name__ == "__main__":
    # Instantiate runtime for an authorized engineering user
    agent_runtime = ProductionAgentRuntime(user_id="user_eng_101", max_step_budget=4)

    # Run Agentic Workflow
    final_execution_summary = agent_runtime.execute_task(
        task_goal="Verify pending production deployments and confirm authorization policies."
    )

    print("\n================ FINAL SYSTEM SUMMARY ================")
    print(json.dumps(final_execution_summary, indent=2))

第6节：生产运营、遥测与治理

可观测性：使用 OpenTelemetry 追踪执行路径

在生产环境中运行智能体系统需要具备普通 APM 仪表板无法提供的追踪能力。由于智能体会遵循多变且非确定性的执行路径，因此您的遥测系统需要重建智能体走过的整个执行路径树，而不仅仅是单一的请求-响应对：

• 跨度粒度：每个智能体轮次应生成一组嵌套的OpenTelemetry跨度，其中不同的跨度分别用于构建系统提示语、测量模型响应所需时间、检查工具参数是否符合预期结构、记录实际工具调用的耗时，以及处理该轮次中发生的任何内存压缩操作。
• 轨迹状态记录：这些跨度需要记录输入上下文的token数量、输出token数量、工具参数所使用的结构规范，以及返回结果的实际长度。正是这样的详细信息才能实现精确的成本归因，并准确找出轨迹中的哪个环节成为了瓶颈。

运营规范与评估（智能体三元组）

要在实际应用中保持智能体的健康状态，就需要持续从三个关键维度进行自动评估：

• 目标完成率：指那些在未耗尽步骤预算或出现未处理的工具异常的情况下，成功达到终止状态的智能体运行次数所占的比例。
• 上下文真实性，或幻觉指数：用于判断智能体最终的总结答案是否确实基于从其检索工具中获取的事实（如 Amazon Kendra 返回的段落），而非仅来自模型自身的参数记忆。

工具执行精度：格式正确且经过授权的工具调用与格式错误、模式验证失败或未经适当授权即尝试调用的比例。无效调用率过高通常表明提示词指令过于宽松，或者工具模式与模型预期已出现偏差。

结论及可行建议

构建企业级智能体AI系统意味着将底层基础模型视为一个概率推理组件，将其置于受严格控制的确定性软件框架之中。那些能够成功将智能体从原型阶段推向实际应用的机构，都是那些在感知、规划、记忆和工具执行之间划清明确架构界限的团队，而非让模型同时自由控制所有这些功能。

架构团队的实用蓝图

• 将推理与执行分离：每次由大语言模型发起的工具调用都应通过一个确定性框架，在执行前根据Pydantic模式验证参数，之后再清晰地捕获错误。
• 利用Amazon Kendra提供上下文支持：调用Kendra的Retrieve API为智能体提供丰富的段落级上下文，同时借助其索引级的文档访问控制功能实现自动的安全管控。
• 明确限制内存使用：采用滚动式上下文压缩机制，并隔离子任务的上下文，从而避免在长时间运行的会话中出现上下文失效、指令偏移以及Token消耗过高的问题。

设定严格的执行限制：对步骤数量和令牌消耗设置严格的上限，并在工具层面实现断路器机制，防止某个出问题的依赖项引发无限循环。

使执行轨迹可观测：采用 OpenTelemetry 标准来追踪智能体推理循环的每个阶段，记录令牌使用情况、工具延迟以及完整的执行路径，从而能够进行持续的离线评估。

相关阅读

• 为何智能体 AI 的成本会激增：一种以架构为重的成本模型 —— 阐述了为何必须按完成的任务而非每次调用来衡量大型语言模型智能体的成本，并介绍了通过模型路由、上下文预算和缓存等架构手段来控制成本的方案。
• 生产级代理型AI系统的九大架构支柱 — 了解涵盖零信任网络、数据分层及证据绑定等内容的九大支柱架构蓝图，用于构建可审计的企业级代理型AI系统。
• 设计符合团队实际工作方式的代理型AI工作空间 — 阐述了构建代理型AI工作空间的十项核心原则，将这些原则与上下文、知识、自主性、工具访问权限及工作流自动化整合为统一的系统。
• 为生产级 RAG 系统选择与调优嵌入模型 — 了解嵌入模型如何将文本转换为可搜索的向量，领域词汇为何会破坏语义搜索，以及如何为生产级 RAG 选择、压缩和微调模型。