psychscanner — Feedback Mechanism Tutorial¶

This notebook shows how to add trial-level feedback to a psychscanner experiment using the FeedbackBase API. The example is built around the Reality Monitoring (RM) word-pair task from examples/rm_msg_injection.py, but every step generalises to any task.

What is feedback in psychscanner?¶

After each trial the LLM produces a response. Feedback lets you inspect that response and inject a short message at the start of the next trial. Typical uses:

• Tell the model whether its last answer was correct.
• Remind it of task rules it violated.
• Provide correctness cues in learning / training paradigms.

What changed in the new API?¶

Old API	New API
feedback_fn instantiated per trial	Instantiated once per simulation
Raw AIMessage passed to handler	Pre-parsed dict passed to handler
parser_status string needed	Not needed — TaskRunner handles parsing
Cross-trial state via module globals	Lives safely in self
feedback='1' string	feedback=True bool (backward-compat with '1')
fb_response not saved	Saved in every trial output dict

In [1]:

# ── Setup: ensure NLTK words corpus is available and examples/ is on the path ──
import sys, os, nltk

nltk.download('words', quiet=True)

# Locate the examples/ folder regardless of where Jupyter was launched from.
# The notebook itself lives in examples/, so we look for rm_msg_injection.py
# starting from cwd and stepping up one level if needed.
_cwd = os.getcwd()
for _candidate in [_cwd, os.path.join(_cwd, 'examples')]:
    if os.path.exists(os.path.join(_candidate, 'rm_msg_injection.py')):
        sys.path.insert(0, _candidate)
        print(f'examples dir: {_candidate}')
        break
else:
    raise RuntimeError(
        'Could not find rm_msg_injection.py. '
        'Open Jupyter from the psychscanner project root or the examples/ folder.'
    )

# ── Setup: ensure NLTK words corpus is available and examples/ is on the path ── import sys, os, nltk nltk.download('words', quiet=True) # Locate the examples/ folder regardless of where Jupyter was launched from. # The notebook itself lives in examples/, so we look for rm_msg_injection.py # starting from cwd and stepping up one level if needed. _cwd = os.getcwd() for _candidate in [_cwd, os.path.join(_cwd, 'examples')]: if os.path.exists(os.path.join(_candidate, 'rm_msg_injection.py')): sys.path.insert(0, _candidate) print(f'examples dir: {_candidate}') break else: raise RuntimeError( 'Could not find rm_msg_injection.py. ' 'Open Jupyter from the psychscanner project root or the examples/ folder.' )

examples dir: /Users/saurabhext/Documents/PSYCHSCANNER/psychscanner/examples

---

1. The FeedbackBase contract¶

Every feedback handler must subclass FeedbackBase and implement one method:

def on_response(self, trial: dict, response: dict) -> str | None:
    ...

Argument	What it contains
trial	Full trial dict from the task JSON (trcode, stimulus, corrAns, …)
response	Model response as a plain Python dict (pydantic .model_dump() for structured output, or {"content": raw_string} for unstructured)

Return a feedback string (usually JSON) or None to skip feedback for that trial.

An optional inject_feedback(input_dict, fb_str) method controls how the feedback string is merged into the next trial's input. The default merges it with the current trial stimulus into a single HumanMessage.

In [2]:

from psychscanner.feedback import FeedbackBase

# Minimal example — always give the same hint
class SimpleFeedback(FeedbackBase):
    def on_response(self, trial, response):
        return 'Remember to follow all task instructions carefully.'

assert issubclass(SimpleFeedback, FeedbackBase)
fb = SimpleFeedback()
print(fb.on_response({}, {}))

from psychscanner.feedback import FeedbackBase # Minimal example — always give the same hint class SimpleFeedback(FeedbackBase): def on_response(self, trial, response): return 'Remember to follow all task instructions carefully.' assert issubclass(SimpleFeedback, FeedbackBase) fb = SimpleFeedback() print(fb.on_response({}, {}))

/opt/anaconda3/envs/psyscan/lib/python3.11/site-packages/langgraph/checkpoint/base/__init__.py:17: LangChainPendingDeprecationWarning: The default value of `allowed_objects` will change in a future version. Pass an explicit value (e.g., allowed_objects='messages' or allowed_objects='core') to suppress this warning.
  from langgraph.checkpoint.serde.jsonplus import JsonPlusSerializer

