当前位置：首页 > news >正文

[Swarm] Result对象 | 智能体切换 | Response对象 | muduo review

news 2025/7/9 7:35:08

第5章：结果对象

欢迎回到swarm！

在上一章第4章：功能中，我们学习了如何通过定义Python功能赋予智能体技能。我们见证了当AI决策调用时，Swarm框架如何执行这些功能。

当前，我们的功能仅返回简单字符串如"武汉当前为晴天"。Swarm框架会将该字符串结果反馈给AI模型，这在简单场景下运作良好。

但当功能需要实现更复杂的操作时该如何处理？例如：

成功完成任务（如用户登录）后需存储新信息（如用户ID）到上下文变量
完成流程（如机票预订）后对话应由其他智能体接管（如行程规划智能体）

简单的字符串返回值无法向Swarm框架传递此类结构化信息。

这正是**结果对象(Result)**的用武之地！

什么是结果对象？

Result对象是可供智能体功能返回的特殊类型，旨在为功能代码提供结构化方式与Swarm框架通信重要结果及状态变更。

如同乐手（功能）不仅演奏音符（返回字符串），还向指挥（Swarm）传递信号：

“演奏完成，以下是需知附加信息，可能需切换其他乐手接续演奏”

Result对象包含三大要素：

value（字符串）：功能主输出，与之前相同。Swarm框架将其格式化为功能调用结果返回AI
context_variables（字典）：包含功能需要更新或添加的上下文变量，将被合并至当前上下文状态
agent（智能体对象）：可选参数，若提供则触发智能体交接机制（第6章详解）

swarm/types.py中的类定义如下：

# File: swarm/types.pyfrom pydantic import BaseModel
from typing import Optional, dictclass Result(BaseModel):"""封装智能体功能可能的返回值"""value: str = ""agent: Optional[Agent] = Nonecontext_variables: dict = {}

为何使用结果对象？

返回Result对象赋予功能更强的对话流程控制能力：

更新共享状态：可结构化更新重要信息（登录状态、购物车商品、流程完成状态）
影响后续步骤：通过更新上下文动态影响AI后续指令或功能调用
触发智能体交接：标记应由其他智能体接管对话

以登录功能更新上下文为例演示应用场景。

应用示例：登录后更新上下文

假设login_user功能需在成功登录后返回确认信息，并将user_id存入上下文供后续操作使用。

定义返回Result的功能：

# 定义返回结果对象的功能
from swarm import Result, Agent, Swarmdef login_user(username: str, password: str):"""尝试用户登录，成功返回用户ID"""print(f"(正在尝试登录用户：{username})")# 模拟凭证校验if username == "test_user" and password == "password123":user_id = "user_12345"print(f"(登录成功，用户ID：{user_id})")# 返回结果对象！return Result(value="登录成功。",context_variables={"user_id": user_id, "is_logged_in": True} # 更新上下文)else:print("(登录失败)")# 无上下文更新时可返回普通字符串return "登录失败，凭证无效"

创建带此功能的智能体：

# 创建带登录功能的智能体
login_agent = Agent(name="登录智能体",instructions="您是登录助手，帮助用户完成登录流程",functions=[login_user]
)

运行Swarm并触发功能调用：

# 创建Swarm实例
client = Swarm()# 初始化登录请求消息
messages = [{"role": "user", "content": "请帮我登录，用户名为test_user，密码是password123"}]# 初始上下文（空或预置状态）
initial_context = {}# 运行对话
print("启动Swarm...")
response = client.run(agent=login_agent,messages=messages,context_variables=initial_context,debug=True
)# 输出最终结果
print("\nAI最终响应：")
print(response.messages[-1]['content'])print("\n最终上下文变量：")
print(response.context_variables)

执行流程解析：

调用client.run()启动流程
Swarm携带初始上下文启动login_agent
AI决策调用login_user并提取参数
Swarm执行Python功能，成功时返回包含新上下文的Result对象
Swarm合并context_variables至内部状态
AI生成自然语言响应，最终上下文包含user_id与is_logged_in

输出示例显示更新后的上下文变量验证状态变更成功。

运行机制：结果对象处理流程

Swarm内部处理结果对象的核心流程如下：
在这里插入图片描述

关键代码逻辑体现在swarm/core.py：

# File: swarm/core.py（简化版）class Swarm:def handle_function_result(self, result, debug) -> Result:# 标准化不同返回类型为结果对象match result:case Result() as result:return resultcase Agent() as agent:return Result(value=json.dumps({"助手": agent.name}), agent=agent)case _:return Result(value=str(result))def handle_tool_calls(...):raw_result = func(**args) # 执行用户功能result = self.handle_function_result(raw_result, debug)# 更新部分响应的上下文及智能体partial_response.context_variables.update(result.context_variables)if result.agent:partial_response.agent = result.agent

小结

结果对象为智能体功能提供了结构化通信机制，使其能够：

