LLM TREND

大模型技术趋势分享：LLM、RAG、Prompt、Agent、Fine-tuning等

Agent

使用LangGraph构建多智能体Swarm：入门指南

By llmtrend #Agent

随着大模型技术的飞速发展，我们越来越希望AI助手能够像人类专家一样，根据不同的需求无缝切换不同的“人格”。 LangGraph的Swarm架构提供了一种强大的解决方案，使我们能够构建一个多智能体系统，让多个专业化的智能体协同工作，自动将任务分配给最合适的专家。本文将带你了解如何使用LangGraph构建一个简单但强大的多智能体Swarm，深入探讨代码，解释每个步骤，并探讨如何将其扩展到实际应用中。

什么是LangGraph中的Swarm？

在LangGraph中，Swarm是一个多智能体系统，它能够将用户查询分配给最专业的智能体。 简单来说，它可以将用户查询路由到最合适的智能体。 智能体之间可以“交接”任务，从而创建一个协作式AI系统。

例如，在客户支持方面，可以有专门处理账单问题的智能体、提供技术支持的智能体以及负责销售的智能体。 在提高工作效率方面，可以有日历智能体、邮件智能体和代码助手。 甚至可以创建一个个人助理，其中包含旅行智能体、天气智能体和购物智能体。

这种Swarm架构的核心优势在于其专业化和协作能力。 每个智能体都专注于特定的领域，可以提供更精准、更高效的服务。 同时，智能体之间的任务交接，使得系统能够处理更复杂、更综合的问题，从而提升整体的用户体验。

构建Swarm的核心步骤

构建一个LangGraph Swarm 主要包含以下几个步骤：

  1. 设置语言模型：选择合适的语言模型是构建智能体的基础。 文章中使用的是 AzureChatOpenAI，你需要替换为你自己的 API 密钥、Azure 端点和部署名称。

    from langchain_openai import AzureChatOpenAI

llm = AzureChatOpenAI(
    openai_api_key="YOUR_API_KEY",
    azure_endpoint="YOUR_AZURE_ENDPOINT",
    azure_deployment="YOUR_DEPLOYMENT_NAME",
    api_version="2025-01-01-preview",
    temperature=0.7,
)

  2. 创建智能体：根据不同的专业领域创建不同的智能体。 文章中创建了三个智能体：旅行智能体、天气智能体和代码智能体。 每个智能体都配备了相应的工具和提示，以使其能够专注于特定任务。

    • 旅行智能体：负责处理与旅行相关的查询，例如搜索航班、预订酒店等。
    • 天气智能体：负责提供天气信息。
    • 代码智能体：负责解答编程问题和解决代码问题。

    下面是创建智能体的代码示例，使用了create_react_agent函数，这个函数会基于指定的LLM，工具，提示词创建一个能够进行ReAct（Reasoning and Acting）的智能体。每个智能体都有对应的工具集，例如旅行智能体有搜索航班、预订航班、搜索酒店和预订酒店等工具：

    from langchain.agents import create_react_agent
from langchain.tools import Tool
from langchain_core.runnables import RunnableConfig
from langchain_community.utilities import GoogleSerperAPIWrapper
import os

# 示例数据（实际应用中需要替换为真实的API调用或数据库查询）
FLIGHTS = [{"id": "F123", "departure_airport": "JFK", "arrival_airport": "LAX", "date": "2024-01-01", "price": 300}]
HOTELS = [{"id": "H456", "location": "LA", "name": "Luxury Hotel", "price": 200}]
RESERVATIONS = {"user_id": {"flight_info": None, "hotel_info": None}}

def search_flights(departure_airport: str, arrival_airport: str, date: str) -> list[dict]:
    """Search for available flights given departure airport, arrival airport, and date."""
    return FLIGHTS

def book_flight(flight_id: str, config: RunnableConfig) -> str:
    """Book a flight by flight_id for the user in config."""
    user_id = config["configurable"].get("user_id")
    flight = [flight for flight in FLIGHTS if flight["id"] == flight_id][0]
    RESERVATIONS[user_id]["flight_info"] = flight
    return "Successfully booked flight"

def search_hotels(location: str) -> list[dict]:
    """Search for available hotels in a given location."""
    return HOTELS

def book_hotel(hotel_id: str, config: RunnableConfig) -> str:
    """Book a hotel by hotel_id for the user in config."""
    user_id = config["configurable"].get("user_id")
    hotel = [hotel for hotel in HOTELS if hotel["id"] == hotel_id][0]
    RESERVATIONS[user_id]["hotel_info"] = hotel
    return "Successfully booked hotel"

search = GoogleSerperAPIWrapper(serper_api_key=os.getenv("SERPAPI_API_KEY"))

