Simulation world building constructs rich digital environments populated by AI agents for the purpose of prediction and analysis. Inspired by approaches like MiroFish, this skill covers the end-to-end process: designing agent personalities, building memory systems (like those powered by Zep Cloud), constructing dual-platform simulation environments, processing seed material into scenarios, injecting events, and exploring outcomes interactively. The goal is to create simulated worlds realistic enough that their emergent behaviors can inform real-world predictions.

## Key Points

1. Key entities (people, organizations, technologies)
2. Main narrative threads
3. Points of contention or uncertainty
4. Potential future developments
5. Stakeholder positions
1. Agent personality design using psychologically validated models (Big Five) produces more realistic behavioral diversity than random attribute generation
2. Multi-layer memory (working, episodic, semantic, procedural) gives agents consistent behavior over time; Zep Cloud provides production-grade persistent memory
3. Dual-platform simulation (microblog + forum) captures both rapid viral dynamics and deeper deliberative reasoning
4. Seed material processing converts real-world news and data into structured simulation scenarios with event schedules
5. Scenario injection lets you model "what if" events and observe emergent reactions across the agent population
6. Population demographics should mirror real-world distributions: mostly followers and lurkers, few influencers and contrarians
7. Agent mood and energy dynamics prevent unrealistic constant engagement; fatigue and emotional contagion drive realistic behavior

Simulation World Building

Overview

Simulation world building constructs rich digital environments populated by AI agents for the purpose of prediction and analysis. Inspired by approaches like MiroFish, this skill covers the end-to-end process: designing agent personalities, building memory systems (like those powered by Zep Cloud), constructing dual-platform simulation environments, processing seed material into scenarios, injecting events, and exploring outcomes interactively. The goal is to create simulated worlds realistic enough that their emergent behaviors can inform real-world predictions.

Agent Personality Design

Creating Psychologically Realistic Agents

The quality of simulation predictions depends directly on the realism of agent personalities. Agents need consistent behavioral patterns, not just random attributes.

import numpy as np
import uuid
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class BigFivePersonality:
    """
    The Big Five (OCEAN) personality model provides a psychologically
    validated framework for agent personality design.
    """
    openness: float = 0.5         # Curiosity, creativity, openness to experience
    conscientiousness: float = 0.5 # Organization, dependability, self-discipline
    extraversion: float = 0.5     # Sociability, assertiveness, positive emotion
    agreeableness: float = 0.5    # Cooperation, trust, altruism
    neuroticism: float = 0.5      # Emotional instability, anxiety, moodiness

    def behavioral_implications(self) -> dict:
        """Translate personality to behavioral tendencies."""
        return {
            'information_seeking': 0.3 + 0.7 * self.openness,
            'consistency_of_views': 0.3 + 0.7 * self.conscientiousness,
            'social_posting_frequency': 0.1 + 0.9 * self.extraversion,
            'conflict_avoidance': 0.2 + 0.8 * self.agreeableness,
            'emotional_reactivity': 0.1 + 0.9 * self.neuroticism,
            'opinion_change_speed': 0.4 + 0.4 * self.openness - 0.2 * self.conscientiousness,
            'influence_susceptibility': 0.3 + 0.4 * self.agreeableness - 0.3 * self.openness,
            'content_creation_rate': 0.1 + 0.5 * self.extraversion + 0.3 * self.openness,
            'fact_checking_tendency': 0.2 + 0.6 * self.conscientiousness + 0.2 * self.openness
        }

    @classmethod
    def from_archetype(cls, archetype: str) -> 'BigFivePersonality':
        """Generate personality from common archetypes."""
        archetypes = {
            'analyst': cls(openness=0.7, conscientiousness=0.9, extraversion=0.3,
                          agreeableness=0.4, neuroticism=0.3),
            'enthusiast': cls(openness=0.9, conscientiousness=0.4, extraversion=0.9,
                             agreeableness=0.7, neuroticism=0.3),
            'skeptic': cls(openness=0.5, conscientiousness=0.7, extraversion=0.4,
                          agreeableness=0.2, neuroticism=0.5),
            'follower': cls(openness=0.3, conscientiousness=0.5, extraversion=0.5,
                           agreeableness=0.8, neuroticism=0.6),
            'contrarian': cls(openness=0.7, conscientiousness=0.5, extraversion=0.6,
                             agreeableness=0.2, neuroticism=0.4),
            'influencer': cls(openness=0.7, conscientiousness=0.6, extraversion=0.9,
                             agreeableness=0.6, neuroticism=0.3),
            'lurker': cls(openness=0.4, conscientiousness=0.5, extraversion=0.1,
                         agreeableness=0.6, neuroticism=0.6),
            'troll': cls(openness=0.4, conscientiousness=0.2, extraversion=0.7,
                        agreeableness=0.1, neuroticism=0.7)
        }
        base = archetypes.get(archetype, cls())
        # Add noise for uniqueness
        return cls(
            openness=np.clip(base.openness + np.random.normal(0, 0.08), 0, 1),
            conscientiousness=np.clip(base.conscientiousness + np.random.normal(0, 0.08), 0, 1),
            extraversion=np.clip(base.extraversion + np.random.normal(0, 0.08), 0, 1),
            agreeableness=np.clip(base.agreeableness + np.random.normal(0, 0.08), 0, 1),
            neuroticism=np.clip(base.neuroticism + np.random.normal(0, 0.08), 0, 1)
        )


