Swarm intelligence with persistent hive mind

from swarm import Swarm, Agent
import httpx
import os

def get_stompy_tools():
    """Create Stompy tool functions for swarm agents."""

    async def recall_context(topic: str) -> str:
        """Recall knowledge from the hive mind."""
        async with httpx.AsyncClient() as client:
            response = await client.post(
                "https://mcp.stompy.ai/sse",
                headers={"Authorization": f"Bearer {os.environ['STOMPY_TOKEN']}"},
                json={"tool": "recall_context", "topic": topic}
            )
            return response.json().get("content", "")

    async def save_learning(topic: str, content: str, priority: str = "reference") -> str:
        """Save knowledge to the hive mind."""
        async with httpx.AsyncClient() as client:
            await client.post(
                "https://mcp.stompy.ai/sse",
                headers={"Authorization": f"Bearer {os.environ['STOMPY_TOKEN']}"},
                json={"tool": "lock_context", "topic": topic,
                      "content": content, "priority": priority}
            )
            return "Saved to hive memory"

    return [recall_context, save_learning]

stompy_tools = get_stompy_tools()

def transfer_to_billing():
    """Hand off to billing specialist."""
    return billing_agent

def transfer_to_technical():
    """Hand off to technical specialist."""
    return technical_agent

triage_agent = Agent(
    name="Triage",
    instructions="""You are the routing specialist. Use recall_context to check:
    - "routing_patterns": Known patterns for efficient routing
    - "customer_{id}_history": Past interactions with returning customers

    After successful routing, save_learning the pattern for future reference.""",
    functions=[*stompy_tools, transfer_to_billing, transfer_to_technical]
)

billing_agent = Agent(
    name="Billing",
    instructions="""You handle billing inquiries. Always recall_context for:
    - "billing_edge_cases": Known edge cases and solutions
    - "customer_{id}_billing": This customer's billing history

    Save novel solutions with save_learning for future agents.""",
    functions=[*stompy_tools, transfer_to_technical]
)

technical_agent = Agent(
    name="Technical",
    instructions="""You solve technical issues. recall_context for:
    - "technical_solutions": Proven solutions to common issues
    - "system_status": Current known issues

    Document new solutions with save_learning.""",
    functions=[*stompy_tools]
)

from swarm import Swarm

client = Swarm()

async def run_with_learning(messages: list, user_id: str = None):
    """Run the swarm with memory and pattern tracking."""

    # Pre-load relevant context for this conversation
    if user_id:
        history = await stompy_tools[0](f"customer_{user_id}_history")
        if history:
            messages[0]["content"] += f"\n\nCUSTOMER HISTORY:\n{history}"

    # Run the swarm
    response = client.run(agent=triage_agent, messages=messages)

    # Track the routing pattern for future optimization
    agents_used = [m.get("sender") for m in response.messages if m.get("sender")]
    routing_path = " -> ".join(agents_used)

    await stompy_tools[1](
        topic=f"routing_log_{datetime.now().strftime('%Y%m%d')}",
        content=f"Query type: {classify_query(messages)}\nPath: {routing_path}\nOutcome: {response.messages[-1]['content'][:200]}",
        priority="reference"
    )

    return response

# The swarm now builds institutional knowledge with every interaction