第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章:结果