[Swarm] Result對象 | 智能體切換 | Response對象

第5章：結果對象

歡迎回到swarm！

在上一章第4章：功能中，我們學習了如何通過定義Python功能賦予智能體技能。我們見證了當AI決策調用時，Swarm框架如何執行這些功能。

當前，我們的功能僅返回簡單字符串如"武漢當前為晴天"。Swarm框架會將該字符串結果反饋給AI模型，這在簡單場景下運作良好。

但當功能需要實現更復雜的操作時該如何處理？例如：

成功完成任務（如用戶登錄）后需存儲新信息（如用戶ID）到上下文變量
完成流程（如機票預訂）后對話應由其他智能體接管（如行程規劃智能體）

簡單的字符串返回值無法向Swarm框架傳遞此類結構化信息。

這正是**結果對象(Result)**的用武之地！

什么是結果對象？

Result對象是可供智能體功能返回的特殊類型，旨在為功能代碼提供結構化方式與Swarm框架通信重要結果及狀態變更。

如同樂手（功能）不僅演奏音符（返回字符串），還向指揮（Swarm）傳遞信號：

“演奏完成，以下是需知附加信息，可能需切換其他樂手接續演奏”

Result對象包含三大要素：

value（字符串）：功能主輸出，與之前相同。Swarm框架將其格式化為功能調用結果返回AI
context_variables（字典）：包含功能需要更新或添加的上下文變量，將被合并至當前上下文狀態
agent（智能體對象）：可選參數，若提供則觸發智能體交接機制（第6章詳解）

swarm/types.py中的類定義如下：

# File: swarm/types.pyfrom pydantic import BaseModel
from typing import Optional, dictclass Result(BaseModel):"""封裝智能體功能可能的返回值"""value: str = ""agent: Optional[Agent] = Nonecontext_variables: dict = {}

為何使用結果對象？

返回Result對象賦予功能更強的對話流程控制能力：

更新共享狀態：可結構化更新重要信息（登錄狀態、購物車商品、流程完成狀態）
影響后續步驟：通過更新上下文動態影響AI后續指令或功能調用
觸發智能體交接：標記應由其他智能體接管對話

以登錄功能更新上下文為例演示應用場景。

應用示例：登錄后更新上下文

假設login_user功能需在成功登錄后返回確認信息，并將user_id存入上下文供后續操作使用。

定義返回Result的功能：

# 定義返回結果對象的功能
from swarm import Result, Agent, Swarmdef login_user(username: str, password: str):"""嘗試用戶登錄，成功返回用戶ID"""print(f"(正在嘗試登錄用戶：{username})")# 模擬憑證校驗if username == "test_user" and password == "password123":user_id = "user_12345"print(f"(登錄成功，用戶ID：{user_id})")# 返回結果對象！return Result(value="登錄成功。",context_variables={"user_id": user_id, "is_logged_in": True} # 更新上下文)else:print("(登錄失敗)")# 無上下文更新時可返回普通字符串return "登錄失敗，憑證無效"

創建帶此功能的智能體：

# 創建帶登錄功能的智能體
login_agent = Agent(name="登錄智能體",instructions="您是登錄助手，幫助用戶完成登錄流程",functions=[login_user]
)

運行Swarm并觸發功能調用：

# 創建Swarm實例
client = Swarm()# 初始化登錄請求消息
messages = [{"role": "user", "content": "請幫我登錄，用戶名為test_user，密碼是password123"}]# 初始上下文（空或預置狀態）
initial_context = {}# 運行對話
print("啟動Swarm...")
response = client.run(agent=login_agent,messages=messages,context_variables=initial_context,debug=True
)# 輸出最終結果
print("\nAI最終響應：")
print(response.messages[-1]['content'])print("\n最終上下文變量：")
print(response.context_variables)

執行流程解析：

調用client.run()啟動流程
Swarm攜帶初始上下文啟動login_agent
AI決策調用login_user并提取參數
Swarm執行Python功能，成功時返回包含新上下文的Result對象
Swarm合并context_variables至內部狀態
AI生成自然語言響應，最終上下文包含user_id與is_logged_in

輸出示例顯示更新后的上下文變量驗證狀態變更成功。

運行機制：結果對象處理流程

Swarm內部處理結果對象的核心流程如下：
在這里插入圖片描述

關鍵代碼邏輯體現在swarm/core.py：

# File: swarm/core.py（簡化版）class Swarm:def handle_function_result(self, result, debug) -> Result:# 標準化不同返回類型為結果對象match result:case Result() as result:return resultcase Agent() as agent:return Result(value=json.dumps({"助手": agent.name}), agent=agent)case _:return Result(value=str(result))def handle_tool_calls(...):raw_result = func(**args) # 執行用戶功能result = self.handle_function_result(raw_result, debug)# 更新部分響應的上下文及智能體partial_response.context_variables.update(result.context_variables)if result.agent:partial_response.agent = result.agent

小結

結果對象為智能體功能提供了結構化通信機制，使其能夠：

通過value 傳遞基礎結果信息
使用context_variables 更新共享狀態
借助agent字段觸發智能體交接（將在第6章詳解）