weather_tool = Tool(
    name="Get_Weather",
    func=lambda location: search.run(f"weather in {location}"),
    description="Get the current weather for a given location. Input should be a city name."
)

travel_agent = create_react_agent(
    llm,
    tools=[search_flights, book_flight, search_hotels, book_hotel, transfer_to_weather_agent, transfer_to_code_agent],
    prompt="You are a travel assistant that can help users search and book flights and hotels.",
    name="travel_agent"
)

# Weather Agent
weather_agent = create_react_agent(
    llm,
    tools=[weather_tool, transfer_to_travel_agent, transfer_to_code_agent],
    prompt="You are a weather assistant. You can answer questions about the weather in any city.",
    name="weather_agent"
)

# Code Agent
code_agent = create_react_agent(
    llm,
    tools=[transfer_to_travel_agent, transfer_to_weather_agent],
    prompt="You are a code assistant. You help users with programming questions and code problems.",
    name="code_agent"
)

  3. 创建交接工具：为了实现智能体之间的任务交接，需要创建交接工具。 这些工具允许智能体将用户查询传递给最合适的智能体。

    from langchain.tools import Tool
def create_handoff_tool(agent_name: str, description: str) -> Tool:
    """Creates a tool that allows an agent to hand off to another agent."""
    def handoff_func():
        return f"Transferring to {agent_name} for {description}."
    return Tool(
    name=f"transfer_to_{agent_name}",
    func=handoff_func,
    description=description,
)

    

    transfer_to_travel_agent = create_handoff_tool(
    
    agent_name="travel_agent",
    
    description="Transfer to the travel assistant for travel-related questions."
    
)
    

    transfer_to_weather_agent = create_handoff_tool(
    
    agent_name="weather_agent",
    
    description="Transfer to the weather assistant for weather-related questions."
    
)
    

    transfer_to_code_agent = create_handoff_tool(
    
    agent_name="code_agent",
    
    description="Transfer to the code assistant for programming questions."
    
)
    

    # 将交接工具添加到各个智能体的工具列表中 (见上面的智能体创建代码)

  4. 创建Swarm Supervisor：Swarm Supervisor 负责管理所有智能体和流程。 它需要列出所有智能体，并指定默认的活动智能体。 此外，还需要一个检查点（checkpointer）来保存会话状态。

    from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage
from typing import List, Dict, TypedDict

class AgentState(TypedDict):
    messages: List[BaseMessage]
    agent: str
    steps: List[str]

from langchain_core.runnables import chain
from langchain_core.tools import BaseTool
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser

def create_react_agent(llm, tools: List[BaseTool], prompt: str, name: str):
    prompt_template = ChatPromptTemplate.from_messages(
    [
        ("system", prompt),
        MessagesPlaceholder(variable_name="messages"),
        MessagesPlaceholder(variable_name="agent_scratchpad"),
    ]
)

llm_with_tool_calls = llm.bind_tools(tools)

agent = (
    {
        "messages": lambda x: x["messages"],
        "agent_scratchpad": lambda x: x["steps"],
    }
    | prompt_template
    | llm_with_tool_calls
    | StrOutputParser()
)
return agent

    

    def agent_node(state: AgentState, agent, name):
    
    """Single Agent Node"""
    
    next_state = agent.invoke(state)
    
    return {"messages": state["messages"] + [next_state]}
    

    def route_to_agent(state):
    
    """Route to the agent"""
    
    return state["agent"]
    

    def create_swarm(agents: List, default_active_agent: str):
    
    """Create a swarm"""
    
    workflow = StateGraph(AgentState)
    

    for agent in agents:
    workflow.add_node(agent.name, chain(
        {
            "messages": lambda x: x["messages"],
            "agent_scratchpad": lambda x: "", # TODO: Use chain intermediate values
        }
        | ChatPromptTemplate.from_messages([
            ("system", agent.prompt),
            MessagesPlaceholder(variable_name="messages"),
            MessagesPlaceholder(variable_name="agent_scratchpad")
        ])
        | llm
        | StrOutputParser()
    ))

workflow.set_entry_point(default_active_agent)

for agent in agents:
    workflow.add_edge(agent.name, END)

return workflow

    

    from langchain_core.tracers import ConsoleCallbackHandler
    
from langchain_core.runnables import chain
    

    class MemorySaver:
    
    """Simple stateful node."""
    

    def __init__(self):
    self.state = {}

def run(self, keys, config):
    thread_id = config["configurable"]["thread_id"]
    # This stores the messages
    if thread_id not in self.state:
        self.state[thread_id] = []

    self.state[thread_id].extend(keys["messages"])
    return {"messages": self.state[thread_id]}

    

    checkpointer = MemorySaver()
    
