Maieutic v1: Immutable Conversation State Store

A Redux-like state management system for LLM conversations and prompt engineering. Provides deterministic, time-travel-capable state management with immutability guarantees.

Features

✅ Implemented (MVP - Feature 1)

• Immutable State Store: Single source of truth for all LLM interactions
• Redux-Style Actions: Type-safe action creators and dispatcher
• LLM Integration: Ollama/LangChain with automatic state management
• Time-Travel Debugging: Complete history with snapshot-based rollback
• Branching: Explore alternative conversation paths
• Observer Pattern: Subscribe to state changes for reactive updates (middleware support)
• Visual Debugging: ASCII tree, Mermaid diagrams, HTML export for state visualization
• Type-Safe Schemas: Strongly typed state, messages, context, and metadata

Architecture

ConversationStore
├── ConversationState (immutable)
│   ├── messages: tuple[Message, ...]
│   ├── context: Context
│   ├── metadata: Metadata
│   └── version: int
├── Snapshots (time-travel)
├── Branches (alternative paths)
└── Subscribers (reactivity)

See ARCHITECTURE.md for detailed design specifications.

Installation

# Install with Poetry
poetry install

# (Optional) Install development dependencies
poetry install --with dev

Quick Start

Option 1: With LLM Integration (Recommended)

from maieutic_v1 import OllamaLLMService

# Initialize service (includes Redux store + dispatcher)
service = OllamaLLMService(model="llama2")

# Chat with automatic state management
service.add_system_message("You are a helpful assistant")
response = service.chat("What is Python?")

# Redux-style time-travel
service.dispatcher.dispatch(rollback_steps(2))

# Create branch for A/B testing
branch_id = service.create_conversation_branch("experiment_A")

Option 2: Pure State Management

from maieutic_v1 import ConversationStore, add_message, MessageRole, Dispatcher

# Initialize store + dispatcher
store = ConversationStore()
dispatcher = Dispatcher(store)

# Dispatch Redux-style actions
action = add_message(MessageRole.USER, "Hello!")
dispatcher.dispatch(action)

# Time-travel
dispatcher.dispatch(rollback_steps(2))

# Subscribe to changes (middleware pattern)
unsubscribe = store.subscribe(lambda state: print(f"State: v{state.version}"))

Usage Examples

Basic Conversation Flow

from maieutic_v1.state import Message, MessageRole, Context, Metadata
from maieutic_v1.store import ConversationStore

store = ConversationStore()

# Add system prompt
system = Message(role=MessageRole.SYSTEM, content="You are helpful.")
store.update_state(store.get_state().with_message(system))

# Add user message
user = Message(role=MessageRole.USER, content="What is Python?")
store.update_state(store.get_state().with_message(user))

# Add assistant response
assistant = Message(role=MessageRole.ASSISTANT, content="Python is a programming language.")
store.update_state(store.get_state().with_message(assistant))

print(f"Conversation has {len(store.get_state().messages)} messages")

Time-Travel Debugging

# Get snapshot history
history = store.get_history()
print(f"Total snapshots: {len(history)}")

# Rollback to specific snapshot
snapshot_id = history[2].snapshot_id
store.rollback_to_snapshot(snapshot_id)

# Or rollback N steps
store.rollback_steps(3)

# Or rollback to timestamp
from datetime import datetime, timedelta
target_time = datetime.utcnow() - timedelta(minutes=5)
store.rollback_to_timestamp(target_time)

Branching for A/B Testing

# Create branch at current state
branch_id = store.create_branch("high_temperature")

# Make changes in this branch
msg = Message(role=MessageRole.ASSISTANT, content="Alternative response")
store.update_state(store.get_state().with_message(msg))

# Switch back to main branch
main_snapshot = store.get_history()[5].snapshot_id
store.rollback_to_snapshot(main_snapshot)

# Switch to branch later
store.switch_to_branch(branch_id)

Reactive Updates

# Subscribe to all state changes
def log_changes(state):
    print(f"State v{state.version}: {len(state.messages)} messages")

unsubscribe = store.subscribe(log_changes)

# Make updates (subscriber will be notified)
msg = Message(role=MessageRole.USER, content="Test")
store.update_state(store.get_state().with_message(msg))

# Unsubscribe when done
unsubscribe()

Context Management

from maieutic_v1.state import Context

# Update conversation context
ctx = Context(
    data={"user_id": "123", "session": "abc"},
    version=1,
    token_count=150
)
new_state = store.get_state().with_context(ctx)
store.update_state(new_state)

