Core Framework#

The Core Framework provides the essential foundation APIs that enable reliable, type-safe agentic system development. These five interconnected systems form the backbone of every capability, infrastructure node, and production deployment in the Alpha Berkeley Framework.

System Architecture#

The Core Framework implements a Type-Safe, Convention-Driven Architecture with five integrated components:

🏗️ Base Components

Foundation & LangGraph Integration

BaseCapability, BaseInfrastructureNode, and decorators for seamless framework integration with error handling and planning.

Base Components

🔄 State & Context Management

Data Persistence & Exchange

AgentState with selective persistence, ContextManager for type-safe data access, and Pydantic-based serialization.

State and Context Management

📋 Registry System

Component Discovery & Management

Configuration-driven component loading with lazy initialization, dependency resolution, and type-safe access.

Registry System

⚙️ Configuration System

Environment & Settings Management

YAML-based configuration with environment resolution, model settings, and LangGraph integration.

Configuration System

💬 Prompt Management

Domain-Agnostic Prompt System

Dependency injection for prompt customization with builder patterns, defaults, and application-specific overrides.

Prompt System

Framework Integration Patterns#

These systems work together to provide a unified development experience:

Startup and Initialization

How the framework initializes and loads components:

from framework.registry import initialize_registry, get_registry
from framework.state import StateManager

# 1. Initialize the global registry system (application startup)
initialize_registry()  # Loads all applications and components

# 2. Access the initialized registry
registry = get_registry()

# 3. Components are now available for use
capability = registry.get_capability('data_analysis')
context_class = registry.get_context_class('ANALYSIS_RESULTS')
data_source = registry.get_data_source('core_user_memory')

State and Context Flow

How state and context work together during execution:

from framework.state import StateManager
from framework.context import ContextManager

# 1. Create fresh state for new conversation
state = StateManager.create_fresh_state(
    user_input="Analyze beam performance data",
    current_state=previous_state  # Preserves context
)

# 2. Access context through ContextManager
context = ContextManager(state)

# 3. Retrieve typed context objects
pv_data = context.get_context('PV_ADDRESSES', 'beam_current')

# 4. Store new context results
return StateManager.store_context(
    state, 'ANALYSIS_RESULTS', 'step_1', analysis_results
)

Configuration Access

How to access configuration values throughout the framework:

from configs.config import (
    get_config_value, get_model_config, get_full_configuration
)

# Simple configuration access with defaults
timeout = get_config_value('execution.timeout', 30)
debug_mode = get_config_value('development.debug', False)

# Model-specific configuration
model_config = get_model_config('framework', 'orchestrator')

# Full configuration for service passing
full_config = get_full_configuration()
service_config = {
    "configurable": {
        **full_config,
        "thread_id": f"my_service_{context_key}"
    }
}

Capability Implementation

Real capability implementation pattern from the framework:

from framework.base import BaseCapability, capability_node
from framework.state import AgentState, StateManager
from framework.context import ContextManager
from configs.config import get_model_config
from applications.als_expert.context_classes import AnalysisResultsContext

@capability_node
class DataAnalysisCapability(BaseCapability):
    name = "data_analysis"
    description = "General data analysis capability"
    provides = ["ANALYSIS_RESULTS"]

    @staticmethod
    async def execute(state: AgentState, **kwargs) -> Dict[str, Any]:
        # Extract current execution step
        step = StateManager.get_current_step(state)

        # Access context manager for input data
        context = ContextManager(state)

        # Get configuration for models/services
        model_config = get_model_config('framework', 'python_code_generator')

        # Process data (actual implementation logic)
        analysis_results = await process_analysis(step, context)

        # Create typed context object
        result_context = AnalysisResultsContext(
            analysis_summary=analysis_results.summary,
            confidence_score=analysis_results.confidence,
            **analysis_results.data
        )

        # Store results in state
        return StateManager.store_context(
            state, 'ANALYSIS_RESULTS',
            step.get('context_key', 'default'),
            result_context
        )