如何讓多個AI Agent高效協作？手把手教你構建企業級AI智能體系統原創

AI博物院

發布于 2025-8-25 09:32

瀏覽

0收藏

在AI技術快速發展的今天，多智能體系統（Multi-Agent Systems）被稱為"AI的第三波浪潮"，而LangGraph作為新一代的智能體編排框架，為構建復雜的多智能體系統提供了強大的基礎設施。

本文將深入探討LangGraph中的兩個核心概念：**Handoffs（交接）和Supervisor（主管）**模式，通過詳細的技術分析和代碼示例，幫助你掌握構建生產級多智能體系統的關鍵技術。

一、LangGraph多智能體架構概述

1.1 什么是多智能體系統？

在多智能體架構中，智能體可以被表示為圖節點。每個智能體節點執行其步驟并決定是完成執行還是路由到另一個智能體，包括可能路由到自己（例如，在循環中運行）。這種設計使得系統能夠動態地在不同專業化的智能體之間協調任務。

1.2 LangGraph的核心優勢

LangGraph基于Google的Pregel系統設計理念，采用事件驅動架構，提供了以下關鍵特性：

狀態管理：內置的狀態持久化和管理機制
流式處理：原生支持token級別的流式輸出
Human-in-the-loop：支持人工干預和審批流程
靈活的控制流：支持單智能體、多智能體、層級式等多種架構模式

二、Handoffs（交接）機制詳解

2.1 Handoffs的核心概念

Handoffs是多智能體交互中的常見模式，其中一個智能體將控制權移交給另一個智能體。Handoffs允許你指定：destination（目標智能體）和payload（要傳遞給該智能體的信息）。

2.2 Command原語的引入

2024年12月，LangGraph引入了??Command??原語，這是一個革命性的改進。Command是LangGraph中的一種特殊類型，當從節點返回時，它不僅指定狀態更新（如常規），還指定接下來要去的節點。

2.3 實現Handoffs的核心代碼

from typing import Literal, Annotated
from langchain_core.tools import tool
from langgraph.types import Command
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import InjectedState

# 創建handoff工具的函數
def create_handoff_tool(*, agent_name: str, description: str = None):
    name = f"transfer_to_{agent_name}"
    description = description orf"Ask {agent_name} for help."
    
    @tool(name, description=description)
    def handoff_tool(
        task_description: Annotated[
            str, 
            "Description of what the next agent should do"
        ],
        state: Annotated[MessagesState, InjectedState],
    ) -> Command:
        # 創建任務描述消息
        task_message = {"role": "user", "content": task_description}
        # 準備下一個智能體的輸入
        agent_input = {**state, "messages": [task_message]}
        
        return Command(
            goto=agent_name,  # 指定下一個智能體
            update=agent_input,  # 傳遞的狀態更新
            graph=Command.PARENT  # 在父圖中導航
        )
    
    return handoff_tool

2.4 智能體節點中使用Command

def agent_node(state: MessagesState) -> Command[Literal["agent_a", "agent_b", "END"]]:
    # 智能體邏輯處理
    messages = state["messages"]
    
    # 基于某種條件決定路由
    if need_expert_help(messages):
        return Command(
            goto="agent_b",  # 路由到agent_b
            update={"messages": messages + [new_message]}
        )
    elif task_completed(messages):
        return Command(
            goto="END",
            update={"messages": messages + [final_message]}
        )
    else:
        # 繼續當前智能體的處理
        return Command(
            goto="agent_a",  # 路由回自己
            update={"messages": messages + [processing_message]}
        )

三、Supervisor（主管）模式深度剖析

3.1 Supervisor架構設計

supervisor模式中，各個智能體由中央主管智能體協調。主管控制所有通信流和任務委派，根據當前上下文和任務要求決定調用哪個智能體。

3.2 使用langgraph-supervisor庫

LangGraph提供了預構建的??langgraph-supervisor??庫，簡化了主管系統的創建：

from langgraph_supervisor import create_supervisor
from langchain.chat_models import init_chat_model

# 初始化模型
supervisor_model = init_chat_model("openai:gpt-4")

# 創建專業智能體
research_agent = create_react_agent(
    model="openai:gpt-4",
    tools=[web_search_tool, document_reader_tool],
    prompt="You are a research expert. Focus on finding accurate information.",
    name="research_expert"
)

math_agent = create_react_agent(
    model="openai:gpt-4",
    tools=[calculator_tool, equation_solver_tool],
    prompt="You are a mathematics expert. Solve complex calculations.",
    name="math_expert"
)

# 創建主管系統
supervisor_app = create_supervisor(
    agents=[research_agent, math_agent],
    model=supervisor_model,
    prompt=(
        "You are a supervisor managing two agents:\n"
        "- a research agent for information gathering\n"
        "- a math agent for calculations\n"
        "Assign work to one agent at a time based on the task."
    ),
    add_handoff_back_messages=True,
    output_mode="full_history"
).compile()

