[Swarm] 上下文变量 | 接入function功能调用 | Mcp

第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 = agent context_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模型关注要素：

1. 名称：函数名（get_current_weather）作为AI的调用标识
2. 文档字符串：函数说明（\"\"\"获取指定地点当前天气...\"\"\"）用于AI理解功能用途，需简明扼要
3. 参数：函数参数（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时，内部流程如下：

1. 调用client.run()
2. Swarm框架启动weather_agent对话轮次
3. 向AI模型通报get_current_weather功能及其参数
4. AI接收用户消息（“巴黎天气如何？”）及功能信息
5. AI解析需调用get_current_weather功能，提取\"巴黎\"作为location参数
6. AI返回\"工具调用\"特殊消息，请求执行get_current_weather并附带{\"location\": \"Paris\"}
7. Swarm框架接收工具调用请求
8. 查找智能体对应的Python函数
9. 执行get_current_weather(location=\"Paris\")，可能看到控制台打印(调用get_current_weather获取Paris天气)
10. 函数返回\"巴黎当前为晴天\"
11. Swarm框架将结果作为\"工具结果\"消息加入对话历史
12. AI接收工具结果并生成自然语言响应
13. AI返回最终文本响应（“巴黎当前天气晴朗”）
14. 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类核心处理逻辑：

1. 调用AI时生成功能schema：

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

2. 处理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) # 更新上下文

3. 执行功能并处理结果：

# 摘自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章：结果