Coverage for agentos/dashboard/tracker.py: 0%
99 statements
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
« prev ^ index » next coverage.py v7.14.3, created at 2026-07-06 10:59 +0800
1"""
2追踪状态管理器 — 记录 Agent 运行历史、会话、步骤。
4数据存储在 ~/.agentos/tracker/ 下,以 JSONL 追加写入。
5"""
7from __future__ import annotations
9import json
10import time
11from dataclasses import dataclass, field, asdict
12from pathlib import Path
15TRACKER_DIR = Path.home() / ".agentos" / "tracker"
18@dataclass
19class StepRecord:
20 """单步执行记录。"""
21 step_index: int
22 step_type: str # "thinking" | "tool_call" | "tool_result" | "final_answer"
23 detail: str # 步骤内容摘要
24 duration_ms: float
25 tokens: int = 0
28@dataclass
29class SessionRecord:
30 """单次会话完整记录。"""
31 session_id: str
32 task: str
33 model: str
34 provider: str
35 started_at: float = field(default_factory=time.time)
36 finished_at: float = 0.0
37 status: str = "running" # "running" | "completed" | "error" | "cancelled"
38 steps: list[StepRecord] = field(default_factory=list)
39 total_tokens: int = 0
40 total_cost_usd: float = 0.0
41 error: str = ""
44class Tracker:
45 """线程安全的追踪记录器(文件级锁)+ 事件发布。
47 支持订阅者模式:外部可 subscribe 回调,在 add_step/finish_session 时收到实时事件推送。
48 """
50 _instance: Tracker | None = None
52 def __init__(self):
53 TRACKER_DIR.mkdir(parents=True, exist_ok=True)
54 self._sessions_file = TRACKER_DIR / "sessions.jsonl"
55 self._active: dict[str, SessionRecord] = {}
56 self._subscribers: list = [] # 回调列表
58 @classmethod
59 def get(cls) -> Tracker:
60 if cls._instance is None:
61 cls._instance = cls()
62 return cls._instance
64 def start_session(self, session_id: str, task: str, model: str = "", provider: str = "") -> SessionRecord:
65 rec = SessionRecord(session_id=session_id, task=task, model=model, provider=provider)
66 self._active[session_id] = rec
67 return rec
69 def add_step(self, session_id: str, step_type: str, detail: str, duration_ms: float = 0.0, tokens: int = 0):
70 rec = self._active.get(session_id)
71 if rec is None:
72 return
73 step = StepRecord(
74 step_index=len(rec.steps),
75 step_type=step_type,
76 detail=detail,
77 duration_ms=duration_ms,
78 tokens=tokens,
79 )
80 rec.steps.append(step)
81 rec.total_tokens += tokens
82 self._notify("step", {"session_id": session_id, "step_type": step_type, "detail": detail, "duration_ms": duration_ms, "tokens": tokens})
84 def finish_session(self, session_id: str, status: str = "completed", error: str = "", total_cost: float = 0.0):
85 rec = self._active.pop(session_id, None)
86 if rec is None:
87 return
88 rec.finished_at = time.time()
89 rec.status = status
90 rec.error = error
91 rec.total_cost_usd = total_cost
92 with open(self._sessions_file, "a") as f:
93 f.write(json.dumps(asdict(rec), ensure_ascii=False) + "\n")
94 self._notify("session_done", asdict(rec))
96 def subscribe(self, callback):
97 """订阅实时事件。callback 接收 (event_type: str, data: dict)。"""
98 self._subscribers.append(callback)
100 def unsubscribe(self, callback):
101 """取消订阅。"""
102 try:
103 self._subscribers.remove(callback)
104 except ValueError:
105 pass
107 def _notify(self, event_type: str, data: dict):
108 for cb in self._subscribers:
109 try:
110 cb(event_type, data)
111 except Exception:
112 pass
114 def list_sessions(self, limit: int = 50) -> list[dict]:
115 sessions = []
116 if self._sessions_file.exists():
117 with open(self._sessions_file, "r") as f:
118 for line in f:
119 if line.strip():
120 sessions.append(json.loads(line))
121 # 倒序,最新的在前
122 sessions.reverse()
123 return sessions[:limit]
125 def get_session(self, session_id: str) -> dict | None:
126 # 先在 active 中找
127 rec = self._active.get(session_id)
128 if rec:
129 return asdict(rec)
130 # 再从文件中找
131 if self._sessions_file.exists():
132 with open(self._sessions_file, "r") as f:
133 for line in f:
134 if line.strip():
135 d = json.loads(line)
136 if d.get("session_id") == session_id:
137 return d
138 return None
140 def clear(self):
141 self._active.clear()
142 if self._sessions_file.exists():
143 self._sessions_file.unlink()