@dataclass
class AgentProfile:
    """Complete agent profile for simulation."""
    id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    name: str = ""
    personality: BigFivePersonality = field(default_factory=BigFivePersonality)
    background: str = ""
    expertise: list = field(default_factory=list)
    values: list = field(default_factory=list)
    demographic: dict = field(default_factory=dict)
    social_role: str = "participant"  # participant, influencer, moderator, observer
    communication_style: str = "neutral"  # formal, casual, aggressive, academic, emotional

    # Dynamic state
    mood: float = 0.0  # -1 to 1
    energy: float = 1.0  # 0 to 1
    beliefs: dict = field(default_factory=dict)
    relationships: dict = field(default_factory=dict)  # agent_id -> sentiment


class AgentFactory:
    """Generate diverse populations of agents for simulation."""

    def __init__(self, llm_client=None):
        self.llm = llm_client

    def generate_population(self, n_agents: int, scenario_context: str,
                            demographics: dict = None) -> list:
        """Create a diverse population matching demographic targets."""
        agents = []

        # Default demographic distribution
        if demographics is None:
            demographics = {
                'archetypes': {
                    'analyst': 0.10,
                    'enthusiast': 0.15,
                    'skeptic': 0.10,
                    'follower': 0.30,
                    'contrarian': 0.08,
                    'influencer': 0.05,
                    'lurker': 0.20,
                    'troll': 0.02
                },
                'expertise_levels': {
                    'expert': 0.05,
                    'knowledgeable': 0.20,
                    'moderate': 0.40,
                    'casual': 0.35
                }
            }

        for archetype, fraction in demographics['archetypes'].items():
            count = int(n_agents * fraction)
            for _ in range(count):
                agent = self._create_agent(archetype, scenario_context)
                agents.append(agent)

        # Fill remaining to reach target
        while len(agents) < n_agents:
            archetype = np.random.choice(list(demographics['archetypes'].keys()))
            agent = self._create_agent(archetype, scenario_context)
            agents.append(agent)

        return agents[:n_agents]

    def _create_agent(self, archetype: str, context: str) -> AgentProfile:
        """Create a single agent with full backstory."""
        personality = BigFivePersonality.from_archetype(archetype)

        # Generate name and background
        agent = AgentProfile(
            personality=personality,
            social_role=self._role_from_archetype(archetype),
            communication_style=self._style_from_personality(personality)
        )

        # In production, use LLM to generate rich backstory
        agent.background = f"A {archetype} personality in the context of {context[:50]}"
        agent.name = f"Agent_{agent.id}"

        return agent

    def _role_from_archetype(self, archetype: str) -> str:
        roles = {
            'analyst': 'participant',
            'enthusiast': 'participant',
            'skeptic': 'participant',
            'follower': 'participant',
            'contrarian': 'participant',
            'influencer': 'influencer',
            'lurker': 'observer',
            'troll': 'participant'
        }
        return roles.get(archetype, 'participant')

    def _style_from_personality(self, personality: BigFivePersonality) -> str:
        if personality.conscientiousness > 0.7 and personality.openness > 0.6:
            return 'academic'
        elif personality.extraversion > 0.7 and personality.agreeableness < 0.4:
            return 'aggressive'
        elif personality.neuroticism > 0.7:
            return 'emotional'
        elif personality.extraversion < 0.3:
            return 'formal'
        return 'casual'

