GitHub - clay-good/worldify: Worldify is a persistent fact store for AI world simulations.

# From source
git clone https://github.com/clay-good/worldify.git
cd worldify
pip install -e .

from worldify import World

with World("my_world.db") as world:
    # Create entities (things in your world)
    alice = world.create_entity("person", "Alice")
    office = world.create_entity("place", "Main Office")

    # Assert facts about entities
    world.assert_fact(alice.id, "occupation", value="Engineer")
    world.assert_fact(alice.id, "works_at", entity_ref=office.id)

    # Query current facts
    for fact in world.get_current_facts(alice.id):
        print(f"{fact.predicate}: {fact.object_value or fact.object_entity_id}")

from worldify import World

class ChatbotMemory:
    def __init__(self, db_path="chatbot_memory.db"):
        self.world = World(db_path)
        self.world.__enter__()

    def remember_user(self, user_id, key, value):
        """Store a fact about a user."""
        # Find or create the user entity
        users = self.world.find_entities(name_pattern=user_id)
        if users:
            user = users[0]
        else:
            user = self.world.create_entity("user", user_id)

        # Store the fact with auto_supersede to update existing values
        self.world.assert_fact(
            user.id,
            key,
            value=value,
            source="conversation",
            multi_value=(key in ["interest", "topic"])  # Allow multiple interests
        )

    def recall(self, user_id, key=None):
        """Retrieve facts about a user."""
        users = self.world.find_entities(name_pattern=user_id)
        if not users:
            return None

        facts = self.world.get_current_facts(users[0].id)
        if key:
            return next((f.object_value for f in facts if f.predicate == key), None)
        return {f.predicate: f.object_value for f in facts}

    def close(self):
        self.world.__exit__(None, None, None)

# Usage
memory = ChatbotMemory()
memory.remember_user("user_123", "name", "Alice")
memory.remember_user("user_123", "preferred_language", "Python")
memory.remember_user("user_123", "interest", "machine learning")
memory.remember_user("user_123", "interest", "web development")

print(memory.recall("user_123"))
# {'name': 'Alice', 'preferred_language': 'Python', 'interest': 'machine learning', ...}

memory.close()

from worldify import World

class GameState:
    def __init__(self, save_file="game.db"):
        self.world = World(save_file)
        self.world.__enter__()
        self.player = None

    def new_game(self, player_name):
        """Start a new game."""
        self.world.clear(confirm=True)
        self.player = self.world.create_entity("player", player_name)
        self.world.assert_typed_fact(self.player.id, "health", 100)
        self.world.assert_typed_fact(self.player.id, "gold", 0)
        self.world.assert_typed_fact(self.player.id, "level", 1)

    def get_stat(self, stat_name):
        """Get a player stat."""
        facts = self.world.get_current_facts(self.player.id)
        for f in facts:
            if f.predicate == stat_name:
                return f.get_typed_value()  # Returns int, not string
        return None

    def update_stat(self, stat_name, new_value, reason=None):
        """Update a player stat, keeping history."""
        # Find current fact
        facts = self.world.query_facts() \
            .for_subject(self.player.id) \
            .with_predicate(stat_name) \
            .execute(self.world.db)

        if facts:
            # Supersede the old value (keeps history)
            self.world.supersede_fact(
                facts[0].id,
                new_value=f"int:{new_value}",  # Typed encoding
                reason=reason
            )
        else:
            self.world.assert_typed_fact(self.player.id, stat_name, new_value)

    def save_checkpoint(self, name):
        """Save current state."""
        return self.world.save_snapshot(name, f"Checkpoint at level {self.get_stat('level')}")

    def load_checkpoint(self, snapshot_id):
        """Restore to a previous state."""
        self.world.restore_snapshot(snapshot_id)
        # Re-find player after restore
        players = self.world.find_entities(entity_type="player")
        if players:
            self.player = players[0]

    def close(self):
        self.world.__exit__(None, None, None)

# Usage
game = GameState()
game.new_game("Hero")

game.update_stat("gold", 50, reason="Found treasure chest")
game.update_stat("health", 80, reason="Took damage from goblin")

checkpoint = game.save_checkpoint("before_boss")

# Attempt boss fight (fails)
game.update_stat("health", 0, reason="Defeated by dragon")

# Player wants to retry - restore checkpoint
game.load_checkpoint(checkpoint.id)
print(f"Health after restore: {game.get_stat('health')}")  # 80

game.close()

from worldify import World

