DeepSeek Harness 系列（09）：可观测性——怎么知道 Agent 在干什么

本文目录

两种 channel 分别做什么
严重程度如何判定

先说一个让人头疼的场景 -----------

你的 Agent 跑了三分钟，最后报了一个错误。

然后你盯着日志——什么都没有。你不知道它调用了哪些工具、哪一步卡住了、Token 到底花在哪里，更不知道为什么失败。

这就是可观测性缺失的典型后果。

为什么可观测性对 Agent 尤其重要

传统服务出了问题，你可以加断点、看堆栈。Agent 不一样。

Agent 是非确定性的。 相同的输入，模型可能产生完全不同的工具调用序列。你没办法在"模型决策"这一步加断点——那是一个黑盒。

调试只能靠日志反推。 模型的"想法"只体现在它输出的文字和工具调用里。你需要把这些全部记下来，事后才能重建它的推理链。

Token 费用不透明。 一个多轮 Agent 对话，到底哪一步最贵？是第三轮工具结果太长？还是 system prompt 占了大头？没有计量，你连优化方向都找不到。

生产环境出问题，你得有证据。 用户说"它给了我一个错误答案"，你需要还原当时的完整执行链——用了哪些工具、返回了什么、模型看到了什么。

Session 日志：最完整的观测数据

回顾第 05 篇讲过的内容：dsh Session 是一份仅追加的类型化事件日志。

这个设计不只是为了持久化，它本身就是最完整的观测数据源。

每个 Session 事件都包含：

每个 SessionEvent 的结构：
  - <span>type</span>：事件类型（如 <span>'tool/call'</span>、<span>'turn/end'</span>、<span>'assistant/message'</span>）
  - <span>seq</span>：单调递增的序号（从 0 开始）
  - time：时间戳（毫秒级 Unix 时间戳）
  - data：类型化的事件数据（不同 <span>type</span> 有不同 data 结构）

这意味着：只要你能读取 Session 日志，你就能重建整个执行过程——每一步做了什么、花了多少时间、有没有出错。

实时监听：session/event

不需要等日志写完再分析。dsh 提供了 session/event 事件，让你实时监听 Session 的每一条记录：

<span>// 监听某个 Session 的所有事件（实时）</span>
ctx.<span>on</span>(<span>'session/event'</span>, <span>(<span>session, event</span>) =></span> {
  <span>// 每条事件都会触发这个回调</span>
  <span>console</span>.<span>log</span>(<span>`[<span>${event.<span>type</span>}</span>] seq=<span>${event.seq}</span> time=<span>${event.time}</span>`</span>)
  
  <span>// 检查是否是工具调用</span>
  <span>if</span> (event.<span>type</span> === <span>'tool/call'</span>) {
    <span>// event.data.name 是工具名</span>
    <span>// event.data.arguments 是原始 JSON 字符串（模型输出的，未解析）</span>
    <span>// event.data.callId 是这次调用的唯一 ID</span>
    <span>console</span>.<span>log</span>(<span>`  Tool: <span>${event.data.name}</span>`</span>)
    <span>console</span>.<span>log</span>(<span>`  Args: <span>${event.data.<span>arguments</span>}</span>`</span>)
  }
  
  <span>// 检查是否是工具执行结果</span>
  <span>if</span> (event.<span>type</span> === <span>'tool/result'</span>) {
    <span>// 通过检查 content block 里有没有 isError: true 判断是否失败</span>
    <span>const</span> isError = event.<span>data</span>.<span>message</span>.<span>content</span>.<span>some</span>(
      <span><span>b</span> =></span> b.<span>type</span> === <span>'tool_result'</span> && b.<span>isError</span>
    )
    <span>console</span>.<span>log</span>(<span>`  Result: <span>${isError ? <span>'ERROR'</span> : <span>'OK'</span>}</span>`</span>)
  }
})

这个监听器在开发调试时非常有用——你能看到 Agent 在实时做什么，不用等它跑完。

Token 计量：ctx.tokenMeter

知道"发生了什么"只是第一步。知道"花了多少钱"同样重要。

dsh 提供了 ctx.tokenMeter，可以测量当前 Session 的 Token 压力：