3.3 自定義Handoff工具

默認情況下，主管使用預構建的create_handoff_tool創建的交接工具。你也可以創建自己的自定義交接工具：

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, 
    name: str = None, 
    description: str = None
) -> BaseTool:
    
    @tool(name, description=description)
    def handoff_to_agent(
        # LLM填充的任務描述
        task_description: Annotated[
            str, 
            "Detailed description of the task including context"
        ],
        # 可以添加額外的參數
        priority: Annotated[
            str,
            "Priority level: high, medium, low"
        ],
        # 注入的狀態
        state: Annotated[dict, InjectedState],
        tool_call_id: Annotated[str, InjectedToolCallId],
    ):
        # 創建工具消息
        tool_message = ToolMessage(
            content=f"Successfully transferred to {agent_name} with priority {priority}",
            name=name,
            tool_call_id=tool_call_id,
        )
        
        # 更新狀態并路由
        return Command(
            goto=agent_name,
            update={
                "messages": state["messages"] + [tool_message],
                "current_task": task_description,
                "priority": priority
            },
            graph=Command.PARENT
        )
    
    return handoff_to_agent

四、高級特性與最佳實踐

4.1 狀態管理與持久化

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.store.memory import InMemoryStore

# 短期記憶（對話狀態）
checkpointer = InMemorySaver()
# 長期記憶（知識存儲）
store = InMemoryStore()

# 編譯時添加持久化支持
app = supervisor_app.compile(
    checkpointer=checkpointer,
    store=store
)

4.2 輸出模式控制

LangGraph提供了靈活的輸出模式控制：

# 創建主管時指定輸出模式
supervisor = create_supervisor(
    agents=[agent1, agent2],
    model=model,
    output_mode="last_message",  # 只返回最后的消息
    # output_mode="full_history"  # 返回完整歷史
)

4.3 消息轉發工具

你可以為主管配備一個工具，直接將從工作智能體收到的最后一條消息轉發到圖的最終輸出：

from langgraph_supervisor.handoff import create_forward_message_tool

# 創建轉發工具
forwarding_tool = create_forward_message_tool("supervisor")

# 在創建主管時添加
workflow = create_supervisor(
    [research_agent, math_agent],
    model=model,
    tools=[forwarding_tool]  # 添加轉發工具
)

4.4 處理復雜的多智能體工作流

class MultiAgentState(TypedDict):
    messages: Annotated[list, add_messages]
    current_agent: str
    task_queue: list
    results: dict

def create_complex_workflow():
    workflow = StateGraph(MultiAgentState)
    
    # 添加智能體節點
    workflow.add_node("supervisor", supervisor_agent)
    workflow.add_node("researcher", research_agent)
    workflow.add_node("analyst", data_analyst_agent)
    workflow.add_node("writer", content_writer_agent)
    
    # 使用Command進行動態路由
    def supervisor_agent(state: MultiAgentState) -> Command:
        task = state["task_queue"][0] if state["task_queue"] elseNone
        
        ifnot task:
            return Command(goto="END", update=state)
        
        # 基于任務類型分配給不同智能體
        if task["type"] == "research":
            return Command(
                goto="researcher",
                update={**state, "current_agent": "researcher"}
            )
        elif task["type"] == "analysis":
            return Command(
                goto="analyst",
                update={**state, "current_agent": "analyst"}
            )
        # ... 更多路由邏輯
    
    return workflow.compile()

五、Swarm模式vs Supervisor模式

5.1 Swarm模式特點

在Swarm架構中，智能體基于其專業化動態地相互傳遞控制權。系統會記住哪個智能體最后處于活動狀態，確保在后續交互中，對話從該智能體恢復。

5.2 選擇合適的模式

Supervisor模式適用于：

需要集中控制和決策的場景
任務分配規則明確的系統
需要嚴格的執行順序控制

Swarm模式適用于：
智能體間平等協作的場景
需要更靈活的動態路由
去中心化的決策制定

六、性能優化與注意事項

6.1 避免狀態膨脹

在長對話中，消息歷史可能變得非常大。考慮實施消息摘要或選擇性傳遞：

def selective_handoff(state: MessagesState) -> Command:
    # 只傳遞最近的N條消息
    recent_messages = state["messages"][-10:]
    
    # 創建摘要
    summary = create_summary(state["messages"][:-10])
    
    return Command(
        goto="next_agent",
        update={"messages": [summary] + recent_messages}
    )

6.2 錯誤處理與恢復

def robust_agent(state: MessagesState) -> Command:
    try:
        # 智能體邏輯
        result = process_task(state)
        return Command(goto="success_handler", update={"result": result})
    except Exception as e:
        # 錯誤恢復
        return Command(
            goto="error_handler",
            update={"error": str(e), "fallback_agent": "supervisor"}
        )

七、實戰案例：構建智能客服系統

讓我們通過一個完整的例子來展示如何使用LangGraph構建一個多智能體客服系統：

