性能调优从定位瓶颈开始

你已经避开了八个常见的反模式，Skills 变得内聚、安全、易维护。但当 10 个 Skill 并发执行时，一个 API 调用慢 300 毫秒就足以拖垮整条用户链路。前一章帮你避免了返工，而这一章要帮你守住响应体验的下线：从瓶颈定位开始，系统性地优化吞吐量和延迟。

在本章中，你将扮演一位负责对话型 Agent 的工程师——系统上线两周，用户反馈“思考时间”越来越长。运营团队发来了平均响应时间趋势图：从 1.8 秒涨到 3.3 秒。我们需要用 Profiling 工具把每个 Skill 的耗时拆解出来，找到根因，然后通过缓存、批处理和模型微调让延迟回落。

读完这一章，你将能够：

• 搭建可落地的 Skill 级耗时追踪方案
• 区分模型推理瓶颈与工具调用瓶颈，并选择不同的优化方向
• 在质量可接受的前提下，将端到端延迟优化 40% 以上

你需要什么

最终成果

我们将得到一个端到端延迟从 3.3 秒降低到 1.9 秒 的对话 Agent，并且会生成一份如下图所示的时间分布火焰图（本地可用 Python matplotlib 绘制），让每一次用户请求的耗时一目了然。

为什么要做这个？ 高性能不是锦上添花，而是体验的底线。一个响应卡顿的 Agent 会让用户放弃继续对话，同时也会增加模型 API 的 Token 账单。定位瓶颈是优化的第一步，也是最重要的一步。

步骤说明

第一步：搭建 Skill 级耗时追踪

我们需要从现有 Agent 的调用链中，提取出每个 Skill 的执行耗时。如果还没有细粒度的日志，不要慌。截至当前调研资料，Anthropic 的 API 响应头中提供了丰富的计时字段，社区常用的 Profiling 方案有三种：

1. API 响应头解析：利用 Claude API 每次请求返回的 x-process-time-ms、x-token-usage-* 等字段，快速区分“模型思考”与“网络传输”。
2. 装饰器埋点：在 Python 代码中，为 Skill 的执行函数加上轻量级计时装饰器，将结果写入结构化日志。
3. OpenTelemetry Span：如果组织内已有分布式追踪系统，直接创建 Span，将 Skill 名作为属性，自动生成甘特图。

我们采用第二种方案，因为它零依赖且适合快速验证。为你的 Skill 入口函数添加如下包裹：

import time
import functools
import logging

logger = logging.getLogger("skill_profiler")

def trace_skill(skill_name: str):
    """记录 Skill 执行耗时的装饰器"""
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            start = time.perf_counter()
            try:
                result = await func(*args, **kwargs)
                return result
            finally:
                elapsed_ms = (time.perf_counter() - start) * 1000
                logger.info(f"skill_trace|{skill_name}|{elapsed_ms:.2f}")
        return wrapper
    return decorator

# 在 Skill 定义中使用
@trace_skill("customer_lookup")
async def customer_lookup_skill(context):
    # 原有逻辑
    pass

然后配置一个简单的聚合脚本，定时解析这些日志，按 Skill 名称统计 P50/P95 延迟。

预期结果：启用装饰器并触发 100 次对话后，你能看到类似下面的输出：

skill_trace|customer_lookup|245.12
skill_trace|summarize_conversation|1234.87
skill_trace|fetch_order_status|410.56

现在，我们可以进入真正的分析阶段了。

⚠️ 踩坑经验：
不要在生产环境强依赖 logger.info 的 I/O 写入。在高并发下，直接写磁盘可能成为新的瓶颈。建议使用异步日志队列（如 aiologger）或将数据发送到内存中的计数器，定期批刷。如果你使用的是 OpenTelemetry，则 SDK 会自动处理缓冲。

第二步：识别瓶颈——模型推理 vs 工具调用

拿到了延迟分布，下一步就是分类。你的每个 Skill 都包含两部分工作：

• 模型推理时间：Claude 理解上下文、生成工具调用决策、输出最终文本所消耗的时间，主要受 Prompt 长度、模型大小和输出 Token 数量影响。
• 工具调用时间：等待外部 API（CRM、数据库、Rest API）返回结果的时间，包括网络往返和远程执行时间。

区分二者，才能对症下药。我们用一个简单的统计来判定：观察每个 Skill 的“Skill 总耗时”与它内部包含的模型 API 调用的 x-process-time-ms 之间的差值。

假设你的 summarize_conversation Skill 包含一次 Claude API 调用，你可以这样记录：

# 伪代码：在调用 Claude 的封装函数中
resp = requests.post(...)
model_time = float(resp.headers.get("x-process-time-ms", 0))
total_skill_time = ...  # 来自装饰器
io_time = total_skill_time - model_time

当聚合 100 次请求后，你可能发现：

结论一目了然：summarize_conversation 是模型推理瓶颈，另外两个则是 I/O 密集的。一次成功的性能调优，一定是先按这种 “锁” 把瓶颈定位到具体类型，再选择策略。

⚠️ 踩坑经验：
有些 Skill 可能包含多步循环（例如“重试直到成功”），这样外部调用耗时会被放大，容易误判。请在聚合时记录“重试次数”维度，单独分析首次调用的平均延迟。

预期结果：你将得到一个清晰的两分类瓶颈地图，能够明确说出 “优化目标是降低模型推理时间” 还是 “并行化外部调用”。