Memory Systems

Agent Memory Architecture

class AgentMemory:
    """
    Multi-layer memory system for simulation agents.
    Inspired by Zep Cloud's approach to persistent agent memory.

    Layers:
    1. Working memory: current context (ephemeral)
    2. Episodic memory: specific events and interactions
    3. Semantic memory: beliefs, knowledge, summaries
    4. Procedural memory: learned behaviors and strategies
    """

    def __init__(self, agent_id: str, max_episodic: int = 1000,
                 max_working: int = 10):
        self.agent_id = agent_id
        self.working_memory = []  # Most recent items
        self.episodic_memory = []  # Specific events
        self.semantic_memory = {}  # Structured knowledge
        self.procedural_memory = {}  # Learned strategies
        self.max_episodic = max_episodic
        self.max_working = max_working

    def add_experience(self, event: dict):
        """Record an experience across memory layers."""
        # Working memory (always current context)
        self.working_memory.append(event)
        if len(self.working_memory) > self.max_working:
            # Oldest working memory moves to episodic
            overflow = self.working_memory.pop(0)
            self._add_episodic(overflow)

        # Direct episodic storage for significant events
        if event.get('significance', 0) > 0.5:
            self._add_episodic(event)

    def _add_episodic(self, event: dict):
        """Add to episodic memory with importance-based retention."""
        event['timestamp'] = event.get('timestamp', len(self.episodic_memory))
        event['recall_count'] = 0
        self.episodic_memory.append(event)

        if len(self.episodic_memory) > self.max_episodic:
            # Remove least important old memories
            self.episodic_memory.sort(
                key=lambda e: e.get('significance', 0) * (1 + e['recall_count'] * 0.1),
                reverse=True
            )
            self.episodic_memory = self.episodic_memory[:self.max_episodic]

    def update_belief(self, topic: str, new_value: float,
                      evidence: str, confidence: float = 0.5):
        """Update semantic memory with a belief change."""
        if topic not in self.semantic_memory:
            self.semantic_memory[topic] = {
                'value': new_value,
                'confidence': confidence,
                'history': [],
                'evidence': [evidence]
            }
        else:
            existing = self.semantic_memory[topic]
            # Bayesian-style update
            old_weight = existing['confidence']
            new_weight = confidence
            total_weight = old_weight + new_weight

            updated_value = (
                existing['value'] * old_weight + new_value * new_weight
            ) / total_weight

            existing['history'].append({
                'old_value': existing['value'],
                'new_value': updated_value,
                'evidence': evidence
            })
            existing['value'] = updated_value
            existing['confidence'] = min(0.99, total_weight / (total_weight + 0.5))
            existing['evidence'].append(evidence)

    def recall(self, query: str, max_results: int = 5) -> list:
        """Retrieve relevant memories based on a query."""
        # Simple keyword matching (in production, use embeddings)
        query_words = set(query.lower().split())
        scored_memories = []

        for memory in self.episodic_memory:
            content = str(memory).lower()
            overlap = len(query_words & set(content.split()))
            if overlap > 0:
                memory['recall_count'] += 1
                scored_memories.append((overlap, memory))

        scored_memories.sort(key=lambda x: -x[0])
        return [m for _, m in scored_memories[:max_results]]

    def get_context_for_llm(self, current_situation: str,
                             max_tokens: int = 500) -> str:
        """Assemble memory context for LLM-driven agent decisions."""
        context_parts = []

        # Working memory (recent context)
        context_parts.append("Recent events:")
        for item in self.working_memory[-5:]:
            context_parts.append(f"  - {item.get('description', str(item)[:100])}")

        # Relevant beliefs
        context_parts.append("\nCurrent beliefs:")
        for topic, belief in list(self.semantic_memory.items())[:10]:
            context_parts.append(
                f"  - {topic}: {belief['value']:.2f} (confidence: {belief['confidence']:.1%})"
            )

        # Relevant episodic memories
        relevant = self.recall(current_situation, max_results=3)
        if relevant:
            context_parts.append("\nRelevant past experiences:")
            for mem in relevant:
                context_parts.append(f"  - {mem.get('description', str(mem)[:100])}")

        return "\n".join(context_parts)

    def get_summary(self) -> dict:
        """Summarize memory state."""
        return {
            'working_memory_items': len(self.working_memory),
            'episodic_memories': len(self.episodic_memory),
            'beliefs': len(self.semantic_memory),
            'top_beliefs': {
                k: v['value']
                for k, v in sorted(
                    self.semantic_memory.items(),
                    key=lambda x: -x[1]['confidence']
                )[:5]
            }
        }

