349 lines
12 KiB
Python
349 lines
12 KiB
Python
"""
|
||
全局 Agent 状态监控脚本
|
||
列出三个 agent 所有会话的实时状态，一眼看出哪些在跑、哪些卡住了。
|
||
|
||
用法：
|
||
python agent_status.py # 查看所有 agent 的所有会话
|
||
python agent_status.py --watch # 实时刷新模式（每 3 秒刷新一次）
|
||
python agent_status.py -w -i 2 # 实时刷新，间隔 2 秒
|
||
python agent_status.py --agent my_agent # 只看指定 agent
|
||
python agent_status.py --limit 20 # 每个 agent 最多显示 20 条
|
||
python agent_status.py -s my_session # 查看特定会话的详细状态
|
||
"""
|
||
import argparse
|
||
import json
|
||
import os
|
||
import sys
|
||
import time
|
||
from datetime import datetime
|
||
|
||
import httpx
|
||
|
||
|
||
# 三个 agent 的 API 地址
|
||
AGENTS = {
|
||
"my_agent": "http://127.0.0.1:8001",
|
||
"luna_agent": "http://127.0.0.1:8002",
|
||
"qwen_agent": "http://127.0.0.1:8003",
|
||
}
|
||
|
||
USER_ID = "codebuddy"
|
||
|
||
|
||
def check_api_alive(url: str) -> bool:
|
||
"""检查 API Server 是否存活"""
|
||
try:
|
||
resp = httpx.get(f"{url}/health", timeout=2.0)
|
||
return resp.status_code == 200
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def list_sessions(api_url: str, app_name: str, user_id: str) -> list[dict]:
|
||
"""获取所有会话列表"""
|
||
try:
|
||
resp = httpx.get(
|
||
f"{api_url}/apps/{app_name}/users/{user_id}/sessions",
|
||
timeout=5.0,
|
||
)
|
||
if resp.status_code == 200:
|
||
data = resp.json()
|
||
# 响应可能是 list 或 {sessions: [...]}
|
||
if isinstance(data, list):
|
||
return data
|
||
return data.get("sessions", [])
|
||
return []
|
||
except Exception:
|
||
return []
|
||
|
||
|
||
def get_session_detail(api_url: str, app_name: str, user_id: str, session_id: str) -> dict | None:
|
||
"""获取会话详情"""
|
||
try:
|
||
resp = httpx.get(
|
||
f"{api_url}/apps/{app_name}/users/{user_id}/sessions/{session_id}",
|
||
timeout=5.0,
|
||
)
|
||
if resp.status_code == 200:
|
||
return resp.json()
|
||
return None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def format_time(timestamp: float) -> str:
|
||
"""格式化时间戳"""
|
||
if not timestamp:
|
||
return "?"
|
||
try:
|
||
return datetime.fromtimestamp(timestamp).strftime("%m-%d %H:%M:%S")
|
||
except Exception:
|
||
return str(timestamp)[:19]
|
||
|
||
|
||
def get_last_event_summary(events: list[dict]) -> tuple[str, str, str, str]:
|
||
"""
|
||
从事件列表提取最后一条事件的摘要信息。
|
||
返回: (角色, 作者, 状态描述, 内容摘要)
|
||
"""
|
||
if not events:
|
||
return ("-", "-", "(空会话)", "")
|
||
|
||
last = events[-1]
|
||
content = last.get("content", {})
|
||
role = content.get("role", "?")
|
||
author = last.get("author", "")
|
||
parts = content.get("parts", [])
|
||
|
||
# 判断状态
|
||
status = ""
|
||
summary = ""
|
||
|
||
role_cn = {
|
||
"user": "用户输入",
|
||
"model": "模型回复",
|
||
"function": "工具调用",
|
||
}.get(role, role)
|
||
|
||
for part in parts:
|
||
if "thought" in part and part.get("thought"):
|
||
status = "💭 思考中"
|
||
text = part.get("text", "")
|
||
summary = text[:60].replace("\n", " ")
|
||
break
|
||
elif "functionCall" in part:
|
||
call = part["functionCall"]
|
||
status = f"📞 调用中: {call.get('name', '?')}"
|
||
args = call.get("args", {})
|
||
# 显示关键参数
|
||
if "path" in args:
|
||
summary = f"path: {args['path'][:50]}"
|
||
elif "command" in args:
|
||
summary = f"cmd: {args['command'][:50]}"
|
||
else:
|
||
args_str = json.dumps(args, ensure_ascii=False)[:60]
|
||
summary = args_str
|
||
break
|
||
elif "functionResponse" in part:
|
||
resp = part["functionResponse"]
|
||
status = f"✅ 工具返回: {resp.get('name', '?')}"
|
||
resp_content = resp.get("content", [])
|
||
text = ""
|
||
for c in resp_content:
|
||
if isinstance(c, dict) and c.get("type") == "text":
|
||
text += c.get("text", "")
|
||
elif isinstance(c, str):
|
||
text += c
|
||
summary = text[:80].replace("\n", " ")
|
||
if resp.get("isError"):
|
||
status = f"❌ 工具错误: {resp.get('name', '?')}"
|
||
break
|
||
elif "text" in part:
|
||
status = f"💬 {role_cn}"
|
||
summary = part["text"][:80].replace("\n", " ")
|
||
break
|
||
|
||
if not status:
|
||
status = f"📨 {role_cn}"
|
||
|
||
return (role, author, status, summary)
|
||
|
||
|
||
def print_status_table(agent_names: list[str], limit: int):
|
||
"""打印所有 agent 的会话状态表格"""
|
||
total_sessions = 0
|
||
active_count = 0
|
||
|
||
for agent_name in agent_names:
|
||
api_url = AGENTS[agent_name]
|
||
alive = check_api_alive(api_url)
|
||
|
||
print(f"\n{'═' * 80}")
|
||
status_icon = "🟢" if alive else "🔴"
|
||
print(f"{status_icon} {agent_name} ({api_url})")
|
||
print(f"{'─' * 80}")
|
||
|
||
if not alive:
|
||
print(" ⚠️ API Server 未启动或无法连接")
|
||
continue
|
||
|
||
sessions = list_sessions(api_url, agent_name, USER_ID)
|
||
total_sessions += len(sessions)
|
||
|
||
if not sessions:
|
||
print(" （暂无会话）")
|
||
continue
|
||
|
||
# 按更新时间倒序（字段名可能是 lastUpdateTime 或 last_update_time）
|
||
sessions.sort(
|
||
key=lambda s: s.get("lastUpdateTime") or s.get("last_update_time", 0),
|
||
reverse=True,
|
||
)
|
||
|
||
# 只显示 limit 条
|
||
shown = sessions[:limit]
|
||
hidden_count = len(sessions) - limit
|
||
|
||
print(f" {'#':>3s} {'最后更新时间':<18s} {'事件数':>5s} {'状态'}")
|
||
print(f" {'─' * 76}")
|
||
|
||
for idx, sess in enumerate(shown, 1):
|
||
sid = sess.get("id", "?")
|
||
last_time = sess.get("lastUpdateTime") or sess.get("last_update_time", 0)
|
||
time_str = format_time(last_time)
|
||
|
||
# 取详情获取最后事件
|
||
detail = get_session_detail(api_url, agent_name, USER_ID, sid)
|
||
events = detail.get("events", []) if detail else []
|
||
event_count = len(events)
|
||
_, _, status, _ = get_last_event_summary(events)
|
||
|
||
# 判断是否活跃（5分钟内有更新）
|
||
now_ts = time.time()
|
||
is_active = (now_ts - last_time) < 300
|
||
if is_active and event_count > 0:
|
||
active_count += 1
|
||
active_icon = "⚡"
|
||
else:
|
||
active_icon = " "
|
||
|
||
sid_short = sid if len(sid) <= 12 else sid[:10] + ".."
|
||
print(f" {active_icon}{idx:>2d}. {time_str} {event_count:>5d} {status[:50]}")
|
||
print(f" id: {sid}")
|
||
|
||
if hidden_count > 0:
|
||
print(f"\n ... 还有 {hidden_count} 个会话未显示（共 {len(sessions)} 个）")
|
||
|
||
print(f"\n{'═' * 80}")
|
||
print(f" 总计: {total_sessions} 个会话 | 活跃中（5分钟内有更新）: {active_count}")
|
||
print(f"{'═' * 80}\n")
|
||
|
||
|
||
def watch_mode(agent_names: list[str], interval: float, limit: int):
|
||
"""实时刷新模式"""
|
||
print(f"\n🔄 实时监控模式（每 {interval} 秒刷新，Ctrl+C 退出）\n")
|
||
try:
|
||
while True:
|
||
# 清屏
|
||
if os.name == "nt":
|
||
os.system("cls")
|
||
else:
|
||
os.system("clear")
|
||
print_status_table(agent_names, limit)
|
||
print(f" 最后刷新: {datetime.now().strftime('%H:%M:%S')} | Ctrl+C 退出")
|
||
time.sleep(interval)
|
||
except KeyboardInterrupt:
|
||
print("\n👋 已退出监控。")
|
||
|
||
|
||
def show_session_detail(agent_name: str, session_id: str):
|
||
"""查看特定会话的详细状态"""
|
||
api_url = AGENTS.get(agent_name, "")
|
||
if not api_url:
|
||
print(f"未知 agent: {agent_name}")
|
||
return
|
||
|
||
alive = check_api_alive(api_url)
|
||
if not alive:
|
||
print(f"⚠️ {agent_name} API Server 未启动（{api_url}）")
|
||
return
|
||
|
||
detail = get_session_detail(api_url, agent_name, USER_ID, session_id)
|
||
if not detail:
|
||
print(f"会话 [{session_id}] 不存在")
|
||
return
|
||
|
||
events = detail.get("events", [])
|
||
event_count = len(events)
|
||
last_time = detail.get("lastUpdateTime") or detail.get("last_update_time", 0)
|
||
# 会话详情里没有 create_time，从第一个事件的时间戳估算
|
||
create_time = detail.get("create_time", 0)
|
||
if not create_time and events:
|
||
create_time = events[0].get("timestamp", 0)
|
||
|
||
print(f"\n{'═' * 80}")
|
||
print(f"📋 会话详情")
|
||
print(f"{'═' * 80}")
|
||
print(f" Agent: {agent_name}")
|
||
print(f" Session: {session_id}")
|
||
print(f" 创建时间: {format_time(create_time)}")
|
||
print(f" 更新时间: {format_time(last_time)}")
|
||
print(f" 事件数: {event_count}")
|
||
|
||
if events:
|
||
duration = last_time - create_time if create_time and last_time else 0
|
||
if duration > 0:
|
||
mins = int(duration // 60)
|
||
secs = int(duration % 60)
|
||
print(f" 运行时长: {mins}分{secs}秒")
|
||
|
||
# 最后 5 条事件
|
||
print(f"\n{'─' * 80}")
|
||
print(f" 最后 5 条事件：")
|
||
print(f"{'─' * 80}")
|
||
|
||
for i, event in enumerate(events[-5:], max(1, event_count - 4)):
|
||
role, author, status, summary = get_last_event_summary([event])
|
||
ts = event.get("timestamp", 0)
|
||
t_str = format_time(ts).split()[-1] if " " in format_time(ts) else format_time(ts)
|
||
print(f"\n #{i} [{t_str}] {status}")
|
||
if summary:
|
||
print(f" {summary[:100]}")
|
||
|
||
print(f"\n{'═' * 80}\n")
|
||
|
||
|
||
def main():
|
||
global USER_ID
|
||
parser = argparse.ArgumentParser(description="全局 Agent 状态监控工具")
|
||
parser.add_argument("--agent", "-a", default=None,
|
||
help="只查看指定 agent（默认查看所有）")
|
||
parser.add_argument("--watch", "-w", action="store_true",
|
||
help="实时刷新模式")
|
||
parser.add_argument("--interval", "-i", type=float, default=3.0,
|
||
help="刷新间隔秒数（默认 3.0）")
|
||
parser.add_argument("--limit", "-l", type=int, default=10,
|
||
help="每个 agent 最多显示的会话数（默认 10）")
|
||
parser.add_argument("--session", "-s", default=None,
|
||
help="查看特定会话的详细状态")
|
||
parser.add_argument("--user", "-u", default="codebuddy",
|
||
help="用户 ID（默认 codebuddy）")
|
||
args = parser.parse_args()
|
||
|
||
global USER_ID
|
||
USER_ID = args.user
|
||
|
||
# 确定要查看的 agent 列表
|
||
if args.agent:
|
||
agent_name = args.agent
|
||
# 支持别名
|
||
aliases = {
|
||
"my": "my_agent", "default": "my_agent", "aq": "my_agent",
|
||
"luna": "luna_agent", "gpt": "luna_agent",
|
||
"qwen": "qwen_agent", "astron": "qwen_agent",
|
||
}
|
||
if agent_name in aliases:
|
||
agent_name = aliases[agent_name]
|
||
if agent_name not in AGENTS:
|
||
print(f"未知 agent: {args.agent}")
|
||
print(f"可用: {list(AGENTS.keys())}")
|
||
sys.exit(1)
|
||
agent_names = [agent_name]
|
||
else:
|
||
agent_names = list(AGENTS.keys())
|
||
|
||
# 查看单个会话详情
|
||
if args.session:
|
||
show_session_detail(agent_names[0], args.session)
|
||
return
|
||
|
||
# 实时刷新模式
|
||
if args.watch:
|
||
watch_mode(agent_names, args.interval, args.limit)
|
||
else:
|
||
print_status_table(agent_names, args.limit)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|