Remember to follow all task instructions carefully.

---

2. The RM-task feedback handler¶

The file examples/rm_msg_injection.py contains Stim_Trial_Injection, a complete feedback handler for the RM word-pair task.

Key design points:

• self.all_words_in_use accumulates every word seen across all trials. Because psychscanner creates one instance per participant, this list resets correctly between participants — no module-level global needed.
• on_response(trial, response) receives a pre-parsed dict — no manual eval() or parser_status string needed.
• generate_fb_response is a standalone pure function so it is easy to unit-test.

In [3]:

from rm_msg_injection import Stim_Trial_Injection, generate_fb_response
import json

# ── Imagined encoding trial ──────────────────────────────────────────────────
# word_2 = '__' means the model had to imagine a second word
trial_imagined = {
    'trcode': 'encode_imagined_01',
    'stimulus': {'Word_Pair': {'word_1': 'apple', 'word_2': '__'}},
    'corrAns': None,
}
response_imagined = {'Word_2': 'mango', 'rating': 75}

words_log = []
fb = generate_fb_response(trial_imagined, response_imagined, words_log)
print('Feedback for imagined trial:')
print(json.dumps(json.loads(fb), indent=2))
print('words_log:', words_log)

from rm_msg_injection import Stim_Trial_Injection, generate_fb_response import json # ── Imagined encoding trial ────────────────────────────────────────────────── # word_2 = '__' means the model had to imagine a second word trial_imagined = { 'trcode': 'encode_imagined_01', 'stimulus': {'Word_Pair': {'word_1': 'apple', 'word_2': '__'}}, 'corrAns': None, } response_imagined = {'Word_2': 'mango', 'rating': 75} words_log = [] fb = generate_fb_response(trial_imagined, response_imagined, words_log) print('Feedback for imagined trial:') print(json.dumps(json.loads(fb), indent=2)) print('words_log:', words_log)

Feedback for imagined trial:
{
  "Feedback on previous response": {
    "response feedback": "**CORRECT: Last trial was an imagined trial type. Your 'Word_2' correctly followed the given instructions.**",
    "rating feedback": "**CORRECT: Your 'rating' value is in the instructed rating scale range.**",
    "overall feedback": "**CORRECT Answer!** Well answered, good job!"
  }
}
words_log: ['apple', 'mango']

In [4]:

# ── Perceived encoding trial ─────────────────────────────────────────────────
# word_2 is a real word — model must reproduce it
trial_perceived = {
    'trcode': 'encode_perceived_01',
    'stimulus': {'Word_Pair': {'word_1': 'chair', 'word_2': 'table'}},
    'corrAns': None,
}

# Correct response
response_correct = {'Word_2': 'table', 'rating': 80}
fb_correct = generate_fb_response(trial_perceived, response_correct, list(words_log))
print('Feedback (correct):')
print(json.dumps(json.loads(fb_correct), indent=2))

# Wrong response
response_wrong = {'Word_2': 'lamp', 'rating': 50}
fb_wrong = generate_fb_response(trial_perceived, response_wrong, list(words_log))
print('\nFeedback (incorrect):')
print(json.dumps(json.loads(fb_wrong), indent=2))

# ── Perceived encoding trial ───────────────────────────────────────────────── # word_2 is a real word — model must reproduce it trial_perceived = { 'trcode': 'encode_perceived_01', 'stimulus': {'Word_Pair': {'word_1': 'chair', 'word_2': 'table'}}, 'corrAns': None, } # Correct response response_correct = {'Word_2': 'table', 'rating': 80} fb_correct = generate_fb_response(trial_perceived, response_correct, list(words_log)) print('Feedback (correct):') print(json.dumps(json.loads(fb_correct), indent=2)) # Wrong response response_wrong = {'Word_2': 'lamp', 'rating': 50} fb_wrong = generate_fb_response(trial_perceived, response_wrong, list(words_log)) print('\nFeedback (incorrect):') print(json.dumps(json.loads(fb_wrong), indent=2))

Feedback (correct):
{
  "Feedback on previous response": {
    "response feedback": "**CORRECT: Last trial was a perceived trial type. Your 'Word_2' correctly followed the trial instruction.**",
    "rating feedback": "**CORRECT: Your 'rating' value is in the instructed rating scale range.**",
    "overall feedback": "**CORRECT Answer!** Well answered, good job!"
  }
}

