---
id: ai.agent-ops.observability-cost
title: Agent 可观测性、成本和延迟：链路追踪、记忆管理、缓存和降级
type: lesson
domain: ai
stack: [agent-ops]
tags: [可观测性, 链路追踪, OpenTelemetry, LangSmith, Langfuse, 成本, 延迟, 记忆, 缓存]
level: advanced
prerequisites: [ai.agent-ops.eval-quality-gate]
est_hours: 3
related: [ai.llm-serving.index, ai.agent-ops.interview]
week: [13]
order: 2
---

# Agent 可观测性、成本和延迟：链路追踪、记忆管理、缓存和降级

## 解释说明

### 可观测性：一次 Agent 运行要能“回放”

一次请求 = 一条链路（trace），里面每一步是一个 span：模型调用（模型名、输入输出 token、耗时、提示词版本）、工具调用（参数、结果摘要、耗时、错误）、检索（命中的文档和分数）。有了它才能回答：“这个答案为什么错”“慢在哪一步”“这个用户花了多少钱”。

| 工具 | 特点（以官方文档为准） |
| --- | --- |
| LangSmith | LangChain 官方，和 LangChain / LangGraph 集成最好 |
| Langfuse | 开源可自部署，支持多框架 |
| OpenTelemetry | 通用标准，可接入已有的监控体系（Jaeger、Grafana 等），有生成式 AI 的语义约定 |

同时要有**指标**：请求量、成功率、P50 / P95 延迟、首 token 延迟、每次请求的 token 和成本、工具错误率、拒答率、转人工率、用户反馈（点赞 / 点踩）。

### 成本和延迟

| 手段 | 说明 |
| --- | --- |
| 模型分级 | 简单任务（分类、路由、改写）用小模型，复杂推理用大模型 |
| 控制上下文 | 历史消息摘要、只放相关的检索结果、工具结果只保留需要的字段 |
| 缓存 | 相同问题的答案缓存（注意权限维度）、厂商的提示词缓存（固定前缀放前面，以厂商文档为准） |
| 并行 | 互不依赖的工具调用、检索并发执行 |
| 流式输出 | 不减少总时间，但首字时间大幅缩短，体验更好 |
| 限制步数和预算 | 每个请求的最大步数、最大 token；超出时降级 |
| 降级 | 主模型超时 / 限流时切换备用模型（见 [模型侧专题](../llm-serving/index.md)） |

### 记忆管理

- **短期记忆**：当前会话的消息历史（LangGraph 中由检查点保存）。超过长度时：保留最近 N 轮 + 更早内容的摘要。
- **长期记忆**：跨会话的用户偏好和事实（LangGraph 的 Store、向量库），写入要有规则（哪些信息值得记、用户能否删除），读取时按相关性检索。
- 记忆也是上下文：越多越贵，也越容易干扰模型。

## 案例

运营报表 Agent 每次请求平均花费 0.3 元、P95 延迟 25 秒。链路追踪显示：60% 的 token 来自每次都把完整的表结构放进提示词，40% 的时间花在串行的 3 次查询上。改进：表结构按问题检索相关的表；查询并行；意图分类改用小模型。成本降到 0.1 元，P95 降到 9 秒，评测集通过率不变（门禁验证）。

## 代码示例

简易链路记录（[examples/eval-gate/agent_eval.py](examples/eval-gate/agent_eval.py) 中的 `Trace`），真实项目中换成 OpenTelemetry / LangSmith / Langfuse：

```python
@dataclass
class Trace:
    case_id: str
    trace_id: str = field(default_factory=lambda: uuid.uuid4().hex[:12])
    spans: list[dict] = field(default_factory=list)

    @contextmanager
    def span(self, name: str, **attrs):
        start = time.perf_counter()
        rec = {"name": name, **attrs}
        try:
            yield rec
        finally:
            rec["ms"] = round((time.perf_counter() - start) * 1000, 2)
            self.spans.append(rec)


with trace.span("llm.answer", model="small", tokens_in=650, tokens_out=60):
    ...
```

成本计算：`成本 = 输入 token / 1000 × 输入单价 + 输出 token / 1000 × 输出单价`，单价按实际使用的模型配置，不要写死在代码里。

## 易错点

- 只记录最终答案，不记录中间步骤，出了问题无法定位。
- 日志里记录完整的提示词和用户数据，没有脱敏和访问控制。
- 只看平均延迟，不看 P95 和首 token 延迟。
- 为了省成本换小模型，没有跑评测集验证效果。
- 答案缓存没有区分用户权限，泄露数据。
- 长期记忆无限增长，旧的错误信息一直影响回答。

## 面试常问

- 线上 Agent 出了错误答案，你怎么定位？
- 怎么降低 Agent 的成本和延迟？
- 短期记忆和长期记忆怎么设计？