Zep Cloud Integration Pattern

class ZepMemoryAdapter:
    """
    Adapter for Zep Cloud's memory service.
    Provides persistent, searchable agent memory with:
    - Automatic summarization of long conversations
    - Semantic search across memories
    - Entity extraction and relationship tracking
    - Session management for multi-turn interactions
    """

    def __init__(self, zep_api_key: str, project_id: str):
        self.api_key = zep_api_key
        self.project = project_id
        # In production: self.client = ZepClient(api_key)

    async def create_agent_session(self, agent_id: str,
                                    metadata: dict = None) -> str:
        """Create a new memory session for an agent."""
        session_id = f"agent_{agent_id}_{uuid.uuid4().hex[:8]}"
        # await self.client.memory.add_session(
        #     session_id=session_id,
        #     metadata=metadata or {}
        # )
        return session_id

    async def store_interaction(self, session_id: str, role: str,
                                 content: str, metadata: dict = None):
        """Store an agent interaction in Zep."""
        # await self.client.memory.add(
        #     session_id=session_id,
        #     messages=[Message(role=role, content=content, metadata=metadata)]
        # )
        pass

    async def search_memory(self, agent_id: str, query: str,
                             limit: int = 5) -> list:
        """Semantic search across an agent's memories."""
        # results = await self.client.memory.search(
        #     session_id=f"agent_{agent_id}",
        #     text=query,
        #     limit=limit
        # )
        return []

    async def get_agent_facts(self, agent_id: str) -> dict:
        """Retrieve extracted facts and entities about an agent."""
        # facts = await self.client.memory.get_facts(session_id=f"agent_{agent_id}")
        return {}

Dual-Platform Simulation

MiroFish-Style Dual-Platform Architecture

class SimulatedPlatform:
    """Base class for simulated social platforms."""

    def __init__(self, platform_type: str, rules: dict = None):
        self.platform_type = platform_type
        self.rules = rules or {}
        self.content = []  # All posts/threads
        self.feeds = {}  # agent_id -> personalized feed

    def post(self, agent_id: str, content: str, content_type: str = 'text',
             reply_to: str = None, metadata: dict = None) -> str:
        """Agent posts content to the platform."""
        post_id = str(uuid.uuid4())[:8]
        post = {
            'id': post_id,
            'author': agent_id,
            'content': content,
            'type': content_type,
            'reply_to': reply_to,
            'timestamp': len(self.content),
            'reactions': {},
            'engagement': 0,
            'metadata': metadata or {}
        }
        self.content.append(post)
        return post_id

    def get_feed(self, agent_id: str, n_items: int = 20) -> list:
        """Generate a personalized feed for an agent."""
        # Simple chronological + engagement-weighted feed
        scored = []
        for post in self.content[-100:]:  # Recent posts
            score = 1.0
            score += post['engagement'] * 0.1
            # Boost content from connections
            if post['author'] in self.feeds.get(agent_id, {}).get('following', []):
                score *= 2
            scored.append((score, post))

        scored.sort(key=lambda x: -x[0])
        return [post for _, post in scored[:n_items]]