Feedback (incorrect):
{
  "Feedback on previous response": {
    "response feedback": "**INCORRECT: Last trial was a perceived trial type. Your 'Word_2' does not match the second word in the pair.**",
    "rating feedback": "**CORRECT: Your 'rating' value is in the instructed rating scale range.**",
    "overall feedback": "**INCORRECT Answer!** Follow all trial instructions more accurately."
  }
}

In [5]:

# ── Recognition / test trial ─────────────────────────────────────────────────
trial_test = {
    'trcode': 'test_01',
    'stimulus': {'prompt': 'Was this word pair perceived or imagined?'},
    'corrAns': 'imagined',
}

response_test_correct = {'Judgment': 'imagined', 'Confidence': 5}
fb_test = generate_fb_response(trial_test, response_test_correct, list(words_log))
print('Feedback (test, correct):')
print(json.dumps(json.loads(fb_test), indent=2))

# ── Recognition / test trial ───────────────────────────────────────────────── trial_test = { 'trcode': 'test_01', 'stimulus': {'prompt': 'Was this word pair perceived or imagined?'}, 'corrAns': 'imagined', } response_test_correct = {'Judgment': 'imagined', 'Confidence': 5} fb_test = generate_fb_response(trial_test, response_test_correct, list(words_log)) print('Feedback (test, correct):') print(json.dumps(json.loads(fb_test), indent=2))

Feedback (test, correct):
{
  "Feedback on previous response": {
    "response feedback": "**CORRECT: The correct 'Judgment' was 'imagined'. Your 'Judgment' followed the task instructions.**",
    "rating feedback": "**CORRECT: Your 'Confidence' value is in the instructed rating scale range (1\u20136).**",
    "overall feedback": "**CORRECT Answer!** Well answered, good job!"
  }
}

In [6]:

# ── Two participants' handlers are completely independent ────────────────────
handler_p1 = Stim_Trial_Injection()
handler_p2 = Stim_Trial_Injection()

# P1 processes a trial
handler_p1.on_response(trial_imagined, {'Word_2': 'river', 'rating': 60})

print('P1 words_in_use:', handler_p1.all_words_in_use)
print('P2 words_in_use:', handler_p2.all_words_in_use)  # still empty — independent!

# ── Two participants' handlers are completely independent ──────────────────── handler_p1 = Stim_Trial_Injection() handler_p2 = Stim_Trial_Injection() # P1 processes a trial handler_p1.on_response(trial_imagined, {'Word_2': 'river', 'rating': 60}) print('P1 words_in_use:', handler_p1.all_words_in_use) print('P2 words_in_use:', handler_p2.all_words_in_use) # still empty — independent!

P1 words_in_use: ['apple', 'river']
P2 words_in_use: []

---

3. Wiring feedback into an experiment¶

Pass the class (not an instance) as feedback_fn in your ExpCard. psychscanner creates one fresh instance per participant via feedback_fn().

from psychscanner import ExpCard, ScannerModel
from rm_msg_injection import Stim_Trial_Injection

card_in = ExpCard(
    model='gpt-4o-mini',
    family='openai',
    memory='Convo',
    task_file='path/to/rm_task.json',
    parser='RMParser',                  # structured output parser
    feedback=True,                      # also accepts '1' for backward compat
    feedback_fn=Stim_Trial_Injection,   # pass the CLASS, not an instance
    projectname='RM_FEEDBACK_STUDY',
)

scanner = ScannerModel(card_in)
results = scanner.run()

psychscanner handles the rest:

1. Creates Stim_Trial_Injection() once at the start of each participant simulation.
2. After each trial, calls handler.on_response(trial, parsed_response).
3. Before the next trial, calls handler.inject_feedback(input_dict, fb_str) to merge the feedback into the LLM prompt.
4. Saves fb_response in every trial output dict.

---

4. Opting individual trials in / out of feedback¶

Add an "fb": false key to any trial in your task JSON to skip feedback for that trial. Trials without the key default to true (feedback on).

{
  "trcode": "practice_01",
  "fb": false,
  "hmsg": { "..." : "..." }
}

Useful for:

• Practice / warm-up trials that should not carry feedback into the first real trial.
• Task-instruction acknowledgement trials where on_response already returns None.
• Any trial where you want a clean prompt slate.

---

5. Customising the injection format¶

