LangGraph Swarm高级定制：自定义状态模式与智能体通信协议

【免费下载链接】langgraph-swarm-py 项目地址: https://gitcode.com/gh_mirrors/la/langgraph-swarm-py

LangGraph Swarm是一个强大的多智能体协作框架，它允许开发者构建复杂的智能体系统，实现智能体之间的高效通信与协作。本文将深入探讨如何通过自定义状态模式和智能体通信协议来高级定制LangGraph Swarm，帮助你构建更灵活、更强大的多智能体系统。

什么是LangGraph Swarm？

LangGraph Swarm是一个基于LangGraph构建的多智能体协作框架，它提供了创建和管理多个智能体的能力，使智能体能够相互通信、协作完成复杂任务。通过LangGraph Swarm，你可以轻松构建具有分工协作能力的智能体系统，如客户支持系统、研究助手等。

图：LangGraph Swarm架构示意图，展示了多智能体之间的通信与协作流程

自定义状态模式

在LangGraph Swarm中，状态模式定义了智能体系统的状态结构。默认情况下，LangGraph Swarm提供了SwarmState类，它继承自MessagesState，并添加了active_agent字段来跟踪当前活动的智能体。

SwarmState基础结构

class SwarmState(MessagesState):
    """State schema for the multi-agent swarm."""
    active_agent: str | None

这个基础状态模式包含了消息列表和当前活动智能体的信息。但在实际应用中，你可能需要扩展这个状态模式以满足特定需求。

扩展自定义状态

要创建自定义状态模式，你可以继承SwarmState并添加新的字段。例如，如果你需要跟踪任务进度，可以添加一个task_progress字段：

class CustomSwarmState(SwarmState):
    task_progress: float = 0.0
    current_task: str | None = None

然后，在创建Swarm时指定这个自定义状态模式：

workflow = create_swarm(
    [alice, bob],
    default_active_agent="Alice",
    state_schema=CustomSwarmState
)

更新状态模式的智能体名称

LangGraph Swarm提供了_update_state_schema_agent_names函数，它可以自动更新状态模式中的active_agent字段，将其类型设置为智能体名称的字面量类型。这有助于在开发过程中提供更好的类型提示和验证。

def _update_state_schema_agent_names(
    state_schema: StateSchemaType,
    agent_names: list[str],
) -> StateSchemaType:
    # 实现细节

智能体通信协议

智能体之间的通信是多智能体系统的核心。LangGraph Swarm提供了灵活的机制来定义和管理智能体之间的通信协议。

基于工具的通信

LangGraph Swarm中的智能体可以通过工具进行通信。create_handoff_tool函数可以创建一个用于在智能体之间切换的工具：

from langgraph_swarm import create_handoff_tool

handoff_tool = create_handoff_tool(
    agent_name="Bob",
    description="Transfer to Bob"
)

这个工具允许智能体将对话转交给另一个智能体，实现智能体之间的无缝协作。

路由机制

add_active_agent_router函数实现了基于当前活动智能体的路由机制。它会根据状态中的active_agent字段将请求路由到相应的智能体：

def add_active_agent_router(
    builder: StateGraph,
    *,
    route_to: list[str],
    default_active_agent: str,
) -> StateGraph:
    # 实现细节

这个路由机制是智能体通信的核心，它确保请求被发送到正确的智能体处理。

消息传递

智能体之间的通信主要通过消息传递实现。SwarmState中的messages字段（继承自MessagesState）存储了对话历史，智能体可以通过读取和写入这个字段来进行通信。

创建自定义Swarm

create_swarm函数是创建自定义Swarm的入口点。它接受智能体列表、默认活动智能体和自定义状态模式等参数，并返回一个配置好的StateGraph：

def create_swarm(
    agents: list[Pregel],
    *,
    default_active_agent: str,
    state_schema: StateSchemaType = SwarmState,
    context_schema: type[Any] | None = None,
    **deprecated_kwargs: Unpack[DeprecatedKwargs],
) -> StateGraph:
    # 实现细节

通过这个函数，你可以轻松创建一个包含多个智能体的Swarm，并自定义其状态模式和通信协议。

实际应用示例

让我们通过一个简单的示例来展示如何使用自定义状态模式和通信协议：

from langgraph.checkpoint.memory import InMemorySaver
from langchain.agents import create_agent
from langgraph_swarm import create_handoff_tool, create_swarm

# 定义工具
def add(a: int, b: int) -> int:
    '''Add two numbers'''
    return a + b

# 创建智能体
alice = create_agent(
    "openai:gpt-4o",
    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(
    "openai:gpt-4o",
    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",
)

# 创建自定义状态模式
class CustomSwarmState(SwarmState):
    task_progress: float = 0.0

# 创建Swarm
checkpointer = InMemorySaver()
workflow = create_swarm(
    [alice, bob],
    default_active_agent="Alice",
    state_schema=CustomSwarmState
)
app = workflow.compile(checkpointer=checkpointer)

# 使用Swarm
config = {"configurable": {"thread_id": "1"}}
turn_1 = app.invoke(
    {"messages": [{"role": "user", "content": "i'd like to speak to Bob"}]},
    config,
)
turn_2 = app.invoke(
    {"messages": [{"role": "user", "content": "what's 5 + 7?"}]},
    config,
)

在这个示例中，我们创建了两个智能体Alice和Bob，它们可以通过create_handoff_tool相互切换。我们还定义了一个自定义状态模式CustomSwarmState，添加了task_progress字段来跟踪任务进度。

总结

通过自定义状态模式和智能体通信协议，你可以充分发挥LangGraph Swarm的强大功能，构建灵活、高效的多智能体系统。无论是简单的任务分配还是复杂的协作流程，LangGraph Swarm都能为你提供坚实的基础。

希望本文能帮助你更好地理解和使用LangGraph Swarm。如果你想深入了解更多细节，可以查看项目中的源代码，特别是langgraph_swarm/swarm.py文件，其中包含了Swarm的核心实现。

开始你的LangGraph Swarm之旅吧，构建属于你的智能体协作系统！

【免费下载链接】langgraph-swarm-py 项目地址: https://gitcode.com/gh_mirrors/la/langgraph-swarm-py