構建多代理系統：從架構設計到落地實踐的完整指南（附代碼解析）
在人工智能從單一模型走向復雜協作的時代，多代理系統（MAS）正成為破解高難度任務的關鍵鑰匙。本文聚焦從理論到代碼的全流程實踐，深度解析如何通過模塊化設計讓多個智能體協同完成復雜目標。
你將學會：
- 三大核心組件（模型選型、工具集成、指令配置）如何支撐代理邏輯；
- 兩種經典架構模式（Supervisor集中管理與Swarm分布式協作）的適用場景與代碼實現；
- 消息流轉、層級管理、流式輸出等工程化細節的落地技巧；
- 隱藏挑戰與行業標準（如MCP、A2A協議）如何突破系統瓶頸。
文中附帶完整Python代碼示例（基于LangGraph框架），從環境搭建到多代理交互全流程可復現，無論是AI開發者、算法工程師，還是技術管理者，都能從中獲取從0到1構建智能協作系統的方法論與實戰經驗。
準備好了嗎？讓我們一起揭開多代理系統的技術面紗，解鎖AI協作的無限可能。
1. 簡介
在評估代理可以增加價值的領域時，請優先考慮那些傳統方法難以自動化、尤其是存在摩擦的工作流：
- 復雜決策：涉及細微判斷、例外情況或上下文敏感決策的工作流，例如客戶服務中的退款審批。
- 難以維護的規則：由于規則集復雜龐大而難以管理的系統，更新成本高或容易出錯，例如供應商安全審查。
- 高度依賴非結構化數據：需要解釋自然語言、從文檔中提取信息或與用戶對話交互的場景，例如處理家庭保險索賠。
在決定構建代理之前，請驗證用例是否明確符合這些標準。否則，確定性解決方案可能已足夠。
2. 代理設計組件
從最基本的形式來看，代理由三個核心組件組成：1. 模型：為代理的推理和決策提供動力的大型語言模型（LLM）。2. 工具：代理可以用來執行操作的外部函數或 API。3. 指令：定義代理行為的明確指導方針和安全機制。
2.1 選擇模型
不同模型在任務復雜度、延遲和成本方面各有優劣。并非所有任務都需要最強大的模型——簡單的檢索或意圖分類任務可以由更小、更快的模型處理，而像決定是否批準退款這樣的復雜任務可能需要更強大的模型。
一個有效的方法是使用最適合每個任務的強大模型構建代理原型，以建立性能基線。然后嘗試替換為更小的模型，觀察是否仍能達到可接受的結果。這種方式可以避免過早限制代理能力，并診斷小模型的優缺點。
總之，選擇模型的原則很簡單：
- 設置評估以建立性能基線。
- 專注于使用可用的最佳模型達到準確性目標。
- 在可能的情況下，通過用更小的模型替換大模型來優化成本和延遲。
函數調用的最佳小型模型是什么？
可以從選擇任何開源權重模型開始，如 Llama、Mistral、Qwen、Claude Opus 等。
2.2 定義工具
函數調用是使大型語言模型（LLM）能夠與工具交互的主要方式。通?！昂瘮怠焙汀肮ぞ摺笨苫Q使用，因為“函數”（可重用的代碼塊）就是代理用來執行任務的“工具”。
工具通過底層應用程序或系統的 API 擴展代理的能力。對于沒有 API 的遺留系統，代理可以依靠計算機使用模型通過 Web 和應用程序界面直接與這些系統交互，就像人類一樣。每個工具都應有標準化定義，以實現工具與代理之間靈活的多對多關系。文檔完善、經過充分測試和可重用的工具可以提高可發現性，簡化版本管理，并避免重復定義。
概括來說，代理需要三種類型的工具：
例如，以下是使用 Agents SDK 為上述代理配置一系列工具的示例：
from agents import Agent, WebSearchTool, function_tool
@function_tool
def save_results(output):
db.insert({"output": output, "timestamp": datetime.time()})
return "File saved"
search_agent = Agent(
name="Search agent",
instructinotallow="Help the user search the internet and save results if asked.",
tools=[WebSearchTool(), save_results],
)當所需工具數量增加時，可考慮將任務分配給多個代理。
2.3 配置指令
高質量的指令對任何 LLM 驅動的應用程序都至關重要，對代理而言更是如此。清晰的指令可以減少歧義，改善代理的決策，從而實現更流暢的工作流執行和更少的錯誤。
可以使用高級模型（如 o1 或 o3-mini）從現有文檔自動生成指令。以下是說明該方法的示例提示：
3. 編排
在基礎組件就緒后，即可考慮編排模式，使代理能夠高效執行工作流。
多代理系統可以建模為圖，代理表示為節點。在管理者和監督者模式中，邊表示工具調用；而在去中心化和群體模式中，邊表示代理之間的執行交接。
AI 代理的工作流程通常遵循以下 4 個步驟：
無論采用何種編排模式，原則都是一致的：保持組件的靈活性、可組合性，并由清晰、結構良好的提示驅動。
在去中心化模式中，代理可以相互“交接”工作流執行。交接是一種單向轉移，允許代理將任務委托給另一個代理。
4. 實現：監督者模式
- 創建一個監督者代理來協調多個專業代理。
pip install langgraph-supervisor langchain-openai
export OPENAI_API_KEY=<your_api_key>
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
from IPython.display import Image, display
import os
# 選擇 LLM
model = ChatOpenAI(model="gpt-4o", api_key=os.getenv("OPENAI_API_KEY"))
# 定義工具
def add(a: float, b: float) -> float:
"""Add two numbers."""
return a + b
def multiply(a: float, b: float):
"""Multiply two numbers."""
return a * b
def divide(a: float, b: float):
"""Divide two numbers."""
return a / b
tavily_api_key = os.getenv("TAVILY_API_KEY", "your_tavily_api_key")
web_search = TavilySearchResults(max_results=3, tavily_api_key=tavily_api_key)
# 創建工作代理
research_agent = create_react_agent(
model=model,
tools=[web_search],
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()
display(Image(supervisor_agent.get_graph().draw_mermaid_png()))
我們將使用 ??pretty_print_messages?? 輔助函數來美觀地呈現流式代理輸出。
from langchain_core.messages import convert_to_messages
def pretty_print_message(message, indent=False):
pretty_message = message.pretty_repr(html=True)
ifnot indent:
print(pretty_message)
return
indented = "\n".join("\t" + c for c in pretty_message.split("\n"))
print(indented)
def pretty_print_messages(update, last_message=False):
is_subgraph = False
if isinstance(update, tuple):
ns, update = update
# 跳過父圖更新的打印
if len(ns) == 0:
return
graph_id = ns[-1].split(":")[0]
print(f"Update from subgraph {graph_id}:")
print("\n")
is_subgraph = True
for node_name, node_update in update.items():
update_label = f"Update from node {node_name}:"
if is_subgraph:
update_label = "\t" + update_label
print(update_label)
print("\n")
messages = convert_to_messages(node_update["messages"])
if last_message:
messages = messages[-1:]
for m in messages:
pretty_print_message(m, indent=is_subgraph)
print("\n")測試流程：
def test_supervisor_functionality():
"""Test the supervisor pattern with a specific GDP query to validate handoffs."""
print(f"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
):
pretty_print_messages(chunk, last_message=True)
print(f"Test completed successfully")
except Exception as e:
print(f"Test failed with error: {str(e)}")
print("=" * 80)
# 運行
if __name__ == "__main__"orTrue:
test_supervisor_functionality()輸出：
Query: find US and New York state GDP in 2024. what % of US GDP was New York state?
--------------------------------------------------------------------------------
Update from node supervisor:
================================= Tool Message =================================
Name: transfer_to_research_agent
Successfully transferred to research_agent
Update from node research_agent:
================================= Tool Message =================================
Name: transfer_back_to_supervisor
Successfully transferred back to supervisor
Update from node supervisor:
================================= Tool Message =================================
Name: transfer_to_math_agent
Successfully transferred to math_agent
Update from node math_agent:
================================= Tool Message =================================
Name: transfer_back_to_supervisor
Successfully transferred back to supervisor
Update from node supervisor:
================================== Ai Message ==================================
Name: supervisor
In 2024, New York State's GDP was approximately 8.15% of the US GDP.
Test completed successfully
================================================================================4.1 消息歷史管理
可以控制如何將代理消息添加到多代理系統的整體對話歷史中：包含代理的完整消息歷史：
supervisor = create_supervisor(
agents=[agent1, agent2],
output_mode="full_history")僅包含代理的最終響應：
supervisor = create_supervisor(
agents=[agent1, agent2],
output_mode="last_message")4.2 多級分層結構
可以通過創建管理多個監督者的超級監督者來構建多級分層系統：
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")4.3 自定義交接工具
默認情況下，監督者使用預構建的 ??create_handoff_tool?? 創建交接工具。您也可以創建自定義交接工具，以下是修改默認實現的一些思路：
- 更改工具名稱和/或描述
- 添加供 LLM 填充的工具調用參數，例如給下一個代理的任務描述
- 更改交接時傳遞給子代理的數據：默認情況下，?
?create_handoff_tool?? 會傳遞完整的消息歷史（截至當前在監督者中生成的所有消息），以及指示交接成功的工具消息。
以下是向 ??create_supervisor?? 傳遞自定義交接工具的示例：
from langgraph_supervisor import create_handoff_tool
workflow = create_supervisor(
[research_agent, math_agent],
tools=[
create_handoff_tool(agent_name="math_expert", name="assign_to_math_expert", descriptinotallow="Assign task to math expert"),
create_handoff_tool(agent_name="research_expert", name="assign_to_research_expert", descriptinotallow="Assign task to research expert")
],
model=model,
)您還可以控制是否將交接工具調用消息添加到狀態中。默認情況下會添加（??add_handoff_messages=True??），但如果需要更簡潔的歷史記錄，可以禁用：
workflow = create_supervisor(
[research_agent, math_agent],
model=model,
add_handoff_messages=False)此外，您可以自定義自動生成的交接工具的前綴：
workflow = create_supervisor(
[research_agent, math_agent],
model=model,
handoff_tool_prefix="delegate_to")
# 這將創建名為：delegate_to_research_expert, delegate_to_math_agent 的工具以下是自定義交接工具的示例代碼：
from typing import Annotated
from langchain_core.tools import tool, InjectedToolCallId
from langgraph.prebuilt import InjectedState
from langgraph.graph import StateGraph, START, MessagesState, END
from langgraph.types import Command
from IPython.display import display, Image
def create_handoff_tool(*, agent_name: str, description: str | None = None):
name = f"transfer_to_{agent_name}"
description = description orf"Ask {agent_name} for help."
@tool(name, descriptinotallow=description)
def handoff_tool(
state: Annotated[MessagesState, InjectedState],
tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
tool_message = {
"role": "tool",
"content": f"Successfully transferred to {agent_name}",
"name": name,
"tool_call_id": tool_call_id,
}
return Command(
goto=agent_name,
update={**state, "messages": state["messages"] + [tool_message]},
graph=Command.PARENT,
)
return handoff_tool
# 交接工具定義
assign_to_research_agent = create_handoff_tool(
agent_name="research_agent",
descriptinotallow="Assign task to a researcher agent.",
)
assign_to_math_agent = create_handoff_tool(
agent_name="math_agent",
descriptinotallow="Assign task to a math agent.",
)
# 監督者代理創建
supervisor_agent = create_react_agent(
model="openai:gpt-4o-mini",
tools=[assign_to_research_agent, assign_to_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."
),
name="supervisor",
)
# 定義多代理監督者圖
supervisor = (
StateGraph(MessagesState)
# 注意：`destinations` 僅用于可視化，不影響運行時行為
.add_node(supervisor_agent, destinatinotallow=("research_agent", "math_agent", END))
.add_node(research_agent)
.add_node(math_agent)
.add_edge(START, "supervisor")
# 始終返回監督者
.add_edge("research_agent", "supervisor")
.add_edge("math_agent", "supervisor")
.compile())
display(Image(supervisor.get_graph().draw_mermaid_png()))
輸出：
for chunk in supervisor.stream(
{
"messages": [
{
"role": "user",
"content": "find US and New York state GDP in 2024. what % of US GDP was New York state?",
}
]
},
):
pretty_print_messages(chunk, last_message=True)
final_message_history = chunk["supervisor"]["messages"]
Update from node supervisor:
================================= Tool Message =================================
Name: transfer_to_research_agent
Successfully transferred to research_agent
Update from node research_agent:
================================== Ai Message ==================================
Name: research_agent
- US GDP in 2024: Approximately $28.18 trillion USD (Statista).
- New York State GDP in 2024: The specific GDP number for New York State in 2024 has not been directly located, but relevant forecasts are provided in the state assembly reports.
For calculations regarding the percentage of the US GDP that New York State comprises, specific figures would need to be sourced.
Update from node supervisor:
================================= Tool Message =================================
Name: transfer_to_math_agent
Successfully transferred to math_agent
Update from node math_agent:
================================== Ai Message ==================================
Name: math_agent
5.93%
Update from node supervisor:
================================== Ai Message ==================================
Name: supervisor
In 2024, the approximate GDP of the United States is $28.18 trillion USD. New York State's GDP is estimated to be around $1.671 trillion USD. Therefore, New York State contributes about 5.93% of the total US GDP.我們已獲得上述 ??final_message_history??。現在，查看完整的消息歷史結果：
for message in final_message_history:
message.pretty_print()輸出：
================================ Human Message =================================
find US and New York state GDP in 2024. what % of US GDP was New York state?
================================== Ai Message ==================================
Name: supervisor
Tool Calls: transfer_to_research_agent (call_9Innk7roc0BYXdvTnvljH267) Call ID: call_9Innk7roc0BYXdvTnvljH267
Args:
================================= Tool Message =================================
Name: transfer_to_research_agent
Successfully transferred to research_agent
================================== Ai Message ==================================
Name: research_agent
Tool Calls: tavily_search_results_json (call_6rXCVDfY6Rg9FiWHsxbyBxQG) Call ID: call_6rXCVDfY6Rg9FiWHsxbyBxQG
Args:
query: US GDP 2024 forecast
tavily_search_results_json (call_hAsCKkl3tMIFwixTTp7eh0zt) Call ID: call_hAsCKkl3tMIFwixTTp7eh0zt
Args:
query: New York state GDP 2024 forecast
================================= Tool Message =================================
Name: tavily_search_results_json
[{"title": "GDP forecast U.S. 2034 - Statista", "url": "https://www.statista.com/statistics/216985/forecast-of-us-gross-domestic-product/", "content": "[](https://www.statista.com/statistics/216985/forecast-of-us-gross-domestic-product/#statisticContainer) This graph shows a forecast of the gross domestic product of the United States of America for fiscal years 2024 to 2034. GDP refers to the market value of all final goods and services produced within a country in a given period. According to the CBO, the United States GDP will increase steadily over the next decade from 28.18 trillion U.S. dollars in 2023 to 41.65 trillion U.S. dollars in [...] * [Economy & Politics](https://www.statista.com/markets/2535/economy-politics/)?\n* [Economy](https://www.statista.com/markets/2535/topic/970/economy/)\n\nU.S. GDP forecast 2024-2034\n===========================\n\n Published by \n\n[Abigail Tierney](https://www.statista.com/aboutus/our-research-commitment/3508/abigail-tierney), \n\n Jul 5, 2024 [...] .%20dollars)%C2%A0https://www.statista.com/statistics/216985/forecast-of-us-gross-domestic-product/)[](mailto:?subject=Statista%C2%A0-%C2%A0Forecast%20of%20the%20gross%20domestic%20product%20of%20the%20United%20States%20from%20fiscal%20year%2024%20%20to%20fiscal%20year%202034%20(in%20billion%20U.S.%20dollars)&body=This%20graph%20shows%20a%20forecast%20of%20the%20gross%20domestic%20product%20of%20the%20United%20States%20of%20America%20for%20fiscal%20years%2024%20to%202034", "score": 0.9245858}, {"title": "United States GDP Growth Rate - Trading Economics", "url": "https://tradingeconomics.com/united-states/gdp-growth", "content": "| [GDP from Services](https://tradingeconomics.com/united-states/gdp-from-services) | 17050.50 | 16949.30 | USD Billion | Dec 2024 |\n| [GDP from Transport](https://tradingeconomics.com/united-states/gdp-from-transport) | 730.50 | 721.40 | USD Billion | Dec 2024 |\n| [GDP from Utilities](https://tradingeconomics.com/united-states/gdp-from-utilities) | 350.80 | 341.40 | USD Billion | Dec 2024 | [...] | [GDP from Manufacturing](https://tradingeconomics.com/united-states/gdp-from-manufacturing) | 2406.80 | 2402.80 | USD Billion | Dec 2024 |\n| [GDP from Mining](https://tradingeconomics.com/united-states/gdp-from-mining) | 343.60 | 337.60 | USD Billion | Dec 2024 |\n| [GDP from Public Administration](https://tradingeconomics.com/united-states/gdp-from-public-administration) | 2653.10 | 2635.50 | USD Billion | Dec 2024 | [...] | [Gross Fixed Capital Formation](https://tradingeconomics.com/united-states/gross-fixed-capital-formation) | 4346.50 | 4265.90 | USD Billion | Mar 2025 |\n| [Gross National Product](https://tradingeconomics.com/united-states/gross-national-product) | 23620.90 | 23427.70 | USD Billion | Dec 2024 |\n| [Real Consumer Spending](https://tradingeconomics.com/united-states/real-consumer-spending) | 1.20 | 4.00 | percent | Mar 2025 |", "score": 0.8458536}, {"title": "A post-cycle economy faces greater policy uncertainty", "url": "https://am.jpmorgan.com/us/en/asset-management/adv/insights/market-insights/investment-outlook/us-economic-outlook/", "content": "This year, the economy saw surprisingly robust growth with real GDP on track to expand at a 1.8% annualized rate in the fourth quarter, closing 2024 with 2.3% annual growth, by our estimates. The biggest driver of this strength has been consumer spending, which contributed an average 78% of real GDP growth in the first three quarters. Despite pushing back on high retail prices, a thriftier consumer managed to stretch their budget and still expand their shopping cart. Inflation-adjusted consumer", "score": 0.83290404}]
================================= Tool Message =================================
Name: tavily_search_results_json
[{"title": "Annual State of the City's Economy and Finances 2024", "url": "https://comptroller.nyc.gov/reports/annual-state-of-the-citys-economy-and-finances-2024/", "content": "[[1]](https://comptroller.nyc.gov/reports/annual-state-of-the-citys-economy-and-finances-2024/#_ftnref1) At that time, the Comptroller’s Office expected five-year cumulative real GDP growth, 2020 to 2024, of 8.7 percent while the mayor expected 9.5 percent.\n\n[[2]](https://comptroller.nyc.gov/reports/annual-state-of-the-citys-economy-and-finances-2024/#_ftnref2) U.S. Bureau of Labor Statistics Local Area Unemployment Statistics (LAUS). [...] Nationally, the economy is expected to grow by 2.3 percent in 2025, stronger than the 1.6 percent that was last forecast in May, according to the Office of the New York City Comptroller’s forecast. Short-term interest rates are projected to decline more rapidly over the next few years, reaching 3 percent by early 2026, while inflation remains near targeted levels. Average wages and overall employment levels in New York City are estimated to have been stronger than had been forecast in 2024. The [...] (constant 2019 dollars)**22,550 15,322 13,596 19,451 21,211 21,096\n**_% change_**(32.1%)(11.3%)43.1%9.0%(0.5%)\n\nSource: NY State Department of Taxation and Finance and Office of the New York City Comptroller", "score": 0.882276}, {"title": "New York by the Numbers Monthly Economic and Fiscal Outlook", "url": "https://comptroller.nyc.gov/newsroom/newsletter/new-york-by-the-numbers-monthly-economic-and-fiscal-outlook-no-101-may-2025/", "content": "| A | Current (2024-25) | Year Earlier (2023-24) | Pre-Pandemic* | E |\n| --- | --- | --- | --- | --- |\n| Jan-24 | 1,335 | 1,810 | 1,307 | |\n| Feb-24 | 1,937 | 1,292 | 1,316 | |\n| Mar-24 | 1,641 | 1,232 | 884 | |\n| Apr-24 | 1,360 | 929 | 947 | |\n| May-24 | 652 | 1,377 | 1,020 | |\n| Jun-24 | 665 | 1,200 | 1,265 | |\n| Jul-24 | 529 | 905 | 1,232 | |\n| Aug-24 | 516 | 754 | 935 | |\n| Sep-24 | 355 | 805 | 806 | |\n| Oct-24 | 375 | 844 | 1,026 | |\n| Nov-24 | 431 | 801 | 1,476 | | [...] | 26-Oct-24 | 5.30% | 12.50% |\n| 2-Nov-24 | 9.40% | 6.90% |\n| 9-Nov-24 | 3.40% | 0.30% |\n| 16-Nov-24 | 64.4 消息轉發
你可以使用??create_forward_message_tool??為 supervisor 配備一個工具，將從 worker 代理接收到的最后一條消息直接轉發到圖表的最終輸出。當 supervisor 確定 worker 的響應已足夠，無需進一步處理或總結時，這非常有用。它為 supervisor 節省了 tokens，并避免了通過轉述可能導致的 worker 響應的誤傳。
from langgraph_supervisor.handoff import create_forward_message_tool
# 假設已定義 research_agent 和 math_agent
forwarding_tool = create_forward_message_tool("supervisor") # 該參數是分配給結果轉發消息的名稱
workflow = create_supervisor(
[research_agent, math_agent],
model=model,
# 將轉發工具與任何其他自定義或默認的交接工具一起傳遞
tools=[forwarding_tool]
)這將創建一個名為??forward_message???的工具，supervisor 可以調用該工具。該工具需要一個參數??from_agent??，用于指定應將哪個代理的最后一條消息直接轉發到輸出。
5. 實現：群模式
以下是使用_langgraph_swarm_庫實現群式多代理系統的示例代碼，該庫構建在 LangGraph 之上，支持專門代理之間的動態協作和交接。
我們將使用與 supervisor 模式中相同的數學代理和研究代理示例。
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_swarm import create_swarm, create_handoff_tool
from IPython.display import Image, display
import os
# 選擇 LLM
model = ChatOpenAI(model="gpt-4o-mini", api_key=os.getenv("OPENAI_API_KEY"))
# 定義工具
def add(a: float, b: float) -> float:
"""添加兩個數字。"""
return a + b
def multiply(a: float, b: float):
"""將兩個數字相乘。"""
return a * b
def divide(a: float, b: float):
"""將兩個數字相除。"""
return a / b
# 群模式的交接工具 - 使用適當的 LangGraph 群交接工具
handoff_to_research_agent = create_handoff_tool(
agent_name="research_agent",
descriptinotallow="將控制權轉移給研究代理，以執行與研究相關的任務、網絡搜索和信息收集。"
)
handoff_to_math_agent = create_handoff_tool(
agent_name="math_agent",
descriptinotallow="將控制權轉移給數學代理，以執行數學計算和計算任務。"
)
tavily_api_key = os.getenv("TAVILY_API_KEY", "your_tavily_api_key")
web_search = TavilySearchResults(max_results=3, tavily_api_key=tavily_api_key)
# 創建 Worker 代理
research_agent = create_react_agent(
model=model,
tools=[web_search, handoff_to_math_agent],
name="research_agent",
prompt=(
"你是一個專門從事網絡研究和信息收集的研究代理。\n\n"
"說明：\n"
"- 處理與研究相關的任務、網絡搜索和信息收集\n"
"- 不要嘗試數學計算 - 任何數學任務都使用 handoff_to_math_agent()\n"
"- 如果用戶要求進行數學計算，立即交接給數學代理\n"
"- 完成研究任務后，提供全面的結果\n"
"- 需要數學計算時使用 handoff_to_math_agent()"
)
)
math_agent = create_react_agent(
model=model,
tools=[add, multiply, divide, handoff_to_research_agent],
name="math_agent",
prompt=(
"你是一個專門從事數學計算和計算的數學代理。\n\n"
"說明：\n"
"- 處理數學計算、算術運算和數值分析\n"
"- 不要嘗試網絡搜索或研究 - 使用 handoff_to_research_agent() 進行研究任務\n"
"- 如果用戶要求研究或網絡搜索，立即交接給研究代理\n"
"- 完成數學任務后，提供清晰的結果和解釋\n"
"- 需要研究或網絡搜索時使用 handoff_to_research_agent()"
)
)
swarm_agent = create_swarm(
agents=[research_agent, math_agent],
default_active_agent="math_agent",
).compile()
display(Image(swarm_agent.get_graph().draw_mermaid_png()))要測試流程：
def test_swarm_functionality():
"""使用特定的 GDP 查詢測試群模式，以驗證交接。"""
print(f"查詢：查找 2024 年美國和紐約州的 GDP。紐約州占美國 GDP 的百分比是多少？")
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
):
pretty_print_messages(chunk, last_message=True)
print(f"Test completed successfully")
except Exception as e:
print(f"Test failed with error: {str(e)}")
print("=" * 80)
# 運行
if __name__ == "__main__"orTrue:
test_swarm_functionality()輸出：
查詢：查找 2024 年美國和紐約州的 GDP。紐約州占美國 GDP 的百分比是多少？
--------------------------------------------------------------------------------
Update from node math_agent:
================================= Tool Message =================================
Name: transfer_to_research_agent
Successfully transferred to research_agent
Update from node research_agent:
================================= Tool Message =================================
Name: transfer_to_math_agent
Successfully transferred to math_agent
Update from node math_agent:
================================== Ai Message ==================================
Name: math_agent
### Results:
- **U.S. GDP in 2024:** Approximately **$28.18 trillion**
- **New York State GDP in 2024:** Approximately **$2.29 trillion**
- **Percentage of U.S. GDP that is New York State:** Approximately **8.13%**
This means that New York State is projected to contribute about 8.13% of the total GDP of the United States in 2024.
Test completed successfully
================================================================================6. 運行代理
6.1 調用模式
代理可以通過兩種主要模式執行：
- 同步使用?
?.invoke()???或??.stream()??：調用者（用戶或系統）等待代理完成任務后再繼續。 - 異步使用?
?await .ainvoke()???或??async for???與??.astream()??結合：代理接收請求，啟動處理，并允許調用者繼續其他任務而無需等待響應。結果通常通過回調、通知或輪詢機制稍后傳遞。
同步調用：
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(...)
response = agent.invoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})異步調用：
from langgraph.prebuilt import create_react_agent
agent = create_react_agent(...)
response = await agent.ainvoke({"messages": [{"role": "user", "content": "what is the weather in sf"}]})
6.2 輸入格式
代理使用期望??messages???列表作為輸入的語言模型。因此，代理的輸入和輸出存儲為代理狀態中??messages??鍵下的消息列表。
您可以在輸入字典中直接提供代理狀態模式中定義的其他字段。這允許基于運行時數據或先前工具輸出的動態行為。
代理輸入必須是帶有??messages??鍵的字典。支持的格式為：
注意：??messages???的字符串輸入會轉換為HumanMessage。此行為不同于??create_react_agent???中的??prompt??參數，當作為字符串傳遞時，該參數會被解釋為SystemMessage。
6.3 輸出格式
代理輸出是一個包含以下內容的字典：
- ?
?messages??：執行期間交換的所有消息的列表（用戶輸入、助手回復、工具調用）。 - 可選的?
?structured_response??（如果配置了結構化輸出）。 - 如果使用自定義?
?state_schema??，輸出中可能還會出現與您定義的字段對應的其他鍵。這些鍵可以保存來自工具執行或提示邏輯的更新狀態值。
6.4 流式輸出
代理支持流式響應，以實現更高的應用響應性。這包括：
- 每一步后的進度更新
- 生成的LLM tokens
- 執行期間的自定義工具消息
流式傳輸可在同步和異步模式下使用：
同步流式傳輸：
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="updates"
):
print(chunk)我們在之前的示例中使用過此方法。
異步流式傳輸：
async for chunk in agent.astream(
{"messages": [{"role": "user", "content": "what is the weather in sf"}]},
stream_mode="updates"
):
print(chunk)6.5 最大迭代次數
這定義了代理在引發錯誤之前可以執行的最大步數。我們可以在運行時或通過??.with_config()??定義代理時進行定義：
運行時：
from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent
max_iterations = 3
recursion_limit = 2 * max_iterations + 1
agent = create_react_agent(
model="anthropic:claude-3-5-haiku-latest",
tools=[get_weather]
)
try:
response = agent.invoke(
{"messages": [{"role": "user", "content": "what's the weather in sf"}]},
{"recursion_limit": recursion_limit},
)
except GraphRecursionError:
print("Agent stopped due to max iterations.").with_config()：
from langgraph.errors import GraphRecursionError
from langgraph.prebuilt import create_react_agent
max_iterations = 3
recursion_limit = 2 * max_iterations + 1
agent = create_react_agent(
model="anthropic:claude-3-5-haiku-latest",
tools=[get_weather]
)
agent_with_recursion_limit = agent.with_config(recursion_limit=recursion_limit)
try:
response = agent_with_recursion_limit.invoke(
{"messages": [{"role": "user", "content": "what's the weather in sf"}]},
)
except GraphRecursionError:
print("Agent stopped due to max iterations.")7. 隱藏的障礙
至于協調多個代理的挑戰，以下幾點值得密切關注：
研究人員考察了五個流行的 MAS 框架，涵蓋超過 150 項任務，并結合了六位專家注釋者的見解。他們確定了 14 種不同的失敗模式，這些模式分為三個主要類別：
- 規范和系統設計失敗：由不明確的任務定義或設計不良的系統架構引起的問題。
- 代理間不一致：由代理之間的溝通不暢或缺乏協調導致的問題。
- 任務驗證和終止：與驗證過程不足和任務完成協議不當相關的失敗。
8. 突破障礙
為解決這些挑戰，作者探索了兩種主要干預措施：
- 改進代理角色的規范：提高代理角色的清晰度和準確性，以防止誤解和重疊。
- 增強協調策略：開發更好的代理間協調機制，以簡化交互和任務執行。
例如，Anthropic 宣布了模型上下文協議 (MCP)，這是針對 AI 系統和流程如何相互集成的首個行業標準嘗試。如果開發人員遵循推薦的代理模式，像這樣的開源標準可能會對上述每個問題產生積極影響。
Google 于 2025 年 4 月 9 日推出的Agent2Agent (A2A) 協議，使 AI 代理能夠使用基于 JSON 的“代理卡”描述其功能、輸入和身份驗證方案，跨不同平臺進行通信和協作。A2A 和 MCP 相互補充：A2A 促進代理之間的互操作性（橫向集成），而 MCP 為代理提供工具訪問（縱向集成）。例如，兩個代理可能使用 A2A 協作完成任務，同時各自利用 MCP 查詢外部數據源，從而創建無縫的多代理生態系統。
9. 構建可擴展 AI 代理的路線圖
將單體系統與代理系統區分開來的核心概念是其推理和正確使用工具的能力。
但如果沒有適當的路線圖，構建這些系統可能會非?；靵y。
1. 選擇您的模型選擇具備以下特點的 LLM：
- 擅長推理和邏輯處理
- 支持逐步思考（思維鏈）
- 提供穩定一致的輸出提示：從 Llama、Claude Opus 或 Mistral 等模型開始（開源權重提供更大的自定義和控制權）。
2. 設計代理的推理過程定義代理處理任務的方式：
- 回答前是否應暫停并思考？
- 應逐步規劃行動還是立即行動？
- 需要額外幫助時是否應調用工具？提示：從簡單的策略開始，如 ReAct 或先計劃后執行框架。
3. 制定操作指南設定明確的交互規則：
- 定義響應行為和語氣
- 指定何時使用外部工具
- 標準化響應格式（例如 JSON、markdown）提示：將此視為代理的內部操作協議。
4. 融入記憶通過以下方式彌補 LLM 缺乏長期記憶的問題：
- 對近期上下文應用滑動窗口記憶
- 總結過去的對話
- 持久化關鍵信息（用戶偏好、決策）提示：MemGPT 或 ZepAI 等工具可以簡化記憶管理。
5. 集成工具和 API使代理能夠采取實際行動：
- 檢索外部數據
- 更新 CRM 和數據庫
- 執行計算或轉換提示：使用 MCP 將工具無縫即插即用地集成到代理工作流程中。
6. 分配明確的目標為代理提供具體任務：“總結用戶反饋并推薦改進措施”“提供幫助”提示： 專注于縮小工作范圍——定義代理不應該做什么比留下開放式指令更有效。
7. 擴展到多代理團隊創建協作的專門代理：
- 一個代理收集信息
- 另一個解釋數據
- 第三個格式化并呈現結果提示：具有明確范圍角色的工作專業化可打造更有效的多代理系統。
Reference
[1] 知乎【柏企】: ??https://www.zhihu.com/people/cbq-91??
[2] 個人網站: ???https://www.chenbaiqi.com??
本文轉載自??柏企閱文??，作者：tailet
已于2025-6-20 11:13:11修改
贊
收藏
回復
分享
微博
QQ
微信
舉報
回復
相關推薦