我們已掌握：

識別需要結構化返回結果的場景
理解結果對象的三元結構
編寫更新上下文變量的功能代碼
掌握Swarm處理結果對象的內核邏輯

下一章將深入探討智能體交接機制，開啟更復雜的多智能體協作場景！

第6章：智能體交接

通過前五章的學習，我們掌握了

指揮家（Swarm框架）
專業樂手（智能體）
樂譜共享注釋（上下文變量）
演奏技法（功能）
技法反饋機制（結果對象）。

現在讓我們探索體現"swarm"核心理念的強大概念：智能體交接。

什么是智能體交接？

假設我們與通用客服助手（智能體A）對話時咨詢賬單問題。該助手識別問題復雜性后，將對話無縫轉接至專業賬單處理助手（智能體B）。

此后所有交互由具備專業知識的智能體B接管。

在swarm中，智能體交接指對話控制權從一個智能體轉移至另一個的機制。

如同指揮家示意不同聲部或獨奏家接管主旋律，實現任務主題的精準切換。

為何需要智能體交接？

智能體交接對構建復雜AI應用至關重要，其優勢包括：

專業化：創建高精度領域專家智能體
模塊化：保持智能體定義簡潔，便于開發維護
高效性：AI模型聚焦當前任務，避免指令過載
性能提升：專用智能體處理任務更精準可靠

交接機制實現原理

交接機制的核心建立在結果對象的agent字段：

# 摘自swarm/types.py
class Result(BaseModel):value: str = ""  # 反饋AI的結果值agent: Optional[Agent] = None  # <<< 交接關鍵字段！context_variables: dict = {}  # 上下文更新

當功能返回的Result對象攜帶新智能體實例時，Swarm框架將激活該智能體作為后續對話的主控者。

基礎應用：語言交接場景

我們通過語言切換示例演示交接流程：當英文智能體檢測到西班牙語輸入時，轉接至西語智能體。

系統需要：

英文智能體：具備西語檢測能力
西語智能體：專精西語交互
交接功能：觸發智能體切換

完整代碼實現如下（參照examples/basic/agent_handoff.py）：

from swarm import Swarm, Agent, Result# 定義英文智能體
english_agent = Agent(name="英文智能體",instructions="您僅使用英語。檢測到西語輸入時調用'transfer_to_spanish_agent'工具",# 功能列表將在下方添加
)# 定義西語智能體
spanish_agent = Agent(name="西語智能體",instructions="您僅使用西班牙語。",
)# 定義交接功能
def transfer_to_spanish_agent():"""將用戶轉接至西語助手"""print("(正在轉接至西語智能體...)")  # 調試信息return Result(value="正在轉接至西語助手",  # AI反饋信息agent=spanish_agent  # <<< 交接標識)# 為英文智能體添加功能
english_agent.functions.append(transfer_to_spanish_agent)# 創建Swarm實例
client = Swarm()# 初始西語消息
messages = [{"role": "user", "content": "Hola. ?Como estás?"}]# 啟動對話流程
print("啟動英文智能體...")
response = client.run(agent=english_agent, messages=messages, debug=True)# 輸出結果
print("\n最終AI響應:")
print(response.messages[-1]["content"])
print("\n最終激活智能體:")
print(response.agent.name)

執行流程解析：

使用english_agent啟動對話
AI檢測西語輸入，調用交接功能
功能返回攜帶spanish_agent的Result對象
Swarm更新active_agent為西語智能體
后續交互由西語智能體接管

輸出

在這里插入圖片描述

內部機制

Swarm框架通過handle_function_result方法處理交接邏輯：
在這里插入圖片描述

關鍵代碼邏輯位于swarm/core.py：

# 簡化版run方法
def run(...):active_agent = agent  # 初始智能體while ...:if partial_response.agent:active_agent = partial_response.agent  # 交接發生點return Response(agent=active_agent)  # 返回最終激活智能體

小結

智能體交接機制使復雜AI應用具備動態角色切換能力，其核心要點包括：

通過Result.agent字段觸發智能體切換
交接功能負責識別切換需求并返回目標智能體
Swarm框架自動更新激活智能體狀態

我們已掌握：

交接機制的應用場景與優勢
多智能體的定義與功能綁定
交接功能的實現方式
Swarm內部的狀態更新邏輯

下一章將深入解析Swarm.run()的輸出核心——響應對象！

第7章：響應對象

第7章：Response

經過前六章的學習，我們已經構建了完整的認知體系：

Swarm作為交響樂團指揮
Agent充當專業樂手
Context Variables如同共享樂譜
Function類比演奏技法
以及通過Result傳遞結構化反饋
借助Agent Handoff實現演奏權轉移。

當Swarm完成多輪交互（執行函數、更新上下文、可能切換代理）后，需要向開發者提供完整交互回復。

此時我們需要獲取的不僅是AI的最終輸出，還包括完整的對話歷史、當前活躍代理以及上下文變量的最終狀態——這正是Response對象的使命！

Response對象是什么？

將Response對象視為復雜操作后的詳盡報告。