class MicroblogPlatform(SimulatedPlatform):
    """Twitter/X-like platform with short posts and retweets."""

    def __init__(self, max_length: int = 280):
        super().__init__('microblog', {'max_length': max_length})
        self.trending = []

    def post(self, agent_id: str, content: str, **kwargs) -> str:
        content = content[:self.rules.get('max_length', 280)]
        post_id = super().post(agent_id, content, **kwargs)

        # Update trending
        self._update_trending(content)
        return post_id

    def repost(self, agent_id: str, original_post_id: str,
               comment: str = '') -> str:
        """Retweet/repost functionality."""
        original = next(
            (p for p in self.content if p['id'] == original_post_id), None
        )
        if original:
            original['engagement'] += 1
            return self.post(agent_id, f"RT: {original['content'][:200]} | {comment}",
                           metadata={'repost_of': original_post_id})
        return ''

    def _update_trending(self, content: str):
        """Extract and track trending topics."""
        words = content.lower().split()
        hashtags = [w for w in words if w.startswith('#')]
        for tag in hashtags:
            found = False
            for trend in self.trending:
                if trend['tag'] == tag:
                    trend['count'] += 1
                    found = True
                    break
            if not found:
                self.trending.append({'tag': tag, 'count': 1})

        self.trending.sort(key=lambda x: -x['count'])
        self.trending = self.trending[:20]


class ForumPlatform(SimulatedPlatform):
    """Reddit-like platform with threads, voting, and long-form discussion."""

    def __init__(self):
        super().__init__('forum')
        self.threads = []

    def create_thread(self, agent_id: str, title: str,
                      body: str, category: str = 'general') -> str:
        thread_id = str(uuid.uuid4())[:8]
        thread = {
            'id': thread_id,
            'author': agent_id,
            'title': title,
            'body': body,
            'category': category,
            'replies': [],
            'upvotes': 0,
            'downvotes': 0,
            'timestamp': len(self.threads)
        }
        self.threads.append(thread)
        return thread_id

    def reply(self, agent_id: str, thread_id: str, content: str,
              parent_reply_id: str = None) -> str:
        reply_id = str(uuid.uuid4())[:8]
        reply = {
            'id': reply_id,
            'author': agent_id,
            'content': content,
            'parent': parent_reply_id,
            'upvotes': 0,
            'downvotes': 0,
            'timestamp': 0
        }

        thread = next((t for t in self.threads if t['id'] == thread_id), None)
        if thread:
            reply['timestamp'] = len(thread['replies'])
            thread['replies'].append(reply)

        return reply_id

    def vote(self, agent_id: str, thread_id: str, reply_id: str = None,
             direction: str = 'up'):
        """Upvote or downvote a thread or reply."""
        thread = next((t for t in self.threads if t['id'] == thread_id), None)
        if thread:
            if reply_id:
                target = next((r for r in thread['replies'] if r['id'] == reply_id), None)
            else:
                target = thread

            if target:
                if direction == 'up':
                    target['upvotes'] += 1
                else:
                    target['downvotes'] += 1


