[Swarm] 上下文變量 | 接入function功能調用

第3章：上下文變量

歡迎回到swarm！

在前兩章中，我們學習了作為對話指揮者的Swarm框架和具備指令與技能的專用AI角色智能體。(智能體就相當于是給用戶問題已經寫好了的提示詞，在用戶提問時自動加入，以此來給用戶更好的體驗）

但若智能體需要記住用戶信息或對話狀態呢？

例如用戶姓名、賬戶類型或之前提到的偏好。
AI模型本身不具備跨輪次完美記憶能力，將所有信息塞入對話歷史又會導致冗長混亂。

這正是**上下文變量(Context Variables)**的用武之地！

什么是上下文變量？

想象一個所有對話參與者都能查看和更新的共享白板——這正是swarm中上下文變量的本質。

它本質上是一個Python字典（dict），伴隨對話流程傳遞，可存儲當前交互相關的動態信息。

具體關系如下：

Swarm框架：對話指揮者，持有這塊白板
智能體：演奏家，通過查看白板（讀取上下文）決定如何演奏（響應或使用工具）
功能：演奏家的技能，執行時可讀取白板內容，也可寫入新信息（更新上下文）

這使得智能體和功能能夠共享并訪問重要信息，而無需將這些信息塞入主對話消息中。

( 有點類似于操作系統對于線程的共享內存的設計）

為何使用上下文變量？

上下文變量解決了跨AI輪次共享動態狀態的需求，確保swarm應用的不同組件（智能體與功能）能可靠訪問共享數據。

典型應用場景包括：

用戶信息存儲：姓名、ID、登錄狀態、偏好設置等
對話狀態追蹤：流程步驟完成狀態、當前主題、已收集信息（如天氣查詢地點）
功能數據傳遞：為外部系統交互功能提供必要數據（賬戶ID或收集的詳細信息）

讓我們通過個性化問候示例演示基礎用法。

基礎應用：個性化問候

我們希望智能體輸出"你好，[用戶姓名]！"而非通用問候。通過上下文變量存儲姓名，并在智能體指令中動態引用。

首先定義接受context_variables參數的指令函數：

from swarm import Agent, Swarm# 使用上下文變量的指令函數
def instructions(context_variables):# 從上下文字典獲取姓名，默認值為"用戶"name = context_variables.get("name", "用戶")# 返回動態生成的指令return f"你是一位樂于助人的助手。請禮貌問候用戶，若知道姓名請使用。已知用戶姓名為{name}。"

該函數通過.get("name", "用戶")安全獲取姓名，未找到時使用默認值。隨后構建動態指令字符串。

創建智能體時傳入該函數：

# 創建使用動態指令的智能體
greeting_agent = Agent(name="問候智能體",instructions=instructions, # 傳入指令函數
)

運行Swarm.run()時提供初始上下文字典：

# 創建Swarm實例
client = Swarm()# 定義初始上下文（包含姓名）
user_context = {"name": "張三"}# 定義初始消息
messages = [{"role": "user", "content": "你好！"}]# 運行對話并傳遞上下文
response = client.run(agent=greeting_agent,messages=messages,context_variables=user_context # 傳遞上下文字典
)# 打印AI響應
print(response.messages[-1]['content'])

執行時，Swarm將user_context字典傳遞給指令函數，生成包含"已知用戶姓名為張三"的指令。AI接收該指令和用戶消息后，可能響應：

你好張三！今天有什么可以幫您的？

若未提供上下文或使用不同姓名，響應將相應變化，體現上下文的動態特性。

上下文變量與功能結合

上下文變量在功能調用中同樣關鍵。以下示例預覽功能如何利用上下文（功能細節將在第四章詳述）：

假設存在需要user_id和name的print_account_details功能：

# 需要上下文變量的功能示例
def print_account_details(context_variables: dict):# 從上下文獲取用戶ID和姓名user_id = context_variables.get("user_id")name = context_variables.get("name")print(f"賬戶詳情：{name} (ID: {user_id})")# 功能應返回字符串或Result對象return "成功獲取賬戶詳情。"

注意該函數顯式接受context_variables參數。

將該功能加入智能體定義：

# 創建具備動態指令和功能的智能體
account_agent = Agent(name="賬戶智能體",instructions=instructions, # 可復用相同指令函數functions=[print_account_details], # 添加功能
)

當用戶請求"顯示我的賬戶詳情"時，AI可能調用該功能。Swarm執行時會自動傳遞當前上下文：

# 定義相同上下文
user_context = {"name": "李四", "user_id": 456}# 新用戶消息請求賬戶詳情
messages = [{"role": "user", "content": "顯示我的賬戶詳情！"}]# 攜帶上下文運行
response = client.run(messages=messages,agent=account_agent,context_variables=user_context, # 再次傳遞上下文
)# AI響應可能包含功能調用結果
print(response.messages[-1]["content"])

AI調用print_account_details時，Swarm檢測功能需要context_variables，自動傳遞包含"李四"和456的上下文字典，供功能使用。

運行機制：Swarm如何管理上下文

Swarm類負責在整個對話輪次中管理上下文字典。核心流程如下：

在這里插入圖片描述

從代碼層面看，swarm/core.py的關鍵處理邏輯如下：

Swarm.run方法初始化并傳遞上下文字典：

# File: swarm/core.py (簡化版run方法)
def run(...):active_agent = agentcontext_variables = copy.deepcopy(context_variables)  # 創建初始副本while ...:  # 對話循環completion = self.get_chat_completion(agent=active_agent,context_variables=context_variables,  # 傳遞上下文...)if message.tool_calls:# 處理功能調用時更新上下文context_variables.update(partial_response.context_variables)return Response(context_variables=context_variables)  # 返回最終上下文

get_chat_completion方法處理動態指令：

# File: swarm/core.py (簡化版get_chat_completion)
def get_chat_completion(...):if callable(agent.instructions):instructions = agent.instructions(context_variables)  # 動態生成指令...

handle_tool_calls方法執行功能時注入上下文：

# File: swarm/core.py (簡化版handle_tool_calls)
def handle_tool_calls(...):if __CTX_VARS_NAME__ in func.__code__.co_varnames:args[__CTX_VARS_NAME__] = context_variables  # 向功能注入上下文raw_result = func(**args)  # 執行功能

小結

上下文變量提供了跨輪次狀態維護和動態信息共享的強大能力。

通過Swarm框架管理的字典結構，我們能夠：

在運行Swarm時定義初始上下文
創建基于上下文動態調整的智能體指令
理解上下文如何傳遞給被調用的功能
掌握Swarm內部傳遞和管理上下文的機制

現在我們已經掌握如何賦予智能體動態響應能力，接下來將深入探索智能體執行具體操作的**功能(Function)**機制！

第四章：功能

第4章：功能

至此我們已經了解：Swarm框架是對話指揮者（第1章），智能體是具備指令配置的AI角色（第2章），而上下文變量助力信息共享（第3章）。

但若智能體需要執行除對話外的實際操作呢？

例如查詢天氣、獲取股價、發送郵件或訪問數據庫？AI模型本身無法直接完成這些操作。

這正是**功能(Function)**的用武之地！

前文傳送：MCP Servers

什么是swarm中的功能？

將智能體再次類比為樂手：他們遵循指令（Agent.instructions），同時具備演奏技能。在swarm中，這些"技能"即功能。

swarm中的功能本質上是用戶編寫的標準Python函數。將功能賦予智能體后，基于用戶請求和對話歷史，AI模型可自主決策*何時及如何*調用這些Python函數。
swarm充當橋梁：將Python函數定義自動轉換為AI可理解的格式（JSON schema）。
當AI決策"調用"功能時（實為請求swarm執行函數），swarm將攔截請求，定位對應Python函數執行，并將結果反饋給AI。

這如同指揮家（Swarm框架）理解樂曲需要特定技法，指示樂手（智能體）執行技法（功能），并將產生的音效融入整體演奏。

為何使用功能？

功能賦予AI系統與現實世界交互的能力，突破純文本生成的局限。典型應用場景包括：

獲取實時數據（天氣、股價、新聞）
執行計算任務
訪問或修改外部系統（數據庫、API、日歷）
發送通信（郵件、消息）

以天氣查詢為例：未配備天氣功能的AI只能回應"無法獲取天氣信息"；而具備該功能的AI可理解請求，調用功能獲取數據后提供真實天氣狀況。

如何定義swarm功能

功能定義與常規Python函數類似，但需注意以下swarm及AI模型關注要素：

名稱：函數名（get_current_weather）作為AI的調用標識
文檔字符串：函數說明（"""獲取指定地點當前天氣..."""）用于AI理解功能用途，需簡明扼要
參數：函數參數（location: str）定義所需信息，類型提示（: str）輔助生成正確schema。AI將嘗試從用戶請求提取必要信息（如地點）作為參數

天氣功能示例：

# 定義獲取天氣數據的Python函數
def get_current_weather(location: str):"""獲取指定地點的當前天氣"""# 實際應用中應調用天氣API# 本例返回靜態字符串print(f"(調用get_current_weather獲取{location}天氣)") # 可選：觀察調用時機if "london" in location.lower():return "倫敦當前為雨天"elif "paris" in location.lower():return "巴黎當前為晴天"else:return f"{location}天氣數據暫不可用"

該標準Python函數的關鍵要素為名稱、文檔字符串及參數。swarm通過function_to_json工具（自動調用，無需手動操作）將其轉換為AI可理解的格式。

為智能體添加功能

定義Python函數后，通過functions列表賦予智能體：

from swarm import Agent# 先定義功能（如上所示）
def get_current_weather(location: str):"""獲取指定地點的當前天氣"""print(f"(調用get_current_weather獲取{location}天氣)")if "london" in location.lower():return "倫敦當前為雨天"elif "paris" in location.lower():return "巴黎當前為晴天"else:return f"{location}天氣數據暫不可用"# 創建智能體并賦予功能
weather_agent = Agent(name="天氣智能體",instructions="您是專業的天氣助手，請使用工具查詢天氣",functions=[get_current_weather] # 添加至功能列表
)

現在weather_agent已具備get_current_weather功能。當swarm使用該智能體與AI交互時，將向AI模型通報此可用功能（基于Python函數生成的JSON schema）。

使用帶功能的智能體

完整示例如下，展示Swarm框架如何協調功能調用：

from swarm import Swarm, Agent# 1. 定義功能
def get_current_weather(location: str):"""獲取指定地點的當前天氣"""print(f"(調用get_current_weather獲取{location}天氣)")if "london" in location.lower():return "倫敦當前為雨天"elif "paris" in location.lower():return "巴黎當前為晴天"else:return f"{location}天氣數據暫不可用"# 2. 創建帶功能的智能體
weather_agent = Agent(name="天氣智能體",instructions="您是專業的天氣助手，請使用工具查詢天氣",functions=[get_current_weather]
)# 3. 創建Swarm實例
client = Swarm() # 需配置OPENAI_API_KEY環境變量# 4. 定義用戶消息
messages = [{"role": "user", "content": "巴黎天氣如何？"}]# 5. 運行對話
print("啟動Swarm...")
response = client.run(agent=weather_agent, messages=messages, debug=True) # 啟用調試模式查看內部流程# 6. 輸出最終響應
print("\nAI最終響應：")
print(response.messages[-1]['content'])

啟用debug=True時，內部流程如下：

調用client.run()
Swarm框架啟動weather_agent對話輪次
向AI模型通報get_current_weather功能及其參數
AI接收用戶消息（“巴黎天氣如何？”）及功能信息
AI解析需調用get_current_weather功能，提取"巴黎"作為location參數
AI返回"工具調用"特殊消息，請求執行get_current_weather并附帶{"location": "Paris"}
Swarm框架接收工具調用請求
查找智能體對應的Python函數
執行get_current_weather(location="Paris")，可能看到控制臺打印(調用get_current_weather獲取Paris天氣)
函數返回"巴黎當前為晴天"
Swarm框架將結果作為"工具結果"消息加入對話歷史
AI接收工具結果并生成自然語言響應
AI返回最終文本響應（“巴黎當前天氣晴朗”）
Swarm框架將最終響應封裝為響應對象返回

啟用調試模式時輸出示例：

在這里插入圖片描述

該流程展示swarm如何將自定義Python代碼無縫集成至AI工作流。

功能與上下文變量

如第3章所述，功能可通過context_variables參數訪問共享狀態：

# 使用上下文變量的功能
def log_user_action(action: str, context_variables: dict):"""記錄用戶操作日志"""user_id = context_variables.get("user_id", "未知用戶")print(f"用戶{user_id}執行操作：{action}")return "操作已記錄"# 創建帶日志功能的智能體
agent_with_logger = Agent(name="日志智能體",instructions="記錄用戶請求日志",functions=[log_user_action]
)# 攜帶上下文運行
user_context = {"user_id": 456}
messages = [{"role": "user", "content": "我已更新個人資料"}]client = Swarm()
response = client.run(agent=agent_with_logger,messages=messages,context_variables=user_context, # 傳遞上下文debug=True
)
# 控制臺將輸出"用戶456執行操作：更新個人資料"

AI調用log_user_action時，swarm自動傳遞當前上下文。

注意：無需在功能描述中聲明context_variables參數，swarm在生成schema時會自動隱藏該參數。

功能還可通過返回結果對象更新上下文變量，該機制將在下一章詳解！

內部機制：功能調用流程

功能調用流程解析（聚焦swarm協調作用）：

在這里插入圖片描述

Swarm類核心處理邏輯：

調用AI時生成功能schema：

# 摘自swarm/core.py（簡化）
def get_chat_completion(...):tools = [function_to_json(f) for f in agent.functions] # 生成JSON schemacreate_params = {"tools": tools or None, # 添加至API調用參數# ...其他參數...}return self.client.chat.completions.create(**create_params)

處理AI返回的工具調用請求：

# 摘自swarm/core.py（簡化run方法）
if message.tool_calls and execute_tools:partial_response = self.handle_tool_calls(...) # 調用處理器history.extend(partial_response.messages) # 添加工具結果context_variables.update(partial_response.context_variables) # 更新上下文

執行功能并處理結果：

# 摘自swarm/core.py（簡化handle_tool_calls方法）
def handle_tool_calls(...):func = function_map[name] # 定位Python函數if __CTX_VARS_NAME__ in func.__code__.co_varnames:args[__CTX_VARS_NAME__] = context_variables # 注入上下文raw_result = func(**args) # 執行用戶函數# 處理結果并更新響應partial_response.messages.append({"role": "tool","content": result.value,})