<span>// TokenMeasurement 接口（来自 packages/llm/token-meter/src/types.ts）</span>
<span>interface</span> <span>TokenMeasurement</span> {
  <span>// 这次计量消费了多少事件（用于缓存，避免重复计算）</span>
  <span>readonly</span> <span>logRevision</span>: <span>SessionLogOffset</span>
  
  <span>// 当前请求的总 token 压力（输入 + 输出之和）</span>
  <span>readonly</span> <span>totalTokens</span>: <span>number</span>
  
  <span>// 当前 surface（模型可见的历史消息）的 token 数量</span>
  <span>readonly</span> <span>surfaceTokens</span>: <span>number</span>
  
  <span>// surface 相对于最后一次成功请求的 token 变化量（有符号，可以是负数）</span>
  <span>readonly</span> <span>surfaceDeltaTokens</span>: <span>number</span>
  
  <span>// 按位置排列的 surface 节点及其 token 数（可以看每条消息占多少）</span>
  <span>readonly</span> <span>nodes</span>: <span>readonly</span> <span>TokenSurfaceNode</span>[]
}

几个关键概念：

surface tokens：模型这次请求实际看到的历史内容有多少 token。这决定了你的 API 费用中"输入 token"这部分。
total tokens：输入 + 输出的总和，反映这次请求的完整费用。
surfaceDeltaTokens：和上一次请求相比，surface 增加了多少。如果这个数字持续增大，说明上下文在膨胀，可能需要压缩策略。

使用示例：

<span>// 在每个 Turn 结束时打印 Token 使用摘要</span>
ctx.<span>on</span>(<span>'session/event'</span>, <span>(<span>session, event</span>) =></span> {
  <span>// 只关心 Turn 结束事件</span>
  <span>if</span> (event.<span>type</span> !== <span>'turn/end'</span>) <span>return</span>
  
  <span>// 调用 measure 获取当前 Session 的 Token 测量结果</span>
  <span>const</span> measurement = ctx.<span>tokenMeter</span>.<span>measure</span>(session)
  
  <span>console</span>.<span>log</span>(<span>`Turn <span>${event.data.turn}</span> ended:`</span>)
  <span>console</span>.<span>log</span>(<span>`  Surface tokens: <span>${measurement.surfaceTokens}</span>`</span>)
  <span>console</span>.<span>log</span>(<span>`  Total tokens:   <span>${measurement.totalTokens}</span>`</span>)
  
  <span>// 显示 delta，正数表示上下文在增长</span>
  <span>const</span> delta = measurement.<span>surfaceDeltaTokens</span>
  <span>const</span> sign = delta > <span>0</span> ? <span>'+'</span> : <span>''</span>
  <span>console</span>.<span>log</span>(<span>`  Delta:          <span>${sign}</span><span>${delta}</span>`</span>)
  
  <span>// 如果上下文增长过快，发出警告</span>
  <span>if</span> (delta > <span>2000</span>) {
    <span>console</span>.<span>warn</span>(<span>'  ⚠ Context growing fast, consider compression'</span>)
  }
})

遥测 Seam：ctx.sessionTelemetry

session/event 监听适合开发调试，但生产环境你需要把数据发到外部系统——比如 Grafana、Datadog、CloudWatch。

dsh 为此设计了一个"遥测 Seam"：ctx.sessionTelemetry。

Seam（接缝）这个词用得很精准——它是一个标准化的接口，让你把遥测数据接入任意后端，同时 harness 本身不依赖任何具体的监控系统。

每条遥测记录的结构：

<span>// SessionTelemetryRecord（来自 packages/session/session-telemetry/src）</span>
<span>interface</span> <span>SessionTelemetryRecord</span> {
  <span>// 两种 channel：</span>
  <span>//   'ledger'：Session 日志事件的完整镜像，和事件一一对应</span>
  <span>//   'ops'：运营信号，只有特殊情况才产生</span>
  <span>channel</span>: <span>'ledger'</span> | <span>'ops'</span>
  
  <span>// 时间戳（毫秒）</span>
  <span>time</span>: <span>number</span>
  
  <span>// 严重程度</span>
  <span>severity</span>: <span>'info'</span> | <span>'warn'</span> | <span>'error'</span>
  
  <span>// 标识属性（用于查询和过滤）</span>
  <span>// 例如：session.id、event.type、event.seq 等</span>
  <span>attributes</span>: <span>Record</span><<span>string</span>, <span>string</span> | <span>number</span>>
  
  <span>// 完整 payload：event.data 的深拷贝</span>
  <span>body</span>: <span>unknown</span>
}

两种 channel 分别做什么

ledger channel：Session 日志的完整镜像。每一条 Session 事件都会产生一条对应的 ledger 记录。这是审计和回放的数据来源。

包括：

每个 assistant/message（含完整流数据）
每个 tool/call 和 tool/result
失败的 assistant/attempt（模型尝试了但最终没用）
所有 turn/start、turn/end 等生命周期事件

ops channel：运营信号，只有两种：

agent-error：Agent 在 Turn 之外失败了（比如初始化报错）
shutdown：Agent 正常关闭

严重程度如何判定