通过value 传递基础结果信息
使用context_variables 更新共享状态
借助agent字段触发智能体交接（将在第6章详解）

我们已掌握：

识别需要结构化返回结果的场景
理解结果对象的三元结构
编写更新上下文变量的功能代码
掌握Swarm处理结果对象的内核逻辑

下一章将深入探讨智能体交接机制，开启更复杂的多智能体协作场景！

第6章：智能体交接

通过前五章的学习，我们掌握了

指挥家（Swarm框架）
专业乐手（智能体）
乐谱共享注释（上下文变量）
演奏技法（功能）
技法反馈机制（结果对象）。

现在让我们探索体现"swarm"核心理念的强大概念：智能体交接。

什么是智能体交接？

假设我们与通用客服助手（智能体A）对话时咨询账单问题。该助手识别问题复杂性后，将对话无缝转接至专业账单处理助手（智能体B）。

此后所有交互由具备专业知识的智能体B接管。

在swarm中，智能体交接指对话控制权从一个智能体转移至另一个的机制。

如同指挥家示意不同声部或独奏家接管主旋律，实现任务主题的精准切换。

为何需要智能体交接？

智能体交接对构建复杂AI应用至关重要，其优势包括：

专业化：创建高精度领域专家智能体
模块化：保持智能体定义简洁，便于开发维护
高效性：AI模型聚焦当前任务，避免指令过载
性能提升：专用智能体处理任务更精准可靠

交接机制实现原理

交接机制的核心建立在结果对象的agent字段：

# 摘自swarm/types.py
class Result(BaseModel):value: str = ""  # 反馈AI的结果值agent: Optional[Agent] = None  # <<< 交接关键字段！context_variables: dict = {}  # 上下文更新

当功能返回的Result对象携带新智能体实例时，Swarm框架将激活该智能体作为后续对话的主控者。

基础应用：语言交接场景

我们通过语言切换示例演示交接流程：当英文智能体检测到西班牙语输入时，转接至西语智能体。

系统需要：

英文智能体：具备西语检测能力
西语智能体：专精西语交互
交接功能：触发智能体切换

完整代码实现如下（参照examples/basic/agent_handoff.py）：

from swarm import Swarm, Agent, Result# 定义英文智能体
english_agent = Agent(name="英文智能体",instructions="您仅使用英语。检测到西语输入时调用'transfer_to_spanish_agent'工具",# 功能列表将在下方添加
)# 定义西语智能体
spanish_agent = Agent(name="西语智能体",instructions="您仅使用西班牙语。",
)# 定义交接功能
def transfer_to_spanish_agent():"""将用户转接至西语助手"""print("(正在转接至西语智能体...)")  # 调试信息return Result(value="正在转接至西语助手",  # AI反馈信息agent=spanish_agent  # <<< 交接标识)# 为英文智能体添加功能
english_agent.functions.append(transfer_to_spanish_agent)# 创建Swarm实例
client = Swarm()# 初始西语消息
messages = [{"role": "user", "content": "Hola. ¿Como estás?"}]# 启动对话流程
print("启动英文智能体...")
response = client.run(agent=english_agent, messages=messages, debug=True)# 输出结果
print("\n最终AI响应:")
print(response.messages[-1]["content"])
print("\n最终激活智能体:")
print(response.agent.name)

执行流程解析：

使用english_agent启动对话
AI检测西语输入，调用交接功能
功能返回携带spanish_agent的Result对象
Swarm更新active_agent为西语智能体
后续交互由西语智能体接管

输出

在这里插入图片描述

内部机制

Swarm框架通过handle_function_result方法处理交接逻辑：
在这里插入图片描述

关键代码逻辑位于swarm/core.py：

# 简化版run方法
def run(...):active_agent = agent  # 初始智能体while ...:if partial_response.agent:active_agent = partial_response.agent  # 交接发生点return Response(agent=active_agent)  # 返回最终激活智能体

小结

智能体交接机制使复杂AI应用具备动态角色切换能力，其核心要点包括：

通过Result.agent字段触发智能体切换
交接功能负责识别切换需求并返回目标智能体
Swarm框架自动更新激活智能体状态

我们已掌握：

交接机制的应用场景与优势
多智能体的定义与功能绑定
交接功能的实现方式
Swarm内部的状态更新逻辑

下一章将深入解析Swarm.run()的输出核心——响应对象！

第7章：响应对象

第7章：Response

经过前六章的学习，我们已经构建了完整的认知体系：

Swarm作为交响乐团指挥
Agent充当专业乐手
Context Variables如同共享乐谱
Function类比演奏技法
以及通过Result传递结构化反馈
借助Agent Handoff实现演奏权转移。

当Swarm完成多轮交互（执行函数、更新上下文、可能切换代理）后，需要向开发者提供完整交互回复。

此时我们需要获取的不仅是AI的最终输出，还包括完整的对话历史、当前活跃代理以及上下文变量的最终状态——这正是Response对象的使命！

Response对象是什么？

