Coverage for merco/observability/observer.py: 89%
162 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-07 14:04 +0800
1"""Observer — 订阅 hooks 事件的轻量可观察性门面"""
3from merco.hooks.registry import HookRegistry
4from merco.observability.metrics import MetricsCollector
7class Observer:
8 """Agent 可观察性门面。
10 两套计数器：
11 - `_live`: 当前运行（/new / /sessions 切换时 reset）
12 - `_acc_map`: 跨运行累计（持久化到 SQLite，从不重置）
13 """
15 def __init__(self, hooks: HookRegistry):
16 self._live = MetricsCollector()
17 self._acc_map: dict[str, int] = {}
18 self._last_merged: dict[str, int] = {} # 记录上次合并时的 live 值
19 self._current_session: str | None = None # 当前会话 ID
21 hooks.on("llm.chat", self._on_llm)
22 hooks.on("tool.after_execute", self._on_tool)
23 hooks.on("tool.error", self._on_error)
24 hooks.on("conversation.turn", self._on_turn)
25 hooks.on("agent.interrupted", self._on_interrupt)
26 hooks.on("agent.start", self._on_agent_start)
27 hooks.on("agent.stop", self._on_agent_stop)
28 hooks.on("context.compact", self._on_context_compact)
29 hooks.on("memory.saved", self._on_memory_saved)
30 hooks.on("plugin.activated", self._on_plugin_activated)
31 hooks.on("plugin.error", self._on_plugin_error)
33 # ── 事件处理 ──────────────────────────────────────────
35 def _on_llm(self, duration: float = 0, tokens_in: int = 0,
36 tokens_out: int = 0, cached_tokens: int = 0,
37 cache_read_tokens: int = 0, **kwargs):
38 self._live.increment("llm_calls")
39 self._live.record_timing("llm", duration)
40 if tokens_in:
41 self._live.increment("tokens_in", tokens_in)
42 if tokens_out:
43 self._live.increment("tokens_out", tokens_out)
44 cache = cached_tokens or cache_read_tokens
45 if cache:
46 self._live.increment("cache_hit_tokens", cache)
48 def _on_tool(self, tool_name: str = "", duration: float = 0, **kwargs):
49 self._live.increment("tool_calls")
50 self._live.increment(f"tool.{tool_name}")
51 self._live.record_timing(f"tool.{tool_name}", duration)
53 def _on_error(self, tool_name: str = "", error: str = "", **kwargs):
54 self._live.increment("errors")
56 def _on_turn(self, **kwargs):
57 self._live.increment("turns")
59 def _on_interrupt(self, interrupted_tools: int = 0, **kwargs):
60 """中断时记录统计。"""
61 # 记录中断的 LLM 调用
62 self._live.increment("llm_calls_interrupted")
63 if interrupted_tools:
64 self._live.increment("tool_calls_interrupted", interrupted_tools)
65 self._live.increment("tool_calls", interrupted_tools)
66 # 中断也算一轮（用户确实发起了请求）
67 self._live.increment("turns")
69 def _on_agent_start(self, session_id: str, **kwargs):
70 """Agent 启动"""
71 self._live.increment("agent_starts")
72 self._current_session = session_id
74 def _on_agent_stop(self, session_id: str, **kwargs):
75 """Agent 停止"""
76 self._live.increment("agent_stops")
78 def _on_context_compact(self, strategy: str, **kwargs):
79 """上下文压缩"""
80 self._live.increment("context_compactions")
82 def _on_memory_saved(self, key: str = "", source: str = "", **kwargs):
83 """记忆保存"""
84 self._live.increment("memories_saved")
86 def _on_plugin_activated(self, plugin_name: str = "", **kwargs):
87 """插件激活"""
88 self._live.increment(f"plugin.{plugin_name}.activations")
90 def _on_plugin_error(self, plugin_name: str = "", **kwargs):
91 """插件错误"""
92 self._live.increment(f"plugin.{plugin_name}.errors")
94 # ── 生命周期 ──────────────────────────────────────────
96 def reset(self, full: bool = False):
97 """清空 live 计数器。full=True 时也清空累计。"""
98 self._live = MetricsCollector()
99 self._last_merged = {} # 清空上次合并的值
100 if full:
101 self._acc_map = {}
103 # ── 持久化 ────────────────────────────────────────────
105 def save(self):
106 """存盘前：把 live 合并到 acc_map"""
107 self._merge_to_acc()
109 def snapshot(self) -> dict:
110 """导出：live 计数器 + acc_map 累计"""
111 return {
112 "live": self._live.get_counters(),
113 "acc": dict(self._acc_map),
114 }
116 def _merge_to_acc(self):
117 # 只累加增量（当前值 - 上次合并的值）
118 for k, v in self._live.get_counters().items():
119 last = self._last_merged.get(k, 0)
120 delta = v - last
121 if delta > 0:
122 self._acc_map[k] = self._acc_map.get(k, 0) + delta
123 self._last_merged[k] = v
125 def restore(self, data: dict):
126 """从快照恢复 acc_map"""
127 self._acc_map = dict(data.get("acc", {}))
129 # ── 报告 ──────────────────────────────────────────────
131 def report(self) -> str:
132 live = self._live
133 acc = self._acc_map
135 turns = live.get_counter("turns")
136 llm_calls = live.get_counter("llm_calls")
137 tool_calls = live.get_counter("tool_calls")
138 errors = live.get_counter("errors")
139 llm_avg = live.get_avg_timing("llm")
140 tokens_in = live.get_counter("tokens_in")
141 tokens_out = live.get_counter("tokens_out")
142 cache_hit = live.get_counter("cache_hit_tokens")
143 agent_starts = live.get_counter("agent_starts")
144 agent_stops = live.get_counter("agent_stops")
145 context_compactions = live.get_counter("context_compactions")
147 acc_turns = acc.get("turns", 0)
148 acc_llm = acc.get("llm_calls", 0)
149 acc_tools = acc.get("tool_calls", 0)
150 acc_errors = acc.get("errors", 0)
151 acc_tokens_in = acc.get("tokens_in", 0)
152 acc_tokens_out = acc.get("tokens_out", 0)
153 acc_agent_starts = acc.get("agent_starts", 0)
154 acc_agent_stops = acc.get("agent_stops", 0)
155 acc_compactions = acc.get("context_compactions", 0)
157 lines = ["[bold]📊 会话报告[/bold]", ""]
159 # 本次
160 lines.append(f" 本次: {turns} 轮 {llm_calls} 次 LLM 平均 {llm_avg:.1f}s")
161 if tokens_in or tokens_out:
162 ratio = f" {cache_hit / tokens_in * 100:.0f}% 缓存命中" if tokens_in and cache_hit else ""
163 lines.append(f" 入 {_fmt_n(tokens_in)} tokens 出 {_fmt_n(tokens_out)} tokens {ratio}")
164 if tool_calls:
165 parts = []
166 for name in sorted(live.get_counters()):
167 if name.startswith("tool.") and name != "tool_calls":
168 cnt = live.get_counters().get(name, 0)
169 avg = live.get_avg_timing(name)
170 parts.append(f"[dim]{name[5:]}[/dim] {cnt}次({avg:.1f}s)")
171 lines.append(f" 工具: {', '.join(parts)}" if parts else f" 工具: {tool_calls} 次")
173 interrupted_tools = live.get_counter("tool_calls_interrupted")
174 interrupted_llm = live.get_counter("llm_calls_interrupted")
175 if interrupted_tools or interrupted_llm:
176 parts = []
177 if interrupted_llm:
178 parts.append(f"{interrupted_llm} 次 LLM")
179 if interrupted_tools:
180 parts.append(f"{interrupted_tools} 次工具调用")
181 lines.append(f" [yellow]中断: {', '.join(parts)}[/yellow]")
183 if errors:
184 lines.append(f" [red]错误: {errors}[/red]")
186 # Agent 启动/停止 & 上下文压缩
187 agent_parts = []
188 if agent_starts:
189 agent_parts.append(f"启动 {agent_starts}")
190 if agent_stops:
191 agent_parts.append(f"停止 {agent_stops}")
192 if context_compactions:
193 agent_parts.append(f"上下文压缩 {context_compactions}")
194 if agent_parts:
195 lines.append(f" Agent: {', '.join(agent_parts)}")
197 # 累计 — acc 里已含上次 merge 的 live 值，只加「未合并增量」避免重复计数
198 lm = self._last_merged
199 u_turns = turns - lm.get("turns", 0)
200 u_llm = llm_calls - lm.get("llm_calls", 0)
201 u_tools = tool_calls - lm.get("tool_calls", 0)
202 u_tokens_in = tokens_in - lm.get("tokens_in", 0)
203 u_tokens_out = tokens_out - lm.get("tokens_out", 0)
204 u_errors = errors - lm.get("errors", 0)
205 u_agent_starts = agent_starts - lm.get("agent_starts", 0)
206 u_agent_stops = agent_stops - lm.get("agent_stops", 0)
207 u_compactions = context_compactions - lm.get("context_compactions", 0)
209 if acc_turns or acc_llm or acc_tools:
210 lines.append("")
211 lines.append(
212 f" 累计: {acc_turns + u_turns} 轮 {acc_llm + u_llm} 次 LLM "
213 f"入 {_fmt_n(acc_tokens_in + u_tokens_in)} tokens "
214 f"出 {_fmt_n(acc_tokens_out + u_tokens_out)} tokens "
215 f"{acc_tools + u_tools} 次工具"
216 + (f" [red]{acc_errors + u_errors} 错误[/red]" if acc_errors + u_errors else "")
217 )
219 # 累计 Agent 启动/停止 & 上下文压缩
220 acc_agent_parts = []
221 total_starts = acc_agent_starts + u_agent_starts
222 total_stops = acc_agent_stops + u_agent_stops
223 total_compactions = acc_compactions + u_compactions
224 if total_starts:
225 acc_agent_parts.append(f"启动 {total_starts}")
226 if total_stops:
227 acc_agent_parts.append(f"停止 {total_stops}")
228 if total_compactions:
229 acc_agent_parts.append(f"上下文压缩 {total_compactions}")
230 if acc_agent_parts:
231 lines.append(f" 累计: Agent {', '.join(acc_agent_parts)}")
233 if not turns and not acc_turns:
234 lines.append(" [dim]暂无数据[/dim]")
236 return "\n".join(lines)
239def _fmt_n(n: int) -> str:
240 if n < 1000:
241 return str(n)
242 return f"{n / 1024:.1f}K"