軌跡ベースの自己改善
成功・失敗したアクション系列から学習し、将来の振る舞いを改善します。
「軌跡ベースの自己改善」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
トラジェクトリとは
トラジェクトリとは、あるタスクについて、エージェントが開始から完了までに取った状態とアクションの完全な последователь sequenceです。最終的な回答だけでなく、エージェントがどのように答えにたどり着いたかも記録します。つまり、どのツールをどの順番で、どのパラメーターとともに呼び出したか、そしてどのような中間結果を得たかを記録します。
トラジェクトリの記録
各エージェントのアクションを、前後の状態を取得するレコーダーでラップします。状態には、現在の目標、メモリの内容、直近の観測結果が含まれます。アクションには、ツール名、パラメーター、結果が含まれます。
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))
トラジェクトリの保存
トラジェクトリは大きくなる場合があります。トラジェクトリごとに1つの圧縮JSONファイルとして保存します。すばやく取得できるよう、結果とタスクの種類を基準にインデックスを作成します。成功したトラジェクトリはfew-shot例になり、失敗したトラジェクトリは学習シグナルになります。
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]成功したトラジェクトリをFew-Shot例として利用する
成功したトラジェクトリは、実行例です。エージェントは、似た新しいタスクに取り組む前にその例を見ることで、ステップの順序を学習できます。最も高いスコアのトラジェクトリを、システムプロンプトのfew-shotプレフィックスとして挿入します。
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)失敗モードの分類体系
失敗したトラジェクトリから失敗モードの分類体系を構築すると、パターンを把握しやすくなります。十分な数の失敗が同じ根本原因を共有している場合、それは1回の実行だけでなく、エージェントのツール、プロンプト、ロジックを修正すべきシグナルです。
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}トラジェクトリからの学習ペアの作成
教師ありファインチューニングには、(input, ideal_output)のペアが必要です。失敗したトラジェクトリからは不適切な出力が得られ、失敗分析からは代わりにどうなるべきだったかが得られます。これらを組み合わせることで、学習ペアになります。
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のファインチューニングAPIは、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回の再試行後に成功したトラジェクトリは、1回目で成功したものよりノイズが多くなります。効率性を基準に、最小ステップ数で成功していること、最終スコアが高いこと、捏造された中間結果がないことを条件にフィルタリングします。
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理解度チェック
失敗したトラジェクトリを分析する主な目的は何ですか。
復習：トラジェクトリベースの自己改善
よくできました。このレッスンの重要なポイントは次のとおりです。
- トラジェクトリ：開始から完了までの(state, action, result)のステップ列
- 成功したトラジェクトリ：似たタスクの前に挿入するfew-shot例
- 失敗したトラジェクトリ：失敗モードを分析 → 学習ペアを作成 → ファインチューニング
- 品質フィルタリング：短く、高スコアで成功したトラジェクトリを優先
- フライホイール：実行 → 記録 → 分析 → ファインチューニング → 改善版を実行
次は、自己改善で起こり得る問題、つまり報酬ハッキング、分布シフト、ガードレールについて学びます。
よくある質問
「軌跡ベースの自己改善」レッスンは無料ですか？
はい。「軌跡ベースの自己改善」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「軌跡ベースの自己改善」で何を学びますか？
成功・失敗したアクション系列から学習し、将来の振る舞いを改善します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「軌跡ベースの自己改善」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。