Gromadzenie i przechowywanie informacji zwrotnych

Dlaczego agenci potrzebują informacji zwrotnych

Bezpośrednie informacje zwrotne: kciuk w górę/w dół

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)

Bezpośrednie informacje zwrotne: oceny w gwiazdkach

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}

Bezpośrednie informacje zwrotne: korekty w wolnym tekście

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)

Pośrednie informacje zwrotne: sygnał ponownego zadania pytania

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)

Pośrednie informacje zwrotne: sygnał edycji wyniku

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

Schemat przechowywania informacji zwrotnych

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

Pisanie klasy zbierającej informacje zwrotne

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

Agregowanie informacji zwrotnych na potrzeby analizy

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

Prywatność i zgoda podczas zbierania informacji zwrotnych

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

Kompletny potok informacji zwrotnych

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

Sprawdzenie wiedzy

Podsumowanie: zbieranie i przechowywanie informacji zwrotnych