signoz 7小时前 · 2026-09-16 19:30:29 · 2 阅读
基于OpenTelemetry和SigNoz的Meta Muse代码监控
Meta Muse Code 是 Meta 推出的终端编程智能体，由 Muse Spark 模型系列驱动。它直接在 shell 里运行，可以读写文件、执行命令、生成子智能体，也就是说，一句提示词可能触发十几次模型调用和工具执行，你才能看到回复。
这正是没有可观测性时难以理解它行为的原因：成本主要来自上下文回放而非用户输入了多少内容，延迟被用户看不见的推理过程占据，而工具失败的表现只是智能体悄然变慢。
本指南将介绍如何用 OpenTelemetry 把 Muse Code 的遥测数据导出到 SigNoz，从而观测会话、轮次、模型调用、token 与提示缓存使用情况，以及工具活动。
前提条件
- 一个带有有效接入密钥的 SigNoz Cloud 账户，或自行部署的 SigNoz 实例
- 已安装并登录 Muse Code，可运行
muse --version确认 - Python 3.9 或更高版本（macOS 和大多数 Linux 发行版都已内置）
用 OpenTelemetry 监控 Meta Muse Code
Muse Code 自带 OpenTelemetry 导出器，但无法发送 SigNoz Cloud 要求的 signoz-ingestion-key 请求头，因此本指南改用 hook 系统。Muse Code 在生命周期的每个关键节点都会执行一个 hook 命令，并以 JSON 格式通过 stdin 把事件传给它。下面的脚本把这些事件转换成 OpenTelemetry span，再通过 OTLP/HTTP 发送到 SigNoz。无需任何 SDK，因为 SigNoz 直接支持 OTLP/HTTP JSON。
第一步：在 ~/.local/share/muse-otel/muse_otel_hook.py 创建 hook 脚本。
#!/usr/bin/env python3
"""将 Meta Muse Code hook 事件导出到 SigNoz，作为 OpenTelemetry spans。"""
import hashlib, json, os, pathlib, sys, time, urllib.request
STATE = pathlib.Path(os.path.expanduser("~/.local/state/muse-otel"))
CFG_PATHS = [pathlib.Path(__file__).resolve().parent / "config.json",
pathlib.Path(os.path.expanduser("~/.config/muse-otel/config.json"))]
def config():
for p in CFG_PATHS:
try:
return json.loads(p.read_text())
except Exception:
continue
return {}
CFG = config()
ENDPOINT = CFG.get("endpoint", "https://ingest.<region>.signoz.cloud:443")
KEY = CFG.get("ingestion_key")
SERVICE = CFG.get("service_name", "muse-code")
def h16(*parts):
return hashlib.sha256("|".join(str(p) for p in parts).encode()).hexdigest()[:16]
def attrs(d):
out = []
for k, v in d.items():
if v is None:
continue
if isinstance(v, bool):
val = {"boolValue": v}
elif isinstance(v, int):
val = {"intValue": str(v)}
elif isinstance(v, (list, tuple)):
val = {"arrayValue": {"values": [{"stringValue": str(x)} for x in v]}}
else:
val = {"stringValue": str(v)}
out.append({"key": k, "value": val})
return out
def post(spans, ev):
if not (KEY and spans):
return
payload = {"resourceSpans": [{
"resource": {"attributes": attrs({"service.name": SERVICE, "surface": "tui"})},
"scopeSpans": [{"scope": {"name": "muse-otel-hook"}, "spans": spans}]}]}
if os.fork() != 0: # 立即返回，永不阻塞 agent
return
os.setsid()
if os.fork() != 0:
os._exit(0)
try:
req = urllib.request.Request(
ENDPOINT.rstrip("/") + "/v1/traces", data=json.dumps(payload).encode(),
headers={"content-type": "application/json", "signoz-ingestion-key": KEY})
urllib.request.urlopen(req, timeout=10).read()
except Exception:
pass
os._exit(0)
def mark(sess, key):
d = STATE / str(sess)
d.mkdir(parents=True, exist_ok=True)
(d / key).write_text(json.dumps({"t": time.time_ns()}))
def take(sess, key):
p = STATE / str(sess) / key
try:
v = json.loads(p.read_text())
p.unlink(missing_ok=True)
return v
except Exception:
return None
def span(name, tid, sid, parent, start, end, a, kind=1, err=False):
s = {"traceId": tid, "spanId": sid, "name": name, "kind": kind,
"startTimeUnixNano": str(int(start)), "endTimeUnixNano": str(int(end)),
"attributes": attrs(a), "status": {"code": 2 if err else 1}}
if parent:
s["parentSpanId"] = parent
return s
def main():
ev = json.loads(sys.stdin.read())
name, sess, turn = ev.get("hook_event_name"), ev.get("session_id"), ev.get("turn_id")
tid = (str(turn).replace("-", "") if turn else hashlib.sha256(
str(sess).encode()).hexdigest()[:32])
root, now = h16("turn", turn or sess), time.time_ns()
tkey = "turn_" + h16(turn or sess)
base = {"session.id": sess, "turn.id": turn, "muse.model": ev.get("model")}
have_root = (STATE / str(sess) / tkey).exists()
if name == "UserPromptSubmit":
mark(sess, tkey)
elif name == "PreLLMCall":
mark(sess, "llm_" + h16(ev.get("request_id"), ev.get("attempt")))
elif name == "PreToolUse":
mark(sess, "tool_" + h16(ev.get("tool_use_id")))
elif name == "PostLLMCall":
st = take(sess, "llm_" + h16(ev.get("request_id"), ev.get("attempt")))
u, status = ev.get("usage") or {}, str(ev.get("status") or "")
sid = h16("llm", ev.get("request_id"))
tp = (ev.get("options") or {}).get("meta.traceparent")
if isinstance(tp, str) and tp.count("-") == 3: # 复用 Muse 自身的 id
_, tp_trace, tp_span, _ = tp.split("-")
tid, sid = tp_trace, tp_span
fr = [ev["finish_reason"]] if ev.get("finish_reason") else (
["stop"] if status == "success" else None)
post([span("chat " + str(ev.get("model")), tid, sid,
root if have_root else None, (st or {}).get("t", now - 1), now,
{**base, "gen_ai.operation.name": "chat",
"gen_ai.request.model": ev.get("model"),
"gen_ai.provider.name": ev.get("model_provider"),
"gen_ai.response.id": ev.get("response_id"),
"gen_ai.response.finish_reasons": fr,
"gen_ai.usage.input_tokens": u.get("input_tokens"),
"gen_ai.usage.output_tokens": u.get("output_tokens"),
"gen_ai.usage.cache_read.input_tokens": u.get("cache_read_tokens"),
"gen_ai.usage.reasoning.output_tokens": u.get("reasoning_tokens"),
"gen_ai.request.reasoning_effort":
(ev.get("options") or {}).get("meta.reasoning.effort"),
"muse.llm.status": status, "muse.llm.attempt": ev.get("attempt")},
kind=3, err=bool(ev.get("error")))], ev)
elif name in ("PostToolUse", "PostToolUseFailure"):
st = take(sess, "tool_" + h16(ev.get("tool_use_id")))
failed = name == "PostToolUseFailure"
post([span("execute_tool " + str(ev.get("tool_name")), tid,
h16("tool", ev.get("tool_use_id")), root if have_root else None,
(st or {}).get("t", now - 1), now,
{**base, "gen_ai.tool.name": ev.get("tool_name"),
"gen_ai.tool.call.id": ev.get("tool_use_id"),
"muse.tool.status": "failed" if failed else "success"},
err=failed)], ev)
elif name in ("Stop", "StopFailure"):
st = take(sess, tkey)
post([span("turn", tid, root, None, (st or {}).get("t", now - 1), now,
{**base, "muse.turn.outcome":
"failed" if name == "StopFailure" else "completed"},
err=name == "StopFailure")], ev)
elif name == "SessionStart":
mark(sess, "session")
elif name == "SessionEnd":
st = take(sess, "session")
post([span("muse session", hashlib.sha256(
("session:" + str(sess)).encode()).hexdigest()[:32], h16("sess", sess),
None, (st or {}).get("t", now - 1), now,
{**base, "muse.session.end_reason": ev.get("reason")})], ev)
print("{}")
if __name__ == "__main__":
try:
main()
except Exception:
print("{}") # 遥测故障绝不应阻塞 agent
sys.exit(0)赋予执行权限：
Copychmod +x ~/.local/share/muse-otel/muse_otel_hook.py第 2 步：在脚本旁创建 ~/.local/share/muse-otel/config.json 配置文件。
{
"endpoint": "https://ingest.<region>.signoz.cloud:443",
"ingestion_key": "<your-ingestion-key>",
"service_name": "muse-code"
}chmod 600 ~/.local/share/muse-otel/config.json核对以下字段值：
<region>：你的 SigNoz Cloud 区域。<your-ingestion-key>：你的 SigNoz Ingestion Key（摄入密钥）。service_name：该 Agent 在 SigNoz 中显示的名称。如果想对比不同团队或仓库的数据，请为各自设置不同的值。
Muse Code 在启动 Hooks 时会使用一个经过清洗的环境。只有 HOME、PATH、PWD、SHELL、TERM、USER 等少数变量会保留下来，因此你在 Shell 中导出的任何变量对 Hook 都不可见。managed_hooks_env_vars 设置也无济于事，因为它仅适用于企业托管的 Hooks。请将 Endpoint 和 Key 保存在 config.json 中。
第 3 步：在 ~/.config/muse/settings.json 中注册 Hooks。
{
"schema_version": 1,
"hooks": {
"SessionStart": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"UserPromptSubmit": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"PreLLMCall": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"PostLLMCall": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"PreToolUse": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"PostToolUse": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"PostToolUseFailure": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"Stop": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }],
"SessionEnd": [{ "hooks": [{ "type": "command", "command": "~/.local/share/muse-otel/muse_otel_hook.py" }] }]
}
}Hooks 必须配置在 settings.json 里，项目级的 .muse/hooks.json 会被直接忽略。
第 4 步：启动 Muse Code 并执行一条 prompt。
Copymuse每轮对话都会产生一个 turn span，其下包含 chat {model} 和 execute_tool {tool} 子 span。由于脚本复用了 Muse Code 在每次模型调用时附加的 meta.traceparent，trace id 会与 agent 自身的 turn id 保持一致，你的 span 也就自然对齐到它的内部 trace 上下文。数据出现会有几秒钟延迟，稍等即可。
大部分步骤都一样。只需把 endpoint
原始来源： signoz