> Agent Harness Patterns 第 11 关，一条讲解 AI 智能体工作原理的模式路线。网页版：https://harnesspatterns.dev/zh/patterns/observability · 全部模式（英文）：https://harnesspatterns.dev/llms.txt

# 可观测性与评估

轨迹（trace）展示智能体在一次运行中做了什么。评估（eval）在许多次运行上给它打分。两者你都需要：评估告诉你有东西坏了，轨迹告诉你为什么。

## 问题

智能体不是一个能从上读到下的函数。模型在运行时决定每一步，同一个提示词明天可能走出另一条路。当用户说“它订错了东西”时，你需要一步一步地知道它做了什么。而当你修改了提示词、某个工具或者模型时，你需要知道它是变好了还是变差了。

这就是 Blindspot，黑箱。它看起来能用，直到某天不能用了，这时谁也说不清发生了什么、花了多少钱、从什么时候开始的。手动试几个提示词，然后说一句“看着挺好”，Blindspot 就是这样被发布上线的。

## 解决方案

**可观测性**指的是把每次运行记录成一条**轨迹**：一棵由 **span** 组成的树，每一块工作一个 span，各自记录耗时和成本。整次运行是一个 span；每一轮、每次模型调用、每个工具也都是。出了问题的 span 会被标记为错误。本站的每个循环都已经在发出事件；tracer 只需要监听 EventBus 并给它们计时。把这些 span 发送到任何支持 OpenTelemetry（轨迹的开放标准）的工具里，你就能得到动画里的那条时间线。

**评估**（evals）就是智能体的测试。一个**数据集**装着若干用例：一个提示词，以及一次好的运行应该是什么样子。你用全新的智能体运行每个用例，再由**评分器**（scorer）给每次运行打分。通过用例所占的比例，就是你要盯住的数字。评分器有四种，好的评估会把它们混合使用：

- **答案**
   把最终文本和用例期望的结果对比：完全一致、必须包含某个词，或者符合某个模式。
   便宜且确定。它只看结果：一个走错了路却得出正确答案的运行照样能通过。
- **轨迹**
   把智能体按顺序调用的工具，和用例期望的工具对比。
   能抓住“路走错了、答案却对了”的情况，比如第 3 洞。太严格的话，会让走了另一条合理路线的运行不及格：在无害的地方允许多出几步。
- **你自己的检查**
   任何把一次运行映射成分数的函数：杆数不超过标准杆、token 预算、输出里的某个字段。
   你的产品在乎什么就检查什么。能保持确定性就尽量保持。
- **让模型当评委**
   另一个模型阅读问题、答案和一份评分标准，然后打分。
   适合没有唯一正确文本的答案：语气、有用程度、一段总结。更慢，要多花一次调用，还需要一份精确的评分标准，并时不时和人工打分做对照。

第 3 洞说明了为什么不能只用一个。答案说“4 杆进洞”，而 4 正是标准杆。只有轨迹评分器看到了：本该用铁杆的地方用了一号木。也只有轨迹展示了那里发生了什么：一个红色的 span，一颗掉进水里的球。

## 角色

还是那群熟悉的角色，这次换到了高尔夫球场上。

- **一个洞** (一个评估用例): 一个提示词，以及一次好的运行应该是什么样子：它的标准杆和该用的球杆。
- **球童** (模型): 研究这个洞，选一支球杆。它从不挥杆。
- **球杆** (工具): `drive`、`iron` 和 `putt`。挥杆的是 Astor。
- **飞行轨迹线** (轨迹): 每一次飞行都留在球场上，每一步都是时间线上的一个 span：模型调用是蓝色，工具是橙色，错误是红色。
- **裁判** (你的评估代码): 把每个用例交给一个全新的智能体，结束后在记分卡上盖章。
- **记分卡** (报告): 每个用例一行，每个评分器一项检查，底部是通过率。

## 代码

**使用 astorlm：**`attachTracer` 把智能体的事件转换成 span，可以交给任何导出器。`runEval` 为每个用例启动一个全新的智能体来运行整个数据集，应用你的评分器，并返回一份带有各评分器通过率的报告。两者都在 `astorlm/experimental` 里。

**从零手写：**第 2 关的循环，在每次模型调用和每个工具外面套一个计时器，再用一个 `for` 遍历用例，每个评分器一个函数。