将Response对象视为复杂操作后的详尽报告。

当我们调用client.run()启动流程时，Swarm会协调所有组件完成交互，并将关键产出封装到Response对象中返回。

Swarm.run()返回的Response对象是本次调用的全景视窗，包含三方面核心属性：

messages：本次运行期间产生的完整对话记录，包含初始消息、AI响应、工具调用请求及其执行结果
agent：运行结束时的活跃Agent实例。若无代理交接，与初始代理相同；发生交接则为新代理
context_variables：所有交互完成后Context Variables的最终状态

以下是swarm/types.py中Response类的定义：

# File: swarm/types.pyfrom pydantic import BaseModel
from typing import List, Optional, dict # 简化导入class Response(BaseModel):"""封装Swarm的运行结果"""messages: List = []      # 本次运行的对话历史agent: Optional[Agent] = None  # 最终活跃代理context_variables: dict = {}   # 最终上下文变量

为何需要Response对象？

仅获取AI的最终文本响应不足以构建动态多轮应用。Response对象提供三大关键价值：

完整历史回溯：延续对话需包含用户输入、AI响应及工具调用痕迹，messages属性为此提供支持
当前代理状态：通过Agent Handoff切换代理后，agent属性确保后续交互正确路由
上下文同步：函数执行产生的关键数据（用户ID、订单状态等）通过context_variables保持状态同步

本质上，Response对象是swarm内部状态与外部应用逻辑的同步枢纽。

Response对象使用实践

通过第1章的简单示例演示如何访问Response对象：

from swarm import Swarm, Agent# 创建基础代理
agent = Agent(name="基础助手")# 初始化Swarm
client = Swarm()# 定义初始消息
messages = [{"role": "user", "content": "你好，最近怎么样？"}]# 运行Swarm交互
print("运行Swarm中...")
response = client.run(agent=agent, messages=messages, debug=True)# 解析Response对象
print("\n--- Response详情 ---")# 1.获取最终消息内容
final_message = response.messages[-1]
print(f"最终消息内容: {final_message.get('content')}")
print(f"消息角色: {final_message.get('role')}")# 2.查看完整对话历史
print("\n本次运行完整消息记录:")
for msg in response.messages:sender = msg.get('sender', msg.get('role', '未知'))content = msg.get('content')if content:print(f"  [{sender}] {content}")elif msg.get('tool_calls'):print(f"  [{sender}] 请求工具调用: {msg['tool_calls']}")# 3.获取当前活跃代理
print(f"\n当前活跃代理: {response.agent.name}")# 4.获取上下文变量
print(f"\n最终上下文变量: {response.context_variables}")

执行结果将展示：

AI的最终响应内容
包含初始消息和AI响应的完整对话历史
当前活跃代理名称（若无交接则为"基础助手"）
上下文变量状态（未调用函数时为空）

底层实现机制

Response对象在Swarm.run()方法返回前构建，收集运行时产生的三大状态：

在这里插入图片描述
我们可以联想到前文的Eventloop: [muduo_1] docs | 配置教程 | EventLoop | Thread

muduo review

Loop

Loop（循环）是让计算机重复执行同一段代码，直到满足某个条件才停止，就像“重复做某件事直到做完为止”。

场景：自动打印100份文件，不用写100行代码，只需1个循环指令。

EventLoop

EventLoop 就像一家快餐店的取餐流程：顾客点单（任务）被放入队列，厨房（主线程）按顺序处理，遇到需要等待的订单（异步任务）就先放一边继续做后面的，等外卖（异步结果）送到了再回来处理。

事件驱动与非阻塞IO

muduo库的核心设计思想是基于事件驱动的非阻塞网络编程，通过Reactor模式高效处理高并发连接，避免线程频繁创建销毁的开销。

Reactor模式通过单线程或多线程事件循环监听I/O事件，结合非阻塞调用和回调机制，实现高并发连接处理，避免为每个连接创建线程的开销。
采用线程池复用固定数量的工作线程，配合事件驱动机制，将连接生命周期管理与业务逻辑解耦，减少线程频繁创建销毁带来的性能损耗。

多线程与对象生命周期管理

采用**one loop per thread**架构，每个线程独立运行事件循环，结合智能指针自动管理资源，保证线程安全与对象生命周期的可控性。

接口

仅依赖Linux原生系统调用（如epoll），避免过度抽象，提供回调机制让开发者聚焦业务逻辑，降低复杂度。

在这里插入图片描述

我们再回过头来看swarm

swarm/core.py中的核心代码片段：

# File: swarm/core.py (简化版run方法)def run(self, agent: Agent, messages: List, context_variables: dict = {}) -> Response:active_agent = agentcontext_variables = copy.deepcopy(context_variables)history = copy.deepcopy(messages)init_len = len(messages)while ...:  # 交互循环# 处理AI响应和工具调用# 更新history/context_variables/active_agent# 构建Response对象return Response(messages=history[init_len:],agent=active_agent,context_variables=context_variables)