crca 1.5.0
CR-CA Wrapper for CR-CA LRM on HF (Depricated)
License: Apache-2.0
Requires Python: >=3.10
Keywords: causal-reasoning, counterfactual-analysis, data-science, governance, llm, machine-learning, optimization, socioeconomic-dynamics, structural-causal-models, trading

# CR-CA: Causal Reasoning and Counterfactual Analysis Framework

**Version**: v1.4.0  
**Repository**: [https://github.com/IlumCI/CR-CA](https://github.com/IlumCI/CR-CA)  
**License**: Apache-2.0

CR-CA is a comprehensive framework for causal reasoning and counterfactual analysis, combining structural causal models (SCMs) with large language model (LLM) integration. The framework enables sophisticated causal inference, automatic variable extraction from natural language, deterministic causal simulation, and comprehensive counterfactual scenario generation. Designed for both research and practical applications, CR-CA provides a modular architecture supporting specialized agents for quantitative trading, socioeconomic dynamics, and corporate governance.

---

## Quick Reference Tables

### Feature Matrix

| Feature | Core CRCAAgent | CRCA-SD | CRCA-CG | CRCA-Q |
|---------|----------------|---------|---------|--------|
| Automatic Variable Extraction | ✓ | ✓ | ✓ | ✓ |
| LLM-Based Causal Analysis | ✓ | ✓ | ✓ | ✓ |
| Deterministic Simulation | ✓ | ✓ | - | ✓ |
| Counterfactual Generation | ✓ | ✓ | ✓ | ✓ |
| Batch Prediction | ✓ | - | - | ✓ |
| Async Operations | ✓ | - | - | ✓ |
| Optimization Methods | ✓ | ✓ | - | ✓ |
| Time-Series Analysis | ✓ | - | - | ✓ |
| Bayesian Inference | ✓ | - | - | ✓ |
| Data Source Integration | - | - | - | ✓ |

### Module Overview

| Module | Purpose | Key Components |
|--------|---------|----------------|
| `CRCA.py` | Core causal reasoning agent | CRCAAgent, automatic extraction, dual-mode operation |
| `utils/` | Utility functions and extensions | Formatter, TUI, router, conversation management |
| `tools/` | External tool integration | MCP client utilities, protocol handling |
| `templates/` | Agent creation framework | Base agents, feature mixins, module registry |
| `schemas/` | Type definitions | MCP schemas, data structures |
| `crca_sd/` | Socioeconomic dynamics | MPC solver, governance, real-time monitoring |
| `crca_cg/` | Corporate governance | CorporateSwarm, ESG scoring, multi-agent orchestration |
| `branches/CRCA-Q.py` | Quantitative trading | QuantTradingAgent, causal signals, portfolio optimization |

### Dependency Matrix

| Dependency | Core | Advanced Analysis | CRCA-SD | CRCA-CG | CRCA-Q |
|------------|------|-------------------|---------|---------|--------|
| numpy | ✓ | ✓ | ✓ | ✓ | ✓ |
| rustworkx | ✓ | ✓ | ✓ | ✓ | ✓ |
| swarms | ✓ | ✓ | ✓ | ✓ | ✓ |
| litellm | ✓ | ✓ | ✓ | ✓ | ✓ |
| pandas | - | ✓ | ✓ | ✓ | ✓ |
| scipy | - | ✓ | ✓ | - | ✓ |
| loguru | ✓ | ✓ | ✓ | ✓ | ✓ |
| pydantic | ✓ | ✓ | ✓ | ✓ | ✓ |
| mcp | ✓ | ✓ | - | - | - |
| openai | ✓ | ✓ | - | - | - |
| ccxt | - | - | - | - | ✓ |
| web3 | - | - | - | - | ✓ |
| curses | - | - | ✓ | - | - |

---

## Table of Contents

1. [Installation & Setup](#installation--setup)
2. [Core CRCAAgent](#core-crcaagent)
3. [Automatic Variable Extraction](#automatic-variable-extraction)
4. [Module Documentation](#module-documentation)
   - [Utils Module](#utils-module)
   - [Tools Module](#tools-module)
   - [Templates Module](#templates-module)
   - [Schemas Module](#schemas-module)
   - [CRCA-SD Module](#crca-sd-module)
   - [CRCA-CG Module](#crca-cg-module)
5. [Policy Engine System](#policy-engine-system)
6. [Branch Integrations](#branch-integrations)
   - [CRCA-Q Integration](#crca-q-integration)
   - [CRCA-SD Integration](#crca-sd-integration)
   - [CRCA-CG Integration](#crca-cg-integration)
7. [Usage Examples](#usage-examples)
8. [Advanced Features](#advanced-features)
9. [Testing](#testing)
10. [CI/CD Workflows](#cicd-workflows)
11. [Project Structure](#project-structure)
12. [Architecture Diagrams](#architecture-diagrams)
13. [Integration Examples](#integration-examples)
14. [API Reference](#api-reference)
15. [Architecture & Design](#architecture--design)
16. [Research & Theory](#research--theory)
17. [Configuration & Customization](#configuration--customization)
18. [Troubleshooting](#troubleshooting)
19. [Contributing](#contributing)
20. [Changelog](#changelog)
21. [References & Resources](#references--resources)
22. [Bolt.diy Web Application (WIP)](#boltdiy-web-application-wip)

---

## Installation & Setup

### Prerequisites

- Python 3.10 or higher
- pip package manager
- (Optional) Git for cloning the repository
- (Optional) Docker for containerized deployment

### Quick Install

**Method 1: From PyPI (Recommended)**

```bash
pip install crca
```

**Method 2: From Source**

Clone the repository:

```bash
git clone https://github.com/IlumCI/CR-CA.git
cd CR-CA
```

Install dependencies:

```bash
pip install -r requirements.txt
```

**Method 3: Development Install**

For development with editable install:

```bash
git clone https://github.com/IlumCI/CR-CA.git
cd CR-CA
pip install -e ".[dev]"
```

### Optional Dependencies

**For Policy Engine:**

```bash
# Install policy engine dependencies
pip install cvxpy ruptures psutil

# Or install all optional dependencies
pip install crca[policy-engine]
```

**For CRCA-SD:**

```bash
# CRCA-SD dependencies are included in requirements.txt
# Additional dependencies may be needed for specific features
```

**For CRCA-CG:**

```bash
# Install TUI dependencies (optional)
pip install rich

# Or install all optional dependencies
pip install crca[tui]
```

**For CRCA-Q:**

```bash
# Install trading dependencies
pip install ccxt web3

# Or install all optional dependencies
pip install crca[trading]
```

### Environment Variables

Create a `.env` file in the project root with required API keys:

```bash
# Required for LLM functionality
OPENAI_API_KEY=your_openai_api_key_here

# Optional: For CRCA-Q trading
KRAKEN_API_KEY=your_kraken_key_here
KRAKEN_API_SECRET=your_kraken_secret_here

# Optional: For alternative data sources
TWITTER_BEARER_TOKEN=your_twitter_token_here
ETHERSCAN_API_KEY=your_etherscan_key_here
THEGRAPH_API_KEY=your_thegraph_key_here

# Optional: For MCP servers
MCP_SERVER_URL=http://localhost:8000
MCP_API_KEY=your_mcp_api_key_here
```

### Docker Setup (Optional)

**Example: Dockerfile for Policy Engine**

```dockerfile
FROM python:3.11-slim

WORKDIR /app

# Install system dependencies
RUN apt-get update && apt-get install -y \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# Copy requirements
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Install policy engine dependencies
RUN pip install cvxpy ruptures psutil

# Copy application
COPY . .

# Set environment variables
ENV PYTHONUNBUFFERED=1

# Run application
CMD ["python", "-m", "CRCA"]
```

**Example: Docker Compose for Full Stack**

```yaml
version: '3.8'

services:
  crca:
    build: .
    environment:
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    volumes:
      - ./policy_ledger.db:/app/policy_ledger.db
      - ./rollback.db:/app/rollback.db
    ports:
      - "8000:8000"
```

### Verification

**Example 1: Basic Verification**

```python
from dotenv import load_dotenv
from CRCA import CRCAAgent

load_dotenv()

agent = CRCAAgent(
    model_name="gpt-4o-mini",
    causal_edges=[("funding", "failure"), ("engineers", "failure"), ("project", "failure")],
    variables=["funding", "engineers", "project", "failure", "success"],
    max_loops=3,
    agent_max_loops=3,
    use_crca_tools=True,
    enable_batch_predict=True,
)

task = "Why do large software projects collapse despite adequate funding?"
result = agent.run(task)
print(result.get("causal_analysis", ""))
```

**Example 2: Verify Policy Engine**

```python
from schemas.policy import DoctrineV1, EpochConfig, MetricSpec, Objective, RiskBudget
from utils.ledger import Ledger

# Create minimal doctrine
doctrine = DoctrineV1(
    epoch=EpochConfig(unit="hours", length=1.0),
    metrics={"test": MetricSpec(extractor_key="test", unit="unit")},
    objectives=[Objective(metric_name="test", direction="minimize", weight=1.0)],
    invariants=[],
    levers={},
    risk_budget=RiskBudget()
)

# Test ledger
ledger = Ledger(db_path=":memory:")
print("Policy engine components verified!")
```

**Example 3: Verify All Modules**

```python
# Test core
from CRCA import CRCAAgent
agent = CRCAAgent(model_name="gpt-4o-mini")
print("✓ Core CRCA verified")

# Test CRCA-SD
try:
    from crca_sd import StateVector
    print("✓ CRCA-SD verified")
except ImportError:
    print("✗ CRCA-SD not available")

# Test CRCA-CG
try:
    from crca_cg.corposwarm import CorporateSwarm
    print("✓ CRCA-CG verified")
except ImportError:
    print("✗ CRCA-CG not available")

# Test Policy Engine
from schemas.policy import DoctrineV1
from utils.ledger import Ledger
print("✓ Policy Engine verified")
```

---

## Core CRCAAgent

### Introduction

CRCAAgent is the core component of the CR-CA framework, providing a lightweight causal reasoning engine with LLM integration. It implements structural causal models (SCMs) based on Judea Pearl's causal hierarchy, enabling both LLM-based causal analysis and deterministic causal simulation.

### Key Features

- **Automatic Variable Extraction**: Extract variables and causal relationships directly from natural language tasks without manual declaration
- **LLM-Based Causal Reasoning**: Sophisticated multi-loop causal analysis using large language models
- **Deterministic Causal Simulation**: Fast, deterministic state evolution using linear/non-linear SCMs
- **Dual-Mode Operation**: Automatically selects LLM mode (task string) or deterministic mode (state dict)
- **Batch Prediction**: Efficient vectorized predictions for ensemble forecasting
- **Async/Await Support**: Non-blocking operations for concurrent analysis
- **Advanced Analysis Methods**: Optimization, time-series analysis, Bayesian inference, root cause analysis

### Architecture

```mermaid
flowchart TB
    Input[Input: Task or State] --> ModeCheck{Mode Detection}
    
    ModeCheck -->|String Task| LLMMode[LLM Mode]
    ModeCheck -->|State Dict| DetMode[Deterministic Mode]
    
    LLMMode --> EmptyCheck{Graph Empty?}
    EmptyCheck -->|Yes| AutoExtract[Automatic Variable Extraction]
    EmptyCheck -->|No| BuildPrompt[Build Causal Prompt]
    
    AutoExtract --> LLMAnalysis[LLM Causal Analysis]
    BuildPrompt --> LLMAnalysis
    LLMAnalysis --> GenerateCF[Generate Counterfactuals]
    
    DetMode --> Standardize[Standardize to z-space]
    Standardize --> TopoSort[Topological Sort]
    TopoSort --> Predict[SCM Evolution]
    Predict --> DeStandardize[De-standardize]
    DeStandardize --> GenerateCF
    
    GenerateCF --> Output[Output: Analysis + Scenarios]
    
    Graph[Causal Graph] --> LLMMode
    Graph --> DetMode
```

### Automatic Variable Extraction

One of the most powerful features introduced in v1.3.0 is automatic variable extraction. When a task is provided without pre-declared variables, CRCAAgent automatically:

1. **Analyzes the task** using LLM-based natural language understanding
2. **Extracts relevant variables** from the task description
3. **Identifies causal relationships** between variables
4. **Builds the causal graph** automatically
5. **Proceeds with causal analysis** using the extracted graph

This eliminates the need to manually declare variables and edges for most use cases, making the framework significantly more accessible.

**How it works:**

- When `run(task)` is called with a string task and the causal graph is empty, the agent triggers automatic extraction
- The LLM analyzes the task and generates structured JSON with variables and edges
- The extraction handler programmatically adds variables and relationships to the graph
- Subsequent analysis uses the automatically constructed graph

**Example:**

```python
from CRCA import CRCAAgent

# No variables declared - automatic extraction will occur
agent = CRCAAgent(model_name="gpt-4o-mini")

# Task automatically triggers variable extraction
task = "Analyze how increasing product price affects demand and revenue"
result = agent.run(task=task)

# Variables and edges were automatically extracted and used
print(f"Extracted {len(agent.get_nodes())} variables")
print(f"Created {len(agent.get_edges())} causal relationships")
```

### Basic Usage

#### LLM-Based Causal Analysis (Automatic Extraction)

```python
from CRCA import CRCAAgent

agent = CRCAAgent(
    model_name="gpt-4o",
    max_loops=3,
    agent_max_loops=3
)

# Automatic variable extraction + causal analysis
task = "Why do software projects fail despite adequate funding?"
result = agent.run(task=task)

print(result["causal_analysis"])
for scenario in result["counterfactual_scenarios"][:3]:
    print(f"{scenario.name}: {scenario.reasoning}")
```

#### Deterministic Simulation (Traditional Approach)

```python
from CRCA import CRCAAgent

agent = CRCAAgent(variables=["price", "demand", "inventory"])
agent.add_causal_relationship("price", "demand", strength=-0.5)
agent.add_causal_relationship("demand", "inventory", strength=-0.2)

state = {"price": 100.0, "demand": 1000.0, "inventory": 5000.0}
result = agent.run(initial_state=state, max_steps=2)

print(result["evolved_state"])
for scenario in result["counterfactual_scenarios"][:3]:
    print(f"{scenario.name}: {scenario.probability:.2f}")
```

#### Hybrid Approach (LLM Extraction + Deterministic Simulation)

```python
from CRCA import CRCAAgent

agent = CRCAAgent(model_name="gpt-4o-mini")

# Step 1: Automatic extraction via LLM
task = "Analyze pricing strategy for a new product"
extraction_result = agent.run(task=task)

# Step 2: Use extracted graph for deterministic simulation
initial_state = {var: 0.0 for var in agent.get_nodes()}
simulation_result = agent.run(initial_state=initial_state, max_steps=5)

print(f"Simulated {len(simulation_result['counterfactual_scenarios'])} scenarios")
```

### Advanced Features

#### Batch Prediction

```python
agent = CRCAAgent(
    variables=["price", "demand"],
    enable_batch_predict=True,
    max_batch_size=32
)

factual_states = [{"price": 100.0, "demand": 1000.0} for _ in range(100)]
interventions_list = [{"price": 100.0 + i} for i in range(100)]

predictions = agent._predict_outcomes_batch(factual_states, interventions_list)
```

#### Async Operations

```python
import asyncio

async def analyze_async():
    agent = CRCAAgent(variables=["x", "y"])
    result = await agent.run_async(
        initial_state={"x": 1.0, "y": 2.0},
        max_steps=3
    )
    return result

result = asyncio.run(analyze_async())
```

#### Optimization Methods

```python
agent = CRCAAgent(variables=["price", "demand", "revenue"])
agent.add_causal_relationship("price", "demand", strength=-0.5)
agent.add_causal_relationship("price", "revenue", strength=1.0)
agent.add_causal_relationship("demand", "revenue", strength=0.8)

initial_state = {"price": 100.0, "demand": 1000.0, "revenue": 100000.0}

# Gradient-based optimization
opt_result = agent.gradient_based_intervention_optimization(
    initial_state=initial_state,
    target="revenue",
    intervention_vars=["price"],
    constraints={"price": (80.0, 120.0)}
)

print(f"Optimal price: {opt_result['optimal_intervention']['price']}")
print(f"Optimal revenue: {opt_result['optimal_target_value']}")
```

---

## Module Documentation

### Utils Module

The `utils/` module provides essential utilities and extensions for the CR-CA framework.

#### Formatter (`utils/formatter.py`)

Provides markdown output formatting and rich text rendering:

```python
from utils.formatter import Formatter

formatter = Formatter(md=True)
formatter.print_markdown(analysis_text, title="Causal Analysis")
```

#### Agent Types (`utils/Agent_types.py`)

Type definitions for agent-related structures.

#### Conversation (`utils/conversation.py`)

Manages conversation history and message handling for agent interactions.

#### Agent Extensions (`utils/agent.py`)

Extended agent functionality building on the Swarms Agent base class, including streaming output, error handling, and tool management.

#### TUI (`utils/tui.py`)

Terminal user interface components for interactive agent interactions, including CorporateSwarmTUI for multi-agent orchestration.

#### Router (`utils/router.py`)

Task routing and distribution system for multi-agent workflows. Supports multiple swarm architectures including sequential, concurrent, hierarchical, and collaborative patterns.

**Example 1: Sequential Workflow**

```python
from utils.router import SwarmRouter, SwarmRouterConfig

# Create sequential workflow router
config = SwarmRouterConfig(
    name="sequential-processor",
    description="Process tasks sequentially",
    swarm_type="SequentialWorkflow",
    task="Process data pipeline: extract, transform, load"
)

router = SwarmRouter(config=config)
result = router.run()

print(f"Sequential workflow completed: {result}")
```

**Example 2: Concurrent Workflow**

```python
from utils.router import SwarmRouter, SwarmRouterConfig

# Create concurrent workflow for parallel processing
config = SwarmRouterConfig(
    name="parallel-processor",
    description="Process multiple tasks in parallel",
    swarm_type="ConcurrentWorkflow",
    task="Analyze multiple datasets concurrently"
)

router = SwarmRouter(config=config)
result = router.run()

print(f"Concurrent workflow completed: {result}")
```

**Example 3: Hierarchical Swarm**

```python
from utils.router import SwarmRouter, SwarmRouterConfig

# Create hierarchical swarm for complex task decomposition
config = SwarmRouterConfig(
    name="hierarchical-manager",
    description="Decompose complex tasks hierarchically",
    swarm_type="HierarchicalSwarm",
    task="Manage multi-level project with dependencies"
)

router = SwarmRouter(config=config)
result = router.run()

print(f"Hierarchical swarm completed: {result}")
```

**Example 4: Collaborative Group Chat**

```python
from utils.router import SwarmRouter, SwarmRouterConfig

# Create collaborative swarm for group discussion
config = SwarmRouterConfig(
    name="collaborative-team",
    description="Collaborative problem solving",
    swarm_type="GroupChat",
    multi_agent_collab_prompt=True,
    task="Design a new feature through team discussion"
)

router = SwarmRouter(config=config)
result = router.run()

print(f"Collaborative discussion completed: {result}")
```

**Example 5: Auto Swarm Selection**

```python
from utils.router import SwarmRouter, SwarmRouterConfig

# Let router automatically select best swarm type
config = SwarmRouterConfig(
    name="auto-router",
    description="Automatic swarm selection",
    swarm_type="auto",  # Automatic selection
    task="Complex task requiring intelligent routing"
)

router = SwarmRouter(config=config)
result = router.run()

print(f"Auto-selected swarm type: {router.selected_swarm_type}")
print(f"Result: {result}")
```

#### HHCS (`utils/HHCS.py`)

Hybrid Hierarchical-Cluster Swarm implementation for intelligent task routing with hierarchical decision-making.

**Example 6: Basic HHCS Setup**

```python
from utils.HHCS import HybridHierarchicalClusterSwarm
from swarms.structs.agent import Agent

# Create agents for different roles
analyst = Agent(agent_name="analyst", model_name="gpt-4o-mini")
researcher = Agent(agent_name="researcher", model_name="gpt-4o-mini")
writer = Agent(agent_name="writer", model_name="gpt-4o-mini")

# Create HHCS with agents
hhcs = HybridHierarchicalClusterSwarm(
    agents=[analyst, researcher, writer],
    task="Research and write a comprehensive report on causal inference"
)

# Run HHCS
result = hhcs.run()

print(f"HHCS completed: {result}")
```

**Example 7: HHCS with Custom Router**

```python
from utils.HHCS import HybridHierarchicalClusterSwarm
from swarms.structs.agent import Agent

# Create specialized agents
agents = [
    Agent(agent_name="data_analyst", model_name="gpt-4o-mini"),
    Agent(agent_name="domain_expert", model_name="gpt-4o-mini"),
    Agent(agent_name="technical_writer", model_name="gpt-4o-mini")
]

# Create HHCS with custom routing
hhcs = HybridHierarchicalClusterSwarm(
    agents=agents,
    task="Analyze market trends and create investment strategy"
)

result = hhcs.run()
print(f"Task routed to: {hhcs.selected_agent}")
print(f"Result: {result}")
```

#### AOP (`utils/aop.py`)

Agent Orchestration Platform with task queue and MCP server deployment for multi-agent coordination.

**Example 8: Task Queue Management**

```python
from utils.aop import TaskQueue, Task
from swarms.structs.agent import Agent

# Create agent
agent = Agent(agent_name="worker", model_name="gpt-4o-mini")

# Create task queue
queue = TaskQueue(
    agent_name="worker",
    agent=agent,
    max_workers=3,  # 3 parallel workers
    max_queue_size=100
)

# Start workers
queue.start_workers()

# Add tasks
task1 = Task(task="Analyze dataset A", priority=10)
task2 = Task(task="Analyze dataset B", priority=5)
task3 = Task(task="Generate report", priority=8)

queue.add_task(task1)
queue.add_task(task2)
queue.add_task(task3)

# Get task status
task = queue.get_task(task1.task_id)
print(f"Task status: {task.status}")

# Get queue statistics
stats = queue.get_stats()
print(f"Total tasks: {stats.total_tasks}")
print(f"Completed: {stats.completed_tasks}")
print(f"Pending: {stats.pending_tasks}")

# Stop workers when done
queue.stop_workers()
```

**Example 9: MCP Server Deployment**

```python
from utils.aop import AOP, AgentToolConfig
from swarms.structs.agent import Agent

# Create agents
agent1 = Agent(agent_name="analyst", model_name="gpt-4o-mini")
agent2 = Agent(agent_name="researcher", model_name="gpt-4o-mini")

# Configure agent tools
configs = [
    AgentToolConfig(
        agent=agent1,
        tool_name="analyze_data",
        description="Analyze datasets and extract insights"
    ),
    AgentToolConfig(
        agent=agent2,
        tool_name="research_topic",
        description="Research topics and gather information"
    )
]

# Create AOP and deploy as MCP server
aop = AOP(agent_configs=configs)

# Start MCP server
aop.start_mcp_server(port=8000)

print("MCP server started on port 8000")
print(f"Available tools: {aop.get_available_tools()}")
```

**Example 10: AOP with Task Queue and MCP**

```python
from utils.aop import AOP, TaskQueue, AgentToolConfig
from swarms.structs.agent import Agent

# Create agents
agents = [
    Agent(agent_name="worker1", model_name="gpt-4o-mini"),
    Agent(agent_name="worker2", model_name="gpt-4o-mini")
]

# Configure AOP
configs = [
    AgentToolConfig(agent=agents[0], tool_name="process_task"),
    AgentToolConfig(agent=agents[1], tool_name="analyze_data")
]

aop = AOP(agent_configs=configs)

# Deploy as MCP server
aop.start_mcp_server(port=8000)

# Create task queues for each agent
for agent in agents:
    queue = TaskQueue(agent_name=agent.agent_name, agent=agent, max_workers=2)
    queue.start_workers()
    
    # Add tasks
    queue.add_task(Task(task=f"Task for {agent.agent_name}", priority=5))
```

#### Ledger (`utils/ledger.py`)

Event-sourced append-only SQLite ledger for audit trail. See Policy Engine examples above for detailed usage.

**Example 11: Ledger Event Recording**

```python
from utils.ledger import Ledger
from schemas.policy import ObservationEvent, DecisionEvent

ledger = Ledger(db_path="audit_ledger.db")

# Record observation
obs = ObservationEvent(epoch=1, metrics={"cpu": 75.0, "memory": 60.0})
event_id = ledger.append(obs)

# Record decision
dec = DecisionEvent(
    epoch=1,
    interventions=[{"lever_id": "throttle", "params": {"value": 5}}],
    rationale="CPU high, throttling",
    decision_hash="abc123"
)
event_id = ledger.append(dec)

# Query events
events = ledger.get_events_by_epoch(epoch=1)
print(f"Found {len(events)} events in epoch 1")
```

#### Rollback (`utils/rollback.py`)

Rollback system for intervention recovery. See Policy Engine examples above for detailed usage.

**Example 12: Checkpoint and Rollback**

```python
from utils.rollback import RollbackManager

rollback = RollbackManager(db_path="rollback.db")

# Create checkpoint
checkpoint_id = rollback.create_checkpoint(epoch=1, state={"value": 100})

# Execute intervention
# ... intervention code ...

# Rollback if needed
rollback.rollback_n_steps(epoch=1, n_steps=1)
restored = rollback.restore_checkpoint(checkpoint_id)
```

#### Canonical (`utils/canonical.py`)

Canonicalization utilities for deterministic hashing. See Policy Engine examples above for detailed usage.

**Example 13: Stable Hashing**

```python
from utils.canonical import stable_hash

obj = {"a": 1, "b": 2, "c": 3}
hash_value = stable_hash(obj)
print(f"Stable hash: {hash_value}")
```

#### AnyToStr (`utils/AnyToStr.py`)

Type conversion utilities for converting various types to strings.

**Example 14: Type Conversion**

```python
from utils.AnyToStr import any_to_str

# Convert various types to string
result1 = any_to_str(123)
result2 = any_to_str([1, 2, 3])
result3 = any_to_str({"key": "value"})

print(f"Int to string: {result1}")
print(f"List to string: {result2}")
print(f"Dict to string: {result3}")
```

#### Out Types (`utils/out_types.py`)

Output type definitions for history and conversation formatting.

**Example 15: Output Type Usage**

```python
from utils.out_types import OutputType, HistoryOutputType

# Define output format
output_format: OutputType = "json"  # or "dict", "string", "yaml", etc.

# Use in conversation formatting
from utils.conversation import Conversation

conv = Conversation()
history = conv.get_history(output_type=output_format)
```

### Tools Module

#### MCP Client Utils (`tools/mcpClientUtils.py`)

Model Context Protocol (MCP) integration for connecting to MCP servers and executing tools:

- **Connection Management**: Establish and manage MCP server connections
- **Tool Execution**: Execute tools via MCP protocol
- **Protocol Handling**: Transform MCP tools to OpenAI-compatible format
- **Error Handling**: Comprehensive error handling for MCP operations

**Example 1: Basic MCP Connection**

```python
from tools.mcpClientUtils import connect_to_mcp_server, execute_mcp_tool

# Connection configuration
connection_config = {
    "server_url": "http://localhost:8000",
    "api_key": "your_api_key"
}

# Connect to MCP server
session = await connect_to_mcp_server(connection_config)

# Execute tool
result = await execute_mcp_tool(
    session=session,
    tool_name="analyze_data",
    parameters={"dataset": "sales_data.csv"}
)

print(f"Tool execution result: {result}")
```

**Example 2: Transform MCP Tools to OpenAI Format**

```python
from tools.mcpClientUtils import transform_mcp_tool_to_openai_tool

# MCP tool schema
mcp_tool = {
    "name": "process_file",
    "description": "Process a file",
    "parameters": {
        "type": "object",
        "properties": {
            "file_path": {"type": "string"},
            "operation": {"type": "string"}
        }
    }
}

# Transform to OpenAI format
openai_tool = transform_mcp_tool_to_openai_tool(mcp_tool)

print(f"OpenAI tool format: {openai_tool}")
```

**Example 3: Error Handling in MCP Operations**

```python
from tools.mcpClientUtils import connect_to_mcp_server, execute_mcp_tool
from loguru import logger

try:
    session = await connect_to_mcp_server(connection_config)
    result = await execute_mcp_tool(session, "tool_name", {})
except ConnectionError as e:
    logger.error(f"MCP connection failed: {e}")
except Exception as e:
    logger.error(f"Tool execution failed: {e}")
```

#### Actuators (`tools/actuators.py`)

Actuator interfaces for executing interventions on system controls, cloud resources, and services. See Policy Engine examples above for detailed usage.

**Example 4: System Control Actuator**

```python
from tools.actuators import SystemControlActuator
from schemas.policy import InterventionSpec

actuator = SystemControlActuator(require_root=False)

# CPU throttling intervention
intervention = InterventionSpec(
    lever_id="cpu_throttle",
    parameters={"nice_value": 10, "pid": 12345}
)

result = actuator.execute([intervention])
print(f"Intervention result: {result}")
```

**Example 5: Custom Actuator Implementation**

```python
from tools.actuators import BaseActuator
from schemas.policy import InterventionSpec
from typing import List, Dict, Any

class CloudResourceActuator(BaseActuator):
    """Actuator for cloud resource management."""
    
    def execute(self, interventions: List[InterventionSpec]) -> Dict[str, Any]:
        """Execute cloud resource interventions."""
        results = []
        
        for intervention in interventions:
            if intervention.lever_id == "scale_instance":
                instance_id = intervention.parameters["instance_id"]
                instance_type = intervention.parameters["instance_type"]
                
                # Scale cloud instance (pseudo-code)
                result = self._scale_instance(instance_id, instance_type)
                results.append({"intervention": intervention.model_dump(), "result": result})
        
        return {"executed": len(results), "results": results}
    
    def _scale_instance(self, instance_id: str, instance_type: str) -> Dict[str, Any]:
        """Scale cloud instance."""
        # Implementation would call cloud API
        return {"status": "success", "instance_id": instance_id, "type": instance_type}

# Use custom actuator
actuator = CloudResourceActuator()
intervention = InterventionSpec(
    lever_id="scale_instance",
    parameters={"instance_id": "i-12345", "instance_type": "t3.large"}
)
result = actuator.execute([intervention])
```

#### Sensors (`tools/sensors.py`)

Sensor interfaces for reading system metrics, cloud resources, and custom metrics. See Policy Engine examples above for detailed usage.

**Example 6: System Metrics Sensor**

```python
from tools.sensors import SystemMetricsSensor

sensor = SystemMetricsSensor()
snapshot = sensor.read()

print(f"CPU: {snapshot['cpu_percent']:.1f}%")
print(f"Memory: {snapshot['memory_percent']:.1f}%")
```

**Example 7: Custom Sensor with Registry**

```python
from tools.sensors import BaseSensor, SensorRegistry
from typing import Dict

class ApplicationMetricsSensor(BaseSensor):
    """Sensor for application-specific metrics."""
    
    def read(self) -> Dict[str, float]:
        """Read application metrics."""
        # Read from application monitoring endpoint
        return {
            "request_rate": 150.0,
            "error_rate": 2.5,
            "response_time": 95.0
        }

# Register sensor
registry = SensorRegistry()
registry.register("app_metrics", ApplicationMetricsSensor())

# Use registered sensor
sensor = registry.get_sensor("app_metrics")
snapshot = sensor.read()
```

#### Code Generator (`tools/code_generator.py`)

Code generation tools with LLM and ML integration for generating, analyzing, and optimizing code.

**Example 8: Basic Code Generation**

```python
from tools.code_generator import LLMCodeGenerator
from swarms.structs.agent import Agent

# Create agent for code generation
agent = Agent(agent_name="code_gen", model_name="gpt-4o-mini")

# Initialize code generator
generator = LLMCodeGenerator(agent=agent)

# Generate code from specification
specification = "Create a Python function that calculates factorial recursively"
code = generator.generate(
    specification=specification,
    language="python"
)

print(f"Generated code:\n{code}")

# Validate generated code
validation = generator.validate(code, language="python")
print(f"Validation result: {validation}")
```

**Example 9: Language-Specific Code Generation**

```python
from tools.code_generator import LLMCodeGenerator

generator = LLMCodeGenerator(agent=agent)

# Generate JavaScript code
js_code = generator.generate(
    specification="Create a React component for a todo list",
    language="javascript"
)

# Generate Python code
python_code = generator.generate(
    specification="Create a data processing pipeline",
    language="python"
)

# Generate TypeScript code
ts_code = generator.generate(
    specification="Create a type-safe API client",
    language="typescript"
)
```

**Example 10: Code Optimization with ML**

```python
from tools.code_generator import MLCodeOptimizer

# Initialize ML-based optimizer
optimizer = MLCodeOptimizer()

# Optimize code
original_code = """
def factorial(n):
    if n <= 1:
        return 1
    return n * factorial(n - 1)
"""

optimized_code = optimizer.optimize(
    code=original_code,
    language="python",
    optimization_target="performance"
)

print(f"Optimized code:\n{optimized_code}")
```

#### Mandate Generator (`tools/mandate_generator.py`)

Convert CorporateProposal to bolt.diy execution mandate for autonomous code generation.

**Example 11: Proposal to Mandate Conversion**

```python
from tools.mandate_generator import proposal_to_mandate
from crca_cg.corposwarm import CorporateProposal, ProposalType

# Create corporate proposal
proposal = CorporateProposal(
    proposal_id="prop-001",
    title="Implement new feature",
    description="Add user authentication system",
    proposal_type=ProposalType.TECHNICAL,
    budget_impact=50000.0,
    risk_level=0.3
)

# Convert to execution mandate
mandate = proposal_to_mandate(
    proposal=proposal,
    mandate_id="mandate-001"
)

print(f"Mandate ID: {mandate['mandate_id']}")
print(f"Objectives: {mandate['objectives']}")
print(f"Budget: {mandate['budget']}")
print(f"Constraints: {mandate['constraints']}")
```

**Example 12: Complete Workflow with Mandate Generation**

```python
from tools.mandate_generator import proposal_to_mandate
from crca_cg.corposwarm import CorporateSwarm, CorporateProposal

# Create corporate swarm
swarm = CorporateSwarm()

# Create and submit proposal
proposal = swarm.create_proposal(
    title="Build API gateway",
    description="Implement microservices API gateway",
    proposal_type=ProposalType.TECHNICAL
)

# Vote on proposal
vote_result = swarm.board.vote_on_proposal(proposal)

if vote_result.passed:
    # Convert approved proposal to mandate
    mandate = proposal_to_mandate(
        proposal=proposal,
        corporate_swarm=swarm
    )
    
    # Execute mandate (would integrate with bolt.diy)
    print(f"Mandate created: {mandate['mandate_id']}")
    print(f"Ready for execution")
```

#### File Operations (`tools/file_operations.py`)

File manipulation tools with LLM integration for intelligent file operations.

**Example 13: Basic File Operations**

```python
from tools.file_operations import FileWriter, FileReader

# Write file
writer = FileWriter()
result = writer.execute(
    operation="write",
    file_path="output.txt",
    content="Hello, World!"
)

# Read file
reader = FileReader()
result = reader.execute(
    operation="read",
    file_path="output.txt"
)

print(f"File content: {result['content']}")
```

**Example 14: LLM-Powered File Operations**

```python
from tools.file_operations import LLMFileOperator
from swarms.structs.agent import Agent

agent = Agent(agent_name="file_ops", model_name="gpt-4o-mini")
operator = LLMFileOperator(agent=agent)

# Generate file content using LLM
result = operator.generate_file(
    specification="Create a Python class for handling user authentication",
    file_path="auth.py",
    language="python"
)

print(f"Generated file: {result['file_path']}")
print(f"Content length: {len(result['content'])} characters")
```

**Example 15: File Analysis and Optimization**

```python
from tools.file_operations import FileAnalyzer

analyzer = FileAnalyzer()

# Analyze file
analysis = analyzer.analyze_file(
    file_path="code.py",
    analysis_type="complexity"  # or "performance", "security", etc.
)

print(f"Complexity score: {analysis['complexity_score']}")
print(f"Issues found: {analysis['issues']}")
print(f"Suggestions: {analysis['suggestions']}")
```

### Templates Module

The templates module provides a framework for creating specialized agents with drag-and-drop feature composition.

#### Base Specialized Agent (`templates/base_specialized_agent.py`)

Base class for creating domain-specific agents with common initialization patterns and memory management.

#### Feature Mixins (`templates/feature_mixins.py`)

Modular feature components that can be mixed into agent classes:

- **GraphFeatureMixin**: Graph operations (nodes, edges, topological sort)
- **PredictionFeatureMixin**: Prediction methods and standardization
- **StatisticsFeatureMixin**: Statistical analysis and data fitting
- **LLMFeatureMixin**: LLM integration patterns
- **FullFeatureMixin**: All features combined

**Usage:**

```python
from templates.base_specialized_agent import BaseSpecializedAgent
from templates.feature_mixins import GraphFeatureMixin, PredictionFeatureMixin

class MyAgent(BaseSpecializedAgent, GraphFeatureMixin, PredictionFeatureMixin):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        self.init_graph_feature(variables=["A", "B", "C"])
        self.init_prediction_feature(use_nonlinear=True)
```

#### Module Registry (`templates/module_registry.py`)

Feature composition system allowing dynamic agent creation:

```python
from templates.module_registry import compose_agent

agent = compose_agent(
    features=["graph", "prediction", "statistics"],
    variables=["x", "y", "z"]
)
```

#### Graph Management (`templates/graph_management.py`)

Graph operations including path finding, cycle detection, and topological sorting.

**Example 1: Graph Operations**

```python
from templates.graph_management import GraphManager

# Create graph manager
graph = GraphManager()
graph.add_node("A")
graph.add_node("B")
graph.add_node("C")
graph.add_edge("A", "B")
graph.add_edge("B", "C")

# Topological sort
topo_order = graph.topological_sort()
print(f"Topological order: {topo_order}")

# Find paths
paths = graph.find_paths("A", "C")
print(f"Paths from A to C: {paths}")

# Check for cycles
has_cycle = graph.has_cycle()
print(f"Has cycle: {has_cycle}")
```

#### Prediction Framework (`templates/prediction_framework.py`)

Prediction methods, standardization, and counterfactual generation patterns.

**Example 2: Prediction with Standardization**

```python
from templates.prediction_framework import PredictionMixin

class MyAgent(PredictionMixin):
    def __init__(self):
        self.init_prediction_feature(use_nonlinear=True)

agent = MyAgent()

# Standardize state
state = {"x": 100.0, "y": 200.0}
standardized = agent.standardize_state(state)

# Predict outcome
prediction = agent.predict_outcome(
    state=state,
    intervention={"x": 110.0}
)

print(f"Predicted outcome: {prediction}")
```

#### Statistical Methods (`templates/statistical_methods.py`)

Statistical analysis including data fitting, uncertainty quantification, and time-series methods.

**Example 3: Statistical Analysis**

```python
from templates.statistical_methods import StatisticsMixin
import numpy as np

class StatsAgent(StatisticsMixin):
    def __init__(self):
        self.init_statistics_feature()

agent = StatsAgent()

# Fit data to model
data = np.array([[1, 2], [2, 4], [3, 6], [4, 8]])
fitted_model = agent.fit_data(data, model_type="linear")

# Quantify uncertainty
uncertainty = agent.quantify_uncertainty(data, fitted_model)
print(f"Uncertainty: {uncertainty}")
```

#### LLM Integration (`templates/llm_integration.py`)

LLM integration patterns for schema definition, prompt building, and multi-loop reasoning.

**Example 4: LLM Integration**

```python
from templates.llm_integration import LLMFeatureMixin

class LLMAgent(LLMFeatureMixin):
    def __init__(self):
        self.init_llm_feature(model_name="gpt-4o-mini")

agent = LLMAgent()

# Build prompt
prompt = agent.build_prompt(
    task="Analyze causal relationships",
    context={"variables": ["A", "B", "C"]}
)

# Execute with multi-loop reasoning
result = agent.execute_with_loops(
    prompt=prompt,
    max_loops=3
)
```

#### Policy Loop Integration (`templates/policy_loop.py`)

Policy loop mixin for temporal policy execution. See Policy Engine example