Архітектура для референсу продуктивних корпоративних систем агентного ШІ високого рівня
Дізнайтеся про основні компоненти, стратегії запам’ятовування, методи відтворення інформації та правила безпеки, необхідні для переходу агентних ШІ від прототипів до надійних корпоративних систем виробництва.
Короткий огляд
Перехід від безстанових застосунків на основі великих мовних моделей (LLM) до продакшн-готових агентних систем штучного інтелекту означає значну зміну у способі створення програмного забезпечення компаніями. Ранні спроби впровадження ШІ в бізнес сильно залежали від базового підходу Retrieval-Augmented Generation (RAG) — завантаження документів у формат ембеддингів та включення їх до контекстного вікна запиту. Цей підхід добре підходить для простого відповідання на запитання, але не дозволяє ефективно приймати автономні рішення, планувати багатоетапні процеси з урахуванням стану, викликати інструменти з високою стійкістю та самостійно коригувати свої дії протягом робочого процесу.
Enterprise Agentic AI закриває цю прогалину, переосмислюючи фундаментальні моделі: замість того, щоб виступати прямим інтерфейсом для користувача, модель стає компонентом міркувань, вбудованим у детерміністичний програмний каркас. Такі системи можуть сприймати своє оточення, розділяти великі цілі на менші кроки, викликати внутрішні корпоративні сервіси та зберігати стан виконання під час багатокрокових операцій. Однак для переходу від функціонального прототипу до системи, яку можна використовувати в промислових умовах, необхідно вирішити проблеми нестабільних недетерміністичних контрольних циклів, занепаду контексту протягом тривалих сеансів, ризиків підвищення привілеїв та неконтрольованої витрати токенів.
У цьому документі описана повна архітектура для архітекторів та керівників проєктів з штучного інтелекту, відповідальних за системи корпоративних агентів. Він охоплює базові шари, необхідні будь-якому агенту, порівнює топології багатоагентних систем, детально розглядає шаблони інтеграції для пошуку на рівні корпоративних систем — з особливою увагою до Amazon Kendra — та містить працездатні приклади на Python, придатні для використання в продакшені.
Розділ 1: Канонічна архітектура корпоративного агента
Основні шари: сприйняття, міркування, планування, пам’ять та виконання завдань за допомогою інструментів
Система агентів, готова до використання в продакшені, розділяє ймовірнісну модель основи від детермінованого середовища виконання, яке її оточує. Цю архітектуру зазвичай складають п’ять основних шарів, усі з яких працюють всередині контейнера керованого середовища виконання:
- Шар сприйняття: Перетворює необроблені корпоративні сигнали — телеметрію, повідомлення користувачів, дані webhook, відповіді API — на чисті, типізовані дані, які може використовувати решта системи. Це включає очищення даних, обмеження частоти запитів та перевірку схеми ще до того, як буде викликаний LLM.
- Двигун міркувань: Когнітивний центр, який працює за допомогою підключеного LLM (наприклад, моделей Anthropic Claude 3.5 Sonnet або AWS Bedrock). Замість того, щоб самостійно вживати дій, цей двигун інтерпретує поточний контекст, оцінює можливі варіанти подальших дій та формулює структуровані заяви про наміри.
Шаблони оркестрації: цикли з одним агентом проти топологій з кількома агентами
Вибір правильного шаблону оркестрації суттєво залежить від складності цільової сфери:
- Цикли Single-Agent ReAct: один цикл міркувань, який проходить через етапи Мислення, Дії та Спостереження. Це підходить для вузьких, лінійних робочих процесів, які використовують менше 5–8 різних інструментів. При більшій кількості інструментів один агент схильний стикатися з перевантаженими вікнами контексту, неточними інструкціями та нерозумінням, який інструмент обрати.
Для використання в корпоративних системах рекомендується за замовчуванням використовувати Supervisor Multi-Agent Topology завдяки чітким межам стану, простоті аудиту та більш передбачуваним витратам на обробку контексту.
Розділ 2: Керування пам’яттю в корпораціях та гігієна контексту
Ієрархія пам’яті: робоча пам’ять, короткостроковий контекст та довгострокова пам’ять
Розглядайте керування пам’яттю агента як проблему бюджетування ресурсів. Неконтрольований зростання пам’яті підвищує витрати на обробку даних, збільшує затримки та призводить до втрати контексту через його погіршення. Добре спроєктований корпоративний агент потребує багатошарової структури пам’яті:
- Робоча пам’ять (Scratchpad): Вікно актуального контексту — поточні інструкції системи, стан активного завдання та результати останніх операцій з інструментами.
Стратегії компактування контексту, його обрізки та ізоляції стану
Щоб контекст не почав «гнити», система потрібно активно дотримуватися правил чистоти:
- Обрізка даних спостережень: Первинні відповіді інструментів — наприклад, 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: Детерміністичні правила, перевірка інструментів та захисні механізми
Контролі до виконання (Feedforward) та після виконання (Feedback)
Автономні агенти потребують чітких меж на рівні програмного забезпечення, щоб запобігти підвищенню привілеїв, некоректному виклику інструментів чи нескінченним циклам виконання:
- Передвідправна верифікація (до виконання): Перш ніж виклик інструменту досягне підсистеми підприємства, движок виконання перевіряє його аргументи за суворими схемами Pydantic — переконуючись, що наявні необхідні поля, числові чи значення з типу enum знаходяться в допустимих діапазонах, а токен авторизації викликаючого справді дозволяє цю дію.
Помилка: Таблиця бази даних ‘users_v2’ не знайдена. Доступні таблиці: ['users', 'orders'] — надаючи агенту щось конкретне, що можна використати для корекції наступних дій.Сандбоксинг, бюджети кроків та автоматичні захисні пристрої
- Ізольований сандбоксинг: Будь-який динамічно створений код — наприклад, Python, створений агентом для аналізу даних — має виконуватися всередині одноразових, ізольованих сандбоксів, таких як контейнери Docker чи мікро-ВМ gVisor, з суворо обмеженим зовнішнім мережевим доступом.
- Бюджети кроків та витрат: Кожен завдання агента має мати жорсткий ліміт на кількість ітерацій виклику інструментів (наприклад, не більше 10), а також максимальну суму витрат на токени. Перевищення будь-якого з цих лімітів має змусити систему припинити виконання та передати завдання людині-оператору.
- Захисні пристрої інструментів: Якщо бекенд-API чи база даних продовжують відмовляти під час послідовних спроб перезапуску, система має активувати захисний пристрій, позначивши цей інструмент як
UNAVAILABLEу каталозі інструментів агента, щоб рівень планування звернувся до альтернативних шляхів замість того, щоб постійно намагатися використати несправну залежність.
Розділ 5: План реалізації в промислових умовах
Наведений нижче приклад — це повна, працездатна Python-застосунок, який ілюструє архітектуру Supervisor Multi-Agent Architecture промислового рівня. Він поєднує в собі перевірку інструментів на основі 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), а не генерується з власної параметричної пам’яті моделі.
Висновки з практичними рекомендаціями
Створення систем AI рівня корпорацій означає розгляд базової моделі як компонента пробабілістичного міркування, який функціонує всередині детермінованої, суворо контрольованої програмної архітектури. Організації, яким вдається перевести агентів з прототипу у продакшн, — це ті, що чітко розділяють архітектурні елементи сприйняття, планування, пам’яті та виконання інструментів, замість того щоб дозволяти моделі контролювати всі вони одночасно.
Конкретний план дій для команд з архітектури
- Розділяйте міркування та виконання: кожен запит до інструменту, ініційований LLM, має проходити через детерміністичний механізм, який перевіряє аргументи за схемою Pydantic перед виконанням та чітко фіксує помилки після нього.
- Використовуйте Amazon Kendra для отримання контексту: використовуйте API
RetrieveKendra, щоб надати агенту детальний контекст на рівні уривків тексту, водночас покладаючись на функції контролю доступу до документів на рівні індексу для автоматичного забезпечення безпеки. - Чітко обмежуйте використання пам’яті: застосовуйте механізми компактування контексту та ізолюйте контекст окремих завдань, щоб у процесі тривалих сеансів не виникало проблем зі старінням контексту, зміною інструкцій чи неконтрольованим витрачанням токенів.
Пов’язана література
- Чому витрати на агентський ШІ стрімко зростають: модель витрат, заснована на архітектурі — пояснює, чому витрати на агентів на основі LLM потрібно вимірювати за кожне завершене завдання, а не за кожен виклик, та описує архітектурні засоби, такі як маршрутизація моделей, бюджети контексту та кешування, для контролю витрат.
- Дев’ять архітектурних принципів для систем AI типу агент на рівні продакшну — Дізнайтеся про план архітектури, який ґрунтується на дев’яти принципах — від мереж з концепцією нульової довіри та рівнів даних до механізмів підтвердження достовірності — для створення перевірюваних систем AI типу агент високого рівня.
- Проектування робочих просторів AI типу агент, що відповідають реальним методам роботи команд — Пояснюється десять основних принципів створення робочих просторів AI типу агент, які поєднують контекст, знання, автономію, доступ до інструментів та автоматизацію робочих процесів у єдину систему.
- Вибір та налаштування моделей ембеддингу для систем RAG у промисловому використанні — Дізнайтеся, як моделі ембеддингу перетворюють текст на вектори, які можна шукати, чому словниковий запас певної галузі заважає семантичному пошуку, та як вибирати, стискувати та налаштовувати моделі для використання у системах RAG.