| Name | langgraph-swarm JSON |
| Version |
0.0.3
JSON |
| download |
| home_page | None |
| Summary | An implementation of a multi-agent swarm using LangGraph |
| upload_time | 2025-02-27 00:13:51 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.10 |
| license | None |
| keywords |
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# 🤖 LangGraph Multi-Agent Swarm
A Python library for creating swarm-style multi-agent systems using [LangGraph](https://github.com/langchain-ai/langgraph). A swarm is a type of [multi-agent](https://langchain-ai.github.io/langgraph/concepts/multi_agent) architecture where agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent.
## Features
- 🤖 **Multi-agent collaboration** - Enable specialized agents to work together and hand off context to each other
- 🛠️ **Customizable handoff tools** - Built-in tools for communication between agents
This library is built on top of [LangGraph](https://github.com/langchain-ai/langgraph), a powerful framework for building agent applications, and comes with out-of-box support for [streaming](https://langchain-ai.github.io/langgraph/how-tos/#streaming), [short-term and long-term memory](https://langchain-ai.github.io/langgraph/concepts/memory/) and [human-in-the-loop](https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/)
## Installation
```bash
pip install langgraph-swarm
```
## Quickstart
```bash
pip install langgraph-swarm langchain-openai
export OPENAI_API_KEY=<your_api_key>
```
```python
from langchain_openai import ChatOpenAI
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.prebuilt import create_react_agent
from langgraph_swarm import create_handoff_tool, create_swarm
model = ChatOpenAI(model="gpt-4o")
def add(a: int, b: int) -> int:
"""Add two numbers"""
return a + b
alice = create_react_agent(
model,
[add, create_handoff_tool(agent_name="Bob")],
prompt="You are Alice, an addition expert.",
name="Alice",
)
bob = create_react_agent(
model,
[create_handoff_tool(agent_name="Alice", description="Transfer to Alice, she can help with math")],
prompt="You are Bob, you speak like a pirate.",
name="Bob",
)
checkpointer = InMemorySaver()
workflow = create_swarm(
[alice, bob],
default_active_agent="Alice"
)
app = workflow.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "1"}}
turn_1 = app.invoke(
{"messages": [{"role": "user", "content": "i'd like to speak to Bob"}]},
config,
)
print(turn_1)
turn_2 = app.invoke(
{"messages": [{"role": "user", "content": "what's 5 + 7?"}]},
config,
)
print(turn_2)
```
## Memory
You can add [short-term](https://langchain-ai.github.io/langgraph/how-tos/persistence/) and [long-term](https://langchain-ai.github.io/langgraph/how-tos/cross-thread-persistence/) [memory](https://langchain-ai.github.io/langgraph/concepts/memory/) to your swarm multi-agent system. Since `create_swarm()` returns an instance of `StateGraph` that needs to be compiled before use, you can directly pass a [checkpointer](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) or a [store](https://langchain-ai.github.io/langgraph/reference/store/#langgraph.store.base.BaseStore) instance to the `.compile()` method:
```python
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore
# short-term memory
checkpointer = InMemorySaver()
# long-term memory
store = InMemoryStore()
model = ...
alice = ...
bob = ...
workflow = create_swarm(
[alice, bob],
default_active_agent="Alice"
)
# Compile with checkpointer/store
app = workflow.compile(
checkpointer=checkpointer,
store=store
)
```
> [!IMPORTANT]
> Adding [short-term memory](https://langchain-ai.github.io/langgraph/concepts/persistence/) is crucial for maintaining conversation state across multiple interactions. Without it, the swarm would "forget" which agent was last active and lose the conversation history. Make sure to always compile the swarm with a checkpointer if you plan to use it in multi-turn conversations; e.g., `workflow.compile(checkpointer=checkpointer)`.
## How to customize
You can customize multi-agent swarm by changing either the [handoff tools](#customizing-handoff-tools) implementation or the [agent implementation](#customizing-agent-implementation).
### Customizing handoff tools
By default, the agents in the swarm are assumed to use handoff tools created with the prebuilt `create_handoff_tool`. You can also create your own, custom handoff tools. Here are some ideas on how you can modify the default implementation:
* change tool name and/or description
* add tool call arguments for the LLM to populate, for example a task description for the next agent
* change what data is passed to the next agent as part of the handoff: by default `create_handoff_tool` passes **full** message history (all of the messages generated in the swarm up to this point), as well as the contents of `Command.update` to the next agent
> [!IMPORTANT]
> If you want to change what messages are passed to the next agent, you **must** use a different state schema key for `messages` in your agent implementation (e.g., `alice_messages`). By default, all agent (subgraph) state updates are applied to the swarm (parent) graph state during the handoff. Since all of the agents by default are assumed to communicate over a single `messages` key, this means that the agent's messages are **automatically combined** into the parent graph's `messages`, unless an agent uses a different key for `messages`. See more on this in the [customizing agent implementation](#customizing-agent-implementation) section.
Here is an example of what a custom handoff tool might look like:
```python
from typing import Annotated
from langchain_core.tools import tool, BaseTool, InjectedToolCallId
from langchain_core.messages import ToolMessage
from langgraph.types import Command
from langgraph.prebuilt import InjectedState
def create_custom_handoff_tool(*, agent_name: str, tool_name: str, tool_description: str) -> BaseTool:
@tool(name=tool_name, description=tool_description)
def handoff_to_agent(
# you can add additional tool call arguments for the LLM to populate
# for example, you can ask the LLM to populate a task description for the next agent
task_description: Annotated[str, "Detailed description of what the next agent should do, including all of the relevant context."],
# you can inject the state of the agent that is calling the tool
state: Annotated[dict, InjectedState],
tool_call_id: Annotated[str, InjectedToolCallId],
):
tool_message = ToolMessage(
content=f"Successfully transferred to {agent_name}",
name=tool_name,
tool_call_id=tool_call_id,
)
# you can use a different messages state key here, if your agent uses a different schema
# e.g., "alice_messages" instead of "messages"
last_agent_message = state["messages"][-1]
# if the tool schema includes task description that LLM generates,
# you can extract it from the last tool call argument and pass it on to the next agent
task_description = last_agent_message.tool_calls[0]["args"]["task_description"]
return Command(
goto=agent_name,
graph=Command.PARENT,
# NOTE: this is a state update that will be applied to the swarm multi-agent graph (i.e., the PARENT graph)
update={
"messages": [last_agent_message, tool_message],
"active_agent": agent_name,
# optionally pass the task description to the next agent
"task_description": task_description,
},
)
return handoff_to_agent
```
> [!IMPORTANT]
> If you are implementing custom handoff tools that return `Command`, you need to ensure that:
(1) your agent has a tool-calling node that can handle tools returning `Command` (like LangGraph's prebuilt [`ToolNode`](https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.tool_node.ToolNode))
(2) both the swarm graph and the next agent graph have the [state schema](https://langchain-ai.github.io/langgraph/concepts/low_level#schema) containing the keys you want to update in `Command.update`
### Customizing agent implementation
By default, individual agents are expected to communicate over a single `messages` key that is shared by all agents and the overall multi-agent swarm graph. This means that **all** of the messages from **all** of the agents will be combined into a single, shared list of messages. This might not be desirable if you don't want to expose an agent's internal history of messages. To change this, you can customize the agent by taking the following steps:
1. use custom [state schema](https://langchain-ai.github.io/langgraph/concepts/low_level#schema) with a different key for messages, for example `alice_messages`
1. write a wrapper that converts the parent graph state to the child agent state and back (see this [how-to](https://langchain-ai.github.io/langgraph/how-tos/subgraph-transform-state/) guide)
```python
from typing_extensions import TypedDict, Annotated
from langchain_core.messages import AnyMessage
from langgraph.graph import StateGraph, add_messages
from langgraph_swarm import SwarmState
class AliceState(TypedDict):
alice_messages: Annotated[list[AnyMessage], add_messages]
# see this guide to learn how you can implement a custom tool-calling agent
# https://langchain-ai.github.io/langgraph/how-tos/react-agent-from-scratch/
alice = (
StateGraph(AliceState)
.add_node("model", ...)
.add_node("tools", ...)
.add_edge(...)
...
.compile()
)
# wrapper calling the agent
def call_alice(state: SwarmState):
# you can put any input transformation from parent state -> agent state
# for example, you can invoke "alice" with "task_description" populated by the LLM
response = alice.invoke({"alice_messages": state["messages"]})
# you can put any output transformation from agent state -> parent state
return {"messages": response["alice_messages"]}
def call_bob(state: SwarmState):
...
```
Then, you can create the swarm manually in the following way:
```python
from langgraph_swarm import add_active_agent_router
workflow = (
StateGraph(SwarmState)
.add_node("Alice", call_alice, destinations=("Bob",))
.add_node("Bob", call_bob, destinations=("Alice",))
)
# this is the router that enables us to keep track of the last active agent
workflow = add_active_agent_router(
builder=workflow,
route_to=["Alice", "Bob"],
default_active_agent="Alice",
)
# compile the workflow
app = workflow.compile()
```
Raw data
{
"_id": null,
"home_page": null,
"name": "langgraph-swarm",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.10",
"maintainer_email": null,
"keywords": null,
"author": null,
"author_email": "Vadym Barda <19161700+vbarda@users.noreply.github.com >",
"download_url": "https://files.pythonhosted.org/packages/94/6c/d9dc3ed0181418f5ea619981cf3d18dc469ca11a0b103d9b1f6a83b187a0/langgraph_swarm-0.0.3.tar.gz",
"platform": null,
"description": "# \ud83e\udd16 LangGraph Multi-Agent Swarm \n\nA Python library for creating swarm-style multi-agent systems using [LangGraph](https://github.com/langchain-ai/langgraph). A swarm is a type of [multi-agent](https://langchain-ai.github.io/langgraph/concepts/multi_agent) architecture where agents dynamically hand off control to one another based on their specializations. The system remembers which agent was last active, ensuring that on subsequent interactions, the conversation resumes with that agent.\n\n\n\n## Features\n\n- \ud83e\udd16 **Multi-agent collaboration** - Enable specialized agents to work together and hand off context to each other\n- \ud83d\udee0\ufe0f **Customizable handoff tools** - Built-in tools for communication between agents\n\nThis library is built on top of [LangGraph](https://github.com/langchain-ai/langgraph), a powerful framework for building agent applications, and comes with out-of-box support for [streaming](https://langchain-ai.github.io/langgraph/how-tos/#streaming), [short-term and long-term memory](https://langchain-ai.github.io/langgraph/concepts/memory/) and [human-in-the-loop](https://langchain-ai.github.io/langgraph/concepts/human_in_the_loop/)\n\n## Installation\n\n```bash\npip install langgraph-swarm\n```\n\n## Quickstart\n\n```bash\npip install langgraph-swarm langchain-openai\n\nexport OPENAI_API_KEY=<your_api_key>\n```\n\n```python\nfrom langchain_openai import ChatOpenAI\n\nfrom langgraph.checkpoint.memory import InMemorySaver\nfrom langgraph.prebuilt import create_react_agent\nfrom langgraph_swarm import create_handoff_tool, create_swarm\n\nmodel = ChatOpenAI(model=\"gpt-4o\")\n\ndef add(a: int, b: int) -> int:\n \"\"\"Add two numbers\"\"\"\n return a + b\n\nalice = create_react_agent(\n model,\n [add, create_handoff_tool(agent_name=\"Bob\")],\n prompt=\"You are Alice, an addition expert.\",\n name=\"Alice\",\n)\n\nbob = create_react_agent(\n model,\n [create_handoff_tool(agent_name=\"Alice\", description=\"Transfer to Alice, she can help with math\")],\n prompt=\"You are Bob, you speak like a pirate.\",\n name=\"Bob\",\n)\n\ncheckpointer = InMemorySaver()\nworkflow = create_swarm(\n [alice, bob],\n default_active_agent=\"Alice\"\n)\napp = workflow.compile(checkpointer=checkpointer)\n\nconfig = {\"configurable\": {\"thread_id\": \"1\"}}\nturn_1 = app.invoke(\n {\"messages\": [{\"role\": \"user\", \"content\": \"i'd like to speak to Bob\"}]},\n config,\n)\nprint(turn_1)\nturn_2 = app.invoke(\n {\"messages\": [{\"role\": \"user\", \"content\": \"what's 5 + 7?\"}]},\n config,\n)\nprint(turn_2)\n```\n\n## Memory\n\nYou can add [short-term](https://langchain-ai.github.io/langgraph/how-tos/persistence/) and [long-term](https://langchain-ai.github.io/langgraph/how-tos/cross-thread-persistence/) [memory](https://langchain-ai.github.io/langgraph/concepts/memory/) to your swarm multi-agent system. Since `create_swarm()` returns an instance of `StateGraph` that needs to be compiled before use, you can directly pass a [checkpointer](https://langchain-ai.github.io/langgraph/reference/checkpoints/#langgraph.checkpoint.base.BaseCheckpointSaver) or a [store](https://langchain-ai.github.io/langgraph/reference/store/#langgraph.store.base.BaseStore) instance to the `.compile()` method:\n\n```python\nfrom langgraph.checkpoint.memory import InMemorySaver\nfrom langgraph.store.memory import InMemoryStore\n\n# short-term memory\ncheckpointer = InMemorySaver()\n# long-term memory\nstore = InMemoryStore()\n\nmodel = ...\nalice = ...\nbob = ...\n\nworkflow = create_swarm(\n [alice, bob],\n default_active_agent=\"Alice\"\n)\n\n# Compile with checkpointer/store\napp = workflow.compile(\n checkpointer=checkpointer,\n store=store\n)\n```\n\n> [!IMPORTANT]\n> Adding [short-term memory](https://langchain-ai.github.io/langgraph/concepts/persistence/) is crucial for maintaining conversation state across multiple interactions. Without it, the swarm would \"forget\" which agent was last active and lose the conversation history. Make sure to always compile the swarm with a checkpointer if you plan to use it in multi-turn conversations; e.g., `workflow.compile(checkpointer=checkpointer)`.\n\n## How to customize\n\nYou can customize multi-agent swarm by changing either the [handoff tools](#customizing-handoff-tools) implementation or the [agent implementation](#customizing-agent-implementation).\n\n### Customizing handoff tools\n\nBy default, the agents in the swarm are assumed to use handoff tools created with the prebuilt `create_handoff_tool`. You can also create your own, custom handoff tools. Here are some ideas on how you can modify the default implementation:\n\n* change tool name and/or description\n* add tool call arguments for the LLM to populate, for example a task description for the next agent\n* change what data is passed to the next agent as part of the handoff: by default `create_handoff_tool` passes **full** message history (all of the messages generated in the swarm up to this point), as well as the contents of `Command.update` to the next agent\n\n> [!IMPORTANT]\n> If you want to change what messages are passed to the next agent, you **must** use a different state schema key for `messages` in your agent implementation (e.g., `alice_messages`). By default, all agent (subgraph) state updates are applied to the swarm (parent) graph state during the handoff. Since all of the agents by default are assumed to communicate over a single `messages` key, this means that the agent's messages are **automatically combined** into the parent graph's `messages`, unless an agent uses a different key for `messages`. See more on this in the [customizing agent implementation](#customizing-agent-implementation) section.\n\nHere is an example of what a custom handoff tool might look like:\n\n```python\nfrom typing import Annotated\n\nfrom langchain_core.tools import tool, BaseTool, InjectedToolCallId\nfrom langchain_core.messages import ToolMessage\nfrom langgraph.types import Command\nfrom langgraph.prebuilt import InjectedState\n\ndef create_custom_handoff_tool(*, agent_name: str, tool_name: str, tool_description: str) -> BaseTool:\n\n @tool(name=tool_name, description=tool_description)\n def handoff_to_agent(\n # you can add additional tool call arguments for the LLM to populate\n # for example, you can ask the LLM to populate a task description for the next agent\n task_description: Annotated[str, \"Detailed description of what the next agent should do, including all of the relevant context.\"],\n # you can inject the state of the agent that is calling the tool\n state: Annotated[dict, InjectedState],\n tool_call_id: Annotated[str, InjectedToolCallId],\n ):\n tool_message = ToolMessage(\n content=f\"Successfully transferred to {agent_name}\",\n name=tool_name,\n tool_call_id=tool_call_id,\n )\n # you can use a different messages state key here, if your agent uses a different schema\n # e.g., \"alice_messages\" instead of \"messages\"\n last_agent_message = state[\"messages\"][-1]\n # if the tool schema includes task description that LLM generates,\n # you can extract it from the last tool call argument and pass it on to the next agent\n task_description = last_agent_message.tool_calls[0][\"args\"][\"task_description\"]\n return Command(\n goto=agent_name,\n graph=Command.PARENT,\n # NOTE: this is a state update that will be applied to the swarm multi-agent graph (i.e., the PARENT graph)\n update={\n \"messages\": [last_agent_message, tool_message],\n \"active_agent\": agent_name,\n # optionally pass the task description to the next agent\n \"task_description\": task_description,\n },\n )\n\n return handoff_to_agent\n```\n\n> [!IMPORTANT]\n> If you are implementing custom handoff tools that return `Command`, you need to ensure that: \n (1) your agent has a tool-calling node that can handle tools returning `Command` (like LangGraph's prebuilt [`ToolNode`](https://langchain-ai.github.io/langgraph/reference/prebuilt/#langgraph.prebuilt.tool_node.ToolNode)) \n (2) both the swarm graph and the next agent graph have the [state schema](https://langchain-ai.github.io/langgraph/concepts/low_level#schema) containing the keys you want to update in `Command.update`\n\n### Customizing agent implementation\n\nBy default, individual agents are expected to communicate over a single `messages` key that is shared by all agents and the overall multi-agent swarm graph. This means that **all** of the messages from **all** of the agents will be combined into a single, shared list of messages. This might not be desirable if you don't want to expose an agent's internal history of messages. To change this, you can customize the agent by taking the following steps:\n\n1. use custom [state schema](https://langchain-ai.github.io/langgraph/concepts/low_level#schema) with a different key for messages, for example `alice_messages`\n1. write a wrapper that converts the parent graph state to the child agent state and back (see this [how-to](https://langchain-ai.github.io/langgraph/how-tos/subgraph-transform-state/) guide)\n\n```python\nfrom typing_extensions import TypedDict, Annotated\n\nfrom langchain_core.messages import AnyMessage\nfrom langgraph.graph import StateGraph, add_messages\nfrom langgraph_swarm import SwarmState\n\nclass AliceState(TypedDict):\n alice_messages: Annotated[list[AnyMessage], add_messages]\n\n# see this guide to learn how you can implement a custom tool-calling agent\n# https://langchain-ai.github.io/langgraph/how-tos/react-agent-from-scratch/\nalice = (\n StateGraph(AliceState)\n .add_node(\"model\", ...)\n .add_node(\"tools\", ...)\n .add_edge(...)\n ...\n .compile()\n)\n\n# wrapper calling the agent\ndef call_alice(state: SwarmState):\n # you can put any input transformation from parent state -> agent state\n # for example, you can invoke \"alice\" with \"task_description\" populated by the LLM\n response = alice.invoke({\"alice_messages\": state[\"messages\"]})\n # you can put any output transformation from agent state -> parent state\n return {\"messages\": response[\"alice_messages\"]}\n\ndef call_bob(state: SwarmState):\n ...\n```\n\nThen, you can create the swarm manually in the following way:\n\n```python\nfrom langgraph_swarm import add_active_agent_router\n\nworkflow = (\n StateGraph(SwarmState)\n .add_node(\"Alice\", call_alice, destinations=(\"Bob\",))\n .add_node(\"Bob\", call_bob, destinations=(\"Alice\",))\n)\n# this is the router that enables us to keep track of the last active agent\nworkflow = add_active_agent_router(\n builder=workflow,\n route_to=[\"Alice\", \"Bob\"],\n default_active_agent=\"Alice\",\n)\n\n# compile the workflow\napp = workflow.compile()\n```\n\n",
"bugtrack_url": null,
"license": null,
"summary": "An implementation of a multi-agent swarm using LangGraph",
"version": "0.0.3",
"project_urls": null,
"split_keywords": [],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "6eb90042a39f4db050c38a09d6a32a531fabcd3c3f56c70ab489822fdfbe9777",
"md5": "0525d461aba7faebf2578fa337ccebb2",
"sha256": "859711e100368e9dd4a05d881a2c26271969bbeb561d80a0228606009683a1cc"
},
"downloads": -1,
"filename": "langgraph_swarm-0.0.3-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0525d461aba7faebf2578fa337ccebb2",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.10",
"size": 8081,
"upload_time": "2025-02-27T00:13:50",
"upload_time_iso_8601": "2025-02-27T00:13:50.562954Z",
"url": "https://files.pythonhosted.org/packages/6e/b9/0042a39f4db050c38a09d6a32a531fabcd3c3f56c70ab489822fdfbe9777/langgraph_swarm-0.0.3-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "946cd9dc3ed0181418f5ea619981cf3d18dc469ca11a0b103d9b1f6a83b187a0",
"md5": "e683998b3703ec26a3fe597573febabe",
"sha256": "249616dffa25c18a17cb6f8277b1516702be57faa386166c100d11675f11edc2"
},
"downloads": -1,
"filename": "langgraph_swarm-0.0.3.tar.gz",
"has_sig": false,
"md5_digest": "e683998b3703ec26a3fe597573febabe",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.10",
"size": 6964,
"upload_time": "2025-02-27T00:13:51",
"upload_time_iso_8601": "2025-02-27T00:13:51.883533Z",
"url": "https://files.pythonhosted.org/packages/94/6c/d9dc3ed0181418f5ea619981cf3d18dc469ca11a0b103d9b1f6a83b187a0/langgraph_swarm-0.0.3.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-02-27 00:13:51",
"github": false,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"lcname": "langgraph-swarm"
}
