识别缓慢且昂贵的步骤
瀑布式分析：智能体的时间和预算都花在哪里？
识别缓慢且昂贵的步骤 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
代理性能分析
代理性能问题分为两类：缓慢的步骤（高延迟）和成本高的步骤（令牌成本高）。两者都会损害用户体验并增加运营成本。第一步是进行测量。
记录每个步骤的耗时
使用 time.perf_counter() 进行高精度计时。它测量包括输入/输出等待在内的挂钟时间——这正是代理步骤延迟所关注的内容。
import time
from contextlib import contextmanager
@contextmanager
def timer(step_name: str, timings: dict):
start = time.perf_counter()
try:
yield
finally:
end = time.perf_counter()
duration_ms = (end - start) * 1000
timings[step_name] = duration_ms
print(f'{step_name}: {duration_ms:.1f}ms')
# Usage
timings = {}
with timer('entity_extraction', timings):
time.sleep(0.05) # Simulate work
with timer('vector_search', timings):
time.sleep(0.12) # Simulate work
with timer('llm_call', timings):
time.sleep(0.80) # Simulate LLM latency
print('\nTimings:', timings)
print('Slowest step:', max(timings, key=timings.get))构建步骤分析器
步骤分析器会包装每个代理步骤，并记录耗时和成本。运行完成后，它会生成一张展示步骤持续时间的瀑布图。
import time
from dataclasses import dataclass, field
from typing import List, Optional
@dataclass
class StepProfile:
name: str
start_ms: float
end_ms: float
duration_ms: float
prompt_tokens: int = 0
completion_tokens: int = 0
cost_usd: float = 0.0
error: Optional[str] = None
class StepProfiler:
def __init__(self):
self.steps: List[StepProfile] = []
self.run_start = time.perf_counter()
def start_step(self, name: str) -> float:
return time.perf_counter()
def end_step(self, name: str, start_time: float, tokens: dict = None, error: str = None):
end = time.perf_counter()
run_elapsed = (start_time - self.run_start) * 1000
duration = (end - start_time) * 1000
profile = StepProfile(
name=name,
start_ms=run_elapsed,
end_ms=run_elapsed + duration,
duration_ms=duration,
error=error
)
if tokens:
profile.prompt_tokens = tokens.get('prompt', 0)
profile.completion_tokens = tokens.get('completion', 0)
self.steps.append(profile)
return profile
profiler = StepProfiler()
t = profiler.start_step('entity_extraction')
time.sleep(0.05)
profiler.end_step('entity_extraction', t, {'prompt': 200, 'completion': 50})
print('Step recorded:', profiler.steps[0].duration_ms)终端中的瀑布图
打印一张简单的 ASCII 步骤计时瀑布图。它以可视化方式展示各步骤何时运行以及耗时多久——类似于浏览器的网络瀑布图，但用于代理。
class Step:
def __init__(self, name, start_ms, end_ms, error=False):
self.name, self.start_ms, self.end_ms, self.error = name, start_ms, end_ms, error
@property
def duration_ms(self):
return self.end_ms - self.start_ms
class StepProfiler:
def __init__(self, steps):
self.steps = steps
def print_waterfall(profiler):
if not profiler.steps:
print('No steps recorded')
return
total_ms = max(s.end_ms for s in profiler.steps)
bar_width = 50
print('\n=== Agent Step Waterfall ===')
for step in profiler.steps:
start_pos = int(step.start_ms / total_ms * bar_width)
end_pos = int(step.end_ms / total_ms * bar_width)
bar = ' ' * start_pos + '#' * max(1, end_pos - start_pos) + ' ' * (bar_width - end_pos)
status = 'ERR' if step.error else ' '
print(f'{status} {step.name:<20} {step.duration_ms:>6.0f}ms |{bar}|')
print(f'\nTotal run: {total_ms:.0f}ms')
profiler = StepProfiler([Step('plan', 0, 120), Step('search', 120, 480), Step('generate', 480, 900, error=True)])
print_waterfall(profiler)收集各次运行的 P95 延迟
单次测量并不足够。在多次运行中收集计时数据，并计算每个步骤的 P50、P95 和 P99 延迟。P95 延迟是 95% 的运行能够在该延迟以内完成的阈值。
import statistics
from collections import defaultdict
class LatencyCollector:
def __init__(self):
self.step_durations = defaultdict(list)
def record(self, step_name: str, duration_ms: float):
self.step_durations[step_name].append(duration_ms)
def percentile(self, data: list, pct: float) -> float:
sorted_data = sorted(data)
index = int(len(sorted_data) * pct / 100)
return sorted_data[min(index, len(sorted_data) - 1)]
def report(self):
print('=== Latency Report (ms) ===')
print(f'{"Step":<30} {"Count":>6} {"P50":>8} {"P95":>8} {"P99":>8} {"Max":>8}')
print('-' * 75)
for step_name, durations in sorted(self.step_durations.items()):
p50 = self.percentile(durations, 50)
p95 = self.percentile(durations, 95)
p99 = self.percentile(durations, 99)
max_d = max(durations)
print(f'{step_name:<30} {len(durations):>6} {p50:>8.0f} {p95:>8.0f} {p99:>8.0f} {max_d:>8.0f}')
collector = LatencyCollector()
import random
for _ in range(100):
collector.record('vector_search', random.gauss(120, 30))
collector.record('llm_call', random.gauss(800, 150))
collector.report()识别 P95 延迟较高的步骤
收集指标后，找出 P95 延迟异常高的步骤。将优化精力集中在那里——缓存、并行化或模型降级都是常见方案。
class LatencyCollector:
def __init__(self, step_durations):
self.step_durations = step_durations
def percentile(self, durations, p):
s = sorted(durations)
k = int(len(s) * p / 100)
return s[min(k, len(s) - 1)]
def find_optimization_targets(collector, p95_threshold_ms=500):
targets = []
for step_name, durations in collector.step_durations.items():
p95 = collector.percentile(durations, 95)
avg = sum(durations) / len(durations)
p95_to_avg_ratio = p95 / avg if avg > 0 else 0
target = {
'step': step_name,
'p95_ms': round(p95),
'avg_ms': round(avg),
'p95_to_avg_ratio': round(p95_to_avg_ratio, 2),
'call_count': len(durations),
'needs_optimization': p95 > p95_threshold_ms
}
if target['needs_optimization']:
if p95_to_avg_ratio > 2.0:
target['suggestion'] = 'High variance: consider timeout and retry or caching'
else:
target['suggestion'] = 'Consistently slow: consider parallel execution or faster model'
targets.append(target)
targets.sort(key=lambda x: x['p95_ms'], reverse=True)
return targets
collector = LatencyCollector({'search': [100, 150, 900], 'generate': [400, 420, 410]})
targets = find_optimization_targets(collector, p95_threshold_ms=500)
for t in targets:
print(f"{t['step']}: P95={t['p95_ms']}ms, Avg={t['avg_ms']}ms, Action: {t.get('suggestion', 'OK')}")缓存高成本工具结果
对于成本高的步骤，最有效的优化是缓存。如果很可能再次进行相同的工具调用（输入相同），就缓存结果，下次立即返回。
import hashlib
import json
import time
from typing import Callable, Any
class ToolResultCache:
def __init__(self, ttl_seconds: int = 300):
self.cache = {}
self.ttl = ttl_seconds
def _make_key(self, tool_name: str, args: dict) -> str:
content = json.dumps({'tool': tool_name, 'args': args}, sort_keys=True)
return hashlib.sha256(content.encode()).hexdigest()[:16]
def get_or_compute(self, tool_name: str, args: dict, compute_fn: Callable) -> Any:
key = self._make_key(tool_name, args)
now = time.time()
if key in self.cache:
entry = self.cache[key]
if now - entry['ts'] < self.ttl:
print(f'Cache HIT for {tool_name}')
return entry['result']
print(f'Cache MISS for {tool_name} - computing...')
start = time.perf_counter()
result = compute_fn(**args)
elapsed = (time.perf_counter() - start) * 1000
print(f'{tool_name} computed in {elapsed:.0f}ms')
self.cache[key] = {'result': result, 'ts': now}
return result
cache = ToolResultCache(ttl_seconds=60)
def expensive_web_search(query: str) -> list:
time.sleep(0.2) # Simulate slow API call
return [f'Result for: {query}']
result1 = cache.get_or_compute('web_search', {'query': 'AI news'}, expensive_web_search)
result2 = cache.get_or_compute('web_search', {'query': 'AI news'}, expensive_web_search) # Cache hit识别令牌消耗大的步骤
识别令牌使用量明显偏高的步骤。大型提示词通常是因为包含了超出需要的上下文，或没有截断检索到的文档。
def find_token_heavy_steps(tracker: 'CostTracker', token_threshold: int = 2000) -> list:
heavy_steps = []
for step in tracker.steps:
total_tokens = step.prompt_tokens + step.completion_tokens
if total_tokens > token_threshold:
heavy_steps.append({
'step': step.step_name,
'total_tokens': total_tokens,
'prompt_tokens': step.prompt_tokens,
'completion_tokens': step.completion_tokens,
'cost_usd': step.cost_usd,
'suggestions': []
})
entry = heavy_steps[-1]
if step.prompt_tokens > token_threshold * 0.9:
entry['suggestions'].append('Prompt is very large: truncate context documents or summarize')
if step.completion_tokens > 1000:
entry['suggestions'].append('Large output: use max_tokens limit if full response not needed')
heavy_steps.sort(key=lambda x: x['total_tokens'], reverse=True)
return heavy_steps
tracker = CostTracker()
tracker.record('answer_generation', 'gpt-4o-mini', 4500, 1200)
tracker.record('entity_extraction', 'gpt-4o-mini', 200, 50)
heavy = find_token_heavy_steps(tracker)
for s in heavy:
print(f"{s['step']}: {s['total_tokens']} tokens, suggestions: {s['suggestions']}")模型降级分析
并非每个步骤都需要最强大的模型。请分析哪些步骤使用了高成本模型，以及更便宜的模型是否足够。简单的提取任务很少需要 GPT-4o。
MODEL_TIERS = {
'gpt-4o': {'tier': 'premium', 'capabilities': ['complex reasoning', 'nuanced writing']},
'gpt-4o-mini': {'tier': 'standard', 'capabilities': ['extraction', 'classification', 'summarization']},
'claude-3-haiku-20240307': {'tier': 'fast', 'capabilities': ['simple tasks', 'routing']}
}
STEP_MODEL_RECOMMENDATIONS = {
'entity_extraction': 'gpt-4o-mini',
'intent_classification': 'gpt-4o-mini',
'simple_summarization': 'gpt-4o-mini',
'complex_reasoning': 'gpt-4o',
'final_answer_generation': 'gpt-4o-mini'
}
def audit_model_usage(tracker: 'CostTracker') -> list:
recommendations = []
for step in tracker.steps:
recommended = STEP_MODEL_RECOMMENDATIONS.get(step.step_name)
if recommended and recommended != step.model:
current_cost = step.cost_usd
# Estimate cost with recommended model (rough calculation)
recommendations.append({
'step': step.step_name,
'current_model': step.model,
'recommended_model': recommended,
'potential_savings': 'significant' if step.model == 'gpt-4o' else 'moderate'
})
return recommendations
print('Model downgrade analysis function defined')在生产环境中进行分析
在生产环境中，不要记录每次运行，而应对分析数据进行抽样。完整记录 10%–20% 的运行，并汇总其余运行的指标。这样可以让存储需求和开销保持在可控范围内。
import random
class SampledProfiler:
def __init__(self, sample_rate: float = 0.1):
self.sample_rate = sample_rate
self.full_profiles = []
self.aggregate_timings = defaultdict(list)
def should_profile_full(self) -> bool:
return random.random() < self.sample_rate
def record_run(self, profiler: 'StepProfiler', full_profile: bool):
# Always record aggregate timing
for step in profiler.steps:
self.aggregate_timings[step.name].append(step.duration_ms)
# Only store full profiles for sampled runs
if full_profile:
self.full_profiles.append(profiler.steps)
def get_summary(self) -> dict:
return {
'full_profiles_stored': len(self.full_profiles),
'steps_tracked': {k: len(v) for k, v in self.aggregate_timings.items()}
}
sampled = SampledProfiler(sample_rate=0.1)
print('Sampled profiler: recording 10% of runs in full detail')
print('Summary:', sampled.get_summary())综合延迟和成本指标
最值得采取行动的优化目标是同时满足缓慢 AND 成本高的步骤。一个缓慢但成本低的步骤（使用小型提示词进行一次 LLM 调用）可能不值得优化。应重点关注两个维度都较高的步骤。
def combined_optimization_priority(profiler: 'StepProfiler', tracker: 'CostTracker') -> list:
# Build combined step data
cost_by_step = {s.step_name: s.cost_usd for s in tracker.steps}
combined = []
for step in profiler.steps:
cost = cost_by_step.get(step.name, 0)
# Priority score: normalize and combine
# High latency + High cost = top priority
priority = (step.duration_ms / 1000) + (cost * 1000) # Rough normalization
combined.append({
'step': step.name,
'duration_ms': round(step.duration_ms),
'cost_usd': round(cost, 6),
'priority_score': round(priority, 3)
})
combined.sort(key=lambda x: x['priority_score'], reverse=True)
return combined
print('Combined optimization priority function defined')
print('High priority = slow + expensive')知识检查：性能分析
请检验您对代理性能分析的理解。
性能分析总结
有效的代理性能分析包括：使用 time.perf_counter() 进行高精度计时；使用瀑布图可视化步骤重叠；在多次运行中收集 P95 延迟；识别令牌消耗大的步骤以进行截断；缓存工具结果以消除重复的高成本调用；分析模型降级以降低步骤执行成本；以及在生产环境中进行抽样分析，以将开销保持在可控范围内。
常见问题解答
「识别缓慢且昂贵的步骤」课时是免费的吗？
是的 — 「识别缓慢且昂贵的步骤」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。
「识别缓慢且昂贵的步骤」这节课中我会学到什么？
瀑布式分析：智能体的时间和预算都花在哪里？ 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。
学习 AI Agents 需要有经验吗？
无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。 这是第 3 节课，共 4 节。
「识别缓慢且昂贵的步骤」课时需要多长时间？
大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。
我能在这节 AI Agents 课中编写并运行代码吗？
能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。