當我們調用client.run()啟動流程時，Swarm會協調所有組件完成交互，并將關鍵產出封裝到Response對象中返回。

Swarm.run()返回的Response對象是本次調用的全景視窗，包含三方面核心屬性：

messages：本次運行期間產生的完整對話記錄，包含初始消息、AI響應、工具調用請求及其執行結果
agent：運行結束時的活躍Agent實例。若無代理交接，與初始代理相同；發生交接則為新代理
context_variables：所有交互完成后Context Variables的最終狀態

以下是swarm/types.py中Response類的定義：

# File: swarm/types.pyfrom pydantic import BaseModel
from typing import List, Optional, dict # 簡化導入class Response(BaseModel):"""封裝Swarm的運行結果"""messages: List = []      # 本次運行的對話歷史agent: Optional[Agent] = None  # 最終活躍代理context_variables: dict = {}   # 最終上下文變量

為何需要Response對象？

僅獲取AI的最終文本響應不足以構建動態多輪應用。Response對象提供三大關鍵價值：

完整歷史回溯：延續對話需包含用戶輸入、AI響應及工具調用痕跡，messages屬性為此提供支持
當前代理狀態：通過Agent Handoff切換代理后，agent屬性確保后續交互正確路由
上下文同步：函數執行產生的關鍵數據（用戶ID、訂單狀態等）通過context_variables保持狀態同步

本質上，Response對象是swarm內部狀態與外部應用邏輯的同步樞紐。

Response對象使用實踐

通過第1章的簡單示例演示如何訪問Response對象：

from swarm import Swarm, Agent# 創建基礎代理
agent = Agent(name="基礎助手")# 初始化Swarm
client = Swarm()# 定義初始消息
messages = [{"role": "user", "content": "你好，最近怎么樣？"}]# 運行Swarm交互
print("運行Swarm中...")
response = client.run(agent=agent, messages=messages, debug=True)# 解析Response對象
print("\n--- Response詳情 ---")# 1.獲取最終消息內容
final_message = response.messages[-1]
print(f"最終消息內容: {final_message.get('content')}")
print(f"消息角色: {final_message.get('role')}")# 2.查看完整對話歷史
print("\n本次運行完整消息記錄:")
for msg in response.messages:sender = msg.get('sender', msg.get('role', '未知'))content = msg.get('content')if content:print(f"  [{sender}] {content}")elif msg.get('tool_calls'):print(f"  [{sender}] 請求工具調用: {msg['tool_calls']}")# 3.獲取當前活躍代理
print(f"\n當前活躍代理: {response.agent.name}")# 4.獲取上下文變量
print(f"\n最終上下文變量: {response.context_variables}")

執行結果將展示：

AI的最終響應內容
包含初始消息和AI響應的完整對話歷史
當前活躍代理名稱（若無交接則為"基礎助手"）
上下文變量狀態（未調用函數時為空）

底層實現機制

Response對象在Swarm.run()方法返回前構建，收集運行時產生的三大狀態：

在這里插入圖片描述
我們可以聯想到前文的Eventloop: [muduo_1] docs | 配置教程 | EventLoop | Thread

muduo review

Loop

Loop（循環）是讓計算機重復執行同一段代碼，直到滿足某個條件才停止，就像“重復做某件事直到做完為止”。

場景：自動打印100份文件，不用寫100行代碼，只需1個循環指令。

EventLoop

EventLoop 就像一家快餐店的取餐流程：顧客點單（任務）被放入隊列，廚房（主線程）按順序處理，遇到需要等待的訂單（異步任務）就先放一邊繼續做后面的，等外賣（異步結果）送到了再回來處理。

事件驅動與非阻塞IO

muduo庫的核心設計思想是基于事件驅動的非阻塞網絡編程，通過Reactor模式高效處理高并發連接，避免線程頻繁創建銷毀的開銷。

Reactor模式通過單線程或多線程事件循環監聽I/O事件，結合非阻塞調用和回調機制，實現高并發連接處理，避免為每個連接創建線程的開銷。
采用線程池復用固定數量的工作線程，配合事件驅動機制，將連接生命周期管理與業務邏輯解耦，減少線程頻繁創建銷毀帶來的性能損耗。

多線程與對象生命周期管理

采用**one loop per thread**架構，每個線程獨立運行事件循環，結合智能指針自動管理資源，保證線程安全與對象生命周期的可控性。

接口

僅依賴Linux原生系統調用（如epoll），避免過度抽象，提供回調機制讓開發者聚焦業務邏輯，降低復雜度。

在這里插入圖片描述

我們再回過頭來看swarm

swarm/core.py中的核心代碼片段：

# File: swarm/core.py (簡化版run方法)def run(self, agent: Agent, messages: List, context_variables: dict = {}) -> Response:active_agent = agentcontext_variables = copy.deepcopy(context_variables)history = copy.deepcopy(messages)init_len = len(messages)while ...:  # 交互循環# 處理AI響應和工具調用# 更新history/context_variables/active_agent# 構建Response對象return Response(messages=history[init_len:],agent=active_agent,context_variables=context_variables)