Production Systems#

Enterprise-grade services that transform research prototypes into production-ready agentic systems. These components provide the security, reliability, and scalability required for high-stakes scientific and industrial environments.

The Five Pillars#

🛡️ Human Approval System

LangGraph-Native Oversight

Production-ready approval workflows with rich context, security analysis, and seamless resumption.

Human Approval

🔄 Data Management Framework

Unified Data Orchestration

Heterogeneous data source integration with provider discovery and concurrent retrieval.

Data Management

🐍 Python Execution Service

Secure Code Execution

Container-isolated Python execution with approval integration and flexible deployment.

Python Execution

🧠 Memory Storage System

Persistent User Memory

Structured memory operations with data source integration and approval workflows.

Memory Storage

🚢 Container Management

Service Orchestration

Podman-based deployment with hierarchical service discovery and template processing.

Container Management

Production Integration#

These systems work together to provide comprehensive production capabilities:

Security Flow

How safety and oversight are maintained:

# Approval system integration
from framework.approval import get_approval_manager
approval_manager = get_approval_manager()

# Secure execution with oversight
from framework.services.python_executor import PythonExecutorService, PythonExecutionRequest
python_service = PythonExecutorService()
request = PythonExecutionRequest(
    user_query="Analyze beam performance data",
    task_objective="Generate comprehensive performance report"
)

# Service automatically pauses for human review when requires_approval: true
config = {"thread_id": "session_123"}
result = await python_service.ainvoke(request, config)
# Execution resumes after approval with audit trail

Data Architecture

Unified data access across systems:

# Data source integration
from framework.data_management import (
    get_data_source_manager,
    create_data_source_request,
    DataSourceRequester
)
data_manager = get_data_source_manager()
request = create_data_source_request(
    state,
    requester=DataSourceRequester(
        capability_name="performance_analysis",
        component_name="beam_analysis"
    ),
    query="beam current trends"
)

# Concurrent retrieval from all providers
result = await data_manager.retrieve_all_context(request)

# Memory integration
from framework.services.memory_storage import get_memory_storage_manager, MemoryContent
from datetime import datetime
memory_manager = get_memory_storage_manager()
memory_entry = MemoryContent(
    timestamp=datetime.now(),
    content=f"Analysis results: {result.context_data}"
)
success = memory_manager.add_memory_entry(user_id, memory_entry)

Deployment Strategy

Container orchestration for scalability:

# Container deployment configuration
deployed_services:
  - framework.jupyter              # Secure execution environment
  - framework.pipelines            # Processing pipeline infrastructure
  - applications.als_expert.logbook  # Application data

framework:
  execution:
    execution_method: "container"   # Isolation by default
    modes:
      write_access:
        requires_approval: true    # Safety first
        allows_writes: true
        kernel_name: "python3-epics-write"