error：工具结果 isError: true、turn/end 带错误原因、agent-error 运营事件
其他情况：info

这个映射让你可以在监控系统里直接过滤 severity === 'error' 来看所有异常，不用自己写判断逻辑。

OpenTelemetry 接入

dsh 提供官方的 OTel Provider 插件：dsh-session-telemetry-otel。

接入方式（概念性）：

<span>// 在你的 Bundle 配置里加入这个插件（伪代码）</span>
<span>// 这会把 ctx.sessionTelemetry 接到 OTel 后端</span>
<span>'@deepseek-ai/dsh-session-telemetry-otel'</span>

<span>// 该插件内部会：</span>
<span>// 1. 注册 ctx.sessionTelemetry 的 OTel 后端实现</span>
<span>// 2. 每条 SessionTelemetryRecord 通过 OTel JS SDK 的 Logger API 发送</span>
<span>// 3. 支持配置不同的 Exporter（OTLP、Console、File 等）</span>

几个设计原则值得了解：

边界公理：harness 只负责调用 emit()，批处理、重试、排队这些属于 OTel SDK 的职责，harness 不插手。这样两边都可以独立演化。

尽力而为：遥测记录可能重复也可能丢失。接收端应该基于 (session.id, format_version, event.seq) 组合来去重 ledger 记录，而不是假设每条记录恰好到达一次。

flush 是可选的：每次 Turn 结束后可以调用 flush()，但 OTel 后端默认不实现（避免并发冲突）。如果你需要强一致性，需要自行配置。

实战：写一个简单的调试插件

把上面的内容整合成一个完整的调试观测插件：

<span>// debug-observer.ts — 调试用的可观测性插件</span>
<span>// 用法：在开发时加入 Bundle，生产时替换为真正的遥测后端</span>

<span>export</span> <span>const</span> name = <span>'debug-observer'</span>

<span>// 声明依赖注入 tokenMeter</span>
<span>export</span> <span>const</span> inject = [<span>'tokenMeter'</span>]

<span>export</span> <span>function</span> <span>apply</span>(<span>ctx: Context</span>): <span>void</span> {
  <span>// ── 1. 监听工具调用 ────────────────────────────────────────</span>
  ctx.<span>on</span>(<span>'session/event'</span>, <span>(<span>session, event</span>) =></span> {
    <span>if</span> (event.<span>type</span> !== <span>'tool/call'</span>) <span>return</span>
    
    <span>console</span>.<span>log</span>(<span>`[Tool Call] <span>${event.data.name}</span>`</span>)
    <span>console</span>.<span>log</span>(<span>`  Call ID: <span>${event.data.callId}</span>`</span>)
    <span>// arguments 是原始 JSON 字符串（模型直接输出的，还没有被解析）</span>
    <span>console</span>.<span>log</span>(<span>`  Args: <span>${event.data.<span>arguments</span>}</span>`</span>)
  })
  
  <span>// ── 2. 监听工具执行结果 ────────────────────────────────────</span>
  ctx.<span>on</span>(<span>'session/event'</span>, <span>(<span>session, event</span>) =></span> {
    <span>if</span> (event.<span>type</span> !== <span>'tool/result'</span>) <span>return</span>
    
    <span>const</span> blocks = event.<span>data</span>.<span>message</span>.<span>content</span>
    <span>const</span> isError = blocks.<span>some</span>(<span><span>b</span> =></span> b.<span>type</span> === <span>'tool_result'</span> && b.<span>isError</span>)
    <span>const</span> icon = isError ? <span>'✗'</span> : <span>'✓'</span>
    
    <span>// 从第一个 block 里拿到对应的 toolUseId（关联 tool/call 事件）</span>
    <span>const</span> toolUseId = blocks[<span>0</span>]?.<span>toolUseId</span> ?? <span>'unknown'</span>
    <span>console</span>.<span>log</span>(<span>`[Tool Result] <span>${icon}</span> (call: <span>${toolUseId}</span>)`</span>)
  })
  
  <span>// ── 3. 每个 Turn 结束时打印 Token 摘要 ────────────────────</span>
  ctx.<span>on</span>(<span>'session/event'</span>, <span>(<span>session, event</span>) =></span> {
    <span>if</span> (event.<span>type</span> !== <span>'turn/end'</span>) <span>return</span>
    
    <span>const</span> reason = event.<span>data</span>.<span>reason</span>.<span>kind</span>  <span>// 'complete' | 'error' | 'interrupted' 等</span>
    <span>const</span> measurement = ctx.<span>tokenMeter</span>.<span>measure</span>(session)
    
    <span>console</span>.<span>log</span>(<span>`\n[Turn <span>${event.data.turn}</span>] ended: <span>${reason}</span>`</span>)
    <span>console</span>.<span>log</span>(<span>`  Surface: <span>${measurement.surfaceTokens}</span> tokens`</span>)
    <span>console</span>.<span>log</span>(<span>`  Total:   <span>${measurement.totalTokens}</span> tokens`</span>)
    
    <span>const</span> delta = measurement.<span>surfaceDeltaTokens</span>
    <span>const</span> sign = delta > <span>0</span> ? <span>'+'</span> : <span>''</span>
    <span>console</span>.<span>log</span>(<span>`  Delta:   <span>${sign}</span><span>${delta}</span>`</span>)
    
    <span>// 如果是错误结束，打印具体的错误信息</span>
    <span>if</span> (reason === <span>'error'</span>) {
      <span>console</span>.<span>error</span>(<span>`  Error: <span>${<span>JSON</span>.stringify(event.data.reason)}</span>`</span>)
    }
  })
  
  <span>// ── 4. 监听 Session 生命周期 ───────────────────────────────</span>
  ctx.<span>on</span>(<span>'session/created'</span>, <span>(<span>session</span>) =></span> {
    <span>console</span>.<span>log</span>(<span>`\n[Session] created: <span>${session.id}</span>`</span>)
  })
  
  ctx.<span>on</span>(<span>'session/disposed'</span>, <span>(<span>session</span>) =></span> {
    <span>console</span>.<span>log</span>(<span>`[Session] disposed: <span>${session.id}</span>`</span>)
  })
}

