LangChain 多智能体架构：深入解析 Swarm 的 Handoffs 模式

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:

# 更新状态 Schema 以包含具体的智能体名称类型
state_schema = _update_state_schema_agent_names(state_schema, agent_names)
builder = StateGraph(state_schema, context_schema)

# 添加核心路由机制
add_active_agent_router(
    builder,
    route_to=agent_names,
    default_active_agent=default_active_agent,
)

# 注册每个智能体为图节点
for agent in agents:
    builder.add_node(
        agent.name,
        agent,
        destinations=tuple(
            # 自动获取 handoff 目的地，支持 Pregel 对象
            get_handoff_destinations(agent)
        ),
    )

uv add langchain-deepseek langchain-openai langgraph langgraph-swarm dotenv

dependencies = [
    "dotenv>=0.9.9",
    "langchain-deepseek>=1.0.1",
    "langchain-openai>=1.1.6",
    "langgraph>=1.0.5",
    "langgraph-swarm>=0.1.0",
    "python-dotenv",
]

import os
import datetime
from collections import defaultdict
from typing import Callable, List, Dict, Any
from dotenv import load_dotenv

from langchain_core.runnables import RunnableConfig
from langchain_core.runnables.config import ensure_config
from langgraph.checkpoint.memory import MemorySaver
from langchain.agents import create_agent
from langchain.agents.middleware import dynamic_prompt
from langchain_deepseek import ChatDeepSeek
from langgraph_swarm import create_handoff_tool, create_swarm

# 加载环境变量
load_dotenv()

# 配置模型
# 确保 .env 文件中有 DEEPSEEK_API_KEY
if not os.getenv("DEEPSEEK_API_KEY"):
    print("警告: 未找到 DEEPSEEK_API_KEY，请确保 .env 文件配置正确。")

model = ChatDeepSeek(model="deepseek-chat")

# 模拟数据
RESERVATIONS = defaultdict(lambda: {"flight_info": {}, "hotel_info": {}})
TOMORROW = (datetime.date.today() + datetime.timedelta(days=1)).isoformat()

FLIGHTS = [
    {
        "departure_airport": "PEK", # 北京首都
        "arrival_airport": "SHA",   # 上海虹桥
        "airline": "中国国际航空",
        "date": TOMORROW,
        "id": "1",
    },
    {
        "departure_airport": "PEK",
        "arrival_airport": "SZX",   # 深圳宝安
        "airline": "南方航空",
        "date": TOMORROW,
        "id": "2",
    }
]

HOTELS = [
    {
        "location": "上海",
        "name": "和平饭店",
        "neighborhood": "外滩",
        "id": "1",
    },
    {
        "location": "深圳",
        "name": "瑞吉酒店",
        "neighborhood": "罗湖",
        "id": "2",
    }
]

# --- 航班工具 ---

def search_flights(
    departure_airport: str,
    arrival_airport: str,
    date: str,
) -> List[Dict]:
    """查询航班。

    Args:
        departure_airport: 出发机场的三字代码 (例如 PEK, SHA).
        arrival_airport: 到达机场的三字代码.
        date: 日期，格式 YYYY-MM-DD.
    """
    # 简单起见，返回所有航班
    return FLIGHTS


def book_flight(
    flight_id: str,
    config: RunnableConfig,
) -> str:
    """预订航班。"""
    user_id = config["configurable"].get("user_id")
    # 查找航班
    matches = [flight for flight in FLIGHTS if flight["id"] == flight_id]
    if not matches:
        return "未找到指定 ID 的航班。"
    
    flight = matches[0]
    RESERVATIONS[user_id]["flight_info"] = flight
    return "成功预订航班！"


# --- 酒店工具 ---

def search_hotels(location: str) -> List[Dict]:
    """查询酒店。

    Args:
        location: 城市名称 (例如 上海, 深圳).
    """
    # 简单起见，返回所有酒店
    return HOTELS


def book_hotel(
    hotel_id: str,
    config: RunnableConfig,
) -> str:
    """预订酒店。"""
    user_id = config["configurable"].get("user_id")
    # 查找酒店
    matches = [hotel for hotel in HOTELS if hotel["id"] == hotel_id]
    if not matches:
        return "未找到指定 ID 的酒店。"
    
    hotel = matches[0]
    RESERVATIONS[user_id]["hotel_info"] = hotel
    return "成功预订酒店！"

# --- 转接工具 ---

