第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,})