フィードバックの収集と保存
エージェントとのやり取りから、明示的な評価と暗黙的な行動シグナルを収集します。
「フィードバックの収集と保存」はCoddyKit上の無料AI Agentsレッスンです。 これはレッスン1/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Agents学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Agentsコースには全4レッスンが含まれています。
エージェントにフィードバックが必要な理由
フィードバックを一度も受け取らないエージェントは、時間が止まったままです。初期トレーニングを超えて改善することができません。フィードバックによって、エージェントの行動とユーザーが実際に求めているものとの間のループが閉じられます。
特に重要なのは、明示的フィードバック（ユーザーが意識的に出力を評価するもの）と、暗黙的フィードバック（ユーザーが言葉で伝えなくても、その行動から品質を示すもの）の2種類です。
明示的フィードバック：高評価・低評価
最も単純な形式は、各応答の後に送る二値のシグナルです。収集や保存は簡単ですが、情報密度は低くなります。
実装パターンとしては、エージェントが応答した後にフィードバックの入力を促し、その結果を会話ターンIDとともに記録します。
import uuid
from datetime import datetime
def collect_thumbs_feedback(turn_id: str, rating: str) -> dict:
"""rating: 'up' or 'down'"""
assert rating in ('up', 'down'), 'Invalid rating'
record = {
'feedback_id': str(uuid.uuid4()),
'turn_id': turn_id,
'type': 'thumbs',
'value': 1 if rating == 'up' else -1,
'created_at': datetime.utcnow().isoformat()
}
return record
feedback = collect_thumbs_feedback('turn_abc123', 'up')
print(feedback)明示的フィードバック：星による評価
1～5個の星による評価は、高評価・低評価よりも細かい粒度で評価できます。かろうじて許容できる（星2個）回答と優れた（星5個）回答を区別できるため、ファインチューニングで利用するシグナルの品質を高めるのに役立ちます。
トレーニングパイプラインで使用する前に、0～1に正規化してください。
def collect_star_feedback(turn_id: str, stars: int) -> dict:
if not 1 <= stars <= 5:
raise ValueError('Stars must be between 1 and 5')
return {
'turn_id': turn_id,
'type': 'star',
'raw_value': stars,
'normalized': (stars - 1) / 4.0 # maps 1->0.0, 5->1.0
}
fb = collect_star_feedback('turn_xyz456', 4)
print(fb)
# {'turn_id': 'turn_xyz456', 'type': 'star', 'raw_value': 4, 'normalized': 0.75}明示的フィードバック：自由記述による修正
自由記述のフィードバックは、最も情報量の多いシグナルです。ユーザーは、求めていた内容を正確に記述できます。たとえば、「要約が長すぎました」、「要点を逃しています」、「通貨が違います。EURを指定しました」などです。
後で教師ありファインチューニングに使用できるよう、元の出力と修正を関連付けて保存し、（悪い出力 → 修正後の出力）のペアを作成できるようにします。
def collect_correction_feedback(
turn_id: str,
original_output: str,
corrected_output: str,
user_note: str = ''
) -> dict:
return {
'turn_id': turn_id,
'type': 'correction',
'original': original_output,
'corrected': corrected_output,
'user_note': user_note
}
fb = collect_correction_feedback(
'turn_789',
'The capital of Australia is Sydney.',
'The capital of Australia is Canberra.',
'Sydney is the largest city but not the capital.'
)
print(fb)暗黙的フィードバック：再質問シグナル
ユーザーが直後に同じ質問を別の言葉で再度尋ねた場合、直前の回答が間違っていたか不十分だったことを示す強い暗黙的シグナルです。ユーザーに何かをクリックしてもらう必要はありません。行動そのものがシグナルになります。
from datetime import datetime, timedelta
def detect_re_ask(
current_msg: str,
conversation_history: list,
similarity_threshold: float = 0.7,
window_seconds: int = 120
) -> bool:
"""
Returns True if the current message is semantically similar
to a recent message, suggesting dissatisfaction.
"""
now = datetime.utcnow()
for turn in conversation_history[-5:]:
age = (now - turn['timestamp']).seconds
if age <= window_seconds and turn['role'] == 'user':
# In production: use embedding cosine similarity
if simple_similarity(current_msg, turn['content']) >= similarity_threshold:
return True
return False
def simple_similarity(a: str, b: str) -> float:
words_a = set(a.lower().split())
words_b = set(b.lower().split())
if not words_a or not words_b:
return 0.0
return len(words_a & words_b) / len(words_a | words_b)
if __name__ == '__main__':
now = datetime.utcnow()
history = [
{'role': 'user', 'content': 'How do I reset my password', 'timestamp': now - timedelta(seconds=30)},
]
result = detect_re_ask('How do I reset my password please', history)
print('Re-ask detected:', result)
暗黙的フィードバック：出力編集シグナル
エージェントがテキストを生成し、ユーザーが使用前に編集した場合、元のテキストと編集後のテキストの差分が暗黙的フィードバックになります。編集後のバージョンが、ユーザーが実際に求めていた内容です。
これはライティングアシスタント、コード生成ツール、メール作成ツールでよく使われます。
import difflib
def extract_edit_feedback(original: str, edited: str) -> dict:
differ = difflib.unified_diff(
original.splitlines(),
edited.splitlines(),
lineterm=''
)
diff_lines = list(differ)
edit_ratio = difflib.SequenceMatcher(None, original, edited).ratio()
return {
'type': 'edit',
'original': original,
'edited': edited,
'edit_distance': 1.0 - edit_ratio, # 0=unchanged, 1=fully rewritten
'diff': '\n'.join(diff_lines)
}
fb = extract_edit_feedback(
'Dear John, I am writing to inform you...',
'Hi John, Just a quick note...'
)
print(f"Edit distance: {fb['edit_distance']:.2f}")フィードバック保存用スキーマ
すべてのフィードバック種別は、種別固有のペイロードフィールドを持つ共通スキーマを使用します。type識別子とpayload JSONカラムを持つ単一のテーブルを使用すると、あらゆるフィードバック種別に対応しながらクエリを簡潔に保てます。
# SQL schema for feedback storage
CREATE_TABLE_SQL = '''
CREATE TABLE agent_feedback (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
session_id TEXT NOT NULL,
turn_id TEXT NOT NULL,
agent_id TEXT NOT NULL,
type TEXT NOT NULL CHECK (type IN ('thumbs','star','correction','re_ask','edit')),
value FLOAT, -- numeric signal: +1/-1, 0-1, edit distance
payload JSONB, -- type-specific data
created_at TIMESTAMPTZ DEFAULT now()
);
CREATE INDEX ON agent_feedback (agent_id, created_at);
CREATE INDEX ON agent_feedback (type);
'''
# Example insert
INSERT_SQL = '''
INSERT INTO agent_feedback (session_id, turn_id, agent_id, type, value, payload)
VALUES ($1, $2, $3, $4, $5, $6)
'''
if __name__ == '__main__':
print('Feedback table schema:')
print(CREATE_TABLE_SQL)
print('Insert statement:')
print(INSERT_SQL)
Feedback Collectorクラスの作成
すべてのフィードバック収集を単一のクラスに集約すると、コードベースの他の部分を整理された状態に保てます。Collectorが重複排除、バッチ処理、非同期書き込みを処理するため、フィードバックによってエージェントのメインループがブロックされることはありません。
import asyncio
from collections import deque
from datetime import datetime
class FeedbackCollector:
def __init__(self, agent_id: str, flush_interval: int = 30):
self.agent_id = agent_id
self.buffer: deque = deque(maxlen=1000)
self.flush_interval = flush_interval
def record(self, turn_id: str, fb_type: str, value: float, payload: dict):
self.buffer.append({
'turn_id': turn_id,
'agent_id': self.agent_id,
'type': fb_type,
'value': value,
'payload': payload,
'created_at': datetime.utcnow().isoformat()
})
async def flush(self, db_client):
while self.buffer:
record = self.buffer.popleft()
await db_client.insert('agent_feedback', record)
async def start_auto_flush(self, db_client):
while True:
await asyncio.sleep(self.flush_interval)
await self.flush(db_client)
if __name__ == '__main__':
fc = FeedbackCollector(agent_id='agent-1')
fc.record('turn-1', 'thumbs', 1.0, {'comment': 'Great answer'})
fc.record('turn-2', 'thumbs', -1.0, {'comment': 'Wrong ticker'})
print(f'Buffered {len(fc.buffer)} feedback records:')
for rec in fc.buffer:
print(' -', rec['type'], rec['value'], rec['payload'])
分析のためのフィードバック集計
生のフィードバックレコードは、改善の判断に役立てる前に集計する必要があります。一般的な集計には、インテント種別ごとの承認率、時間の経過に伴う修正率、最も編集された出力カテゴリなどがあります。
from collections import defaultdict
from statistics import mean
def aggregate_feedback(records: list) -> dict:
by_type = defaultdict(list)
for r in records:
by_type[r['type']].append(r['value'])
summary = {}
if 'thumbs' in by_type:
values = by_type['thumbs']
summary['approval_rate'] = (values.count(1) / len(values)) * 100
if 'star' in by_type:
summary['avg_star'] = mean(by_type['star']) * 4 + 1 # denormalize
if 'edit' in by_type:
summary['avg_edit_distance'] = mean(by_type['edit'])
if 'correction' in by_type:
summary['correction_count'] = len(by_type['correction'])
return summary
records = [
{'type': 'thumbs', 'value': 1},
{'type': 'thumbs', 'value': -1},
{'type': 'star', 'value': 0.75},
{'type': 'edit', 'value': 0.3}
]
print(aggregate_feedback(records))フィードバック収集におけるプライバシーと同意
フィードバックには、機密性の高いユーザーデータが含まれることがよくあります。ベストプラクティスは、自由記述の修正を記録する前に明示的な同意を得ること、分析前にセッションIDを匿名化すること、保持期間の上限（例：90日後に削除）を設定すること、フィードバックのペイロードにPIIを決して記録しないことです。
import hashlib
import re
def anonymise_feedback(record: dict) -> dict:
"""Anonymise feedback record before storing for training."""
safe = record.copy()
# Hash the session_id so it can't be traced back to a user
if 'session_id' in safe:
safe['session_id'] = hashlib.sha256(
safe['session_id'].encode()
).hexdigest()[:16]
# Strip emails and phone numbers from correction text
if 'payload' in safe and 'corrected' in safe['payload']:
text = safe['payload']['corrected']
text = re.sub(r'[\w.+-]+@[\w-]+\.[\w.]+', '[EMAIL]', text)
text = re.sub(r'\+?[0-9][\s\-().]{7,}[0-9]', '[PHONE]', text)
safe['payload'] = dict(safe['payload'], corrected=text)
return safe
if __name__ == '__main__':
record = {
'session_id': 'sess-abc123',
'payload': {'corrected': 'Contact me at jane@example.com or 555-123-4567'}
}
print('Anonymised record:', anonymise_feedback(record))
エンドツーエンドのフィードバックパイプライン
すべてを組み合わせると、収集 → 匿名化 → バッファリング → フラッシュ → 集計 → レポートという流れになります。パイプラインは本番環境でエージェントと並行して動作し、承認率が最も低いインテントを示す週次の改善レポートを生成します。
# Simplified end-to-end feedback pipeline sketch
class FeedbackPipeline:
def __init__(self, agent_id: str):
self.collector = FeedbackCollector(agent_id)
self.records = []
def on_thumbs(self, turn_id: str, rating: str):
value = 1.0 if rating == 'up' else -1.0
record = self.collector.record(turn_id, 'thumbs', value, {})
self.records.append(record)
def on_edit(self, turn_id: str, original: str, edited: str):
fb = extract_edit_feedback(original, edited)
record = self.collector.record(
turn_id, 'edit', fb['edit_distance'], fb
)
self.records.append(record)
def weekly_report(self) -> dict:
return aggregate_feedback(
[r for r in self.records]
)理解度チェック
ユーザーの意識的な操作を必要としないフィードバックシグナルはどれですか？
総括：フィードバックの収集と保存
よくできました。このレッスンでは次のことを学びました。
- 明示的フィードバック：高評価・低評価（バイナリ）、星（段階評価）、修正（ペア化したトレーニングデータ）
- 暗黙的フィードバック：再質問の検出と出力編集の距離により、ユーザーの操作なしでシグナルの品質を測定
- 保存用スキーマ：種別識別子とJSONBペイロードを持つ単一テーブル
- パイプライン：収集 → 匿名化 → バッファリング → フラッシュ → 集計
- プライバシー：セッションIDのハッシュ化、PIIの除去、保持期間の上限設定
次のレッスンでは、エージェントが自身のパフォーマンスを振り返り、その振り返りをエピソード記憶として保存する方法を学びます。
よくある質問
「フィードバックの収集と保存」レッスンは無料ですか？
はい。「フィードバックの収集と保存」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Agentsコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Agentsコースには全4レッスンが含まれています。
「フィードバックの収集と保存」で何を学びますか？
エージェントとのやり取りから、明示的な評価と暗黙的な行動シグナルを収集します。 ブラウザで直接実行するハンズオンコードでAI Agentsを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Agentsを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Agentsは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン1/4です。
「フィードバックの収集と保存」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Agentsレッスンでコードを書いて実行できますか？
はい。すべてのAI Agentsレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。
このコースのすべてのレッスン
- フィードバックの収集と保存
- リフレクションと自己批評のループ
- 軌跡ベースの自己改善
- 自己改善がうまくいかないとき