coreason-human-layer 0.1.0

coreason-human-layer

coreason-human-layer

Domain: HITL, Durable Execution, & RLHF Data Capture Role: The "Switchboard" & "Labeling Factory"

coreason-human-layer is the dedicated bridge between autonomous execution and human oversight, operationalizing Online Alignment through a Multiverse Branching Strategy. Instead of a simple "Stop/Go" model, it allows humans to fork agent reality, capturing decisions as DPO Triplets to fuel the model improvement loop.

Features

1. The Stasis Engine (Durable Execution)

Implements an event-sourcing pattern for durable execution.

• Snapshot: Stores a pointer to the last event ID rather than a massive state blob.
• Rehydration: Replays events to rebuild context.
• Storage: Hot storage in Redis Streams, cold archival in S3.

2. The Branch Manager (Cognitive Forking)

Manages the tree topology of execution branches.

• Forking: create_fork API allows instantiating new execution paths with human overrides.
• Prefix Caching: Optimizes costs and latency by reusing the KV Cache of the shared prefix (cached_prefix_id) on the LLM backend.
• Topology: Exposes the conversation tree structure for visualization.

3. The Super-Prompt Injector (System Injection)

Ensures the model obeys human overrides.

• Frame Injection: Injects a high-priority [SYSTEM PRIORITY INTERRUPT] block.
• Variable Patching: Parses and hard-updates variables from human overrides (e.g., "Dose = 50mg").

4. The Learning Bridge (Auto-DPO)

Converts operational actions into training assets.

• Trigger: Merging a fork (approving it) while abandoning another.
• Auto-DPO: Automatically constructs a DPO object ({ prompt, chosen, rejected }) and pushes it to coreason-synthesis.

Installation

pip install coreason-human-layer

Usage

from uuid import uuid4
from coreason_human_layer.branch_manager import BranchManager
from coreason_human_layer.stasis import InMemoryStasisEngine

# Initialize engines
stasis = InMemoryStasisEngine()
manager = BranchManager(stasis_engine=stasis)

# Create a root branch
root_id = uuid4()
branch_id = manager.create_fork(
    parent_branch_id=None,
    parent_event_id=None,
    root_id=root_id,
    human_override_text="Initial Start",
)

# ... (Agent runs and events are appended to stasis) ...

# Create a Human Override Fork
fork_id = manager.create_fork(
    parent_branch_id=branch_id,
    parent_event_id=None, # or specific event ID
    root_id=root_id,
    human_override_text="Don't be so formal. Use a friendly tone.",
)

# ... (New branch runs) ...

# Merge the successful fork (triggers Feedback Signal)
feedback = manager.merge_branches(winning_branch_id=fork_id, losing_branch_id=branch_id)
print(f"Feedback Signal Sent: {feedback.timestamp}")

License

Copyright (c) 2025 CoReason, Inc. Licensed under the Prosperity Public License 3.0.