**使用 astorlm**

```ts
import { OpenAIProvider, createLocalAgent } from 'astorlm'
import { attachTracer, createInMemoryExporter } from 'astorlm/experimental/tracing'
import { contains, runEval, toolTrajectory, type EvalCase, type Scorer } from 'astorlm/experimental/evals'

// Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy…
const LLM = { baseURL: 'http://localhost:11434/v1', apiKey: 'YOUR_API_KEY' } // local servers usually ignore the key

const newGolfer = () =>
  createLocalAgent({
    provider: new OpenAIProvider({ ...LLM, model: 'your-model' }), // e.g. 'llama3.1', 'gpt-4o-mini'
    tools: [drive, iron, putt], // your code: each swing moves the ball and says where it landed
    maxTurns: 10,
  })

// 1. SEE one run: a tracer turns the agent's events into spans (run → turn → model call / tool).
const golfer = await newGolfer()
const exporter = createInMemoryExporter() // in production: an OpenTelemetry exporter instead
attachTracer(golfer, { exporter })
await golfer.run('Hole 3: par 4, 330 m to the pin. Water crosses the fairway at 200 m.')
for (const span of exporter.spans) {
  console.log(span.name, span.endTime! - span.startTime, 'ms', span.status) // e.g. "tool drive 320 ms error"
}

// 2. GRADE many runs: a dataset of cases, and scorers that decide pass or fail.
const dataset: EvalCase[] = [
  { id: 'hole-1', input: 'Hole 1: par 3, 150 m to the pin. Play it out and report your score.', expected: { par: 3, clubs: ['iron', 'putt'] } },
  { id: 'hole-2', input: 'Hole 2: par 4, 360 m to the pin. Play it out and report your score.', expected: { par: 4, clubs: ['drive', 'iron', 'putt'] } },
  {
    id: 'hole-3',
    input: 'Hole 3: par 4, 330 m to the pin. Water crosses the fairway at 200 m. Play it out and report your score.',
    expected: { par: 4, clubs: ['iron', 'iron', 'putt'] }, // lay up short of the water
  },
]
type Expected = { par: number; clubs: string[] }

// Each case expects its own clubs, so wrap the built-in trajectory scorer.
const path: Scorer = {
  name: 'trajectory',
  score: (result) => toolTrajectory((result.case.expected as Expected).clubs, { mode: 'ordered-subset' }).score(result),
}
// Your own scorer: any function from a run to a 0..1 score.
const par: Scorer = {
  name: 'par',
  score: (result) => {
    const strokes = Number(/in (\d+)/.exec(result.output)?.[1] ?? Infinity)
    const passed = strokes <= (result.case.expected as Expected).par
    return { scorer: 'par', score: passed ? 1 : 0, passed, details: `${strokes} strokes` }
  },
}

const report = await runEval({
  dataset,
  createAgent: () => newGolfer(), // a fresh agent per case: no shared history
  scorers: [contains('Holed out'), path, par], // add llmJudge({ provider, rubric }) for fuzzy answers
})

console.log(report.summary) // { total: 3, passed: 2, passRate: 0.67, byScorer: { … } }
```

**TypeScript**