The default inject_feedback wraps feedback and the current stimulus into one JSON HumanMessage. Override it to change the format:

In [7]:

import json
from langchain_core.messages import HumanMessage
from psychscanner.feedback import FeedbackBase

class PlainTextFeedback(FeedbackBase):
    """Inject feedback as a plain-text prefix rather than JSON."""

    def on_response(self, trial, response):
        return f"Previous trial feedback: {response.get('content', '')[:40]}"

    def inject_feedback(self, input_dict, fb_str):
        trial_content = input_dict['inputs'][0].content
        combined = f"{fb_str}\n\n---\n{trial_content}"
        input_dict['inputs'] = [HumanMessage(combined)]
        return input_dict

# Demo
fb = PlainTextFeedback()
input_dict = {'inputs': [HumanMessage('Current trial: word pair [apple / __]')], 'system_message': 'sys'}
updated = fb.inject_feedback(input_dict, 'Previous trial feedback: CORRECT')
print(updated['inputs'][0].content)

import json from langchain_core.messages import HumanMessage from psychscanner.feedback import FeedbackBase class PlainTextFeedback(FeedbackBase): """Inject feedback as a plain-text prefix rather than JSON.""" def on_response(self, trial, response): return f"Previous trial feedback: {response.get('content', '')[:40]}" def inject_feedback(self, input_dict, fb_str): trial_content = input_dict['inputs'][0].content combined = f"{fb_str}\n\n---\n{trial_content}" input_dict['inputs'] = [HumanMessage(combined)] return input_dict # Demo fb = PlainTextFeedback() input_dict = {'inputs': [HumanMessage('Current trial: word pair [apple / __]')], 'system_message': 'sys'} updated = fb.inject_feedback(input_dict, 'Previous trial feedback: CORRECT') print(updated['inputs'][0].content)

Previous trial feedback: CORRECT

---
Current trial: word pair [apple / __]

---

6. Inspecting saved feedback in results¶

Every trial output dict now contains an fb_response key with the feedback string generated after that trial (and injected before the next one).

In [8]:

# Hypothetical post-run inspection
# results = scanner.run()   # list of per-participant data

# for participant_data in results:
#     for trial_record in participant_data:
#         trcode = trial_record['trcode']
#         fb = trial_record.get('fb_response')   # None when no feedback was generated
#         print(f"{trcode}: feedback generated = {fb is not None}")

print('Access feedback via:  trial_record["fb_response"]')
print('It is None when on_response returned None for that trial.')

# Hypothetical post-run inspection # results = scanner.run() # list of per-participant data # for participant_data in results: # for trial_record in participant_data: # trcode = trial_record['trcode'] # fb = trial_record.get('fb_response') # None when no feedback was generated # print(f"{trcode}: feedback generated = {fb is not None}") print('Access feedback via: trial_record["fb_response"]') print('It is None when on_response returned None for that trial.')

Access feedback via:  trial_record["fb_response"]
It is None when on_response returned None for that trial.

---

7. Migration from the old API¶

Before (old)¶

# module-level global — breaks between participants!
all_words_in_use = []

class OldFeedback:
    def __init__(self, trial_data):          # re-instantiated every trial
        self.trial_data = trial_data

    def generate_feedback(self, trdata, pred_dict, input_dict, parser_status, trial_item_collector):
        message = pred_dict.content
        if parser_status == '1':
            message = eval(message)          # manually parse
        ...

    def update_trial_stim(self, finput, fb_response):
        ...

card_in = ExpCard(feedback='1', feedback_fn=OldFeedback)

After (new)¶

from psychscanner.feedback import FeedbackBase

class NewFeedback(FeedbackBase):             # instantiated once per simulation
    def __init__(self):
        self.all_words_in_use = []           # instance variable — safe across participants

    def on_response(self, trial, response):  # response is already a plain dict
        word_2 = response.get('Word_2', '')
        ...
        return feedback_string_or_none

card_in = ExpCard(feedback=True, feedback_fn=NewFeedback)  # True or '1' both work

Summary of changes:

1. Subclass FeedbackBase instead of plain object.
2. Rename generate_feedback → on_response; signature simplifies to (self, trial, response).
3. Move module-level globals into self.__init__.
4. Remove update_trial_stim — the base class inject_feedback handles injection (or override it).
5. Change feedback='1' → feedback=True (both still work).