Distributed Hive Mind — Tutorial¶

from amplihack.agents.goal_seeking.hive_mind.hive_graph import (
    InMemoryHiveGraph,
    HiveFact,
    create_hive_graph,
)

# Create a hive
hive = create_hive_graph("memory", hive_id="my-hive")

# Register agents
hive.register_agent("alice", domain="security")
hive.register_agent("bob", domain="infrastructure")

# Alice promotes a fact
hive.promote_fact("alice", HiveFact(
    fact_id="", content="SSH runs on port 22", concept="networking", confidence=0.95,
))

# Bob promotes a fact
hive.promote_fact("bob", HiveFact(
    fact_id="", content="Nginx default port is 80", concept="networking", confidence=0.9,
))

# Query the hive
results = hive.query_facts("port networking", limit=10)
for fact in results:
    print(f"  [{fact.confidence:.0%}] {fact.content}")

from amplihack.agents.goal_seeking.hive_mind.hive_graph import (
    InMemoryHiveGraph,
    HiveFact,
)

# Create a tree: root -> [security, infrastructure, data]
root = InMemoryHiveGraph("root")

security = InMemoryHiveGraph("security")
infra = InMemoryHiveGraph("infrastructure")
data = InMemoryHiveGraph("data")

for child in [security, infra, data]:
    root.add_child(child)
    child.set_parent(root)

# Each hive has its own agents and facts
security.register_agent("sec-1", domain="security")
security.promote_fact("sec-1", HiveFact(
    fact_id="", content="CVE-2024-1234 affects OpenSSL 3.x",
    concept="vulnerabilities", confidence=0.95,
))

infra.register_agent("infra-1", domain="infrastructure")
infra.promote_fact("infra-1", HiveFact(
    fact_id="", content="Server prod-db-01 runs on 10.0.1.5 port 5432",
    concept="servers", confidence=0.9,
))

data.register_agent("data-1", domain="data")
data.promote_fact("data-1", HiveFact(
    fact_id="", content="Users table has 2.5M rows with daily growth of 10K",
    concept="schema", confidence=0.85,
))

# Federated query from root finds facts across ALL hives
results = root.query_federated("server port infrastructure", limit=10)
for fact in results:
    print(f"  [{fact.confidence:.0%}] {fact.content}")

# Federated query from a child also traverses the tree
results = security.query_federated("server database port", limit=10)
print(f"\nSecurity hive found {len(results)} results across federation")

from pathlib import Path
from amplihack.agents.goal_seeking.learning_agent import LearningAgent
from amplihack.agents.goal_seeking.hive_mind.hive_graph import InMemoryHiveGraph

# Create shared hive
hive = InMemoryHiveGraph("shared-hive")
hive.register_agent("agent_a")
hive.register_agent("agent_b")

# Create agents with hive_store
agent_a = LearningAgent(
    agent_name="agent_a",
    storage_path=Path("/tmp/agent_a_db"),
    use_hierarchical=True,
    hive_store=hive,  # <-- Auto-promotes facts to hive
)
agent_b = LearningAgent(
    agent_name="agent_b",
    storage_path=Path("/tmp/agent_b_db"),
    use_hierarchical=True,
    hive_store=hive,
)

# Agent A learns biology
agent_a.learn_from_content(
    "Photosynthesis converts sunlight into chemical energy in chloroplasts."
)

# Agent B learns chemistry
agent_b.learn_from_content(
    "Water (H2O) has a bent molecular geometry with a 104.5 degree bond angle."
)

# Agent B can answer questions about biology (facts shared via hive)
answer = agent_b.answer_question("How do plants convert sunlight to energy?")
print(answer)

agent_a.close()
agent_b.close()

# Requires: pip install kuzu
# Requires: amplihack-memory-lib installed

from amplihack.agents.goal_seeking.hive_mind.distributed import AgentNode
from amplihack.agents.goal_seeking.hive_mind.event_bus import create_event_bus

# Create event bus (local for testing, "redis" or "azure" for production)
bus = create_event_bus("local")

# Create agents with their own Kuzu databases
agent_a = AgentNode(
    agent_id="bio-agent",
    domain="biology",
    db_path="/tmp/hive_eval/bio-agent",
    hive_store=None,  # or pass a HiveGraphStore for federation
    event_bus=bus,
)

# Learn and query
agent_a.learn("Mitosis is cell division", tags=["biology"])
results = agent_a.query("cell division")
print(f"Found: {results}")

agent_a.close()