AI Agents-8 | 構建Muti-Agent系統 原創
在 AI 的世界里，智能體（Agent）就像是一個個擁有獨立思考能力的小機器人，它們能夠根據任務需求做出決策、執行操作，甚至與其他智能體協作完成復雜的任務。今天，就讓我們一起走進多智能體架構的世界，看看這些“變形金剛”是如何通過不同的模式和策略，為我們的生活和工作帶來巨大變革的。
一、智能體的價值：從復雜任務中脫穎而出
當我們思考智能體能夠為哪些場景帶來價值時，不妨先看看那些傳統自動化方法難以觸及的領域：
- 復雜決策：那些需要細膩判斷、處理異常情況或者依賴上下文的任務，比如在客戶服務流程中決定是否批準退款。
- 難以維護的規則：當系統因為龐大的規則集變得難以更新，且每次更新都可能引入錯誤時，智能體或許能提供更靈活的解決方案。例如，對供應商進行安全審查時，復雜的規則判斷往往讓人頭疼。
- 依賴非結構化數據：如果任務涉及自然語言理解、文檔內容提取或者與用戶進行對話交互，那么智能體可能就是最佳選擇。比如處理家庭保險索賠時，需要理解和處理大量的文本信息。
在決定開發智能體之前，一定要明確你的用例是否符合這些場景。如果不符合，可能傳統的確定性解決方案就已經足夠了。
二、智能體設計的三大核心組件
智能體雖然功能強大，但它的基本構成其實并不復雜，主要包括以下三個核心部分：
（一）模型：智能體的“大腦”
模型是驅動智能體推理和決策的大語言模型（LLM）。不同的模型有不同的優勢，比如有的模型擅長處理復雜的任務，而有的模型則在速度和成本上更具優勢。并不是所有任務都需要最強大的模型，簡單的任務（如檢索信息或意圖分類）可以用較小、較快的模型來完成，而復雜的任務（如決定是否批準退款）則可能需要更強大的模型。
一個實用的方法是：先用最強大的模型為每個任務構建智能體原型，以此建立性能基線。然后，嘗試用更小的模型替換，看看是否仍能達到可接受的結果。這樣既不會過早限制智能體的能力，也能幫助你診斷出哪些任務可以用小模型完成，哪些不行。
選擇模型的原則其實很簡單：
- 設置評估來建立性能基線。
- 用最好的模型來達到你的準確率目標。
- 在可能的情況下，用更小的模型替換大模型，以優化成本和延遲。
（二）工具：智能體的“手腳”
工具是智能體用來執行操作的外部函數或 API。通過調用這些工具，智能體可以與外部世界進行交互。例如，一個智能體可以調用一個 API 來獲取天氣信息，或者調用一個數據庫接口來存儲數據。
工具的定義應該是標準化的，這樣可以實現智能體與工具之間的靈活、多對多的關系。工具的文檔清晰、經過充分測試且可復用，能夠提高工具的可發現性，簡化版本管理，并避免重復定義。
一般來說，智能體需要以下三種類型的工具：
- 信息獲取工具：用于從外部獲取信息，如搜索引擎、數據庫查詢工具等。
- 數據處理工具：用于對數據進行處理和分析，如數據清洗、統計分析工具等。
- 交互工具：用于與用戶或其他系統進行交互，如聊天界面、API 調用工具等。
（三）指令：智能體的行為準則
指令是明確指導智能體行為的規則和限制。高質量的指令對于任何基于 LLM 的應用都至關重要，尤其是對于智能體來說。清晰的指令可以減少歧義，提高智能體的決策能力，從而使工作流程更加順暢，減少錯誤。
你可以使用先進的模型（如 o1 或 o3-mini）來自動生成指令，也可以手動編寫指令。以下是一個自動生成指令的示例：
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from IPython.display import display, Image
# 定義一個簡單的工具
@tool
def get_weather(location: str) -> str:
"""獲取指定地點的天氣信息"""
returnf"{location} 的天氣是晴天"
# 創建智能體
agent = create_react_agent(
model="gpt-4o-mini",
tools=[get_weather],
name="weather_agent",
prompt=(
"You are a weather agent.\n\n"
"INSTRUCTIONS:\n"
"- Assist ONLY with weather-related tasks\n"
"- After you're done with your tasks, respond directly to the user\n"
"- Respond ONLY with the results of your work, do NOT include ANY other text."
)
)三、協同作戰：多智能體的編排模式
當智能體的基本組件準備就緒后，接下來要考慮的就是如何讓它們協同工作，以高效地執行工作流程。多智能體系統可以被建模為一個圖，其中智能體是節點，而節點之間的邊則代表智能體之間的交互。
常見的編排模式包括：
（一）集中式模式：監督者說了算
在這種模式下，所有智能體都由一個監督者（Supervisor）智能體來管理和協調。監督者根據任務需求，將任務分配給不同的智能體，并收集它們的輸出結果。這種模式的優點是管理集中，易于控制，但缺點是監督者可能會成為瓶頸。
（二）去中心化模式：智能體之間的“接力棒”
在這種模式下，智能體之間可以直接進行“交接”（Handoff），將任務的執行權從一個智能體轉移到另一個智能體。這種模式的優點是靈活性高，智能體之間的協作更加自然，但缺點是可能會導致控制流的混亂。
（三）層次化模式：分級管理，各司其職
在這種模式下，智能體被組織成一個層次化的結構，每個層級的智能體負責管理下一層級的智能體。這種模式的優點是能夠更好地應對復雜的任務，但缺點是層級過多可能會導致信息傳遞的延遲。
四、實戰演練：構建一個監督者模式的多智能體系統
接下來，讓我們通過一個具體的例子來感受一下多智能體系統的魅力。我們將構建一個監督者模式的多智能體系統，用來解答一個關于 GDP 的問題。
（一）準備工作
首先，我們需要安裝必要的庫，并設置 OpenAI 的 API 密鑰：
pip install langgraph-supervisor langchain-openai
import os
os.environ["OPENAI_API_KEY"] = "<your_api_key>"（二）定義智能體和工具
接下來，我們定義兩個智能體：一個研究智能體（Research Agent），負責進行網絡搜索和信息收集；一個數學智能體（Math Agent），負責進行數學計算。
from langchain_openai import ChatOpenAI
from langgraph.graph import START, END
from langchain_community.tools.tavily_search import TavilySearchResults
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor
# 選擇 LLM
model = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))
# 定義工具
def add(a: float, b: float) -> float:
"""加法"""
return a + b
def multiply(a: float, b: float) -> float:
"""乘法"""
return a * b
def divide(a: float, b: float) -> float:
"""除法"""
return a / b
# 創建研究智能體
research_agent = create_react_agent(
model=model,
tools=[TavilySearchResults(max_results=3, tavily_api_key=os.getenv("TAVILY_API_KEY"))],
name="research_agent",
prompt=(
"You are a research agent.\n\n"
"INSTRUCTIONS:\n"
"- Assist ONLY with research-related tasks, DO NOT do any math\n"
"- After you're done with your tasks, respond to the supervisor directly\n"
"- Respond ONLY with the results of your work, do NOT include ANY other text."
)
)
# 創建數學智能體
math_agent = create_react_agent(
model=model,
tools=[add, multiply, divide],
name="math_agent",
prompt=(
"You are a math agent.\n\n"
"INSTRUCTIONS:\n"
"- Assist ONLY with math-related tasks\n"
"- After you're done with your tasks, respond to the supervisor directly\n"
"- Respond ONLY with the results of your work, do NOT include ANY other text."
)
)（三）創建監督者智能體
現在，我們創建一個監督者智能體來協調這兩個智能體的工作：
# 創建監督者智能體
supervisor_agent = create_supervisor(
model=model,
agents=[research_agent, math_agent],
prompt=(
"You are a supervisor managing two agents:\n"
"- a research agent. Assign research-related tasks to this agent\n"
"- a math agent. Assign math-related tasks to this agent\n"
"Assign work to one agent at a time, do not call agents in parallel.\n"
"Do not do any work yourself."
),
add_handoff_back_messages=True,
output_mode="full_history"
).compile()（四）運行測試
最后，我們來測試一下這個多智能體系統是否能夠正確地回答問題：
def test_supervisor_functionality():
"""測試監督者模式的功能"""
print("Query: find US and New York state GDP in 2024. what % of US GDP was New York state?")
print("-" * 80)
try:
for chunk in supervisor_agent.stream(
{
"messages": [
{
"role": "user",
"content": "find US and New York state GDP in 2024. what % of US GDP was New York state?",
}
]
},
subgraphs=False
):
print(chunk)
print("Test completed successfully")
except Exception as e:
print(f"Test failed with error: {str(e)}")
print("=" * 80)
if __name__ == "__main__":
test_supervisor_functionality()運行測試后，我們可以看到智能體們是如何一步步協作完成任務的：
- 監督者智能體將任務分配給研究智能體，研究智能體進行網絡搜索，收集關于美國和紐約州 GDP 的信息。
- 研究智能體將收集到的信息傳遞給監督者智能體，監督者智能體再將任務分配給數學智能體，數學智能體進行計算，得出紐約州 GDP 占美國 GDP 的百分比。
- 最終，監督者智能體將結果呈現給用戶。
五、智能體的“記憶管理”：消息歷史的控制
在多智能體系統中，每個智能體的消息歷史管理是非常重要的。你可以選擇將智能體的完整消息歷史添加到整個系統的對話歷史中，也可以只添加智能體的最終響應。
例如，如果你想保留完整的歷史記錄，可以在創建監督者時設置 ??output_mode="full_history"???；如果你想只保留最終響應，可以設置 ??output_mode="last_message"??。
supervisor = create_supervisor(
agents=[agent1, agent2],
output_mode="last_message"
)六、多層級架構：構建“智能體的團隊”
你可以通過創建管理多個監督者的監督者，來構建多層級的架構。例如，你可以創建一個研究團隊和一個寫作團隊，每個團隊都有自己的監督者，然后創建一個頂層監督者來管理這兩個團隊。
research_team = create_supervisor(
[research_agent, math_agent],
model=model,
supervisor_name="research_supervisor"
).compile(name="research_team")
writing_team = create_supervisor(
[writing_agent, publishing_agent],
model=model,
supervisor_name="writing_supervisor"
).compile(name="writing_team")
top_level_supervisor = create_supervisor(
[research_team, writing_team],
model=model,
supervisor_name="top_level_supervisor"
).compile(name="top_level_supervisor")七、自定義交接工具：讓智能體的“交接”更靈活
默認情況下，監督者使用預定義的交接工具來在智能體之間傳遞任務。你也可以自定義這些交接工具，例如修改工具的名稱和描述，或者添加工具調用的參數。
以下是一個自定義交接工具的示例：
from langgraph_supervisor import create_handoff_tool
# 創建一個自定義的交接工具
assign_to_research_agent = create_handoff_tool(
agent_name="research_agent",
descriptinotallow="Assign task to a researcher agent.",
)
# 將自定義工具傳遞給監督者
supervisor_agent = create_react_agent(
model="openai:gpt-4o-mini",
tools=[assign_to_research_agent],
prompt=(
"You are a supervisor managing two agents:\n"
"- a research agent. Assign research-related tasks to this agent\n"
"- a math agent. Assign math-related tasks to this agent\n"
"Assign work to one agent at a time, do not call agents in parallel.\n"
"Do not do any work yourself."
),
name="supervisor",
)八、實戰演練：構建一個“蜂群模式”的多智能體系統
除了監督者模式，我們還可以構建一個“蜂群模式”的多智能體系統。在這種模式下，智能體之間可以直接進行交互和協作，而不需要一個中央監督者來協調。
以下是一個使用 ??langgraph_swarm?? 庫構建蜂群模式的示例：
from langgraph_swarm import create_swarm, create_handoff_tool
# 創建蜂群模式的智能體
swarm_agent = create_swarm(
agents=[research_agent, math_agent],
default_active_agent="math_agent",
).compile()
# 測試蜂群模式的功能
def test_swarm_functionality():
print("Query: find US and New York state GDP in 2024. what % of US GDP was New York state?")
print("-" * 80)
try:
for chunk in swarm_agent.stream(
{
"messages": [
{
"role": "user",
"content": "find US and New York state GDP in 2024. what % of US GDP was New York state?",
}
]
},
subgraphs=False
):
print(chunk)
print("Test completed successfully")
except Exception as e:
print(f"Test failed with error: {str(e)}")
print("=" * 80)
if __name__ == "__main__":
test_swarm_functionality()九、智能體的“召喚”：調用模式與輸入輸出格式
智能體可以通過同步或異步的方式被調用。同步調用會阻塞調用者，直到智能體完成任務；而異步調用則允許調用者在智能體處理任務的同時繼續執行其他任務。
（一）同步調用
response = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})（二）異步調用
response = await agent.ainvoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})智能體的輸入是一個包含消息列表的字典，消息列表中每個消息都是一個字典，包含角色（如用戶或助手）和內容。智能體的輸出也是一個字典，包含消息列表和可能的結構化響應。
此外，智能體還支持流式響應，可以在處理過程中逐步返回結果，這對于需要實時交互的應用非常有用。
十、智能體的“成長之路”：構建可擴展智能體的路線圖
構建可擴展的智能體系統并不是一件容易的事，但以下路線圖或許能為你提供一些思路：
（一）選擇合適的模型
選擇一個擅長推理和邏輯處理的 LLM，支持逐步思考，并且輸出穩定一致。可以從 Llama、Claude Opus 或 Mistral 等開源模型開始，它們提供了更大的定制化和控制能力。
（二）設計智能體的推理過程
定義智能體如何處理任務，例如是否在回答前暫停并反思，是否逐步規劃行動，或者是否在需要幫助時調用工具。可以從簡單的 ReAct 或 Plan-then-Execute 框架開始。
（三）建立操作指南
為智能體設定清晰的交互規則，包括響應行為、語氣、何時使用外部工具以及標準化響應格式（如 JSON 或 Markdown）。
（四）整合記憶
補償 LLM 缺乏長期記憶的問題，例如應用滑動窗口記憶以保留最近的上下文，總結過去的對話，以及持久化關鍵信息（如用戶偏好、決策等）。可以使用 MemGPT 或 ZepAI 等工具簡化記憶管理。
（五）集成工具和 API
使智能體能夠執行實際操作，例如檢索外部數據、更新 CRM 和數據庫、執行計算或轉換。可以使用 MCP（Model Context Protocol）無縫地將工具集成到智能體工作流中。
（六）分配明確的目標
為智能體提供具體任務，例如“總結用戶反饋并推薦改進”，而不是模糊的“提供幫助”。專注于縮小工作范圍——定義智能體不應該做什么，而不是讓指令過于開放。
（七）擴展到多智能體團隊
創建具有明確角色的智能體團隊，例如一個智能體收集信息，另一個智能體解釋數據，第三個智能體格式化并呈現結果。明確的角色分工將使多智能體系統更加高效。
十一、智能體的“挑戰與突破”：面對困難，勇往直前
盡管智能體系統帶來了巨大的潛力，但也面臨著一些挑戰。研究人員在 150 多個任務中測試了 5 種流行的多智能體系統框架，并從 6 位專家標注者的見解中識別出 14 種不同的失敗模式，這些失敗模式被分為以下三大類：
（一）規范和系統設計失敗
由于任務定義不清晰或系統架構設計不佳而導致的問題。
（二）智能體之間的不協調
由于智能體之間的溝通不暢或缺乏協調而導致的問題。
（三）任務驗證和終止失敗
由于驗證流程不足或任務完成協議不當而導致的問題。
為了應對這些挑戰，研究人員探索了以下兩種主要干預措施：
（一）改進智能體角色的規范
提高智能體角色的清晰度和精確度，以防止誤解和重疊。
（二）增強編排策略
開發更好的協調機制，以簡化智能體之間的互動和任務執行。
例如，Anthropic 公司宣布了模型上下文協議（MCP），這是首次嘗試為 AI 系統和流程之間的集成制定行業標準。如果開發者遵循推薦的智能體模式，這樣的開源標準可能會對上述提到的每個問題產生積極影響。
谷歌于 2025 年 4 月 9 日推出的 Agent2Agent（A2A）協議，使 AI 智能體能夠使用基于 JSON 的“智能體卡片”在不同平臺之間進行通信和協作，這些卡片描述了它們的能力、輸入和認證方案。A2A 和 MCP 互為補充：A2A 促進了智能體之間的互操作性（水平集成），而 MCP 為智能體提供了工具訪問權限（垂直集成）。例如，兩個智能體可以使用 A2A 協議協作完成任務，同時分別利用 MCP 查詢外部數據源，從而創建一個無縫的多智能體生態系統。
十二、結語：開啟智能體的無限可能
你已經走過了構建可擴展智能體系統的藝術與科學之旅，從基礎設計到突破隱藏的障礙！現在，輪到你來掌控全局了。不妨思考一下：你的下一個智能體將由哪種模型驅動——快速簡單的模型還是強大復雜的模型？監督者模式或蜂群模式將如何改變你項目的效率？你是否可以利用我們探索過的異步處理模式來為你的用例解鎖新的潛力？
構建可擴展智能體系統的路線圖就在你手中，但挑戰依然存在。你預見到你的設計或實施中會有哪些障礙嗎？在評論區分享你的想法吧——告訴我們你將如何定制交接工具或調整迭代以取得成功！嘗試從編排到流式傳輸的各種策略，并加入創新者社區，共同突破界限，構建適應性強、強大的智能體系統。你的可擴展解決方案現在就開始——你的下一步是什么？
本文轉載自??Halo咯咯?? 作者：基咯咯