Visual Debugging

from maieutic_v1 import ConversationVisualizer

# Initialize visualizer
visualizer = ConversationVisualizer(store)

# ASCII tree of conversation state
print(visualizer.render_ascii_tree())

# Mermaid diagram for documentation
print(visualizer.render_mermaid_diagram())

# Message flow visualization
print(visualizer.render_message_flow())

# Export to interactive HTML
visualizer.export_to_html()

# Statistics and analytics
print(visualizer.render_stats())

Running Tests

# Run tests with Poetry
poetry run pytest tests/ -v

# Run tests with coverage report
poetry run pytest tests/ --cov=maieutic_v1

Running Examples

# Basic state management demo
python examples/basic_usage.py

# Ollama LLM integration
python examples/ollama_integration.py

# Redux-style integration (recommended)
python examples/redux_integration.py

# Visual debugging demo
python examples/visualizer_demo.py

See REDUX_INTEGRATION.md for detailed architecture documentation.

API Reference

Core Classes

ConversationStore

Main store for managing conversation state.

Methods:

• get_state() -> ConversationState: Get current immutable state
• update_state(state, create_snapshot=True, snapshot_label=None): Update state
• subscribe(listener) -> Callable: Subscribe to state changes
• get_history() -> List[StateSnapshot]: Get all snapshots
• rollback_to_snapshot(snapshot_id) -> bool: Rollback to snapshot
• rollback_steps(n) -> bool: Rollback N steps
• create_branch(name, from_snapshot_id=None) -> UUID: Create branch
• switch_to_branch(branch_id) -> bool: Switch to branch

ConversationState

Immutable root state container.

Fields:

• messages: tuple[Message, ...]: Message sequence
• context: Context: Conversation context
• metadata: Metadata: Conversation metadata
• version: int: State version counter
• id: UUID: Conversation identifier

Methods:

• with_message(msg) -> ConversationState: Add message
• with_context(ctx) -> ConversationState: Update context
• with_metadata(meta) -> ConversationState: Update metadata

Message

Immutable message representation.

Fields:

• role: MessageRole: system | user | assistant | function | tool
• content: str: Message content
• id: UUID: Unique identifier
• timestamp: datetime: Creation time
• metadata: Dict[str, Any]: Extensible metadata

Roadmap

Completed ✅

• Immutable state store
• Time-travel debugging
• Snapshot system
• Branching
• Observer pattern
• Comprehensive test suite
• Redux-style action system
• Dispatcher with action handlers
• LLM service with action dispatch
• Deterministic replayer with action dispatch
• Conversation visualizer

v2.2 Milestones (In Progress) 🚧

• Milestone 1: Deterministic Replay Core (✅ Complete + Redux refactored)
• Milestone 2: Context Window Management
• Milestone 3: Response Caching (middleware-based)
• Milestone 4: Divergence Analysis (semantic similarity)

Future Features 🔮

• Middleware pipeline (logging, token counting, persistence)
• Selectors with memoization
• Async actions (thunks)
• Persistence layer
• Redux DevTools integration

Examples

Complete Workflows

See the examples/ directory for complete demonstrations:

• replayer_demo.py: Deterministic replay with temperature enforcement, model validation, and what-if scenarios
• visualizer_demo.py: Conversation visualization with ASCII trees, Mermaid diagrams, and HTML export
• replayer_with_visualizer.py: Integrated workflow combining replayer and visualizer for debugging
• branching_replayer_with_visualizer.py: 🆕 Advanced branching workflow - combines explicit branch creation with sequential replays to build complex conversation trees. See BRANCHING_WORKFLOW.md for detailed guide.
• replayer_real_llm.py: Real-world example with actual Ollama LLM calls

All examples use real LLM (deepseek-r1:8b) for live testing.

Run examples:

# First, start Ollama and pull the model
ollama serve
ollama pull deepseek-r1:8b

# Then run any example
poetry run python examples/replayer_demo.py
poetry run python examples/replayer_real_llm.py
poetry run python examples/replayer_with_visualizer.py
poetry run python examples/visualizer_demo.py

Note: Unit tests use mocks and don't require Ollama.

Design Principles

1. Immutability: All state is immutable by default
2. Predictability: State changes are explicit and traceable
3. Time-Travel: Complete history with rollback capabilities
4. Composability: Modular design for extensibility
5. Type Safety: Strong typing throughout

Contributing

This is a proof-of-concept implementation. Contributions welcome!

License

MIT