第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,})
小結
- 掌握功能是賦予智能體現實交互能力的核心。
- 通過定義Python函數并添加至智能體的
functions
列表,可擴展AI的實用技能。 swarm
框架自動處理AI通信,將Python代碼轉換為模型可理解的描述,并在AI決策調用時執行對應函數。
同時,我們了解到上下文變量如何賦能功能訪問共享狀態。但功能執行后如何反饋狀態更新?這正是結果對象的職責,將在下一章深入探討!
第5章:結果