基于轨迹的自我改进
从成功和失败的操作序列中学习，以改进未来行为
第 3 / 4 课13 个步骤
基于轨迹的自我改进 是 CoddyKit 上的免费 AI Agents 课时。 这是第 3 节课，共 4 节。 你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。 这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。
什么是轨迹？
轨迹是智能体针对特定任务从开始到结束所经历的完整状态与行动序列。它记录的不仅是最终答案，还记录了智能体如何得到这个答案：调用了哪些工具、调用顺序如何、使用了什么参数，以及观察到了哪些中间结果。
记录轨迹
请用记录器包装智能体的每个行动，捕获行动前后的状态。状态包括：当前目标、记忆内容和近期观察结果。行动包括：工具名称、参数和结果。
from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
@dataclass
class TrajectoryStep:
step_index: int
state_summary: str # short description of world state
action_name: str # tool or reasoning step name
action_params: dict
result: Any
timestamp: str = ''
def __post_init__(self):
if not self.timestamp:
self.timestamp = datetime.utcnow().isoformat()
@dataclass
class Trajectory:
trajectory_id: str
task: str
steps: list = field(default_factory=list)
outcome: str = 'unknown' # 'success', 'failure', 'partial'
final_score: float = 0.0
def add_step(self, step: TrajectoryStep):
self.steps.append(step)
def mark_success(self, score: float = 1.0):
self.outcome = 'success'
self.final_score = score
def mark_failure(self, reason: str = ''):
self.outcome = 'failure'
self.final_score = 0.0
if __name__ == '__main__':
traj = Trajectory(trajectory_id='t-1', task='Book a flight to Tokyo')
traj.add_step(TrajectoryStep(
step_index=0, state_summary='searching flights',
action_name='search_flights', action_params={'dest': 'NRT'}, result='5 options found'
))
traj.mark_success(score=0.95)
print(f'Trajectory {traj.trajectory_id}: outcome={traj.outcome}, score={traj.final_score}')
print('Steps recorded:', len(traj.steps))
存储轨迹
轨迹可能很大。请将每条轨迹分别存储为压缩的 JSON 文件。按结果和任务类型为轨迹建立索引，以便快速检索。成功轨迹可作为少样本示例；失败轨迹可作为训练信号。
import json
import os
from dataclasses import asdict
TRAJECTORY_DIR = 'trajectories'
def save_trajectory(traj: Trajectory):
os.makedirs(TRAJECTORY_DIR, exist_ok=True)
filename = f'{TRAJECTORY_DIR}/{traj.trajectory_id}_{traj.outcome}.json'
data = asdict(traj)
with open(filename, 'w') as f:
json.dump(data, f, indent=2)
print(f'Saved trajectory: {filename}')
def load_successful_trajectories(task_type: str, n: int = 5) -> list:
results = []
for fname in os.listdir(TRAJECTORY_DIR):
if '_success.json' not in fname:
continue
with open(os.path.join(TRAJECTORY_DIR, fname)) as f:
traj = json.load(f)
if task_type.lower() in traj['task'].lower():
results.append(traj)
results.sort(key=lambda t: t['final_score'], reverse=True)
return results[:n]将成功轨迹用作少样本示例
成功轨迹就是一个完整示例：智能体可以先观察示例，再学习其中的步骤序列，然后尝试新的类似任务。请将得分最高的轨迹作为少样本前缀注入系统提示词。
def trajectory_to_few_shot(traj: dict) -> str:
lines = [f'Example task: {traj["task"]}', 'Steps taken:']
for step in traj['steps']:
lines.append(
f' [{step["step_index"]}] {step["action_name"]}'
f'({json.dumps(step["action_params"])}) -> {str(step["result"])[:80]}'
)
lines.append(f'Outcome: {traj["outcome"]} (score={traj["final_score"]:.2f})')
return '\n'.join(lines)
def build_few_shot_system_prompt(task_type: str) -> str:
successful = load_successful_trajectories(task_type, n=2)
if not successful:
return 'Complete the following task step by step.'
examples = '\n\n---\n\n'.join(
trajectory_to_few_shot(t) for t in successful
)
return (
'Here are examples of successfully completed similar tasks:\n\n'
+ examples
+ '\n\n---\n\nNow complete the new task using the same approach.'
)分析失败轨迹
失败轨迹同样有很高的价值。请分析它们，找出失败模式：哪一步出错了，以及为什么出错。常见失败模式包括：选择了错误的工具、工具选择正确但参数错误、臆造中间结果，以及循环未终止。
def analyze_failure(traj: dict, client) -> dict:
steps_str = json.dumps(traj['steps'], indent=2)
prompt = (
f'Task: {traj["task"]}\n\n'
f'Agent trajectory (failed):\n{steps_str}\n\n'
'Identify the failure mode. Return JSON:\n'
'{"failure_step": 0, "failure_mode": "", "root_cause": "", '
'"prevention": ""}'
)
import anthropic
client_obj = anthropic.Anthropic(api_key='YOUR_API_KEY')
result = client_obj.messages.create(
model='claude-opus-4-5',
max_tokens=512,
messages=[{'role': 'user', 'content': prompt}]
)
import json
return json.loads(result.content[0].text)失败模式分类体系
根据失败轨迹构建失败模式分类体系，有助于发现其中的规律。当足够多的失败具有相同的根本原因时，这说明应该修复智能体的工具、提示词或逻辑，而不只是修复某一次运行。
from collections import Counter
def build_failure_taxonomy(failed_trajectories: list, client) -> dict:
failure_modes = []
for traj in failed_trajectories:
analysis = analyze_failure(traj, client)
failure_modes.append(analysis['failure_mode'])
counts = Counter(failure_modes)
total = len(failure_modes)
taxonomy = [
{
'failure_mode': mode,
'count': count,
'percentage': round(count / total * 100, 1)
}
for mode, count in counts.most_common()
]
print('Failure Mode Taxonomy:')
for entry in taxonomy:
print(f' {entry["failure_mode"]}: {entry["count"]} ({entry["percentage"]}%)')
return {'taxonomy': taxonomy, 'total_failures': total}从轨迹创建训练对
进行监督式微调时，您需要（输入、理想输出）对。失败轨迹会提供错误输出；失败分析会说明本应发生什么。两者结合后就形成了一个训练对。
def trajectory_to_training_pair(
failed_traj: dict,
failure_analysis: dict
) -> dict:
"""
Creates an SFT-ready training pair:
input = task + context at failure step
output = what the agent should have done
"""
fail_step_idx = failure_analysis['failure_step']
steps = failed_traj['steps']
# Context up to (but not including) the failure step
context_steps = steps[:fail_step_idx]
context_str = '\n'.join(
f'Step {s["step_index"]}: {s["action_name"]}({s["action_params"]})'
for s in context_steps
)
return {
'input': f'Task: {failed_traj["task"]}\n\nPrevious steps:\n{context_str}\n\nNext action:',
'output': failure_analysis['prevention'], # correct action
'source': 'failure_trajectory',
'trajectory_id': failed_traj.get('trajectory_id', 'unknown')
}
if __name__ == '__main__':
failed_traj = {
'task': 'Cancel subscription',
'trajectory_id': 'traj-7',
'steps': [
{'step_index': 0, 'action_name': 'find_account', 'action_params': {'user': 'u1'}},
{'step_index': 1, 'action_name': 'delete_account', 'action_params': {'user': 'u1'}},
]
}
failure_analysis = {'failure_step': 1, 'prevention': 'call cancel_subscription(user="u1") instead'}
pair = trajectory_to_training_pair(failed_traj, failure_analysis)
print('Training input:')
print(pair['input'])
print('Expected output:', pair['output'])
根据轨迹进行微调
当您拥有足够多的高质量训练对后（针对范围较窄的任务通常需要 50–500 对），就可以微调较小的模型，使其内化成功模式。OpenAI 的微调接口接受采用 messages 格式的 JSONL 文件。
import json
def export_fine_tuning_jsonl(
training_pairs: list,
output_file: str,
system_prompt: str = 'You are an efficient AI agent.'
):
with open(output_file, 'w') as f:
for pair in training_pairs:
record = {
'messages': [
{'role': 'system', 'content': system_prompt},
{'role': 'user', 'content': pair['input']},
{'role': 'assistant', 'content': pair['output']}
]
}
f.write(json.dumps(record) + '\n')
print(f'Exported {len(training_pairs)} training pairs to {output_file}')
# Upload via OpenAI API (pseudocode):
# client.files.create(file=open('train.jsonl','rb'), purpose='fine-tune')
# client.fine_tuning.jobs.create(training_file='file-id', model='gpt-4o-mini')
if __name__ == '__main__':
import tempfile, os
pairs = [{'input': 'Task: Cancel subscription\n\nNext action:', 'output': 'cancel_subscription(user="u1")'}]
out_path = os.path.join(tempfile.gettempdir(), 'demo_training.jsonl')
export_fine_tuning_jsonl(pairs, out_path)
筛选高质量轨迹
并非所有成功轨迹的质量都相同。重试 15 次后才成功的轨迹，比第一次尝试就成功的轨迹包含更多噪声。请按效率进行筛选：以最少步骤成功、最终得分较高，并且没有臆造的中间结果。
def filter_high_quality_trajectories(
trajectories: list,
max_steps: int = 8,
min_score: float = 0.85
) -> list:
high_quality = []
for traj in trajectories:
if traj['outcome'] != 'success':
continue
if traj['final_score'] < min_score:
continue
if len(traj['steps']) > max_steps:
continue
high_quality.append(traj)
# Sort by (score DESC, steps ASC)
high_quality.sort(
key=lambda t: (-t['final_score'], len(t['steps']))
)
print(f'High-quality trajectories: {len(high_quality)} / {len(trajectories)}')
return high_quality
if __name__ == '__main__':
trajectories = [
{'outcome': 'success', 'final_score': 0.92, 'steps': [1, 2, 3]},
{'outcome': 'failure', 'final_score': 0.10, 'steps': [1]},
{'outcome': 'success', 'final_score': 0.60, 'steps': [1, 2]},
]
filter_high_quality_trajectories(trajectories)
通过比较轨迹获得洞察
比较同一任务类型下的一条成功轨迹和一条失败轨迹，可以准确发现两者从哪里开始出现分歧。分歧点是改进智能体决策能力最有效的位置。
def compare_trajectories(success_traj: dict, failure_traj: dict) -> dict:
s_steps = {s['step_index']: s for s in success_traj['steps']}
f_steps = {s['step_index']: s for s in failure_traj['steps']}
divergences = []
for idx in sorted(set(s_steps) & set(f_steps)):
s_action = s_steps[idx]['action_name']
f_action = f_steps[idx]['action_name']
if s_action != f_action:
divergences.append({
'step': idx,
'success_action': s_action,
'failure_action': f_action
})
break # First divergence is most important
return {
'first_divergence': divergences[0] if divergences else None,
'success_steps': len(s_steps),
'failure_steps': len(f_steps)
}
# Usage:
# comparison = compare_trajectories(good_traj, bad_traj)
# print('First divergence:', comparison['first_divergence'])
if __name__ == '__main__':
good_traj = {'steps': [{'step_index': 0, 'action_name': 'search'}, {'step_index': 1, 'action_name': 'summarize'}]}
bad_traj = {'steps': [{'step_index': 0, 'action_name': 'search'}, {'step_index': 1, 'action_name': 'delete'}]}
comparison = compare_trajectories(good_traj, bad_traj)
print('First divergence:', comparison['first_divergence'])
基于轨迹的改进飞轮
完整的飞轮流程是：运行智能体 → 记录轨迹 → 评估结果 → 存入轨迹库 → 分析失败 → 创建训练对 → 微调或更新提示词 → 运行改进后的智能体 → 记录新轨迹。每个循环都会改进智能体。
class TrajectorySelfImprovement:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.trajectory_lib = []
def run_and_record(self, task: str, agent_fn) -> Trajectory:
traj = Trajectory(
trajectory_id=f'{self.agent_id}_{len(self.trajectory_lib)}',
task=task
)
result = agent_fn(task, traj) # agent_fn appends steps to traj
# Evaluate result (e.g., via reflection or user rating)
score = evaluate_result(result)
if score >= 0.7:
traj.mark_success(score)
else:
traj.mark_failure('Low quality score')
save_trajectory(traj)
self.trajectory_lib.append(traj)
return traj
def improvement_cycle(self, client):
failed = [t for t in self.trajectory_lib if t.outcome == 'failure']
if len(failed) >= 10:
taxonomy = build_failure_taxonomy([vars(t) for t in failed], client)
print('Improvement cycle complete:', taxonomy)
def evaluate_result(result: str) -> float:
return 0.8 # placeholder知识检查
分析失败轨迹的主要目的是什么？
回顾：基于轨迹的自我改进
做得很好！本课的要点如下：
- 轨迹：从开始到结束的一系列（状态、行动、结果）步骤
- 成功轨迹：在类似任务前注入的少样本示例
- 失败轨迹：用于分析失败模式 → 创建训练对 → 进行微调
- 质量筛选：优先选择步骤少且得分高的成功轨迹
- 飞轮：运行 → 记录 → 分析 → 微调 → 运行改进后的智能体
下一步：了解自我改进可能出现的问题——奖励投机、分布偏移和安全防护机制。
免费开始
用 AI 导师学习 AI Agents — 免费
在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。
- 课程
- 60
- 课程
- 239
常见问题解答
「基于轨迹的自我改进」课时是免费的吗？
是的 — 「基于轨迹的自我改进」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 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 反馈 — 无需本地设置。