AI Agents · 课时

反馈收集与存储

Q: 「反馈收集与存储」这节课中我会学到什么？

捕获智能体互动中的明确评分和隐式行为信号 你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

捕获智能体互动中的明确评分和隐式行为信号

第 1 / 4 课13 个步骤

反馈收集与存储 是 CoddyKit 上的免费 AI Agents 课时。这是第 1 节课，共 4 节。你可以在下方免费阅读本课时的完整内容 — 然后在浏览器中使用内置代码编辑器和全天候 AI 导师进行实践。这是 AI Agents 学习路径的一部分，你的进度在网页和 CoddyKit 应用中同步。 AI Agents 课程共包含 4 节课。

代理为什么需要反馈

从未收到反馈的代理会停留在原地，无法超越初始训练水平。反馈在代理的行为与用户真正的需求之间建立了闭环。

最重要的有两类：显式 feedback（用户有意识地评价输出）和隐式 feedback（用户行为在不直接表达的情况下传递质量信号）。

显式反馈：点赞/点踩

最简单的形式是每次响应后的二元信号。它易于收集和存储，但信息密度较低。

实现模式：代理响应后，提供反馈提示，并将结果与对话轮次 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}")

反馈存储架构

所有 feedback 类型共享一个通用架构，同时包含特定于类型的负载字段。使用包含 type 判别字段和 payload 数据列的单个表，可以简化查询，同时支持任何 feedback 类型。

# 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 收集集中到单个类之后，可以让代码库的其余部分保持整洁。收集器负责去重、批处理和异步写入，因此 feedback 不会阻塞代理主循环。

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'])

聚合反馈以供分析

原始 feedback 记录需要先聚合，才能用于改进决策。常见的聚合方式包括：按意图类型统计批准率、统计一段时间内的纠正率，以及统计编辑次数最多的输出类别。

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))

反馈收集中的隐私与同意

Feedback 通常包含敏感的用户数据。最佳实践包括：在记录自由文本纠正前获得明确同意，在分析前将会话 ID 匿名化，设置保留期限（例如 90 天后删除），并且绝不在 feedback 负载中记录 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))

端到端反馈流程

将所有内容串联起来：收集 → anonymise → 缓冲 → flush → 聚合 → 报告。该流程在生产环境中与代理并行运行，并生成每周改进报告，显示哪些意图的批准率最低。

# 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]
        )

知识检查

哪种 feedback 信号不需要用户有意识地采取行动？

在交互式课程中回答

回顾：反馈收集与存储

做得很好！以下是您在本课中学到的内容：

显式 feedback：点赞（二元）、星级（分级）、纠正（配对的训练数据）
隐式 feedback：重新提问检测和输出编辑距离可以在无需用户操作的情况下反映信号质量
存储架构：包含类型判别字段和 JSONB 负载的单个表
流程：收集 → anonymise → 缓冲 → flush → 聚合
隐私：对会话 ID 进行哈希处理，移除 PII，设置保留期限

下一课中，您将学习代理如何反思自身表现，并将这些反思存储为情景记忆。

免费开始

用 AI 导师学习 AI Agents — 免费

在浏览器中编写并运行真实代码，获得全天候 AI 导师的即时帮助，并在网页或应用中继续学习。

课程: 60
课程: 239

免费开始学习查看完整课程

常见问题解答

「反馈收集与存储」课时是免费的吗？

是的 — 「反馈收集与存储」的完整文本可在网页上免费阅读。要进行交互式练习（内置代码编辑器和全天候 AI 导师）并解锁 AI Agents 课程的其余内容，请升级到 CoddyKit PRO。 AI Agents 课程共包含 4 节课。

「反馈收集与存储」这节课中我会学到什么？

捕获智能体互动中的明确评分和隐式行为信号你通过在浏览器中直接运行的动手代码来练习 AI Agents，全天候 AI 导师会在你学习这节课的过程中回答你的问题。

学习 AI Agents 需要有经验吗？

无需任何先前经验。CoddyKit 上的 AI Agents 课程适合初学者到高级学习者，你可以从这里开始或从头开始，按照自己的节奏学习。这是第 1 节课，共 4 节。

「反馈收集与存储」课时需要多长时间？

大多数 CoddyKit 课程大约需要 5–10 分钟。每节课都很精短且互动，所以你能稳步进步，并在网页和应用中从离开的地方继续。

我能在这节 AI Agents 课中编写并运行代码吗？

能。每节 AI Agents 课都包含内置代码编辑器，你可以在浏览器中直接编写并运行真实代码，并获得即时 AI 反馈 — 无需本地设置。

此课程中的所有课时

← 返回 AI Agents