这个插件在开发时可以快速加入 Bundle，看到完整的运行轨迹。生产环境则换成 dsh-session-telemetry-otel 插件，数据流向监控系统。

调试技巧：读 JSONL 日志文件

dsh 默认把 Session 日志持久化为 JSONL 文件（每行一个 JSON 对象，即一条 SessionEvent）。

以下是一些常用的命令行分析技巧：

<span># 查看所有工具调用（提取工具名列表）</span>
<span>cat</span> session.jsonl | grep <span>'"type":"tool/call"'</span> | jq <span>'.data.name'</span>

<span># 查看失败的助手尝试（模型生成了但最终没用到的内容）</span>
<span>cat</span> session.jsonl | grep <span>'"type":"assistant/attempt"'</span> | jq <span>'.'</span>

<span># 统计每轮的 token 用量（从 assistant/message 里的 usage 字段）</span>
<span>cat</span> session.jsonl | grep <span>'"type":"assistant/message"'</span> | jq <span>'.data.usage'</span>

<span># 查看所有 Turn 的结束原因（是正常完成还是出错）</span>
<span>cat</span> session.jsonl | grep <span>'"type":"turn/end"'</span> | jq <span>'.data.reason.kind'</span>

<span># 检查有没有工具执行失败</span>
<span>cat</span> session.jsonl | grep <span>'"type":"tool/result"'</span> | jq <span>'select(.data.message.content[].isError == true)'</span>

这些命令假设你有 jq 工具。如果是 Windows 环境，可以用 PowerShell 的 ConvertFrom-Json 做类似的分析。

可观测性层次总结

四个层次，覆盖从开发到生产：

实时观测（开发调试）
  └─ session/event 监听器 → 每条事件即时打印到控制台

审计与回放（事后分析）
  └─ JSONL 日志文件 → 完整重建执行链，配合 jq 分析

Token 用量分析
  └─ ctx.tokenMeter.measure(session) → 每个 surface 节点的 token 计量
<span>     → 找到上下文膨胀的罪魁祸首
</span>
生产监控（系统级）
  └─ ctx.sessionTelemetry + OTel 插件 → 接入 Grafana / Datadog / CloudWatch
<span>     → 告警、看板、错误追踪全部打通
</span>

小结

可观测性不是"有了更好"的附加项，对 Agent 来说它是调试的唯一手段。

dsh 在设计上就考虑到了这一点：Session 本身是事件日志，事件日志天然就是审计数据；ctx.tokenMeter 让 Token 消耗不再是黑盒；ctx.sessionTelemetry 提供标准化接缝，让你自由选择后端。

核心模式很简单：Session 事件监听 → JSONL 持久化 → 遥测 Seam → OTel 后端。你用哪一层取决于你的场景，但这几层可以同时运行，互不干扰。

下一篇是系列的最后一篇，我们会把前面学到的所有机制整合起来，完整地写一个生产级插件——从工具注册、Session 管理、错误处理，到可观测性，一起落地。

在 PrimeSkills 可以找到已在真实企业场景验证过的 AI Agent 技能和工作流，不是演示级的，是用在实际项目里的。

更多内容见我的个人主页