swarm = create_swarm(
    
    agents=[travel_agent, weather_agent, code_agent],
    
    default_active_agent="travel_agent",
    
)
    
app = swarm.compile(checkpointer=checkpointer)

  5. 运行Swarm：编写代码来运行 Swarm，并与多智能体系统进行交互。 你可以使用 app.invoke 方法来提交用户查询，并获取结果。

    import uuid

config = {"configurable": {"thread_id": str(uuid.uuid4()), "user_id": "1"}}

while True:
    user_input = input("\nEnter your request (or 'exit' to quit): ")
    if user_input.lower() == 'exit':
        print("Goodbye!")
        break
    result = app.invoke(
    {"messages": [{"role": "user", "content": user_input}]},
    config,
)

for m in result["messages"]:
    print(m.content)

Swarm的强大之处

LangGraph Swarm 架构的强大之处在于以下几个方面：

  • 智能体专业化：每个智能体都专注于特定的领域，可以提供更精准、更高效的服务。 例如，旅行智能体可以专注于航班和酒店预订，而天气智能体可以专注于提供天气信息。
  • 可扩展性：可以根据需要添加更多的智能体，以覆盖更多的领域。 例如，可以添加购物智能体、健康智能体等。 这使得系统能够不断扩展，以满足不断变化的需求。
  • 智能体之间的协作：智能体可以相互交接任务，从而创建一个更智能、更有帮助的助手。 例如，如果用户询问“明天北京的天气怎么样？”，系统可以首先将问题传递给天气智能体，获取天气信息，然后将信息传递给旅行智能体，询问是否有飞往北京的航班，并将所有信息整合后返回给用户。

如何进一步优化Swarm？

为了进一步提升 Swarm 的性能和用户体验，可以考虑以下几个方面：

  • 添加更多智能体：根据实际需求，添加更多专业领域的智能体，例如旅行智能体、数学智能体、健康智能体等。 领域覆盖的越全面，能解决的问题就越多。
  • 使用更高级的交接逻辑：目前的交接逻辑比较简单，可以基于意图检测等技术，实现更智能的任务分配。 例如，可以使用自然语言处理（NLP）技术来分析用户查询的意图，并将查询传递给最合适的智能体。
  • 集成外部API：为每个智能体的专业领域集成外部 API，以获取更丰富的数据和功能。 例如，旅行智能体可以集成航班预订 API 和酒店预订 API，以便用户可以直接在系统中预订航班和酒店。天气智能体可以集成专业的API来获取更准确和实时的天气信息。

实际应用案例

Swarm 架构在实际应用中具有广泛的应用前景。 以下是一些示例：

  • 客户支持：构建一个包含账单智能体、技术支持智能体和销售智能体的 Swarm，以提供更全面、更高效的客户支持服务。 用户可以通过自然语言与系统进行交互，系统会自动将问题分配给最合适的智能体，从而提高客户满意度。
  • 智能助手：构建一个包含日历智能体、邮件智能体和代码助手的 Swarm，以提高工作效率。 例如，用户可以通过语音或文本指令安排日程、发送邮件、生成代码等。
  • 教育领域：构建一个包含数学智能体、物理智能体和历史智能体的 Swarm，以提供个性化的学习辅导。 学生可以向系统提问，系统会将问题分配给最合适的智能体，并提供详细的解答和指导。

结论

LangGraph Swarm 架构是一种强大而灵活的解决方案，可以用于构建多智能体系统。 它的专业化、可扩展性和协作能力使其能够处理更复杂、更综合的问题，从而提升整体的用户体验。 通过不断添加新的智能体、优化交接逻辑和集成外部 API，我们可以构建出更智能、更有帮助的 AI 助手。 Swarm 架构代表了 AI 发展的一个重要方向，它将我们从通用 AI 推向一个更加专业化和更有能力的未来。 它使得我们可以很容易添加其他专业智能体，将成为未来的趋势。希望本文能帮助大家理解和应用LangGraph Swarm 架构，共同探索 AI 的无限可能。

By llmtrend

Related Post

Agent

本地构建强大AI智能体：无需云端，无需订阅，只需代码

llmtrend
Agent

AI通信协议深度解析：MCP、ACP与A2A的对比与选择

llmtrend
Agent

大模型时代：探索Agentic Patterns，构建高效的自主Agent协作模式

llmtrend

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注

You Missed

LLM

pydiode：揭秘 ChatGPT Python Canvas 的运行时真相

LLM

大语言模型 (LLM)：聊天机器人背后的智能大脑，理解 Tokens、Transformers 以及 ChatGPT 的工作原理

LLM

大模型时代的“天鹅绒牢笼”：重塑智识主权宣言

LLM

利用生成式AI（GenAI）进行数据增强与合成：提升模型性能的实战指南