第三步：缓存与批处理策略

现在我们根据瓶颈类型，实施第一个优化套餐。对于 I/O 密集型的 Skill，主要矛盾是“等待”。解决思路有两个：减少不必要的请求（缓存）和把多次请求合并（批处理）。

缓存模型响应（针对模型推理瓶颈）

对于 summarize_conversation 这类输入高度重复的场景（比如用户在短时间内多次要求“总结我们刚才说的”），可以直接缓存最终的摘要结果。使用 Redis 或内存 LRU 缓存，Key 为输入文本内容 SHA256 哈希。

import hashlib
from cachetools import LRUCache, cached

cache = LRUCache(maxsize=100)

def hash_prompt(prompt: str) -> str:
    return hashlib.sha256(prompt.encode()).hexdigest()

@cached(cache, key=lambda content: hash_prompt(content))
async def summarize_text(content: str) -> str:
    # 调用 Claude 生成摘要
    pass

但注意：缓存不仅可用于模型推理，更常用于外部 API 响应。比如 customer_lookup 每次用同样的 ID 查客户信息，在 TTL 60 秒内完全可以复用。

批处理多个 API 请求（针对工具调用瓶颈）

当一个 Skill 需要查询多个订单的状态时，典型的错误写法是串行：

# 错误示例
for order_id in order_ids:
    status = await fetch_status(order_id)

这会让延迟线性累加。正确的做法是使用 asyncio.gather 并发请求，或者将能批量查询的 API 一次性调用。

# 正确：并发请求
statuses = await asyncio.gather(*[fetch_status(oid) for oid in order_ids])

如果后端 API 支持批量接口，务必优先使用：

# 更优：一个请求返回所有结果
statuses = await fetch_status_batch(order_ids)

预期结果：实施缓存后，重复的 customer_lookup 调用延迟降至 1 ms 以下；并发改造后，原本需要 800 ms 的 4 个顺序查询，现在只需 200 ms 左右。

第四步：模型选择与参数微调

前一步优化了工具调用，现在回到模型推理瓶颈。summarize_conversation 消耗 1.1 秒，这是否必须？我们从两个维度压缩：

1. 输入与输出 Token 数：检查 Prompt 是否因携带了过多历史对话而膨胀。你可以通过限制携带的最近消息轮次（例如只保留最近 5 轮），或要求 Claude 生成更短的总结。
2. 模型切换：如果你的对话 Agent 对速度要求高于绝对推理深度，完全可以将 summarize_conversation 使用的模型从 claude-3-opus 切换为 claude-3-haiku，后者延迟通常降低 60%~70%，而总结质量在多数场景下仍然足够。
3. 温度与 max_tokens：降低 temperature 值（例如 0 到 0.3）能让输出更确定，有时会加速推理；调低 max_tokens 避免模型生成不必要的长篇大论，既减延迟又省 Token。

实际调整参数如下：

# 优化后配置
response = claude.messages.create(
    model="claude-3-haiku-20240307",
    max_tokens=150,   # 原来是 500
    temperature=0.1,
    messages=history[-5:]  # 只带最近 5 轮
)

踩坑提醒：模型切换前，务必对 Skill 的输出质量做小批量人工评估，或者设置自动对比测试。我曾见过一次将客服意图分类从 Opus 降到 Haiku，准确率从 98% 掉到 91%，这在某些场景不可接受。这里的原则是：先在离线环境用 50 个样本测试，确定质量损失在容忍范围内。

预期结果：summarize_conversation 的模型推理延迟从 1.1 s 降低到 350 ms，整体端到端延迟显著下降。

回顾

本章我们完成了从“症状”到“根因”到“疗效验证”的完整链路：

1. 搭建 Skill 级耗时追踪，用装饰器 + 日志获得延迟分布。
2. 识别瓶颈类型，明确模型推理 vs. 工具调用，锁定两个不同方向的优化对象。
3. 实施缓存与批处理，消除冗余等待，让 I/O 密集型 Skill 获得数量级提升。
4. 模型微调，通过减少上下文、切换轻量模型、限制 Token 数，将推理延迟压缩到可接受水平。

现在，你的 Agent 在相同功能下，端到端延迟已经回到 2 秒以内，运营团队不再需要为性能报表发愁。

行动清单（下一步你可以做的）：

• [ ] 为当前线上的 3 个核心 Skill 挂上延迟装饰器，收集 24 小时数据。
• [ ] 绘制 P50/P95/P99 延迟曲线，识别长尾请求。
• [ ] 对 P95 超过 2 秒的 Skill，分析其模型调用 vs 外部调用比例。
• [ ] 为只读性质的外部查询添加 30 秒 TTL 缓存。
• [ ] 针对总结类 Skill，实验 haiku 模型并对比 50 个样本的质量。

性能优化从来不是一次性动作，而是需要持续监测的工程项。当你下次调整 Prompt 或新增 Skill 时，记得回头看一眼火焰图，确保没有引入新的“慢站点”。

下一章，我们将进入对话型 Skills 的另一个核心课题——幻觉缓解是对话型 Skills 的永恒课题。在毫秒级响应和低延迟背后，如果 Agent 提供了错误信息，一切优化都将失去意义。我们将从提示、检索和多模型校验三个层面，探讨如何降低事实错误，让你的 Agent 不仅快，而且可信。