class DualPlatformSimulation:
    """
    MiroFish-style simulation with agents interacting on two platforms.
    The microblog captures rapid, viral dynamics.
    The forum captures deeper analysis and deliberation.
    """

    def __init__(self, scenario: str, n_agents: int = 500):
        self.scenario = scenario
        self.microblog = MicroblogPlatform()
        self.forum = ForumPlatform()
        self.agent_factory = AgentFactory()
        self.agents = self.agent_factory.generate_population(n_agents, scenario)
        self.memories = {a.id: AgentMemory(a.id) for a in self.agents}
        self.time_step = 0

    def inject_scenario(self, event: dict):
        """Inject a scenario event into both platforms."""
        # Breaking news on microblog
        self.microblog.post(
            'system_news',
            f"BREAKING: {event['headline']}",
            content_type='news'
        )

        # Discussion thread on forum
        self.forum.create_thread(
            'system_news',
            event['headline'],
            event.get('full_text', event['headline']),
            category='breaking'
        )

    async def run_step(self):
        """Execute one simulation step."""
        for agent in self.agents:
            behavior = agent.personality.behavioral_implications()

            # Decide whether to engage
            if np.random.random() < behavior['social_posting_frequency'] * 0.1:
                await self._agent_takes_action(agent, behavior)

            # Update mood based on content consumed
            self._update_agent_state(agent)

        self.time_step += 1

    async def _agent_takes_action(self, agent: AgentProfile, behavior: dict):
        """Have an agent take an action based on personality and context."""
        # Choose platform
        if agent.personality.extraversion > 0.6:
            platform_choice = 'microblog' if np.random.random() < 0.7 else 'forum'
        else:
            platform_choice = 'forum' if np.random.random() < 0.6 else 'microblog'

        # Get feed
        if platform_choice == 'microblog':
            feed = self.microblog.get_feed(agent.id)
        else:
            feed = [{'content': t['title'] + ' ' + t['body']}
                    for t in self.forum.threads[-10:]]

        # Prepare memory context
        memory = self.memories[agent.id]
        context = memory.get_context_for_llm(str(feed[:3]))

        # In production, call LLM to generate agent response
        # response = await self.llm.generate_agent_action(agent, context, feed)

        # For now, simple rule-based action
        if feed:
            latest = feed[0]
            if behavior['content_creation_rate'] > np.random.random():
                if platform_choice == 'microblog':
                    self.microblog.post(agent.id, f"Responding to: {latest.get('content', '')[:50]}...")
                else:
                    if self.forum.threads:
                        self.forum.reply(agent.id, self.forum.threads[-1]['id'],
                                        f"My perspective on this...")

    def _update_agent_state(self, agent: AgentProfile):
        """Update agent mood and energy based on simulation state."""
        feed = self.microblog.get_feed(agent.id, n_items=5)
        sentiment = 0
        for post in feed:
            # Simple sentiment proxy
            content = post.get('content', '').lower()
            if any(w in content for w in ['good', 'great', 'positive', 'win']):
                sentiment += 0.1
            elif any(w in content for w in ['bad', 'terrible', 'negative', 'lose']):
                sentiment -= 0.1

        agent.mood = np.clip(agent.mood + sentiment * agent.personality.neuroticism, -1, 1)
        agent.energy = max(0, agent.energy - 0.01)  # Gradual fatigue

    def extract_predictions(self) -> dict:
        """Extract prediction signals from simulation state."""
        all_beliefs = {}
        for agent in self.agents:
            memory = self.memories[agent.id]
            for topic, belief in memory.semantic_memory.items():
                if topic not in all_beliefs:
                    all_beliefs[topic] = []
                all_beliefs[topic].append(belief['value'])

        predictions = {}
        for topic, values in all_beliefs.items():
            predictions[topic] = {
                'mean': np.mean(values),
                'median': np.median(values),
                'std': np.std(values),
                'consensus': 1 - np.std(values) * 2,
                'n_agents': len(values)
            }

        # Platform-level signals
        predictions['_platform_signals'] = {
            'microblog_volume': len(self.microblog.content),
            'forum_threads': len(self.forum.threads),
            'trending_topics': self.microblog.trending[:5],
            'simulation_steps': self.time_step
        }

        return predictions

Seed Material Processing

class SeedMaterialProcessor:
    """Process real-world information into simulation scenarios."""

    def __init__(self, llm_client=None):
        self.llm = llm_client

    async def process_news_articles(self, articles: list) -> dict:
        """Convert news articles into simulation seed material."""
        prompt = f"""Analyze these news articles and extract:
1. Key entities (people, organizations, technologies)
2. Main narrative threads
3. Points of contention or uncertainty
4. Potential future developments
5. Stakeholder positions

Articles:
{chr(10).join(a['text'][:500] for a in articles[:5])}

Respond as structured JSON."""

        # In production, call LLM
        return {
            'entities': [],
            'narratives': [],
            'contentions': [],
            'future_developments': [],
            'stakeholder_map': {}
        }

    def create_scenario_from_seed(self, seed_material: dict) -> dict:
        """Convert processed seed material into a simulation scenario."""
        return {
            'title': seed_material.get('primary_narrative', 'Scenario'),
            'description': 'Generated from seed material',
            'initial_conditions': seed_material.get('current_state', {}),
            'key_actors': seed_material.get('entities', []),
            'tension_points': seed_material.get('contentions', []),
            'event_schedule': self._generate_event_schedule(seed_material),
            'measurement_points': self._define_measurements(seed_material)
        }

    def _generate_event_schedule(self, seed: dict) -> list:
        """Generate a sequence of events to inject during simulation."""
        events = []
        for i, development in enumerate(seed.get('future_developments', [])):
            events.append({
                'time_step': (i + 1) * 10,
                'event': {
                    'headline': development,
                    'full_text': f'Development: {development}',
                    'affected_topics': seed.get('contentions', [])[:2]
                },
                'probability': 0.7  # Some events may or may not fire
            })
        return events

    def _define_measurements(self, seed: dict) -> list:
        """Define what to measure during simulation."""
        measurements = [
            {'metric': 'overall_sentiment', 'frequency': 'every_step'},
            {'metric': 'topic_consensus', 'frequency': 'every_10_steps'},
            {'metric': 'polarization_index', 'frequency': 'every_10_steps'},
        ]
        for contention in seed.get('contentions', []):
            measurements.append({
                'metric': f'belief_about_{contention[:30]}',
                'frequency': 'every_step'
            })
        return measurements

