UFO 模块化会话状态中枢深入解析 Context 类型安全共享状态容器ufo.module.context【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOContext是 UFO 会话Session内部用于跨轮次Round共享状态的类型安全容器它集中管理会话的请求文本、执行步骤、LLM 调用成本、应用窗口句柄、日志器与执行元数据并通过ContextNames枚举保证属性名与默认值的一致性。本文以 documents/docs/infrastructure/modules/context.md 为核心骨架结合 ufo/module/context.py 的源码实现完整讲解 Context 的属性体系、核心方法、自动同步机制、序列化能力以及在真实会话与 Agent 处理链路中的调用方式读完即可在自己的 UFO 会话扩展或 Agent 处理器中熟练读写上下文状态。一、Context 是什么会话级共享状态的设计目标在 UFO 的多智能体架构中一个 Session 由多个 Round 组成每个 Round 又包含多个 StepHostAgent 负责任务规划AppAgent 负责在具体应用中执行操作。这些组件需要共同读写当前用户请求是什么本轮已经花掉多少成本正在操作哪个应用窗口等信息。Context就是为此而生的跨 Round、跨 Agent 的共享状态容器其核心设计目标如下类型安全所有属性名由ContextNames枚举定义杜绝魔法字符串拼写错误默认值每个属性自动初始化出合理默认值读写前无需判空自动同步current_round_*三个属性与按轮次索引的字典ROUND_STEP、ROUND_COST、ROUND_SUBTASK_AMOUNT自动双向同步序列化支持与 dict 互转便于落盘持久化与断点恢复Dispatcher 挂载可挂载命令分发器使 Round/Agent 能经由 Context 执行工具命令。从源码结构看Context是一个 dataclass见 ufo/module/context.py其内部存储_context: Dict[str, Any]字段通过default_factory依据枚举逐个生成默认值dataclass class Context: _context: Dict[str, Any] field( default_factorylambda: {name.name: name.default_value for name in ContextNames} ) command_dispatcher: Optional[BasicCommandDispatcher] None架构总览Context 容器与访问模式、自动同步属性及共享方之间的关系可用下图概括二、ContextNames 枚举30 属性的类型安全入口所有 Context 属性都定义在ContextNames枚举中见 ufo/module/context.py。使用时通过枚举成员而非字符串从根本上避免键名不一致from ufo.module.context import ContextNames # 类型安全的属性名 request context.get(ContextNames.REQUEST) context.set(ContextNames.SESSION_COST, 0.42)枚举成员除名称外还携带两个关键元数据这是自动默认值与类型校验的底层来源default_value属性按类别返回默认值例如字符串类返回、计数器返回0、成本返回0.0、字典类返回{}、列表类返回[]Logger/窗口类返回Nonetype属性返回属性类型str、int、float、dict、list等其中 Logger 类为字符串FileWriter、窗口类为 Windows 平台才导入的UIAWrapper、织造模式为字符串WeavingMode均以字符串形式返回以避免循环导入。值得注意的边界实现STRUCTURAL_LOGS的默认值是defaultdict(lambda: defaultdict(list))用于支撑按轮次 × 子任务组织结构化日志WEAVING_MODEGalaxy 织造模式默认值在源码中延迟导入from galaxy.agents.schema import WeavingMode并返回WeavingMode.CREATION若导入失败则回退为字符串creation见 ufo/module/context.py这是为避免循环导入而设计的防御式写法APPLICATION_WINDOW的UIAWrapper仅在TYPE_CHECKING或 Windows 平台下导入非 Windows 环境置为None保证模块可在 Linux/macOS 上被安全导入。属性分类总表Context 的 30 个属性在文档中按 7 个逻辑类别组织1. 标识与会话模式Identifiers ModeAttributeTypeDefaultDescriptionIDint0Session IDMODEstr执行模式normal、service 等CURRENT_ROUND_IDint0当前轮次编号2. 执行状态Execution StateAttributeTypeDefaultDescriptionREQUESTstr当前用户请求SUBTASKstr当前 AppAgent 处理的子任务PREVIOUS_SUBTASKSList[]历史子任务记录HOST_MESSAGEList[]HostAgent → AppAgent 的消息ROUND_RESULTstr当前轮次结果3. 成本追踪Cost TrackingAttributeTypeDefaultDescriptionSESSION_COSTfloat0.0会话总成本美元ROUND_COSTDict[int, float]{}各轮次成本CURRENT_ROUND_COSTfloat0.0当前轮次成本自动同步4. 步数统计Step CountingAttributeTypeDefaultDescriptionSESSION_STEPint0会话总步数ROUND_STEPDict[int, int]{}各轮次步数CURRENT_ROUND_STEPint0当前轮次步数自动同步ROUND_SUBTASK_AMOUNTDict[int, int]{}各轮次子任务数CURRENT_ROUND_SUBTASK_AMOUNTint0当前轮次子任务数自动同步5. 应用上下文Application ContextAttributeTypeDefaultDescriptionAPPLICATION_WINDOWUIAWrapperNone当前应用窗口对象APPLICATION_WINDOW_INFOAny-窗口元数据APPLICATION_PROCESS_NAMEstr进程名如WINWORD.EXEAPPLICATION_ROOT_NAMEstrUI 根元素名称CONTROL_REANNOTATIONList[]控件重标注信息6. 日志LoggingAttributeTypeDefaultDescriptionLOG_PATHstr日志目录路径LOGGERLoggerNone会话日志器REQUEST_LOGGERLoggerNoneLLM 请求日志器EVALUATION_LOGGERLoggerNone评估日志器STRUCTURAL_LOGSdefaultdictdefaultdict(...)结构化日志7. 工具与通信Tools CommunicationAttributeTypeDefaultDescriptionTOOL_INFODict{}可用工具元数据DEVICE_INFOList[]已连接设备信息GalaxyCONSTELLATIONTaskConstellationNone任务星座GalaxyWEAVING_MODEWeavingModeCREATION织造模式Galaxy完整属性参考源码级以下完整清单与 ufo/module/context.py 中的枚举定义一一对应注释标注了类型与默认值class ContextNames(Enum): # Identifiers ID ID # int, default: 0 MODE MODE # str, default: CURRENT_ROUND_ID CURRENT_ROUND_ID # int, default: 0 # Requests Tasks REQUEST REQUEST # str, default: SUBTASK SUBTASK # str, default: PREVIOUS_SUBTASKS PREVIOUS_SUBTASKS # List, default: [] HOST_MESSAGE HOST_MESSAGE # List, default: [] ROUND_RESULT ROUND_RESULT # str, default: # Costs SESSION_COST SESSION_COST # float, default: 0.0 ROUND_COST ROUND_COST # Dict, default: {} CURRENT_ROUND_COST CURRENT_ROUND_COST # float, default: 0.0 # Steps SESSION_STEP SESSION_STEP # int, default: 0 ROUND_STEP ROUND_STEP # Dict, default: {} CURRENT_ROUND_STEP CURRENT_ROUND_STEP # int, default: 0 ROUND_SUBTASK_AMOUNT ROUND_SUBTASK_AMOUNT # Dict, default: {} CURRENT_ROUND_SUBTASK_AMOUNT CURRENT_ROUND_SUBTASK_AMOUNT # int, default: 0 # Application APPLICATION_WINDOW APPLICATION_WINDOW # UIAWrapper, default: None APPLICATION_WINDOW_INFO APPLICATION_WINDOW_INFO # Any APPLICATION_PROCESS_NAME APPLICATION_PROCESS_NAME # str, default: APPLICATION_ROOT_NAME APPLICATION_ROOT_NAME # str, default: CONTROL_REANNOTATION CONTROL_REANNOTATION # List, default: [] # Logging LOG_PATH LOG_PATH # str, default: LOGGER LOGGER # Logger, default: None REQUEST_LOGGER REQUEST_LOGGER # Logger, default: None EVALUATION_LOGGER EVALUATION_LOGGER # Logger, default: None STRUCTURAL_LOGS STRUCTURAL_LOGS # defaultdict # Tools Devices TOOL_INFO TOOL_INFO # Dict, default: {} DEVICE_INFO DEVICE_INFO # List, default: [] CONSTELLATION CONSTELLATION # TaskConstellation, default: None WEAVING_MODE WEAVING_MODE # WeavingMode, default: CREATION三、Context 核心方法详解3.1get()读取上下文值读取时内部会先执行_sync_round_values()将当前轮次的步数、成本、子任务数同步到对应CURRENT_ROUND_*字段保证读到的是最新一致值随后按枚举名从字典取值。若属性尚未被显式赋值则返回初始化时的默认值见 ufo/module/context.py。request context.get(ContextNames.REQUEST) # 未设置时返回 cost context.get(ContextNames.SESSION_COST) # 未设置时返回 0.0文档与源码差异提示接口文档中的签名写作get(name, defaultNone)而当前仓库源码的实际签名为get(self, key: ContextNames)无default参数。默认值由ContextNames.default_value自动注入因此推荐直接依赖内置默认值而不是额外传默认参数。3.2set()写入上下文值写入时先校验枚举名是否合法非法键名会抛出KeyError随后存入字典若写入的是三个CURRENT_ROUND_*字段之一会进一步触发对应的自动同步属性将值回写到按轮次索引的字典见 ufo/module/context.pycontext.set(ContextNames.REQUEST, Send an email to John) context.set(ContextNames.SESSION_COST, 0.42) context.set(ContextNames.APPLICATION_PROCESS_NAME, WINWORD.EXE)3.3update_dict()合并更新字典型属性update_dict用于向某个字典类型的属性批量合并新增键值要求目标属性当前值与传入值都必须是 dict否则抛出TypeError见 ufo/module/context.py# 为当前轮次初始化步数/成本/子任务计数BaseRound._init_context 的真实用法 context.update_dict(ContextNames.ROUND_STEP, {self.id: 0}) context.update_dict(ContextNames.ROUND_COST, {self.id: 0}) context.update_dict(ContextNames.ROUND_SUBTASK_AMOUNT, {self.id: 0}) # 向 TOOL_INFO 中合并新的工具条目 context.update_dict(ContextNames.TOOL_INFO, {new_tool: {...}})文档与源码差异提示接口文档给出的批量示例context.update_dict({ContextNames.REQUEST: New task, ...})与当前源码签名不符——实际签名是update_dict(self, key: ContextNames, value: Dict[str, Any])一次只针对一个字典型属性做合并更新。若需同时设置多个不同类型的属性应改用多次set()。3.4to_dict()序列化为字典to_dict深拷贝内部字典后返回见 ufo/module/context.py。接口文档描述仅返回可 JSON 序列化的值源码则通过可选参数ensure_serializable控制默认False时原样返回含 Logger、窗口对象等置True时利用 ufo/utils/init.py 中的is_json_serializable内部即json.dumps试运行逐字段检查不可序列化的字段统一置为None并输出 warning 日志。context_dict context.to_dict(ensure_serializableTrue) # 保存到文件 import json json.dump(context_dict, open(context.json, w))被排除置空的典型对象包括日志器LOGGER、REQUEST_LOGGER、EVALUATION_LOGGER、窗口对象APPLICATION_WINDOW以及任何无法json.dumps的对象。3.5from_dict()从字典恢复from_dict逐个枚举名从传入字典中取回字段并写回内部存储随后调用_sync_round_values()重建当前轮次的一致性见 ufo/module/context.pydata json.load(open(context.json)) context Context() context.from_dict(data) # 注意当前源码中为实例方法原地恢复不返回新对象文档与源码差异提示接口文档中from_dict写作staticmethod且返回新的Context而当前源码实现为实例方法、返回None、直接修改自身内部存储。落盘恢复的推荐写法是先构造一个空的Context()再调用from_dict。3.6attach_command_dispatcher()挂载命令分发器Context 承载一个可选字段command_dispatcher类型为 ufo/module/dispatcher.py 中定义的抽象基类BasicCommandDispatcher挂载后 Round 与 Agent 即可经由context.command_dispatcher.execute_commands(commands, timeout6000)异步执行工具命令如截图、获取 UI 树、调用 MCP 工具。本地默认实现是LocalCommandDispatcher其构造需要 Session 与MCPServerManagerfrom ufo.module.dispatcher import LocalCommandDispatcher from ufo.client.mcp.mcp_server_manager import MCPServerManager mcp_server_manager MCPServerManager() dispatcher LocalCommandDispatcher(session, mcp_server_manager) context.attach_command_dispatcher(dispatcher) # 现在轮次/Agent 可以通过 context 执行命令 result await context.command_dispatcher.execute_commands([command])在 ufo/module/sessions/session.py 的Session._init_context中这正是 UFO 标准会话的真实初始化流程设置MODE后立即创建 dispatcher 并挂载到 Context。四、Auto-Syncing 自动同步属性三个current_round_*属性是 Context 中最具特色的机制它们既是当前轮次值的便捷读写入口又与按轮次索引的字典ROUND_STEP/ROUND_COST/ROUND_SUBTASK_AMOUNT保持双向一致。读写时通过CURRENT_ROUND_ID定位当前轮次。4.1current_round_stepproperty def current_round_step(self) - int: 获取当前轮次步数。 return self._context.get(ContextNames.ROUND_STEP.name).get( self._context.get(ContextNames.CURRENT_ROUND_ID.name), 0 ) current_round_step.setter def current_round_step(self, value: int) - None: 设置当前轮次步数并写回 ROUND_STEP 字典。 current_round_id self._context.get(ContextNames.CURRENT_ROUND_ID.name) self._context[ContextNames.ROUND_STEP.name][current_round_id] value使用方式# 读取 steps context.current_round_step # 写入同时更新 ROUND_STEP 字典 context.current_round_step 54.2current_round_cost# 读取 cost context.current_round_cost # 写入累加到 ROUND_COST 中当前轮次条目 context.current_round_cost 0.014.3current_round_subtask_amount# 读取 subtasks context.current_round_subtask_amount # 写入 context.current_round_subtask_amount 1同步机制源码剖析文档描述写入时同时更新 dict 与 CURRENT_ROUND_* 字段源码中的实际闭环由get()/set()与_sync_round_values()协作完成见 ufo/module/context.pyget()每次读取前调用_sync_round_values()把属性从 dict 中按CURRENT_ROUND_ID取出的值回写到CURRENT_ROUND_STEP/CURRENT_ROUND_COST/CURRENT_ROUND_SUBTASK_AMOUNT三个单值字段set()在写入CURRENT_ROUND_STEP/CURRENT_ROUND_COST/CURRENT_ROUND_SUBTASK_AMOUNT时反向调用对应属性 setter 把单值写回 dict 的当前轮次条目。因此无论从哪个方向修改两处存储都会在下次读写时收敛一致。这也意味着请优先使用属性读写当前轮次值而不是手工同时维护两个存储# ✅ 推荐 —— 自动同步 context.current_round_cost 0.01 # ❌ 不推荐 —— 必须手工同时更新两处容易遗漏 round_id context.get(ContextNames.CURRENT_ROUND_ID) context.attributes[ContextNames.ROUND_COST][round_id] 0.01 context.attributes[ContextNames.CURRENT_ROUND_COST] 0.01五、结构化日志add_to_structural_logs 与 filter_structural_logs接口文档未展开、但源码中值得一提的扩展能力是结构化日志见 ufo/module/context.py。STRUCTURAL_LOGS的默认结构为defaultdict(lambda: defaultdict(list))即按Round→SubtaskIndex→ 日志条目列表的二级键组织数据def add_to_structural_logs(self, data: Dict[str, Any]) - None: round_key data.get(Round, None) subtask_key data.get(SubtaskIndex, None) if round_key is None or subtask_key is None: return remaining_items {key: data[key] for key in data} self._context[ContextNames.STRUCTURAL_LOGS.name][round_key][subtask_key].append( remaining_items ) def filter_structural_logs(self, round_key, subtask_key, keys): structural_logs self._context[ContextNames.STRUCTURAL_LOGS.name][round_key][subtask_key] if isinstance(keys, str): return [log[keys] for log in structural_logs] # 提取单字段 elif isinstance(keys, list): return [{key: log[key] for key in keys} for log in structural_logs] # 提取多字段注意add_to_structural_logs要求数据自带Round与SubtaskIndex两个键否则直接忽略写入的日志条目中Round/SubtaskIndex本身也会作为字段一并保留。六、典型使用模式可直接复用的代码骨架模式 1会话初始化from ufo.module.context import Context, ContextNames # 创建 context内部已按 ContextNames 填充全部默认值 context Context() # 初始化会话元数据 context.set(ContextNames.ID, 0) context.set(ContextNames.MODE, normal) context.set(ContextNames.LOG_PATH, ./logs/task_001/) context.set(ContextNames.REQUEST, Send an email)模式 2轮次执行# 轮次开始 context.set(ContextNames.CURRENT_ROUND_ID, round_id) # 轮次过程中 context.current_round_step 1 context.current_round_cost agent_cost # Agent 读取共享状态 request context.get(ContextNames.REQUEST) process_name context.get(ContextNames.APPLICATION_PROCESS_NAME)模式 3成本追踪# Agent 产生成本 agent_cost llm_call_cost() context.current_round_cost agent_cost # 会话总成本同步累加 context.set( ContextNames.SESSION_COST, context.get(ContextNames.SESSION_COST, 0.0) agent_cost ) # 打印汇总 print(fRound cost: ${context.current_round_cost:.4f}) print(fSession total: ${context.get(ContextNames.SESSION_COST):.4f})这一模式正是 Agent 处理链路的真实做法在 ufo/agents/processors/core/processor_framework.py 中每次 LLM 调用返回后都会执行SESSION_COST累加与SESSION_STEP自增然后由 Round/Session 的属性读取汇总展示。模式 4应用跟踪# Agent 选定应用 context.set(ContextNames.APPLICATION_PROCESS_NAME, WINWORD.EXE) context.set(ContextNames.APPLICATION_ROOT_NAME, Document1 - Word) context.set(ContextNames.APPLICATION_WINDOW, word_window) # 后续轮次复用同一窗口 app_window context.get(ContextNames.APPLICATION_WINDOW) if app_window: app_window.set_focus()在真实链路中ufo/agents/processors/host_agent_processor.py 会在 HostAgent 选定应用后写入APPLICATION_PROCESS_NAMEAppAgent 侧则从 ufo/agents/processors/app_agent_processor.py 读取SUBTASK与进程名组装提示词。模式 5日志挂载# 配置日志器 context.set(ContextNames.LOGGER, session_logger) context.set(ContextNames.REQUEST_LOGGER, request_logger) # 会话内任意位置使用 logger context.get(ContextNames.LOGGER) logger.info(Round started) request_logger context.get(ContextNames.REQUEST_LOGGER) request_logger.write(prompt \n response)在 ufo/module/basic.py 的BaseSession._init_context中会话初始化时会基于LOG_PATH创建三个绕过全局 logging 配置的FileWriterresponse.log、request.log、evaluation.log见 ufo/module/basic.py分别写入LOGGER、REQUEST_LOGGER、EVALUATION_LOGGER保证即使全局日志被关闭也能落盘。模式 6持久化与断点恢复# 保存上下文状态 context_dict context.to_dict(ensure_serializableTrue) with open(checkpoint.json, w) as f: json.dump(context_dict, f, indent2) # 从检查点恢复 with open(checkpoint.json) as f: data json.load(f) restored_context Context() restored_context.from_dict(data)七、Context 在会话生命周期中的真实调用链Context 并非孤立的工具类而是被 Session 与 Round 两层生命周期反复驱动的共享对象见 ufo/module/basic.pyBaseSession.__init__创建Context()L498随后_init_context()L588-L616写入ID、LOG_PATH挂载三个日志FileWriter并初始化SESSION_COST 0、SESSION_STEP 0Session._init_contextufo/module/sessions/session.py覆盖模式并挂载LocalCommandDispatcherFollowerSession将模式置为followerL182-L191FromFileSession置为batch_normalL282-L288BaseRound._init_contextufo/module/basic.py每轮开始时用update_dict为该轮初始化ROUND_STEP、ROUND_COST、ROUND_SUBTASK_AMOUNT三个字典条目并set(REQUEST, ...)、set(CURRENT_ROUND_ID, ...)BaseRound.run循环中await self.agent.handle(self.context)把 Context 传给 Agent轮次结束返回context.get(ContextNames.ROUND_RESULT)L163-L191结束判定BaseRound.is_finished()依赖context.get(ContextNames.SESSION_STEP)与配置的max_step比较L193-L201BaseSession.is_finished()同样检查SESSION_STEP/total_roundsL813-L828。换言之从用户请求注入、Agent 决策、成本累计、步数统计到轮次结果回收全部经由同一个 Context 对象完成这也是它被称为会话状态中枢的原因。相关的配套概念可继续阅读 Session会话生命周期、Round轮次执行与 模块系统架构总览。八、最佳实践始终使用ContextNames枚举而非字符串# ✅ 推荐 context.get(ContextNames.REQUEST) # ❌ 不推荐 context.attributes[REQUEST]枚举除了防止拼写错误还自动携带default_value与type元数据是类型安全与默认值机制的入口。善用内置默认值无需判空# 未设置时自动返回 0.0 cost context.get(ContextNames.SESSION_COST) # 计数器属性天然带 0 起点 steps context.get(ContextNames.SESSION_STEP)当前轮次值一律走自动同步属性不要手工同时维护单值与字典两处存储见第四节说明。理解文档与源码的签名差异本文档系列中的方法签名属于教学简化如get的default参数、update_dict的批量字典参数、from_dict的静态构造风格。实际以 ufo/module/context.py 中的实现为准——get无默认参数、update_dict(key, value)按字典合并、from_dict为实例方法原地恢复。撰写依赖这些 API 的扩展代码前建议先对照源码确认签名。持久化时务必开启ensure_serializableTrue否则 Logger、窗口对象等不可 JSON 序列化的值会导致json.dump失败。参考与延伸阅读核心实现ufo/module/context.pyContext dataclass 与 ContextNames 枚举命令分发抽象ufo/module/dispatcher.pyBasicCommandDispatcher/LocalCommandDispatcher会话与轮次基类ufo/module/basic.pyBaseSession/BaseRound对 Context 的初始化与驱动平台会话ufo/module/sessions/session.pySession/FollowerSession/FromFileSession的 Context 使用成本与步数累计ufo/agents/processors/core/processor_framework.pyLLM 调用后更新SESSION_COST/SESSION_STEP序列化辅助函数ufo/utils/init.pyis_json_serializable配套文档Session、Round、Overview【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考