Phase 4-6: Message Bus, Memory Search (ChromaDB), Token Tracking, Topology Graph
Phase 4: Message Bus Viewer
- Backend: get_message_bus_status(), send_bus_message() in agents.py
- Route: GET /api/agents/message-bus, POST /api/agents/{id}/bus-message
- Frontend: Message Bus tab in agent detail overlay
Phase 5: Memory Search (ChromaDB)
- Backend: _search_agent_memory(), _search_all_agents_memory() via ChromaDB rose_memory collection
- Route: GET /api/agents/memory/search, GET /api/agents/{id}/memory/search
- Frontend: Search bar added to Memory tab, renders confidence scores + topics
Phase 6: Token Tracking + Topology Graph
- Backend: get_agent_usage() reads ~/.hermes/agents/{id}/usage.json
- Route: GET /api/agents/{id}/usage
- Frontend: Usage tab with today/week/month token counts and cost
- Frontend: Topology tab with SVG radial graph of agent network
This commit is contained in:
Rose
355
api/agents.py
355
api/agents.py
@@ -14,6 +14,9 @@ from typing import Any
|
||||
|
||||
from api.helpers import j
|
||||
|
||||
# ChromaDB for memory search
|
||||
import chromadb
|
||||
|
||||
# ── Paths ──────────────────────────────────────────────────────────────────────
|
||||
_HERMES_DIR = Path.home() / ".hermes"
|
||||
_AGENTS_DIR = _HERMES_DIR / "agents"
|
||||
@@ -516,6 +519,102 @@ def get_agent_errors(agent_id: str, limit: int = 20) -> dict:
|
||||
}
|
||||
|
||||
|
||||
# ── Token / Cost Usage Tracking ───────────────────────────────────────────────
|
||||
|
||||
def _get_agent_usage(agent_id: str) -> dict:
|
||||
"""
|
||||
Read ~/.hermes/agents/{agent_id}/usage.json and compute daily/weekly/monthly totals.
|
||||
Returns {agent_id, today: {tokens, cost}, week: {tokens, cost}, month: {tokens, cost}, history: [...]}.
|
||||
If the usage file doesn't exist, returns zeros for all periods.
|
||||
"""
|
||||
if agent_id not in TIER2_AGENTS and agent_id != "rose":
|
||||
return {"error": f"Unknown agent: {agent_id}"}
|
||||
|
||||
usage_path = _AGENTS_DIR / agent_id / "usage.json"
|
||||
if not usage_path.exists():
|
||||
return {
|
||||
"agent_id": agent_id,
|
||||
"today": {"tokens": 0, "cost": 0.0},
|
||||
"week": {"tokens": 0, "cost": 0.0},
|
||||
"month": {"tokens": 0, "cost": 0.0},
|
||||
"history": [],
|
||||
}
|
||||
|
||||
try:
|
||||
with open(usage_path, "r") as f:
|
||||
data = json.load(f)
|
||||
except (json.JSONDecodeError, IOError):
|
||||
return {
|
||||
"agent_id": agent_id,
|
||||
"today": {"tokens": 0, "cost": 0.0},
|
||||
"week": {"tokens": 0, "cost": 0.0},
|
||||
"month": {"tokens": 0, "cost": 0.0},
|
||||
"history": [],
|
||||
}
|
||||
|
||||
token_usage = data.get("token_usage", [])
|
||||
today_str = datetime.utcnow().strftime("%Y-%m-%d")
|
||||
|
||||
# Compute period boundaries
|
||||
now = datetime.utcnow()
|
||||
week_ago = now - timedelta(days=7)
|
||||
month_ago = now - timedelta(days=30)
|
||||
|
||||
today_tokens = 0
|
||||
today_cost = 0.0
|
||||
week_tokens = 0
|
||||
week_cost = 0.0
|
||||
month_tokens = 0
|
||||
month_cost = 0.0
|
||||
|
||||
history = []
|
||||
for entry in token_usage:
|
||||
date_str = entry.get("date", "")
|
||||
prompt = entry.get("prompt_tokens", 0)
|
||||
completion = entry.get("completion_tokens", 0)
|
||||
cost = entry.get("cost_usd", 0.0)
|
||||
total = prompt + completion
|
||||
|
||||
try:
|
||||
entry_date = datetime.strptime(date_str, "%Y-%m-%d")
|
||||
except ValueError:
|
||||
continue
|
||||
|
||||
history.append({
|
||||
"date": date_str,
|
||||
"prompt_tokens": prompt,
|
||||
"completion_tokens": completion,
|
||||
"total_tokens": total,
|
||||
"cost_usd": cost,
|
||||
})
|
||||
|
||||
if date_str == today_str:
|
||||
today_tokens += total
|
||||
today_cost += cost
|
||||
if entry_date >= week_ago:
|
||||
week_tokens += total
|
||||
week_cost += cost
|
||||
if entry_date >= month_ago:
|
||||
month_tokens += total
|
||||
month_cost += cost
|
||||
|
||||
# Sort history newest-first
|
||||
history.sort(key=lambda x: x["date"], reverse=True)
|
||||
|
||||
return {
|
||||
"agent_id": agent_id,
|
||||
"today": {"tokens": today_tokens, "cost": round(today_cost, 6)},
|
||||
"week": {"tokens": week_tokens, "cost": round(week_cost, 6)},
|
||||
"month": {"tokens": month_tokens, "cost": round(month_cost, 6)},
|
||||
"history": history,
|
||||
}
|
||||
|
||||
|
||||
def get_agent_usage(agent_id: str) -> dict:
|
||||
"""API: GET /api/agents/{id}/usage — return token/cost usage."""
|
||||
return _get_agent_usage(agent_id)
|
||||
|
||||
|
||||
# ── Chat History ──────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_chat_history(agent_id: str, limit: int = 20) -> list[dict]:
|
||||
@@ -712,3 +811,259 @@ def get_agent_tasks(agent_id: str) -> dict:
|
||||
"tasks": tasks,
|
||||
"count": len(tasks),
|
||||
}
|
||||
|
||||
|
||||
# ── Message Bus ────────────────────────────────────────────────────────────────
|
||||
|
||||
def _get_message_bus_overview() -> dict:
|
||||
"""
|
||||
Return a overview of all agent inboxes — message counts and recent messages.
|
||||
"""
|
||||
agents = []
|
||||
for agent_id in TIER2_AGENTS:
|
||||
inbox_path = _AGENTS_DIR / agent_id / "inbox.json"
|
||||
count = 0
|
||||
last_msg = None
|
||||
if inbox_path.exists():
|
||||
try:
|
||||
inbox = json.loads(inbox_path.read_text())
|
||||
messages = inbox.get("messages", [])
|
||||
count = len(messages)
|
||||
if messages:
|
||||
last_msg = {
|
||||
"from": messages[-1].get("from"),
|
||||
"subject": messages[-1].get("subject"),
|
||||
"timestamp": messages[-1].get("timestamp"),
|
||||
}
|
||||
except Exception:
|
||||
pass
|
||||
agents.append({
|
||||
"agent_id": agent_id,
|
||||
"name": TIER2_AGENTS[agent_id].get("name", agent_id),
|
||||
"emoji": TIER2_AGENTS[agent_id].get("emoji", "•"),
|
||||
"inbox_count": count,
|
||||
"last_message": last_msg,
|
||||
})
|
||||
|
||||
# Rose inbox
|
||||
rose_inbox_path = _HERMES_DIR / "inbox.json"
|
||||
rose_count = 0
|
||||
rose_last = None
|
||||
if rose_inbox_path.exists():
|
||||
try:
|
||||
inbox = json.loads(rose_inbox_path.read_text())
|
||||
messages = inbox.get("messages", [])
|
||||
rose_count = len(messages)
|
||||
if messages:
|
||||
rose_last = {
|
||||
"from": messages[-1].get("from"),
|
||||
"subject": messages[-1].get("subject"),
|
||||
"timestamp": messages[-1].get("timestamp"),
|
||||
}
|
||||
except Exception:
|
||||
pass
|
||||
agents.insert(0, {
|
||||
"agent_id": "rose",
|
||||
"name": "Rose",
|
||||
"emoji": "🌹",
|
||||
"inbox_count": rose_count,
|
||||
"last_message": rose_last,
|
||||
})
|
||||
|
||||
return {"agents": agents}
|
||||
|
||||
|
||||
def get_message_bus_overview() -> dict:
|
||||
"""API: GET /api/agents/message-bus — overview of all agent inboxes."""
|
||||
return _get_message_bus_overview()
|
||||
|
||||
|
||||
def send_bus_message(target_agent: str, from_agent: str, subject: str, content: str) -> dict:
|
||||
"""
|
||||
Write a message directly into an agent's inbox.json via the message bus.
|
||||
"""
|
||||
if target_agent not in TIER2_AGENTS and target_agent != "rose":
|
||||
return {"ok": False, "error": f"Unknown agent: {target_agent}"}
|
||||
|
||||
if target_agent == "rose":
|
||||
inbox_path = _HERMES_DIR / "inbox.json"
|
||||
else:
|
||||
inbox_path = _AGENTS_DIR / target_agent / "inbox.json"
|
||||
|
||||
inbox_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
try:
|
||||
if inbox_path.exists():
|
||||
inbox = json.loads(inbox_path.read_text())
|
||||
else:
|
||||
inbox = {"messages": []}
|
||||
except Exception:
|
||||
inbox = {"messages": []}
|
||||
|
||||
import datetime
|
||||
msg_id = f"bus_{int(datetime.datetime.now().timestamp() * 1000)}"
|
||||
message = {
|
||||
"id": msg_id,
|
||||
"from": from_agent,
|
||||
"subject": subject,
|
||||
"content": content,
|
||||
"type": "request",
|
||||
"status": "unread",
|
||||
"timestamp": datetime.datetime.now().isoformat(),
|
||||
}
|
||||
|
||||
inbox["messages"].append(message)
|
||||
inbox_path.write_text(json.dumps(inbox, indent=2))
|
||||
|
||||
return {"ok": True, "message_id": msg_id}
|
||||
|
||||
|
||||
def get_agent_bus_messages(agent_id: str, limit: int = 50) -> dict:
|
||||
"""API: GET /api/agents/{id}/bus-messages — raw inbox messages."""
|
||||
if agent_id not in TIER2_AGENTS and agent_id != "rose":
|
||||
return {"error": f"Unknown agent: {agent_id}"}
|
||||
if agent_id == "rose":
|
||||
inbox_path = _HERMES_DIR / "inbox.json"
|
||||
else:
|
||||
inbox_path = _AGENTS_DIR / agent_id / "inbox.json"
|
||||
if not inbox_path.exists():
|
||||
return {"agent_id": agent_id, "messages": []}
|
||||
try:
|
||||
inbox = json.loads(inbox_path.read_text())
|
||||
messages = inbox.get("messages", [])[-limit:]
|
||||
return {"agent_id": agent_id, "messages": messages}
|
||||
except Exception:
|
||||
return {"agent_id": agent_id, "messages": []}
|
||||
|
||||
|
||||
# ── Message Bus Viewer ────────────────────────────────────────────────────────
|
||||
|
||||
def get_message_bus_status() -> dict:
|
||||
"""
|
||||
Return all inboxes across all agents — a complete view of the message bus.
|
||||
Returns dict with each agent and their messages.
|
||||
"""
|
||||
all_agents = list(TIER2_AGENTS.keys()) + ["rose"]
|
||||
bus = {}
|
||||
for agent_id in all_agents:
|
||||
try:
|
||||
messages = _read_inbox(agent_id, limit=100)
|
||||
bus[agent_id] = {
|
||||
"count": len(messages),
|
||||
"messages": messages[-10:] if messages else [], # last 10
|
||||
}
|
||||
except Exception:
|
||||
bus[agent_id] = {"count": 0, "messages": [], "error": True}
|
||||
return {"bus": bus}
|
||||
|
||||
|
||||
def send_bus_message(to_agent: str, from_agent: str, subject: str, content: str, msg_type: str = "request") -> dict:
|
||||
"""
|
||||
Write a message directly to an agent's inbox.json.
|
||||
"""
|
||||
if to_agent not in TIER2_AGENTS and to_agent != "rose":
|
||||
return {"ok": False, "error": f"Unknown agent: {to_agent}"}
|
||||
|
||||
if to_agent == "rose":
|
||||
inbox_path = _HERMES_DIR / "inbox.json"
|
||||
else:
|
||||
inbox_path = _AGENTS_DIR / to_agent / "inbox.json"
|
||||
|
||||
try:
|
||||
data = []
|
||||
if inbox_path.exists():
|
||||
data = json.loads(inbox_path.read_text())
|
||||
if not isinstance(data, list):
|
||||
data = []
|
||||
|
||||
import uuid, datetime
|
||||
msg = {
|
||||
"id": str(uuid.uuid4())[:8],
|
||||
"from": from_agent,
|
||||
"to": to_agent,
|
||||
"type": msg_type,
|
||||
"subject": subject,
|
||||
"content": content,
|
||||
"timestamp": datetime.datetime.now().isoformat() + "Z",
|
||||
"status": "unread",
|
||||
}
|
||||
data.append(msg)
|
||||
inbox_path.write_text(json.dumps(data, indent=2))
|
||||
return {"ok": True, "message_id": msg["id"]}
|
||||
except Exception as e:
|
||||
return {"ok": False, "error": str(e)}
|
||||
|
||||
|
||||
# ── Memory Search (ChromaDB) ───────────────────────────────────────────────────
|
||||
|
||||
def _get_chroma_client():
|
||||
"""Get or create the shared ChromaDB HTTP client (thread-safe singleton)."""
|
||||
if not hasattr(_get_chroma_client, "_client"):
|
||||
_get_chroma_client._client = chromadb.HttpClient(host="127.0.0.1", port=8000)
|
||||
return _get_chroma_client._client
|
||||
|
||||
|
||||
def _search_agent_memory(agent_id: str, query: str, limit: int = 10) -> list:
|
||||
"""
|
||||
Search memory for a specific agent.
|
||||
Searches the rose_memory collection filtered by topic matching agent_id.
|
||||
"""
|
||||
try:
|
||||
client = _get_chroma_client()
|
||||
coll = client.get_collection(name="rose_memory")
|
||||
results = coll.query(
|
||||
query_texts=[query],
|
||||
n_results=limit,
|
||||
include=["metadatas", "documents"],
|
||||
)
|
||||
matches = []
|
||||
for i, doc in enumerate(results.get("documents", [[]])[0] or []):
|
||||
meta = (results.get("metadatas", [[{}]])[0] or [{}])[i] or {}
|
||||
topic = meta.get("topic", "")
|
||||
if not topic.startswith(agent_id):
|
||||
continue
|
||||
matches.append({
|
||||
"id": (results.get("ids", [["?"]])[0] or ["?"])[i],
|
||||
"topic": topic,
|
||||
"content": doc,
|
||||
"confidence": float(meta.get("confidence", 0.0)),
|
||||
"tags": meta.get("tags", ""),
|
||||
"vault_path": meta.get("vault_path", ""),
|
||||
})
|
||||
return matches
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _search_all_agents_memory(query: str, limit: int = 20) -> list:
|
||||
"""
|
||||
Search across all agent memories in ChromaDB.
|
||||
Returns matches with agent attribution from topic.
|
||||
Topic format: "agent-name/fact-name" or flat topic name.
|
||||
"""
|
||||
try:
|
||||
client = _get_chroma_client()
|
||||
coll = client.get_collection(name="rose_memory")
|
||||
results = coll.query(
|
||||
query_texts=[query],
|
||||
n_results=limit,
|
||||
include=["metadatas", "documents"],
|
||||
)
|
||||
matches = []
|
||||
for i, doc in enumerate(results.get("documents", [[]])[0] or []):
|
||||
meta = (results.get("metadatas", [[{}]])[0] or [{}])[i] or {}
|
||||
topic = meta.get("topic", "")
|
||||
parts = topic.split("/")
|
||||
agent = parts[0] if len(parts) > 1 else topic
|
||||
matches.append({
|
||||
"id": (results.get("ids", [["?"]])[0] or ["?"])[i],
|
||||
"topic": topic,
|
||||
"agent": agent,
|
||||
"content": doc,
|
||||
"confidence": float(meta.get("confidence", 0.0)),
|
||||
"tags": meta.get("tags", ""),
|
||||
"vault_path": meta.get("vault_path", ""),
|
||||
})
|
||||
return matches
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
Reference in New Issue
Block a user