Interactive Exploration

class SimulationExplorer:
    """Tools for interactively exploring simulation results."""

    def __init__(self, simulation: DualPlatformSimulation):
        self.sim = simulation

    def query_agent(self, agent_id: str) -> dict:
        """Inspect a specific agent's state and history."""
        agent = next((a for a in self.sim.agents if a.id == agent_id), None)
        if not agent:
            return {'error': f'Agent {agent_id} not found'}

        memory = self.sim.memories[agent_id]
        return {
            'profile': {
                'name': agent.name,
                'personality': agent.personality.behavioral_implications(),
                'social_role': agent.social_role,
                'mood': agent.mood,
                'energy': agent.energy
            },
            'memory': memory.get_summary(),
            'beliefs': agent.beliefs,
            'recent_activity': memory.working_memory[-5:]
        }

    def narrative_summary(self) -> dict:
        """Generate a narrative summary of the simulation so far."""
        total_posts = len(self.sim.microblog.content)
        total_threads = len(self.sim.forum.threads)
        trending = self.sim.microblog.trending[:5]

        # Mood distribution
        moods = [a.mood for a in self.sim.agents]

        return {
            'time_steps': self.sim.time_step,
            'total_microblog_posts': total_posts,
            'total_forum_threads': total_threads,
            'trending_topics': trending,
            'population_mood': {
                'mean': np.mean(moods),
                'positive_fraction': np.mean(np.array(moods) > 0),
                'negative_fraction': np.mean(np.array(moods) < 0),
            },
            'predictions': self.sim.extract_predictions()
        }

    def what_if(self, event: dict) -> dict:
        """Explore what-if scenarios by injecting events and observing changes."""
        # Snapshot current state
        pre_state = self.sim.extract_predictions()

        # Inject event
        self.sim.inject_scenario(event)

        # Run a few steps
        # In production: await self.sim.run_steps(10)

        # Compare
        post_state = self.sim.extract_predictions()

        return {
            'event': event,
            'pre_state': pre_state,
            'post_state': post_state,
            'changes': self._compute_changes(pre_state, post_state)
        }

    def _compute_changes(self, pre: dict, post: dict) -> dict:
        changes = {}
        for topic in pre:
            if topic.startswith('_'):
                continue
            if topic in post:
                if isinstance(pre[topic], dict) and 'mean' in pre[topic]:
                    changes[topic] = {
                        'belief_shift': post[topic]['mean'] - pre[topic]['mean'],
                        'consensus_change': post[topic].get('consensus', 0) - pre[topic].get('consensus', 0)
                    }
        return changes

Key Takeaways

1. Agent personality design using psychologically validated models (Big Five) produces more realistic behavioral diversity than random attribute generation
2. Multi-layer memory (working, episodic, semantic, procedural) gives agents consistent behavior over time; Zep Cloud provides production-grade persistent memory
3. Dual-platform simulation (microblog + forum) captures both rapid viral dynamics and deeper deliberative reasoning
4. Seed material processing converts real-world news and data into structured simulation scenarios with event schedules
5. Scenario injection lets you model "what if" events and observe emergent reactions across the agent population
6. Population demographics should mirror real-world distributions: mostly followers and lurkers, few influencers and contrarians
7. Agent mood and energy dynamics prevent unrealistic constant engagement; fatigue and emotional contagion drive realistic behavior
8. Interactive exploration (querying individual agents, narrative summaries, what-if scenarios) is where the predictive insights emerge