UFO Galaxy 会话指标观测器SessionMetricsObserver深度解析星座执行的性能度量与统计分析【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFOSessionMetricsObserver 是 UFO Galaxy 框架观察者体系Observer System中的核心指标组件负责在星座Constellation执行过程中收集任务执行时间、星座生命周期、结构修改等三类原始度量并在查询时计算成功率、平均耗时、修改分布等统计摘要。本文以 metrics_observer.md 为主线结合 base_observer.py、events.py、galaxy_session.py 等源码实现系统讲解该观察者的架构、事件处理链路、统计计算逻辑、集成方式与实战用法帮助读者理解并复用在 Galaxy 星座执行中做性能评估与瓶颈分析的关键能力。一、定位与设计目标为什么需要 Metrics ObserverUFO Galaxy 采用事件驱动的观察者模式来解耦星座执行过程中的事件生产与消费。在 overview.md 描述的观察者体系中Orchestrator、Agent等事件发布者只需向全局EventBus发布事件而观测者Observer则按需订阅并处理这些事件。SessionMetricsObserver正是专门负责性能度量和统计的观察者。其核心设计目标可归纳为四点性能跟踪Performance Tracking——度量任务与星座的执行耗时成功率监控Success Rate Monitoring——追踪完成与失败的数量和比率修改分析Modification Analytics——监控星座结构的动态变更统计摘要Statistical Summaries——在查询时聚合计算各类统计指标供研究、评测与优化使用。从源码结构看SessionMetricsObserver实现了IEventObserver抽象接口定义于 events.py与ConstellationProgressObserver一同位于 base_observer.py 中并在galaxy/session/observers/__init__.py中对外导出。它是评估 Galaxy 性能、定位瓶颈、分析星座修改模式的关键基础设施。二、架构总览事件源、事件总线与输出原文档以 mermaid 图描述了指标的流转路径Orchestrator发布任务事件、Agent发布星座事件二者汇入全局EventBusEventBus通知SessionMetricsObserver观察者内部通过任务事件处理器与星座事件处理器将原始数据写入Metrics Storage查询时由Statistics Computer计算统计摘要最终输出到result.json与日志。对照源码该图的每个环节都有真实对应事件总线EventBus是 events.py 中的单例维护_observers按事件类型索引与_all_observers两组订阅集合。发布事件时通过asyncio.gather并发通知所有相关观察者单个观察者的异常不会阻断其他观察者。事件分派SessionMetricsObserver.on_event依据事件类型分派——TaskEvent走任务处理分支ConstellationEvent走星座处理分支见 base_observer.py。统计计算get_metrics()在返回时即时调用_compute_task_statistics()、_compute_constellation_statistics()、_compute_modification_statistics()三个私有方法完成聚合见 base_observer.py。观察者在体系中的位置在观察者体系中SessionMetricsObserver与其余四个观察者并行工作、互不干扰详见 overview.md观察者源码文件核心职责SessionMetricsObserverbase_observer.py性能度量耗时、成功率、修改统计ConstellationProgressObserverbase_observer.py任务进度跟踪、向 Agent 投递完成事件DAGVisualizationObserverdag_visualization_observer.py星座拓扑实时可视化ConstellationModificationSynchronizerconstellation_sync_observer.py修改同步与竞态防护AgentOutputObserveragent_output_observer.pyAgent 响应与动作的实时展示三、采集的指标体系SessionMetricsObserver将指标划分为任务指标、星座指标、修改指标三大类。前两类在事件到达时实时累加Real-time第三类及衍生指标在get_metrics()调用时计算Computed。3.1 任务指标Task Metrics跟踪单个任务的执行过程Metric描述计算时机task_count启动的任务总数Real-timecompleted_tasks成功完成的任务数Real-timefailed_tasks失败的任务数Real-timetotal_execution_time所有任务执行耗时之和Real-timetask_timings字典task_id → {start, end, duration}Real-timesuccess_ratecompleted / total 任务数Computedfailure_ratefailed / total 任务数Computedaverage_task_duration每个任务的平均执行时间Computedmin_task_duration最快任务执行时间Computedmax_task_duration最慢任务执行时间Computed3.2 星座指标Constellation Metrics监控星座的完整生命周期Metric描述计算时机constellation_count处理的星座总数Real-timecompleted_constellations成功完成的星座数Real-timefailed_constellations失败的星座数Real-timetotal_constellation_time星座总执行时间Real-timeconstellation_timings字典constellation_id → 时序数据Real-timeconstellation_success_ratecompleted / total 星座数Computedaverage_constellation_duration平均星座执行时间Computedmin_constellation_duration最快星座Computedmax_constellation_duration最慢星座Computedaverage_tasks_per_constellation每个星座的平均任务数Computed3.3 修改指标Modification Metrics跟踪星座的结构性变更动态 DAG 调整Metric描述计算时机constellation_modifications字典constellation_id → 修改记录列表Real-timetotal_modifications修改总次数Computedconstellations_modified发生过修改的星座数Computedaverage_modifications_per_constellation平均每个星座的修改次数Computedmax_modifications_for_single_constellation单星座最大修改次数Computedmost_modified_constellation被修改最多的星座 IDComputedmodification_types_breakdown按修改类型统计的次数分布Computed3.4 内部度量存储结构上述原始指标由构造函数初始化的字典统一维护见 base_observer.py与文档描述完全一致self.metrics: Dict[str, Any] { session_id: session_id, # Task metrics task_count: 0, completed_tasks: 0, failed_tasks: 0, total_execution_time: 0.0, task_timings: {}, # task_id - {start, end, duration} # Constellation metrics constellation_count: 0, completed_constellations: 0, failed_constellations: 0, total_constellation_time: 0.0, constellation_timings: {}, # constellation_id - timing data # Modification tracking constellation_modifications: {} # constellation_id - [modifications] }四、初始化与订阅接入事件总线4.1 构造与订阅SessionMetricsObserver只需两个参数即可创建必填的session_id会话唯一标识和可选的logger不传则自动创建默认 Logger。from galaxy.session.observers import SessionMetricsObserver import logging # Create metrics observer metrics_observer SessionMetricsObserver( session_idgalaxy_session_20231113, loggerlogging.getLogger(__name__) ) # Subscribe to event bus from galaxy.core.events import get_event_bus event_bus get_event_bus() event_bus.subscribe(metrics_observer)构造参数参数类型必填描述session_idstr是会话唯一标识用于区分不同执行批次loggerlogging.Logger否Logger 实例为 None 时自动创建默认实例需要注意EventBus.subscribe(observer)不传event_types时订阅全部事件见 events.py。由于on_event内部通过isinstance判断事件类型SessionMetricsObserver会过滤掉AgentEvent、DeviceEvent等无关事件仅处理TaskEvent与ConstellationEvent。若希望更精确也可用subscribe(observer, {EventType.TASK_COMPLETED, ...})形式订阅指定事件类型以降低通知开销。4.2 在 GalaxySession 中的自动装配在实际项目中观察者通常由GalaxySession自动装配而非手动创建。在 galaxy_session.py 的_setup_observers()方法中# Metrics observer for performance tracking self._metrics_observer SessionMetricsObserver( session_idfgalaxy_session_{self._id}, loggerself.logger ) self._observers.append(self._metrics_observer)会话执行结束后get_metrics()的完整结果被写入self._session_results[metrics]见 galaxy_session.py可通过session.session_results属性直接访问见 galaxy_session.py。这是将指标与整次会话结果一体化交付的标准路径。五、事件处理链路从事件到指标SessionMetricsObserver.on_event是唯一的入口根据事件类型分派到六个内部处理方法async def on_event(self, event: Event) - None: if isinstance(event, TaskEvent): await self._handle_task_event(event) elif isinstance(event, ConstellationEvent): await self._handle_constellation_event(event)5.1 任务事件处理任务事件覆盖TASK_STARTED、TASK_COMPLETED、TASK_FAILED三种类型EventType枚举定义见 events.py对应的处理逻辑见 base_observer.pydef _handle_task_started(self, event: TaskEvent) - None: Handle TASK_STARTED event. self.metrics[task_count] 1 self.metrics[task_timings][event.task_id] {start: event.timestamp} def _handle_task_completed(self, event: TaskEvent) - None: Handle TASK_COMPLETED event. self.metrics[completed_tasks] 1 if event.task_id in self.metrics[task_timings]: duration ( event.timestamp - self.metrics[task_timings][event.task_id][start] ) self.metrics[task_timings][event.task_id][duration] duration self.metrics[task_timings][event.task_id][end] event.timestamp self.metrics[total_execution_time] duration def _handle_task_failed(self, event: TaskEvent) - None: Handle TASK_FAILED event. self.metrics[failed_tasks] 1 # Also calculate duration for failed tasks if event.task_id in self.metrics[task_timings]: duration ( event.timestamp - self.metrics[task_timings][event.task_id][start] ) self.metrics[task_timings][event.task_id][duration] duration self.metrics[total_execution_time] duration要点失败任务同样计入耗时统计——即使任务失败其从开始到失败的时间差也会写入task_timings并累加进total_execution_time保证耗时统计不丢失失败样本。TaskEvent的task_id、timestamp、status、result、error字段定义于 events.py。5.2 星座事件处理星座生命周期事件包括CONSTELLATION_STARTED、CONSTELLATION_COMPLETED、CONSTELLATION_MODIFIEDCONSTELLATION_FAILED枚举存在观察者当前未为其设置独立分支失败星座的统计依赖生命周期事件与状态字段。def _handle_constellation_started(self, event: ConstellationEvent) - None: Handle CONSTELLATION_STARTED event. self.metrics[constellation_count] 1 constellation_id event.constellation_id constellation event.data.get(constellation) # Store initial statistics self.metrics[constellation_timings][constellation_id] { start_time: event.timestamp, initial_statistics: ( constellation.get_statistics() if constellation else {} ), processing_start_time: event.data.get(processing_start_time), processing_end_time: event.data.get(processing_end_time), processing_duration: event.data.get(processing_duration), } def _handle_constellation_completed(self, event: ConstellationEvent) - None: Handle CONSTELLATION_COMPLETED event. self.metrics[completed_constellations] 1 constellation_id event.constellation_id constellation event.data.get(constellation) duration ( event.timestamp - self.metrics[constellation_timings][constellation_id][start_time] if constellation_id in self.metrics[constellation_timings] else None ) if constellation_id in self.metrics[constellation_timings]: self.metrics[constellation_timings][constellation_id].update({ end_time: event.timestamp, duration: duration, final_statistics: ( constellation.get_statistics() if constellation else {} ), })值得注意的实现细节星座开始事件会同时记录initial_statistics初始快照完成事件则记录final_statistics终态快照两者对比即可还原星座从创建到完成的演化事件data中若携带processing_start_time/processing_end_time/processing_duration会原样透传存储便于与 Agent 的处理阶段耗时对齐constellation.get_statistics()来自 task_constellation.py返回total_tasks、total_dependencies、task_status_counts、longest_path_length、max_width以及并行度指标L、W、P等结构化统计是后续计算每个星座平均任务数的数据来源。5.3 修改跟踪Modification Tracking星座的动态修改是 Galaxy 的进阶能力Agent 在任务完成后可增删任务、调整依赖从而重构图谱。_handle_constellation_modified借助VisualizationChangeDetector做细粒度变更检测def _handle_constellation_modified(self, event: ConstellationEvent) - None: Handle CONSTELLATION_MODIFIED event. constellation_id event.constellation_id if constellation_id not in self.metrics[constellation_modifications]: self.metrics[constellation_modifications][constellation_id] [] if hasattr(event, data) and event.data: old_constellation event.data.get(old_constellation) new_constellation event.data.get(new_constellation) changes None if old_constellation and new_constellation: changes VisualizationChangeDetector.calculate_constellation_changes( old_constellation, new_constellation ) modification_record { timestamp: event.timestamp, modification_type: event.data.get(modification_type, unknown), on_task_id: event.data.get(on_task_id, []), changes: changes, new_statistics: ( new_constellation.get_statistics() if new_constellation else {} ), processing_start_time: event.data.get(processing_start_time), processing_end_time: event.data.get(processing_end_time), processing_duration: event.data.get(processing_duration), } self.metrics[constellation_modifications][constellation_id].append( modification_record )VisualizationChangeDetector.calculate_constellation_changes定义于 change_detector.py返回结构化的变更明细added_tasks/removed_tasks/modified_tasks通过新旧星座的 task_id 集合差集与属性比对得出added_dependencies/removed_dependencies/modified_dependencies依赖以from_task_id-to_task_id字符串形式描述同样先做集合差集、再做属性比对modification_type根据变更内容归纳为constellation_created、任务/依赖增删改等类型。每条修改记录还携带触发修改的目标任务on_task_id、修改时间戳与修改后的星座统计快照为哪次修改发生在哪个任务完成后、带来了什么结构变化提供了完整可追溯的审计链。六、统计计算get_metrics() 与三类统计摘要get_metrics()是观察者的对外查询接口返回原始度量与三类计算统计的组合字典def get_metrics(self) - Dict[str, Any]: Get collected metrics with computed statistics. metrics self.metrics.copy() metrics[task_statistics] self._compute_task_statistics() metrics[constellation_statistics] self._compute_constellation_statistics() metrics[modification_statistics] self._compute_modification_statistics() return metrics返回值包含全部原始指标计数、时序等task_statisticsconstellation_statisticsmodification_statistics。6.1 任务统计Task Statistics{ total_tasks: 10, completed_tasks: 8, failed_tasks: 2, success_rate: 0.8, failure_rate: 0.2, average_task_duration: 2.5, min_task_duration: 0.5, max_task_duration: 5.2, total_task_execution_time: 25.0 }源码实现要点见 base_observer.pydurations从task_timings中提取所有带duration字段的记录success_rate、failure_rate在task_count 0时计算否则返回0.0规避除零异常。6.2 星座统计Constellation Statistics{ total_constellations: 1, completed_constellations: 1, failed_constellations: 0, success_rate: 1.0, average_constellation_duration: 30.5, min_constellation_duration: 30.5, max_constellation_duration: 30.5, total_constellation_time: 30.5, average_tasks_per_constellation: 10.0 }实现要点见 base_observer.pyaverage_tasks_per_constellation并非简单的任务总数除以星座数而是遍历每个星座的initial_statistics[total_tasks]求和后除以含该字段的星座数——以初始任务快照为分子口径避免后续动态修改造成统计失真。6.3 修改统计Modification Statistics{ total_modifications: 3, constellations_modified: 1, average_modifications_per_constellation: 3.0, max_modifications_for_single_constellation: 3, most_modified_constellation: const_123, modifications_per_constellation: { const_123: 3 }, modification_types_breakdown: { add_tasks: 2, modify_dependencies: 1 } }实现要点见 base_observer.pymodification_types_breakdown对每条记录的modification_type字段做计数聚合缺失时归入unknownmost_modified_constellation通过max(..., keylambda x: x[1])找出修改次数最多的星座 ID无修改时返回None。6.4 典型读取示例# After constellation execution metrics metrics_observer.get_metrics() # Access task statistics print(fTotal tasks: {metrics[task_statistics][total_tasks]}) print(fSuccess rate: {metrics[task_statistics][success_rate]:.2%}) print(fAvg duration: {metrics[task_statistics][average_task_duration]:.2f}s) # Access constellation statistics print(fTotal constellations: {metrics[constellation_statistics][total_constellations]}) print(fAvg tasks per constellation: {metrics[constellation_statistics][average_tasks_per_constellation]:.1f}) # Access modification statistics print(fTotal modifications: {metrics[modification_statistics][total_modifications]}) print(fModification types: {metrics[modification_statistics][modification_types_breakdown]})七、完整使用示例示例 1基础指标采集在星座执行前后订阅观察者并读取汇总结果import asyncio from galaxy.core.events import get_event_bus from galaxy.session.observers import SessionMetricsObserver async def collect_metrics(): Collect and display metrics for constellation execution. # Create and subscribe metrics observer metrics_observer SessionMetricsObserver(session_iddemo_session) event_bus get_event_bus() event_bus.subscribe(metrics_observer) # Execute constellation (orchestrator will publish events) await orchestrator.execute_constellation(constellation) # Retrieve metrics metrics metrics_observer.get_metrics() # Display summary print(\n Execution Summary ) print(fSession: {metrics[session_id]}) print(fTasks: {metrics[task_count]} total, f{metrics[completed_tasks]} completed, f{metrics[failed_tasks]} failed) print(fTotal execution time: {metrics[total_execution_time]:.2f}s) # Display task statistics task_stats metrics[task_statistics] print(f\nTask Success Rate: {task_stats[success_rate]:.1%}) print(fAverage Task Duration: {task_stats[average_task_duration]:.2f}s) print(fFastest Task: {task_stats[min_task_duration]:.2f}s) print(fSlowest Task: {task_stats[max_task_duration]:.2f}s) # Clean up event_bus.unsubscribe(metrics_observer) asyncio.run(collect_metrics())示例 2性能分析定位瓶颈利用task_timings排序找出最慢任务并结合修改统计分析 Agent 的动态调整行为def analyze_performance(metrics_observer: SessionMetricsObserver): Analyze performance metrics and identify bottlenecks. metrics metrics_observer.get_metrics() task_timings metrics[task_timings] # Find slowest tasks sorted_tasks sorted( task_timings.items(), keylambda x: x[1].get(duration, 0), reverseTrue ) print(\n Top 5 Slowest Tasks ) for task_id, timing in sorted_tasks[:5]: duration timing.get(duration, 0) print(f{task_id}: {duration:.2f}s) # Analyze modification patterns mod_stats metrics[modification_statistics] if mod_stats[total_modifications] 0: print(f\n Modification Analysis ) print(fTotal Modifications: {mod_stats[total_modifications]}) print(fAverage per Constellation: f{mod_stats[average_modifications_per_constellation]:.1f}) print(fMost Modified: {mod_stats[most_modified_constellation]}) print(\nModification Types:) for mod_type, count in mod_stats[modification_types_breakdown].items(): print(f {mod_type}: {count})示例 3导出 JSON 供离线分析将统计结果序列化为 JSON 文件便于后续研究或报表生成import json from pathlib import Path def export_metrics(metrics_observer: SessionMetricsObserver, output_path: str): Export metrics to JSON file for analysis. metrics metrics_observer.get_metrics() # Convert to JSON-serializable format output_data { session_id: metrics[session_id], task_statistics: metrics[task_statistics], constellation_statistics: metrics[constellation_statistics], modification_statistics: metrics[modification_statistics], raw_metrics: { task_count: metrics[task_count], completed_tasks: metrics[completed_tasks], failed_tasks: metrics[failed_tasks], total_execution_time: metrics[total_execution_time], constellation_count: metrics[constellation_count], } } # Write to file output_file Path(output_path) output_file.parent.mkdir(parentsTrue, exist_okTrue) with open(output_file, w) as f: json.dump(output_data, f, indent2) print(fMetrics exported to: {output_file})八、最佳实践1. Session ID 命名使用描述性会话 ID 便于后续跨批次对比分析# ✅ Good: Descriptive session ID session_id fgalaxy_session_{task_type}_{timestamp} # ❌ Bad: Generic session ID session_id session_12. 无论成败都导出指标将指标导出放进finally即使执行异常也能保留诊断数据try: await orchestrator.execute_constellation(constellation) finally: # Always export metrics, even if execution failed metrics metrics_observer.get_metrics() export_metrics(metrics, results/metrics.json)3. 长会话的内存管理task_timings与constellation_timings会随任务量线性增长长跑会话处理完指标后应清理# After processing metrics metrics_observer.metrics[task_timings].clear() metrics_observer.metrics[constellation_timings].clear()4. 生命周期管理与其余观察者一致会话结束或执行异常后应统一unsubscribe避免观察者持续接收事件造成内存泄漏与多余开销模式详见 overview.md 中的 Observer Lifecycle Management 章节。5. 复用现有装配而非重复创建在GalaxySession场景下观察者已被 galaxy_session.py 自动创建并订阅应通过session_results[metrics]读取而非在外部重复实例化。九、测试与验证仓库中为观察者体系提供了多层次的测试覆盖可作为理解行为与验证自定义扩展的参考test_session_observers.py —— 会话观察者的功能测试test_modular_observers.py 与 test_observer_modular_structure.py —— 观察者模块化结构的单元测试test_event_system.py ——EventBus、EventType与订阅/发布机制的测试test_galaxy_framework_summary.py —— 框架级汇总测试覆盖会话结果与指标输出。这些测试一方面验证了事件分派、计数累加与统计计算的正确性另一方面也展示了如何在自定义场景中组合EventBus与观察者进行可复现的指标采集。十、总结SessionMetricsObserver是 UFO Galaxy 观察者体系中负责性能度量和统计分析的关键组件通过订阅全局EventBus在星座执行期间被动采集任务与星座的生命周期事件实时维护计数与时序数据并在查询时聚合出三类统计摘要收集完整的性能指标任务数、完成/失败数、执行耗时跟踪任务与星座的执行时间及星座初始/终态统计快照监控星座结构修改模式借助VisualizationChangeDetector记录细粒度的 DAG 变更明细计算成功率、平均/最值耗时、每星座平均任务数、修改类型分布等统计摘要导出结构化数据JSON / session_results供性能评估、瓶颈定位与科研分析。配合事件系统核心文档event_system.md与观察者体系总览overview.md即可完整理解并复用这套指标采集能力。在涉及星座执行与编排的深入场景中可进一步参阅 constellation_orchestrator 目录 下的编排实现与文档。【免费下载链接】UFOUFO³: Weaving the Digital Agent Galaxy项目地址: https://gitcode.com/GitHub_Trending/uf/UFO创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考