```ts
// Tracing and evals, from scratch. Plain fetch, no SDK.

// Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy…
const LLM = {
  baseURL: 'http://localhost:11434/v1', // e.g. Ollama's default address
  model: 'your-model', // e.g. 'llama3.1', 'gpt-4o-mini'
  apiKey: 'YOUR_API_KEY', // local servers usually ignore it
}

// 1. A span: something that took time, with what you want to know about it.
type Span = { name: string; ms: number; tokens?: number; error?: boolean }

type ToolFn = (args: Record<string, number>) => { output: string; isError?: boolean }
const tools: Record<string, ToolFn> = { drive, iron, putt } // your code: each swing moves the ball
const toolSchemas = [/* one JSON Schema per club: drive(meters), iron(meters), putt(meters) */]

type ToolCall = { id: string; function: { name: string; arguments: string } }
type Message =
  | { role: 'user'; content: string }
  | { role: 'assistant'; content: string | null; tool_calls?: ToolCall[] }
  | { role: 'tool'; tool_call_id: string; content: string }

// 2. The loop from level 2, timing every model call and every tool run.
async function runAgent(prompt: string, maxTurns = 10) {
  const spans: Span[] = []
  const toolCalls: string[] = []
  const messages: Message[] = [{ role: 'user', content: prompt }]
  for (let turn = 1; turn <= maxTurns; turn++) {
    const start = performance.now()
    const res = await fetch(`${LLM.baseURL}/chat/completions`, {
      method: 'POST',
      headers: { 'content-type': 'application/json', authorization: `Bearer ${LLM.apiKey}` },
      body: JSON.stringify({ model: LLM.model, messages, tools: toolSchemas }),
    })
    const body = await res.json()
    spans.push({ name: 'model', ms: performance.now() - start, tokens: body.usage?.total_tokens })
    const [choice] = body.choices
    const reply: Message = choice.message
    messages.push(reply)
    if (choice.finish_reason !== 'tool_calls') return { output: reply.content ?? '', spans, toolCalls }

    for (const call of reply.tool_calls ?? []) {
      const t0 = performance.now()
      const run = tools[call.function.name]
      const result = run ? run(JSON.parse(call.function.arguments)) : { output: `Unknown tool: ${call.function.name}`, isError: true }
      spans.push({ name: call.function.name, ms: performance.now() - t0, error: result.isError })
      toolCalls.push(call.function.name)
      messages.push({ role: 'tool', tool_call_id: call.id, content: result.output })
    }
  }
  throw new Error(`No answer after ${maxTurns} turns`)
}

// 3. The eval: cases with what a good run looks like, and one function per scorer.
const cases = [
  { input: 'Hole 1: par 3, 150 m to the pin. Play it out and report your score.', par: 3, clubs: ['iron', 'putt'] },
  { input: 'Hole 2: par 4, 360 m to the pin. Play it out and report your score.', par: 4, clubs: ['drive', 'iron', 'putt'] },
  { input: 'Hole 3: par 4, 330 m to the pin. Water crosses the fairway at 200 m. Play it out and report your score.', par: 4, clubs: ['iron', 'iron', 'putt'] },
]
type Case = (typeof cases)[number]
type Run = Awaited<ReturnType<typeof runAgent>>

// Every expected club shows up, in order (extra swings allowed).
const inOrder = (expected: string[], actual: string[]): boolean => {
  let i = 0
  for (const name of actual) if (name === expected[i]) i++
  return i === expected.length
}
const scorers: Record<string, (run: Run, c: Case) => boolean> = {
  holed: (run) => run.output.includes('Holed out'),
  trajectory: (run, c) => inOrder(c.clubs, run.toolCalls),
  par: (run, c) => Number(/in (\d+)/.exec(run.output)?.[1] ?? Infinity) <= c.par,
}

let passed = 0
for (const c of cases) {
  const run = await runAgent(c.input) // every case starts with an empty history
  const checks = Object.entries(scorers).map(([name, score]) => [name, score(run, c)] as const)
  if (checks.every(([, ok]) => ok)) passed++
  console.log(c.input.slice(0, 6), checks, run.spans) // failed a check? its spans say why
}
console.log(`passRate ${(passed / cases.length).toFixed(2)}`)
```

**Python**