transfer_to_hotel_assistant = create_handoff_tool(
    agent_name="hotel_assistant",
    description="将用户转接给负责查询和预订酒店的酒店助手。",
)

transfer_to_flight_assistant = create_handoff_tool(
    agent_name="flight_assistant",
    description="将用户转接给负责查询和预订航班的航班助手。",
)

# --- 智能体提示词 ---

def make_prompt_middleware(base_system_prompt: str):
    @dynamic_prompt
    def prompt(request) -> str:
        config = ensure_config()
        user_id = config["configurable"].get("user_id")
        current_reservation = RESERVATIONS[user_id]
        system_prompt = (
            base_system_prompt
            + f"\n\n用户当前的预订信息: {current_reservation}"
            + f"\n今天是: {datetime.datetime.now().strftime('%Y-%m-%d')}"
            + "\n请使用中文回答用户的问题。"
        )
        return system_prompt
    return prompt


# --- 定义智能体 ---

flight_assistant = create_agent(
    model,
    tools=[search_flights, book_flight, transfer_to_hotel_assistant],
    middleware=[make_prompt_middleware("你是一个乐于助人的航班预订助手。")],
    name="flight_assistant",
)

hotel_assistant = create_agent(
    model,
    tools=[search_hotels, book_hotel, transfer_to_flight_assistant],
    middleware=[make_prompt_middleware("你是一个乐于助人的酒店预订助手。")],
    name="hotel_assistant",
)

# --- 编译 Swarm ---

checkpointer = MemorySaver()
builder = create_swarm(
    [flight_assistant, hotel_assistant], 
    default_active_agent="flight_assistant"
)

app = builder.compile(checkpointer=checkpointer)

# --- 运行演示 ---

def print_stream(stream):
    """辅助函数：打印流式输出"""
    for ns, update in stream:
        for node, node_updates in update.items():
            # 过滤根图层面的 Agent 消息冒泡，避免重复打印
            if not ns and node in ["flight_assistant", "hotel_assistant"]:
                continue

            if node_updates is None:
                continue

            if isinstance(node_updates, (dict, tuple)):
                node_updates_list = [node_updates]
            elif isinstance(node_updates, list):
                node_updates_list = node_updates
            else:
                raise ValueError(node_updates)

            for node_update in node_updates_list:
                if isinstance(node_update, tuple):
                    continue
                
                # 尝试获取消息并打印
                messages_key = next(
                    (k for k in node_update.keys() if "messages" in k), None
                )
                if messages_key is not None:
                    last_msg = node_update[messages_key][-1]
                    role = "AI" if last_msg.type == "ai" else last_msg.type
                    content = last_msg.content
                    tool_calls = getattr(last_msg, "tool_calls", [])
                    
                    display_name = ns[-1].split(":")[0] if ns else node
                    print(f"\n--- {display_name} ({role}) ---")
                    if content:
                        print(content)
                    if tool_calls:
                        for tc in tool_calls:
                            print(f"[调用工具]: {tc['name']} 参数: {tc['args']}")
                else:
                    # 可能是状态更新，例如 active_agent 改变
                    print(f"\n--- {node} (State Update) ---")
                    print(node_update)


def main():
    import uuid
    thread_id = str(uuid.uuid4())
    config = {"configurable": {"thread_id": thread_id, "user_id": "user_1"}}

    print(f"开始对话 (Thread ID: {thread_id})...\n")

    # 1. 用户查询航班
    user_input_1 = "我想查一下明天从北京(PEK)到上海(SHA)的航班"
    print(f"\n>>> 用户: {user_input_1}")
    
    events = app.stream(
        {"messages": [{"role": "user", "content": user_input_1}]},
        config,
        subgraphs=True,
    )
    print_stream(events)

    # 2. 用户确认预订
    user_input_2 = "请帮我预订中国国际航空的那一班"
    print(f"\n\n>>> 用户: {user_input_2}")
    
    events = app.stream(
        {"messages": [{"role": "user", "content": user_input_2}]},
        config,
        subgraphs=True,
    )
    print_stream(events)

    # 3. 用户查询酒店 (触发 Handoff)
    user_input_3 = "好的，我也需要订一个上海的酒店"
    print(f"\n\n>>> 用户: {user_input_3}")
    
    events = app.stream(
        {"messages": [{"role": "user", "content": user_input_3}]},
        config,
        subgraphs=True,
    )
    print_stream(events)
    
    # 4. 用户预订酒店
    user_input_4 = "帮我预订和平饭店"
    print(f"\n\n>>> 用户: {user_input_4}")
    
    events = app.stream(
        {"messages": [{"role": "user", "content": user_input_4}]},
        config,
        subgraphs=True,
    )
    print_stream(events)

