🤖 LangGraph Multi-Agent Swarm

一个使用 LangGraph 创建蜂群式多智能体 (multi-agent) 系统的 Python 库。蜂群是一种 多智能体 架构类型，其中智能体根据其专长动态地将控制权移交给彼此。系统会记住最后活跃的智能体是谁，确保在后续交互中，对话能与该智能体继续。

Features

• 🤖 多智能体协作 - 启用专用智能体协同工作并相互传递上下文
• 🛠️ 可自定义的交接工具 - 内置用于智能体间通信的工具

本库基于 LangGraph 构建，这是一个用于构建智能体应用的强大框架，并开箱即用地支持 流式传输 (streaming)、短期和长期记忆 (memory) 以及 人机回环 (human-in-the-loop)

Installation

pip install langgraph-swarm

Quickstart

pip install langgraph-swarm langchain-openai

export OPENAI_API_KEY=<your_api_key>

from langchain_openai import ChatOpenAI

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_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_agent(
    model,
    tools=[
        add,
        create_handoff_tool(
            agent_name="Bob",
            description="Transfer to Bob",
        ),
    ],
    system_prompt="You are Alice, an addition expert.",
    name="Alice",
)

bob = create_agent(
    model,
    tools=[
        create_handoff_tool(
            agent_name="Alice",
            description="Transfer to Alice, she can help with math",
        ),
    ],
    system_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)

[!TIP] 对于开发、调试和部署 AI 智能体和 大语言模型 (LLM) 应用，请参阅 LangSmith。

Memory

您可以为您的蜂群多智能体系统添加 短期 (short-term) 和 长期 (long-term) 记忆 (memory)。由于 create_swarm() 返回一个需要在编译后才能使用的 状态图 (StateGraph) 实例，您可以直接将 检查点保存器 (checkpointer) 或 存储 (store) 实例传递给 .compile() 方法：

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] 添加 短期记忆 (short-term memory) 对于跨多次交互维持对话状态至关重要。如果没有它，蜂群将“忘记”最后活跃的智能体是谁并丢失对话历史。如果您计划在多轮对话中使用它，请确保始终使用检查点保存器编译蜂群；例如，workflow.compile(checkpointer=checkpointer)。

How to customize

您可以通过更改 交接工具 (handoff tools) 实现或 智能体 (agent) 实现来自定义多智能体蜂群。

自定义交接工具

默认情况下，群集（swarm）中的代理（agents）假定使用通过预构建的 create_handoff_tool 创建的交接工具（handoff tools）。你也可以创建自己的自定义交接工具。以下是一些关于如何修改默认实现的想法：

• 更改工具名称和/或描述
• 添加供大语言模型（LLM）填充的工具调用参数，例如下一个代理的任务描述
• 更改作为交接一部分传递给下一个代理的数据：默认情况下，create_handoff_tool 传递完整的消息历史（直到当前点群集中生成的所有消息），以及一个指示交接成功的工具消息。

以下是自定义交接工具可能的外观示例：

from typing import Annotated

from langchain.tools import tool, BaseTool, InjectedToolCallId
from langchain.messages import ToolMessage
from langgraph.types import Command
from langgraph.prebuilt import InjectedState

def create_custom_handoff_tool(*, agent_name: str, name: str | None, description: str | None) -> BaseTool:

    @tool(name, description=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=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"
        messages = state["messages"]
        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": messages + [tool_message],
                "active_agent": agent_name,
                # optionally pass the task description to the next agent
                "task_description": task_description,
            },
        )

    return handoff_to_agent

[!IMPORTANT] 如果你正在实现返回 Command 的自定义交接工具，你需要确保：
(1) 你的代理拥有一个可以处理返回 Command 的工具的调用节点（例如 LangGraph 预构建的 ToolNode）
(2) 群集图和下一个代理图都拥有包含你希望在 Command.update 中更新的键的 状态架构

自定义代理实现

默认情况下，各个代理预期通过一个由所有代理和整体多代理群集图共享的单一 messages 键进行通信。这意味着来自所有代理的消息将被合并到一个单一的共享消息列表中。如果你不希望暴露代理的内部消息历史，这可能不是理想的方案。要更改这一点，你可以通过以下步骤自定义代理：

1. 使用自定义的 状态架构，为消息使用不同的键，例如 alice_messages
2. 编写一个包装器，将父图状态转换为子代理状态并转换回来（参见此 操作指南 指南）

from typing_extensions import TypedDict, Annotated

from langchain.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):
    ...

然后，你可以按以下方式手动创建群集：

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()

🤖 langgraph-swarm-py 快速上手指南

LangGraph Multi-Agent Swarm 是一个基于 LangGraph 的 Python 库，用于构建蜂群式多智能体系统。它允许智能体根据专长动态交接控制权，并自动记录最后活跃的智能体以维持上下文。

1. 环境准备

• Python 版本：建议 3.8 及以上
• 前置依赖：需配置 OpenAI API 密钥（示例基于 OpenAI 模型）
• 网络环境：建议配置国内镜像源以加速依赖下载

2. 安装步骤

使用国内镜像源安装核心库及 OpenAI 适配器：

pip install -i https://pypi.tuna.tsinghua.edu.cn/simple langgraph-swarm langchain-openai

设置环境变量（请替换为真实密钥）：

export OPENAI_API_KEY=<your_api_key>

3. 基本使用

以下示例演示了如何创建两个智能体（Alice 和 Bob），定义交接工具，并编译支持多轮对话记忆的工作流。

from langchain_openai import ChatOpenAI

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_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_agent(
    model,
    tools=[
        add,
        create_handoff_tool(
            agent_name="Bob",
            description="Transfer to Bob",
        ),
    ],
    system_prompt="You are Alice, an addition expert.",
    name="Alice",
)

bob = create_agent(
    model,
    tools=[
        create_handoff_tool(
            agent_name="Alice",
            description="Transfer to Alice, she can help with math",
        ),
    ],
    system_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)

💡 关键说明

• 记忆持久化：务必在 workflow.compile() 时传入 checkpointer（如 InMemorySaver）。否则系统将无法记住上一轮活跃的智能体，导致多轮对话中断。
• 智能体交接：通过 create_handoff_tool 定义智能体间的通信规则，LLM 会根据上下文自动调用交接工具切换智能体。