```python
# Tracing and evals, from scratch. Standard library only, no SDK.
import json
import re
import time
import urllib.request

# Any OpenAI-compatible endpoint: OpenAI, Ollama, LM Studio, vLLM, a proxy...
LLM = {
    "base_url": "http://localhost:11434/v1",  # e.g. Ollama's default address
    "model": "your-model",  # e.g. "llama3.1", "gpt-4o-mini"
    "api_key": "YOUR_API_KEY",  # local servers usually ignore it
}

def post(path, payload):
    request = urllib.request.Request(
        f"{LLM['base_url']}{path}",
        data=json.dumps(payload).encode(),
        headers={"Content-Type": "application/json", "Authorization": f"Bearer {LLM['api_key']}"},
    )
    with urllib.request.urlopen(request) as response:
        return json.load(response)

TOOLS = {"drive": drive, "iron": iron, "putt": putt}  # your code: each returns (output, is_error)
TOOL_SCHEMAS = [...]  # one JSON Schema per club: drive(meters), iron(meters), putt(meters)

# 1 + 2. The loop from level 2, timing every model call and every tool run as a span.
def run_agent(prompt, max_turns=10):
    spans, tool_calls = [], []
    messages = [{"role": "user", "content": prompt}]

    for _ in range(max_turns):
        start = time.perf_counter()
        body = post("/chat/completions", {"model": LLM["model"], "messages": messages, "tools": TOOL_SCHEMAS})
        spans.append({"name": "model", "ms": (time.perf_counter() - start) * 1000, "tokens": body.get("usage", {}).get("total_tokens")})
        choice = body["choices"][0]
        reply = choice["message"]
        messages.append(reply)
        if choice["finish_reason"] != "tool_calls":
            return {"output": reply.get("content") or "", "spans": spans, "tool_calls": tool_calls}

        for call in reply.get("tool_calls", []):
            name = call["function"]["name"]
            t0 = time.perf_counter()
            output, is_error = TOOLS[name](**json.loads(call["function"]["arguments"]))
            spans.append({"name": name, "ms": (time.perf_counter() - t0) * 1000, "error": is_error})
            tool_calls.append(name)
            messages.append({"role": "tool", "tool_call_id": call["id"], "content": output})

    raise RuntimeError(f"No answer after {max_turns} turns")

# 3. The eval: cases with what a good run looks like, and one function per scorer.
CASES = [
    {"input": "Hole 1: par 3, 150 m to the pin. Play it out and report your score.", "par": 3, "clubs": ["iron", "putt"]},
    {"input": "Hole 2: par 4, 360 m to the pin. Play it out and report your score.", "par": 4, "clubs": ["drive", "iron", "putt"]},
    {
        "input": "Hole 3: par 4, 330 m to the pin. Water crosses the fairway at 200 m. Play it out and report your score.",
        "par": 4,
        "clubs": ["iron", "iron", "putt"],
    },
]

def in_order(expected, actual):
    """Every expected club shows up, in order (extra swings allowed)."""
    i = 0
    for name in actual:
        if i < len(expected) and name == expected[i]:
            i += 1
    return i == len(expected)

def strokes(output):
    match = re.search(r"in (\d+)", output)
    return int(match.group(1)) if match else float("inf")

SCORERS = {
    "holed": lambda run, case: "Holed out" in run["output"],
    "trajectory": lambda run, case: in_order(case["clubs"], run["tool_calls"]),
    "par": lambda run, case: strokes(run["output"]) <= case["par"],
}

passed = 0
for case in CASES:
    run = run_agent(case["input"])  # every case starts with an empty history
    checks = {name: score(run, case) for name, score in SCORERS.items()}
    passed += all(checks.values())
    print(case["input"][:6], checks, run["spans"])  # failed a check? its spans say why
print(f"passRate {passed / len(CASES):.2f}")
```

## 注意事项

- **每次改动都跑评估。**一个新的提示词、一段工具描述或者一个模型版本，可能修好一个用例，又弄坏两个。把数据集放在仓库里，在 CI 中运行，通过率一下降就让构建失败。
- **用真实的失败来扩充数据集。**用户报告的每个 bug 都变成一个用例，用它的轨迹作为证据。只有简单用例的数据集会永远通过，什么也证明不了。
- **运行结果会波动。**同一个用例今天通过，明天可能失败。重要的用例要多跑几次，看比率，而不是看单次结果。
- **轨迹里存着你用户的数据。**提示词、工具的输入和输出都会进到里面。该遮的要遮（用钩子，第 6 关），并把轨迹存储当作一个装有个人数据的数据库来对待。
- **成本也是一项指标。**记录每个 span 的 token。一个用两倍调用次数通过所有用例的智能体，是一次你的通过率显示不出来的退步。

## 相关模式

- [6 · 钩子](https://harnesspatterns.dev/zh/patterns/hooks.md)
- [5 · 循环中的错误](https://harnesspatterns.dev/zh/patterns/errors-in-the-loop.md)
- [3 · 设计一个工具](https://harnesspatterns.dev/zh/patterns/designing-a-tool.md)
- [10 · 每圈重新开始](https://harnesspatterns.dev/zh/patterns/fresh-laps.md)
- [12 · 先规划，再反思](https://harnesspatterns.dev/zh/patterns/plan-and-reflect.md)