❓ How to build a dynamic agentic Swarm workflow in LangGraph?

[BoyPit]

Hi everyone

I'm trying to build a dynamic Swarm-based LangGraph workflow where the number of agents can vary at runtime, based on a configuration.

The challenge I'm facing is that I currently need to recompile the whole Swarm graph every time a message is received, just to adapt to the updated set of agents. That doesn't feel right and introduces overhead I’d like to avoid.

I'm looking for a cleaner architectural approach that would let me:

Dynamically inject agents into the swarm,

Avoid recompiling the entire workflow on each user interaction,

Stay compatible with subgraphs (each agent has its own MCP workflow graph).

Has anyone managed to implement a dynamic agentic swarm that avoids full recompilation?
Here’s a simplified version of my setup:
Each agent uses its own subgraph (via create_graph_builder), and I build custom handoff tools dynamically to allow inter-agent communication.

Here’s the coordinator logic and Swarm setup:

# Coordinator node creates agent subgraphs dynamically
async def swarm_coordinator(state: SwarmState, config):
    agents_config = config["configurable"].get("agents", [])
    default_agent = config["configurable"].get("default_active_agent", "Toto")
    
    agent_list = []
    for agent_conf in agents_config:
        current_name = agent_conf["name"]
        handoff_tools = [
            create_custom_handoff_tool(
                agent_name=agent["name"],
                name=f"transfer_to_{agent['name']}",
                description=f"use to talk with {agent['name']}"
            )
            for agent in agents_config if agent["name"] != current_name
        ]
        builder = await create_graph_builder(agent_conf, agent_conf["name"], handoff_tools)
        agent_graph = builder.compile()
        agent_graph.name = agent_conf["name"]
        agent_list.append(agent_graph)

    swarm_workflow = create_swarm(
        agent_list,
        state_schema=SwarmState,
        default_active_agent=default_agent
    )
    compiled_swarm = swarm_workflow.compile(checkpointer=InMemorySaver())
    return await compiled_swarm.ainvoke({"messages": state["messages"]})

Thank you all !