if __name__ == "__main__":
    main()

uv run python customer_support_zh.py
开始对话 (Thread ID: 1c4e0536-4913-4911-b7a7-227910062e21)...


>>> 用户: 我想查一下明天从北京(PEK)到上海(SHA)的航班

--- flight_assistant (AI) ---
我来帮您查询明天从北京(PEK)到上海(SHA)的航班。今天是2026年1月2日，明天就是2026年1月3日。
[调用工具]: search_flights 参数: {'departure_airport': 'PEK', 'arrival_airport': 'SHA', 'date': '2026-01-03'}

--- flight_assistant (tool) ---
[{"departure_airport": "PEK", "arrival_airport": "SHA", "airline": "中国国际航空", "date": "2026-01-03", "id": "1"}, {"departure_airport": "PEK", "arrival_airport": "SZX", "airline": "南方航空", "date": "2026-01-03", "id": "2"}]

--- flight_assistant (AI) ---
我为您查询到了明天（2026年1月3日）从北京(PEK)到上海(SHA)的航班信息：

1. **中国国际航空**航班
   - 航班ID：1
   - 出发机场：PEK（北京首都国际机场）
   - 到达机场：SHA（上海虹桥国际机场）
   - 日期：2026年1月3日

另外还有一个航班是到深圳的，但您要的是到上海的。

请问您需要预订这个中国国际航空的航班吗？如果需要预订，我可以帮您办理。

>>> 用户: 请帮我预订中国国际航空的那一班

--- flight_assistant (AI) ---
好的，我来帮您预订中国国际航空的航班（航班ID：1）。
[调用工具]: book_flight 参数: {'flight_id': '1'}

--- flight_assistant (tool) ---
成功预订航班！

--- flight_assistant (AI) ---
太好了！我已经成功为您预订了中国国际航空的航班。

**预订详情：**
- 航空公司：中国国际航空
- 出发机场：北京首都国际机场 (PEK)
- 到达机场：上海虹桥国际机场 (SHA)
- 日期：2026年1月3日
- 航班ID：1

您的航班预订已经完成！请问您还需要查询或预订酒店吗？我可以帮您转接到酒店助手。

>>> 用户: 好的，我也需要订一个上海的酒店

--- flight_assistant (AI) ---
好的，我来帮您转接到酒店助手，他们可以帮您查询和预订上海的酒店。
[调用工具]: transfer_to_hotel_assistant 参数: {}

--- hotel_assistant (AI) ---
您好！我是酒店预订助手。我看到您已经预订了明天（2026年1月3日）从北京到上海的航班，现在需要预订上海的酒店。

让我为您查询上海的酒店信息。
[调用工具]: search_hotels 参数: {'location': '上海'}

--- hotel_assistant (tool) ---
[{"location": "上海", "name": "和平饭店", "neighborhood": "外滩", "id": "1"}, {"location": "深圳", "name": "瑞吉酒店", "neighborhood": "罗湖", "id": "2"}]

--- hotel_assistant (AI) ---
我为您查询到了上海的酒店信息：

1. **和平饭店**
   - 位置：上海
   - 区域：外滩
   - 酒店ID：1

另外还有一个深圳的酒店，但您需要的是上海的酒店。

请问您想要预订上海外滩的和平饭店吗？如果需要预订，我可以帮您办理。

>>> 用户: 帮我预订和平饭店

--- hotel_assistant (AI) ---
好的，我来帮您预订上海的和平饭店（酒店ID：1）。
[调用工具]: book_hotel 参数: {'hotel_id': '1'}

--- hotel_assistant (tool) ---
成功预订酒店！

--- hotel_assistant (AI) ---
太好了！我已经成功为您预订了上海的和平饭店。

**预订详情：**
- 酒店名称：和平饭店
- 位置：上海
- 区域：外滩
- 酒店ID：1

**您的完整行程安排：**
1. **航班**：中国国际航空，2026年1月3日从北京(PEK)飞往上海(SHA)
2. **酒店**：上海和平饭店，位于外滩区域

您的航班和酒店预订都已经完成了！请问您还需要其他帮助吗？