Architektura referencyjna dla systemów sztucznej inteligencji agentowej klasy przemysłowej

Sandboxing, budżety kroków i zautomatyzowane wyłączniki obwodowe

• Izolowany sandboxing: Każdy dynamicznie generowany kod — na przykład Python wygenerowany przez agenta do analizy danych — powinien być uruchamiany w jednorazowych, izolowanych środowiskach typu kontenery Docker lub mikro-VM gVisor, z ściśle kontrolowanym dostępem do sieci wyjściowej.
• Budżety kroków i kosztów: Każde zadanie agenta powinno mieć sztywny limit liczby iteracji wywołań narzędzi (na przykład nie więcej niż 10) oraz maksymalny wydatek na tokeny. Przekroczenie któregokolwiek z tych limitów powinno zmusić system do zatrzymania wykonywania zadania i przekazania go operatorowi ludzkiemu.
• Bezpieczniki narzędzi: Jeśli API lub baza danych w tle nadal zawodzi podczas kolejnych prób ponownego wysłania żądania, system powinien aktywować bezpiecznik, oznaczając dane narzędzie jako UNAVAILABLE w katalogu narzędzi agenta, dzięki czemu warstwa planowania będzie skłonna wybrać alternatywne ścieżki zamiast bezskutecznie próbować używać uszkodzonej zależności.

Rozdział 5: Plan wdrożenia w produkcji

Poniższy przykład to kompletna, działająca aplikacja w Pythonie, która ilustruje Supervisor Multi-Agent Architecture przeznaczoną do użycia w produkcji. Łączy ona walidację narzędzi opartą na Pydantic, planowanie zasobów na poszczególne kroki, deterministyczne radzenie sobie z błędami oraz narzędzie otaczające proces wyszukiwania w 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))

Sekcja 6: Operacje w produkcji, telemetria i zarządzanie

Obserwowalność: śledzenie trajektorii za pomocą OpenTelemetry

Rozwijanie systemów opartych na agentach w środowisku produkcyjnym wymaga poziomu śledzenia, którego nie mogą zapewnić zwykłe panele APM. Ponieważ agenci podążają za zmieniającymi się, niedeterministycznymi ścieżkami wykonywania, pakiety telemetrii muszą odtworzyć cały drzewo trajektorii, którą agent przebył, a nie tylko pojedynczą parę żądanie-odpowiedź:

• Granularność spanów: Każda tura agenta powinna generować zbiór nawarstwionych spanów OpenTelemetry, przy czym poszczególne spany powinny być przeznaczone na tworzenie promptu systemowego, mierzenie czasu potrzebnego modelowi na odpowiedź, sprawdzanie, czy argumenty narzędzia odpowiadają oczekiwanemu schematowi, mierzenie czasu rzeczywistego wywołania narzędzia oraz przeprowadzanie kompresji pamięci, która ma miejsce w trakcie tej tury.
• Zapisywanie stanu trajektorii: Spany muszą rejestrować liczbę tokenów wejściowych i wyjściowych, schematy używane dla parametrów narzędzia oraz surową długość zwróconych obserwacji. To poziom szczegółowości umożliwia dokładne przypisywanie kosztów i precyzyjne określenie, który krok w trajektorii stał się wąskim gardłem.

Zasady operacyjne i ocena (Trójca agentowa)

Zachowanie zdrowia agenta w środowisku produkcyjnym oznacza ciągłe przeprowadzanie automatycznej oceny w trzech kluczowych wymiarach:

• Wskaźnik ukończenia celu: udział wykonań agenta, które osiągają pomyślny stan końcowy bez wyczerpania budżetu kroków lub wystąpienia nieobsłużonego błędu narzędzia.
• Związek z kontekstem, czyli indeks halucynacji: sprawdza to, czy ostateczna, streszczona odpowiedź agenta faktycznie opiera się na faktach pochodzących z jego narzędzi wyszukiwania (takich jak fragmenty tekstowe zwracane przez Amazon Kendra), a nie jest generowana z własnej pamięci parametrycznej modelu.

Wniosek z praktycznymi rekomendacjami

Budowa systemów AI typu agentic na poziomie korporacyjnym oznacza traktowanie podstawowego modelu jako komponentu rozumowania probabilistycznego, który funkcjonuje w ramach deterministycznego, ściśle kontrolowanego oprogramowania. Organizacje, które z powodzeniem przenoszą agenty z fazy prototypu do produkcji, to te, które wyznaczają wyraźne granice architektoniczne pomiędzy percepcją, planowaniem, pamięcią a wykonywaniem narzędzi, zamiast pozwalać modelowi swobodnie kontrolować je wszystkie jednocześnie.

Praktyczny plan działania dla zespołów architektury

• Rozdzielanie rozumowania od wykonywania: każde wezwanie narzędzia inicjowane przez LLM powinno przechodzić przez deterministyczny mechanizm, który przed wykonaniem weryfikuje argumenty według schematu Pydantic, a następnie czysto łapie błędy.
• Wykorzystanie Amazon Kendra do uzyskiwania kontekstu: użyj API Retrieve Kendry, aby dostarczyć agentowi bogaty kontekst na poziomie akapitów, polegając jednocześnie na jego mechanizmie filtrowania uprawnień dokumentów na poziomie indeksu, aby automatycznie zapewniać bezpieczeństwo.
• Jasne ograniczenie zużycia pamięci: zastosuj mechanizm stopniowej kompresji kontekstu oraz izoluj kontekst poszczególnych zadań, aby podczas długotrwałych sesji nie doszło do degradacji kontekstu, odchylenia od instrukcji ani nadmiernego zużycia tokenów.

Literatura pokrewna

• Dlaczego koszty AI agencyjnej rosną gwałtownie: model kosztów oparty na architekturze — Wyjaśnia, dlaczego koszty agentów LLM należy mierzyć według ukończonych zadań, a nie według połączeń, i przedstawia takie elementy architektury jak routowanie modeli, budżety kontekstu oraz cache, które pomagają kontrolować wydatki.
• Dziewięć elementów architektonicznych dla systemów sztucznej inteligencji agencyjnej klasy produkcyjnej — Poznaj plan architektury oparty na dziewięciu elementach – obejmujący sieci zero trust, poziomy danych oraz mechanizmy wiążące dowody – służący do budowy systemów sztucznej inteligencji agencyjnej nadających się do audytu i przeznaczonych dla dużych przedsiębiorstw.
• Projektowanie przestrzeni pracy sztucznej inteligencji agencyjnej dostosowanych do sposobu działania zespołów — Wyjaśnia dziesięć podstawowych zasad budowy przestrzeni pracy sztucznej inteligencji agencyjnej, które łączą kontekst, wiedzę, autonomię, dostęp do narzędzi oraz automatyzację procesów w jednolity system.
• Wybieranie i dostosowywanie modeli embeddingowych dla systemów RAG w produkcji — Dowiedz się, jak modele embeddingowe przekształcają tekst w wektory nadające się do wyszukiwania, dlaczego słownictwo specjalistyczne psuje funkcjonowanie wyszukiwania semantycznego oraz jak wybierać, kompresować i dopasowywać modele do zastosowań w systemach RAG w produkcji.