class AgentKnowledge:
    def __init__(self, db_path=":memory:"):
        self.world = World(db_path)
        self.world.__enter__()

    def learn(self, subject, predicate, obj, confidence=1.0, source="observation"):
        """Learn a new fact."""
        # Get or create subject entity
        subjects = self.world.find_entities(name_pattern=subject)
        if subjects:
            subj_entity = subjects[0]
        else:
            subj_entity = self.world.create_entity("concept", subject)

        # Check if object is another entity
        objects = self.world.find_entities(name_pattern=obj)

        if objects:
            # Relationship between entities
            self.world.assert_fact(
                subj_entity.id, predicate,
                entity_ref=objects[0].id,
                confidence=confidence,
                source=source
            )
        else:
            # Just a value
            self.world.assert_fact(
                subj_entity.id, predicate,
                value=obj,
                confidence=confidence,
                source=source
            )

    def query(self, subject, predicate=None):
        """Query what we know about a subject."""
        subjects = self.world.find_entities(name_pattern=subject)
        if not subjects:
            return []

        q = self.world.query_facts().for_subject(subjects[0].id)
        if predicate:
            q = q.with_predicate(predicate)

        results = []
        for fact in q.execute(self.world.db):
            if fact.object_entity_id:
                obj_entity = self.world.get_entity(fact.object_entity_id)
                obj = obj_entity.name if obj_entity else fact.object_entity_id
            else:
                obj = fact.object_value

            results.append({
                "predicate": fact.predicate,
                "object": obj,
                "confidence": fact.confidence,
                "source": fact.source
            })
        return results

    def close(self):
        self.world.__exit__(None, None, None)

# Usage
kb = AgentKnowledge()

# Agent learns facts from different sources
kb.learn("Python", "is_a", "programming language", source="wikipedia")
kb.learn("Python", "created_by", "Guido van Rossum", source="wikipedia")
kb.learn("Python", "good_for", "data science", confidence=0.9, source="observation")
kb.learn("Python", "good_for", "web development", confidence=0.85, source="observation")

# Query what we know
print(kb.query("Python"))
# [{'predicate': 'is_a', 'object': 'programming language', ...}, ...]

print(kb.query("Python", "good_for"))
# [{'predicate': 'good_for', 'object': 'data science', 'confidence': 0.9}, ...]

kb.close()

entity = world.create_entity("person", "Alice", metadata={"age": 28})

# Value fact
world.assert_fact(person.id, "mood", value="happy")

# Relationship fact (linking two entities)
world.assert_fact(person.id, "works_at", entity_ref=company.id)

# Fact with confidence
world.assert_fact(person.id, "location", value="office", confidence=0.7)

# Query facts valid at a specific time
past_facts = world.query_facts() \
    .for_subject(person.id) \
    .valid_at("2024-01-15T00:00:00Z") \
    .execute(world.db)

# Fact caused by an event
world.assert_fact(
    person.id, "mood", value="happy",
    cause_event="Received promotion"
)

# Trace why something is true
print(world.explain_fact(fact.id))

results = world.query_facts() \
    .for_subject(entity_id) \        # Filter by subject entity
    .for_subject_type("person") \    # Filter by subject entity type
    .with_predicate("works_at") \    # Exact predicate match
    .with_predicate_like("skill_%") \ # Pattern match (SQL LIKE)
    .valid_at(timestamp) \           # Valid at specific time
    .valid_between(start, end) \     # Valid during time range
    .with_min_confidence(0.8) \      # Minimum confidence threshold
    .from_source("observation") \    # Filter by source
    .include_history() \             # Include superseded facts
    .limit(10) \                     # Limit results
    .offset(5) \                     # Skip first N results
    .execute(world.db)               # Execute and get QueryResult

# QueryResult methods
results.as_dict()           # Convert to list of dictionaries
results.as_json()           # Convert to JSON string
results.group_by_predicate() # Group facts by predicate
results.group_by_subject()  # Group facts by subject
results.count(world.db)     # Get count without fetching
results.first(world.db)     # Get first result or None

# Initialize a new database
worldify init myworld.db

# Entity commands
worldify entity create myworld.db --type person --name "Alice"
worldify entity get myworld.db <entity_id>
worldify entity find myworld.db --type person
worldify entity delete myworld.db <entity_id> --cascade

# Fact commands
worldify fact assert myworld.db --subject <id> --predicate occupation --value Engineer
worldify fact get myworld.db <fact_id>
worldify fact query myworld.db --subject <id> --predicate occupation
worldify fact retract myworld.db <fact_id> --reason "No longer valid"

# Query (shorthand)
worldify query myworld.db subject=<id> predicate=occupation

# Snapshot commands
worldify snapshot create myworld.db --name "checkpoint_1"
worldify snapshot list myworld.db
worldify snapshot export myworld.db <snapshot_id> backup.json
worldify snapshot import myworld.db backup.json --mode merge

# Other
worldify summary myworld.db
worldify explain myworld.db <fact_id>

# Output formats
worldify entity find myworld.db --format json  # Also: table, csv

worldify/
    worldify/           # Main package
        __init__.py     # Public API exports
        world.py        # World class (main interface)
        entity.py       # Entity management
        fact.py         # Fact management
        query.py        # Query builder
        causal.py       # Causal chain tracking
        snapshot.py     # Export/import functionality
        session.py      # Agent session tracking
        database.py     # SQLite connection management
        schema.py       # Database schema definition
        exceptions.py   # Custom exception classes
        utils.py        # Utility functions
        aggregates.py   # Aggregate query functions
        cli.py          # Command-line interface
    tests/              # Test suite (84 tests)
    docs/               # Detailed documentation
    examples/           # Working example scripts

python examples/01_basic_usage.py

# Run all tests
python -m unittest discover tests

# Run with coverage
python scripts/run_tests.py --coverage

# Run specific test file
python -m unittest tests.test_core