from langgraph_supervisor import create_supervisor
from langgraph.prebuilt import create_react_agent
import os

# 設置API密鑰
os.environ["OPENAI_API_KEY"] = "your-api-key"

# 1. 創建專業智能體
# FAQ智能體
faq_agent = create_react_agent(
    model="openai:gpt-3.5-turbo",
    tools=[search_faq_tool, get_product_info_tool],
    prompt="You handle frequently asked questions about products and services.",
    name="faq_specialist"
)

# 技術支持智能體
tech_agent = create_react_agent(
    model="openai:gpt-4",
    tools=[diagnose_issue_tool, check_system_status_tool],
    prompt="You are a technical support specialist. Help users solve technical problems.",
    name="tech_specialist"
)

# 訂單處理智能體
order_agent = create_react_agent(
    model="openai:gpt-3.5-turbo",
    tools=[check_order_status_tool, process_return_tool],
    prompt="You handle order-related inquiries and returns.",
    name="order_specialist"
)

# 2. 創建主管系統
customer_service_system = create_supervisor(
    agents=[faq_agent, tech_agent, order_agent],
    model="openai:gpt-4",
    prompt="""
    You are a customer service supervisor managing three specialists:
    - FAQ specialist: handles general questions
    - Tech specialist: handles technical issues
    - Order specialist: handles order and shipping issues
    
    Analyze the customer's query and delegate to the appropriate specialist.
    If the query spans multiple areas, handle them sequentially.
    """,
    add_handoff_back_messages=True,
    output_mode="last_message"# 返回最終響應
)

# 3. 編譯并運行
app = customer_service_system.compile(
    checkpointer=InMemorySaver()  # 添加對話記憶
)

# 4. 處理客戶查詢
asyncdef handle_customer_query(query: str, thread_id: str):
    config = {"configurable": {"thread_id": thread_id}}
    
    result = await app.ainvoke(
        {"messages": [{"role": "user", "content": query}]},
        config=config
    )
    
    return result["messages"][-1]["content"]

# 使用示例
response = await handle_customer_query(
    "My laptop won't turn on and I want to check if it's still under warranty",
    thread_id="customer_123"
)

八、最佳實踐建議

始終為智能體定義清晰的職責邊界
實施適當的錯誤處理和回退機制
使用checkpointer保持對話狀態的連續性
監控和優化智能體間的交互效率
定期評估和調整智能體的提示詞和工具配置

本文轉載自???AI 博物院??? 作者：longyunfeigu

?著作權歸作者所有，如需轉載，請注明出處，否則將追究法律責任

標簽

AI Agent

AI智能體

LangGraph

贊

回復

舉報

回復

51CTO

51CTO博客

51CTO學堂

如何讓多個AI Agent高效協作？手把手教你構建企業級AI智能體系統原創

一、LangGraph多智能體架構概述

1.1 什么是多智能體系統？

1.2 LangGraph的核心優勢

二、Handoffs（交接）機制詳解

2.1 Handoffs的核心概念

2.2 Command原語的引入

2.3 實現Handoffs的核心代碼

2.4 智能體節點中使用Command

三、Supervisor（主管）模式深度剖析

3.1 Supervisor架構設計

3.2 使用langgraph-supervisor庫

3.3 自定義Handoff工具

四、高級特性與最佳實踐

4.1 狀態管理與持久化

4.2 輸出模式控制

4.3 消息轉發工具

4.4 處理復雜的多智能體工作流

五、Swarm模式vs Supervisor模式

5.1 Swarm模式特點

5.2 選擇合適的模式

六、性能優化與注意事項

6.1 避免狀態膨脹

6.2 錯誤處理與恢復

七、實戰案例：構建智能客服系統

八、最佳實踐建議

目錄

51CTO

51CTO博客

51CTO學堂

如何讓多個AI Agent高效協作？手把手教你構建企業級AI智能體系統 原創

一、LangGraph多智能體架構概述

1.1 什么是多智能體系統？

1.2 LangGraph的核心優勢

二、Handoffs（交接）機制詳解

2.1 Handoffs的核心概念

2.2 Command原語的引入

2.3 實現Handoffs的核心代碼

2.4 智能體節點中使用Command

三、Supervisor（主管）模式深度剖析

3.1 Supervisor架構設計

3.2 使用langgraph-supervisor庫

3.3 自定義Handoff工具

四、高級特性與最佳實踐

4.1 狀態管理與持久化

4.2 輸出模式控制

4.3 消息轉發工具

4.4 處理復雜的多智能體工作流

五、Swarm模式vs Supervisor模式

5.1 Swarm模式特點

5.2 選擇合適的模式

六、性能優化與注意事項

6.1 避免狀態膨脹

6.2 錯誤處理與恢復

七、實戰案例：構建智能客服系統

八、最佳實踐建議

目錄

如何讓多個AI Agent高效協作？手把手教你構建企業級AI智能體系統原創