Архитектура для производственных корпоративных систем искусственного интеллекта с агентными функциями
Изучите основные компоненты, стратегии запоминания, методы восстановления информации и ограничительные меры, необходимые для того, чтобы превратить агентные ИИ из прототипов в надежные корпоративные производственные системы.
Краткое резюме
Переход от приложений на основе больших языковых моделей без состояния к производственным решениям с агентными технологиями представляет собой значительное изменение в подходе компаний к разработке программного обеспечения. Первоначальные попытки внедрения ИИ в бизнесе в значительной степени опирались на базовую технологию Retrieval-Augmented Generation (RAG) — преобразование документов в эмбеддинги и включение их в окно контекста запроса. Такой подход хорошо справляется с простыми задачами ответов на вопросы, но не позволяет осуществлять автономное принятие решений, планирование с несколькими шагами и учетом состояния, надежное вызов инструментов, а также самокоррекцию в рамках рабочего процесса.
Enterprise Agentic AI устраняет этот разрыв путем переосмысления функций базовых моделей: вместо того чтобы выступать прямым интерфейсом для пользователя, модель становится компонентом для логических рассуждений, встроенным в детерминистическую программную среду. Такие системы могут воспринимать окружающую среду, разбивать крупные цели на более мелкие шаги, вызывать внутренние корпоративные сервисы и сохранять состояние выполнения в ходе многоэтапных операций. Однако для перехода от рабочего прототипа к продукту, готовому к использованию в производственных условиях, необходимо решить проблемы нестабильных недетерминистических циклов управления, снижения качества обработки информации в течение длительных сессий, рисков повышения привилегий и чрезмерного расхода токенов.
В этом материале представлена полная схема референтной архитектуры, предназначенная для архитекторов и руководителей проектов по ИИ, отвечающих за системы корпоративных агентов. В нем рассматриваются основные слои, необходимые любому агенту, сравниваются топологии множественных агентов, подробно описываются шаблоны интеграции для поиска корпоративного уровня — с особым акцентом на Amazon Kendra — а также приводятся рабочие примеры на Python, пригодные для использования в производстве.
Раздел 1: Каноническая архитектура корпоративного агента
Основные слои: восприятие, рассуждение, планирование, память и выполнение задач с использованием инструментов
Система агентов, готовая к использованию в производстве, разделяет вероятностную модель основы и детерминированную среду выполнения, окружающую её. Обычно такая архитектура состоит из пяти основных слоев, все из которых работают внутри управляемого контейнера среды выполнения:
- Слой восприятия: Преобразует необработанные корпоративные сигналы — телеметрию, сообщения пользователей, данные webhook, ответы API — в чистые, типизированные данные, которые могут быть использованы остальной частью системы. Это включает очистку данных, ограничение частоты запросов и раннюю проверку схемы, причем все эти действия выполняются ещё до вызова LLM.
- Механизм рассуждений: Когнитивный центр, работающий на основе LLM (например, моделей Anthropic Claude 3.5 Sonnet или AWS Bedrock). Вместо того чтобы самостоятельно принимать действия, этот механизм анализирует текущий контекст, оценивает возможные варианты дальнейших действий и формулирует структурированные заявления о намерениях.
Шаблоны оркестрации: циклы с одним агентом против топологий с множеством агентов
Выбор подходящего шаблона оркестрации во многом зависит от степени сложности целевой области:
- Циклы ReAct с одним агентом: один цикл рассуждений, проходящий через этапы мышления, действия и наблюдения. Этот подход подходит для узких, линейных рабочих процессов, в которых используется менее 5–8 различных инструментов. При большем количестве инструментов один агент склонен сталкиваться с перегруженными окнами контекста, неточными инструкциями и неопределенностью относительно выбора инструмента.
Для использования в корпоративных производственных средах рекомендуемым стандартным вариантом является Топология множества агентов-супервайзеров благодаря четким границам состояний, упрощенной возможности аудита и более предсказуемым затратам на обработку контекста.
Раздел 2: Управление памятью в корпорациях и гигиена контекста
Иерархия памяти: рабочая память, краткосрочная траектория и долгосрочная память
Управление памятью агента можно рассматривать как проблему бюджетирования ресурсов. Неограниченное увеличение объема памяти приводит к росту затрат на выводы, увеличению задержек и потере контекста по мере его ухудшения, что заставляет модель терять нить размышлений. Хорошо спроектированный корпоративный агент нуждается в многоуровневой структуре памяти:
- Рабочая память (рабочая поверхность): окно актуального контекста — текущие действующие системные инструкции, состояние активной задачи и самые свежие результаты работы инструментов.
Стратегии сжатия контекста, его обрезки и изоляции состояния
Чтобы контекст не портился, движок должен активно соблюдать правила очистки:
- Обрезка данных наблюдений: необработанные ответы инструментов — например, JSON-данные с 500 записями — никогда не должны попадать напрямую в рабочую память. Движок должен очистить, обрезать или свести к краткому изложению данные, возвращаемые инструментом, прежде чем они попадут в двигатель обработки информации.
- Сжатие с использованием «окна скроллинга»: Как только объём временной памяти превышает установленный лимит (например, 20% от общего лимита контекста модели), происходит шаг сжатия, в ходе которого предыдущие сообщения диалога преобразуются в краткие семантические резюме, а оригинальные тексты сообщений удаляются.
- Изоляция состояния подзадачи: Когда руководитель задаёт задание подагенту, он создаёт для этого подагента новый контекст, содержащий только конкретную цель и необходимые параметры — никакая информация из истории рассуждений самого руководителя не просачивается.
Раздел 3: Подробный обзор: использование Amazon Kendra для поиска в корпоративных системах
Amazon Kendra как слой поиска в производственных системах
Создание системы поиска на общей векторной базе данных обычно подразумевает самостоятельную разработку механизмов ввода документов, логики их разбиения на части, генерации векторных представлений и слоя слияния результатов гибридного поиска с нуля. Amazon Kendra, в свою очередь, предлагает полностью управляемый корпоративный поисковик, который решает эти задачи из коробки. Он оснащен встроенным многоэтапным пониманием естественного языка, структурным анализом, учитывающим таблицы, заголовки и подзаголовки, а также гибридной моделью ранжирования, сочетающей лексические и семантические признаки.
В составе корпоративных агентских систем Kendra обычно выступает в роли основной подсистемы обеспечения фактологической основы знаний — компонента, отвечающего за возможность агентов получать проверенный фактологический контекст из различных внутренних источников данных, без риска фрагментации, возникающего при примитивном разделении данных в векторных хранилищах.
Расширенная настройка релевантности и динамическое сокращение метаданных
Для обеспечения безопасного автономного принятия решений системам поиска корпоративного уровня необходима точная обработка разрешений и настраиваемые механизмы контроля релевантности:
- Встроенные списки контроля доступа (ACL): Kendra автоматически загружает и преобразует списки ACL на уровне документов, определенные в исходных системах — будь то SharePoint, Confluence или S3. Когда агент отправляет запрос, он передает идентификатор авторизованного пользователя в виде токена
UserContext, и Kendra применяет механизмы безопасности на уровне индекса, чтобы ни один фрагмент информации, доступ к которому у запрашивающего пользователя отсутствует, не попадал к агенту.
_last_updated_at, по категориям бизнеса или при точном совпадении в пользовательских полях, таких как Department или ProjectCode.Retrieve вместо универсального API Query. API Retrieve полностью игнорирует метаданные поиска, ориентированные на интерфейс, и вместо этого возвращает подробные выдержки на уровне абзацев, специально предназначенные для непосредственного ввода в окно контекста больших языковых моделей.Раздел 4: Детерминистичные ограничения, проверка инструментов и защитные механизмы
Контроль до выполнения (предварительный) и после выполнения (обратная связь)
Автономным агентам необходимы четкие границы на уровне программного обеспечения для предотвращения повышения привилегий, вызовов некорректных инструментов или бесконечных циклов выполнения:
- Валидация до выполнения: Прежде чем запрос к инструменту достигнет внутренней корпоративной системы, движок выполнения проверяет его аргументы согласно строгим схемам Pydantic — убеждаясь, что все необходимые поля присутствуют, числовые или значения из перечня находятся в допустимых диапазонах, и токен авторизации вызывающего пользователя действительно разрешает данное действие.
Ошибка: таблица базы данных 'users_v2' не найдена. Доступные таблицы: ['users', 'orders'] — предоставляя агенту конкретные данные, которые он может использовать для корректировки своих дальнейших действий.Сандбоксирование, лимиты на шаги и автоматические защитные механизмы
- Изолированное сандбоксирование: Любой динамически генерируемый код — например, Python, созданный агентом для анализа данных — должен выполняться в временных, изолированных средах вроде контейнеров Docker или микро-ВМ gVisor с строго ограниченным внешним сетевым доступом.
- Бюджеты шагов и затрат: Каждая задача агента должна иметь строгий лимит на количество итераций вызова инструментов (например, не более 10) а также максимальную сумму расходов на токены. Превышение любого из этих лимитов должно привести к остановке выполнения и передаче задачи человеку-оператору.
- Защитные механизмы инструментов: Если бэкенд-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, причем отдельные спаны должны быть предназначены для формирования системного запроса, измерения времени ответа модели, проверки соответствия аргументов инструмента ожидаемой схеме, отслеживания времени реального вызова инструмента и выполнения любых операций сжатия памяти, происходящих в течение этого хода.
- Запись состояния траектории: Спаны должны фиксировать количество токенов входного контекста, количество токенов выходных данных, схемы параметров инструмента и первоначальную длину возвращаемых результатов. Именно такой уровень детализации позволяет точно определять затраты и выявлять конкретный этап в траектории, ставший узким местом.
Операционные ограничения и оценка (Триада агентов)
Для поддержания здоровья агента в производственной среде необходимо постоянно проводить автоматизированную оценку в трех ключевых аспектах:
- Уровень выполнения целей: доля запусков агента, которые достигают успешного конечного состояния без исчерпания лимита шагов или возникновения неконтролируемых исключений.
- Основанность на контексте, или индекс галлюцинаций: этот показатель проверяет, подкреплена ли окончательная краткая ответ агента фактами, полученными из его инструментов поиска (например, отрывками из данных Amazon Kendra), или же он сгенерирован на основе собственной параметрической памяти модели.
Выводы с практическими рекомендациями
Создание систем искусственного интеллекта класса enterprise подразумевает рассмотрение основной модели как компонента вероятностного вывода, находящегося внутри детерминированной, строго контролируемой программной среды. Организации, которым удается перевести агентов с этапа прототипирования в производственную эксплуатацию, — это те, которые четко разграничивают архитектурные элементы восприятия, планирования, памяти и выполнения инструментов, вместо того чтобы позволять модели свободно управлять всеми ими одновременно.
Практический план действий для команд по архитектуре
- Разделяйте процесс обоснования и выполнения: каждый вызов инструмента, инициированный с помощью LLM, должен проходить через детерминистичный механизм, который проверяет аргументы согласно шаблону Pydantic перед выполнением и четко фиксирует возникающие ошибки.
- Используйте Amazon Kendra для обеспечения контекста: вызывайте API
RetrieveKendra, чтобы предоставить агенту подробный контекст на уровне отдельных фрагментов текста, одновременно полагаясь на его механизмы управления правами доступа к документам на уровне индекса для автоматического соблюдения требований к безопасности. - Явно ограничивайте использование памяти: применяйте механизмы постепенной компактации контекста и изолируйте контекст отдельных подзадач, чтобы при длительной работе не возникало проблем с устареванием контекста, отклонениями в выполнении инструкций и чрезмерным расходом токенов.
Связанные материалы
- Почему стоимость агентных ИИ растет в геометрической прогрессии: модель затрат, основанная на архитектуре — объясняет, почему стоимость работы агентов на основе больших языковых моделей должна измеряться по завершенным задачам, а не по вызовам, и описывает архитектурные подходы, такие как маршрутизация моделей, лимиты контекста и кэширование, для контроля расходов.
- Девять архитектурных принципов для производственных систем искусственного интеллекта типа агент — Ознакомьтесь с планом архитектуры, основанным на девяти принципах, включающим сети нулевого доверия, уровни данных и механизмы связывания доказательств, для создания проверяемых систем искусственного интеллекта типа агент корпоративного уровня.
- Проектирование рабочих пространств искусственного интеллекта типа агент, соответствующих реальным методам работы команд — Рассматриваются десять основных принципов построения таких рабочих пространств, которые объединяют контекст, знания, автономию, доступ к инструментам и автоматизацию рабочих процессов в единую систему.
- Выбор и настройка моделей эмбеддингов для производственных систем RAG — Узнайте, как модели эмбеддингов преобразуют текст в векторы, подлежащие поиску, почему словарный запас конкретной области мешает семантическому поиску, и как выбирать, компрессировать и дорабатывать модели для использования в производственных системах RAG.