PsychScanner Parser System — ps-parser Branch Guide¶

This notebook is the definitive reference for the ps-parser branch API.
It covers every parser mode, the new unified registry, and the feedback integration.

What changed from the archive (old) API¶

Topic	Old / archive API	ps-parser API
Discover parsers	`inspect.getmembers(parser_mod)`	`from psychscanner.parsers import list_parsers`
Look up by name	`eval(name)` — fragile, unsafe	`get_parser(name)` — registry lookup, clear errors
Per-trial routing	`parser='dynamic'` (hardcoded RM only)	callable dispatch or per-trial `"parser"` field in task JSON
Resolve any spec	ad-hoc `if/elif` in scanner	`resolve_parser(value)` — single entry point
Feedback toggle	`card.feedback = '1'` (string)	`card.feedback = True` (bool; `'1'` still accepted)
Feedback handler	module-level globals	`FeedbackBase` subclass — instance-per-simulation, no globals

Outline¶

Setup
Parser registry: discover and look up parsers
Module layout: three import paths
Parser modes in ExpCard
- Mode A: no parser (None / "0")
- Mode B: resolve from task JSON ("1")
- Mode C: string name ("DefaultLiteralVivid15")
- Mode D: direct Pydantic class
- Mode E: callable dispatch (replaces old "dynamic")
Custom parsers
Per-trial routing: Form A vs Form B
Diagnostic options: parser_raw and parser_config
Feedback API
Quick-reference summary

1. Setup¶

In [ ]:

Copied!





import json
import shutil
from pathlib import Path
from pprint import pprint

import psychscanner as psy
from psychscanner import ExpCard, ExpCardInit, ScannerModel

print("psychscanner version:", psy.__version__)

# ── Backend — change these two lines to use OpenAI / Anthropic / Groq, etc. ──
MODEL_FAMILY = "ollama"
MODEL_NAME   = "smollm2:360m-instruct-fp16"

# ── Output dir ────────────────────────────────────────────────────────────────
PROJECT_DIR = Path.cwd() / "_ps_parser_guide_runs"
if PROJECT_DIR.exists():
    shutil.rmtree(PROJECT_DIR)
PROJECT_DIR.mkdir()

TASKS_DIR = Path.cwd() / "tasks"
SURVEY_TASK = TASKS_DIR / "example_survey.json"

print(f"Outputs: {PROJECT_DIR}")
print(f"Survey task: {SURVEY_TASK}")
import json
import shutil
from pathlib import Path
from pprint import pprint

import psychscanner as psy
from psychscanner import ExpCard, ExpCardInit, ScannerModel

print("psychscanner version:", psy.__version__)

# ── Backend — change these two lines to use OpenAI / Anthropic / Groq, etc. ──
MODEL_FAMILY = "ollama"
MODEL_NAME   = "smollm2:360m-instruct-fp16"

# ── Output dir ────────────────────────────────────────────────────────────────
PROJECT_DIR = Path.cwd() / "_ps_parser_guide_runs"
if PROJECT_DIR.exists():
    shutil.rmtree(PROJECT_DIR)
PROJECT_DIR.mkdir()

TASKS_DIR = Path.cwd() / "tasks"
SURVEY_TASK = TASKS_DIR / "example_survey.json"

print(f"Outputs: {PROJECT_DIR}")
print(f"Survey task: {SURVEY_TASK}")

2. Parser Registry¶

The psychscanner.parsers top-level module exposes three registry helpers:

Function	Purpose
`list_parsers()`	Sorted list of all registered parser class names
`get_parser(name)`	Look up a class by name; raises `KeyError` with helpful suggestions on typos
`resolve_parser(value)`	One-stop resolver for any parser specification (see §4)

All bundled parsers in parser_tasks and parser_general are auto-registered at import time.

In [ ]:

Copied!





from psychscanner.parsers import list_parsers, get_parser, resolve_parser, PARSER_REGISTRY

all_names = list_parsers()
print(f"{len(all_names)} parsers registered:")
for name in all_names:
    print(f"  {name}")
from psychscanner.parsers import list_parsers, get_parser, resolve_parser, PARSER_REGISTRY

all_names = list_parsers()
print(f"{len(all_names)} parsers registered:")
for name in all_names:
    print(f"  {name}")

In [ ]:

Copied!





# Look up a class by name
cls = get_parser("DefaultLiteralVivid15")
print("Class:", cls)
print("Module:", cls.__module__)
print()

# Inspect its schema — useful when building task JSON
schema = cls.model_json_schema()
print(json.dumps(schema, indent=2))
# Look up a class by name
cls = get_parser("DefaultLiteralVivid15")
print("Class:", cls)
print("Module:", cls.__module__)
print()

# Inspect its schema — useful when building task JSON
schema = cls.model_json_schema()
print(json.dumps(schema, indent=2))

In [ ]:

Copied!





# get_parser raises a clear KeyError (with the full available-names list) on typos
try:
    get_parser("DefaultLiteralVivid16")   # doesn't exist
except KeyError as e:
    print("KeyError:", str(e)[:120], "...")
# get_parser raises a clear KeyError (with the full available-names list) on typos
try:
    get_parser("DefaultLiteralVivid16")   # doesn't exist
except KeyError as e:
    print("KeyError:", str(e)[:120], "...")

2a. Splitting parsers by paradigm¶

The registry merges two themed modules.
You can query them separately to see the split:

In [ ]:

Copied!





import inspect
from pydantic import BaseModel
from psychscanner.datasets.prompts import parser_tasks as pt_mod, parser_general as pg_mod

def _names_from(mod):
    return sorted(
        name for name, obj in inspect.getmembers(mod, inspect.isclass)
        if obj is not BaseModel
        and issubclass(obj, BaseModel)
        and obj.__module__ == mod.__name__
    )

task_parsers    = _names_from(pt_mod)
general_parsers = _names_from(pg_mod)

print(f"parser_tasks    ({len(task_parsers)} classes — vividness + RM):")
for n in task_parsers:
    print(f"  {n}")

print(f"\nparser_general  ({len(general_parsers)} classes — Likert, ratings, word-class, readiness):")
for n in general_parsers:
    print(f"  {n}")
import inspect
from pydantic import BaseModel
from psychscanner.datasets.prompts import parser_tasks as pt_mod, parser_general as pg_mod

def _names_from(mod):
    return sorted(
        name for name, obj in inspect.getmembers(mod, inspect.isclass)
        if obj is not BaseModel
        and issubclass(obj, BaseModel)
        and obj.__module__ == mod.__name__
    )

task_parsers    = _names_from(pt_mod)
general_parsers = _names_from(pg_mod)

print(f"parser_tasks    ({len(task_parsers)} classes — vividness + RM):")
for n in task_parsers:
    print(f"  {n}")

print(f"\nparser_general  ({len(general_parsers)} classes — Likert, ratings, word-class, readiness):")
for n in general_parsers:
    print(f"  {n}")

3. Module Layout — Three Import Paths¶

All three paths are equivalent — they return the same class object (no copies).

psychscanner.parsers                              ← recommended for new code
  └─ re-exports everything + registry helpers

psychscanner.datasets.prompts.parser_tasks        ← vividness + RM parsers
psychscanner.datasets.prompts.parser_general      ← Likert, generic, word-class

psychscanner.datasets.prompts.parser              ← backward-compat shim (old name)
psychscanner.datasets.prompts.parser_extra        ← backward-compat shim (old name)

Existing code importing from .parser or .parser_extra keeps working unchanged.

In [ ]:

Copied!





# All four import paths return the same class
from psychscanner.parsers import DefaultLiteralVivid15 as A
from psychscanner.datasets.prompts.parser_tasks   import DefaultLiteralVivid15 as B
from psychscanner.datasets.prompts.parser         import DefaultLiteralVivid15 as C   # shim

assert A is B is C, "They must all be the same class object"
print("All paths return the same class:", A)
# All four import paths return the same class
from psychscanner.parsers import DefaultLiteralVivid15 as A
from psychscanner.datasets.prompts.parser_tasks   import DefaultLiteralVivid15 as B
from psychscanner.datasets.prompts.parser         import DefaultLiteralVivid15 as C   # shim

assert A is B is C, "They must all be the same class object"
print("All paths return the same class:", A)

4. Parser Modes in `ExpCard`¶

ExpCard.parser accepts five different specifications, resolved by resolve_parser():

Mode	`parser=` value	Structured output?	Requires real LLM?
A	`None` or `"0"`	No — raw `AIMessage`	No (works with mock-llm)
B	`"1"`	Yes — name read from task JSON `"parser"` field	Yes
C	`"DefaultLiteralVivid15"`	Yes — direct name string	Yes
D	`DefaultLiteralVivid15`	Yes — Pydantic class	Yes
E	`lambda trcode: ...`	Yes — per-trial routing	Yes

Mode A — No parser (`None` / `"0"`)¶

The raw AIMessage is returned unchanged. Works with any model, including the built-in mock-llm.

In [ ]:

Copied!





card_a = ExpCardInit(
    model   ="mock-chat-model",
    family  ="mock-llm",
    task_file=SURVEY_TASK,
    parser  ="0",              # ← Mode A
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeA",
    tunnel_status="0",
)
exp_a = ExpCard(card_a)
results_a = ScannerModel(expcard=exp_a).run()

r0 = results_a[0][0]["pred_resp"]
print("Type:", type(r0).__name__)
print("Content (first trial):", getattr(r0, 'content', str(r0))[:80])
card_a = ExpCardInit(
    model   ="mock-chat-model",
    family  ="mock-llm",
    task_file=SURVEY_TASK,
    parser  ="0",              # ← Mode A
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeA",
    tunnel_status="0",
)
exp_a = ExpCard(card_a)
results_a = ScannerModel(expcard=exp_a).run()

r0 = results_a[0][0]["pred_resp"]
print("Type:", type(r0).__name__)
print("Content (first trial):", getattr(r0, 'content', str(r0))[:80])

Mode B — Resolve from task JSON (`"1"`)¶

The task JSON carries a "parser" key at the top level.
Setting card.parser = "1" tells ExpCard to read that name and look it up in the registry.

{ "parser": "DefaultLiteralAgree", ... }

This is the most common mode for task-driven experiments where parsers are declared in the task file.

In [ ]:

Copied!

# Inspect what parser the task JSON declares
task_data = json.loads(SURVEY_TASK.read_text())
print("Top-level parser declared in task JSON:", repr(task_data["parser"]))
# Inspect what parser the task JSON declares
task_data = json.loads(SURVEY_TASK.read_text())
print("Top-level parser declared in task JSON:", repr(task_data["parser"]))

In [ ]:

Copied!





card_b = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  ="1",              # ← Mode B — read name from task JSON
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeB",
    tunnel_status="0",
)
exp_b = ExpCard(card_b)
print("Parser resolved to:", exp_b.parser.__name__)

results_b = ScannerModel(expcard=exp_b).run()
print("\nFirst trial response:", results_b[0][0]["pred_resp"])
card_b = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  ="1",              # ← Mode B — read name from task JSON
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeB",
    tunnel_status="0",
)
exp_b = ExpCard(card_b)
print("Parser resolved to:", exp_b.parser.__name__)

results_b = ScannerModel(expcard=exp_b).run()
print("\nFirst trial response:", results_b[0][0]["pred_resp"])

Mode C — String name (`"DefaultLiteralVivid15"`)¶

Any registered parser name can be passed directly as a string — no need for "1" or the task JSON.

Under the hood, resolve_parser calls get_parser(name), so you get the same helpful KeyError on typos.

In [ ]:

Copied!





card_c = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  ="DefaultLiteralVivid15",  # ← Mode C — string name
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeC",
    tunnel_status="0",
)
exp_c = ExpCard(card_c)
print("Parser resolved to:", exp_c.parser.__name__)

results_c = ScannerModel(expcard=exp_c).run()
print("First trial response:", results_c[0][0]["pred_resp"])
card_c = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  ="DefaultLiteralVivid15",  # ← Mode C — string name
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeC",
    tunnel_status="0",
)
exp_c = ExpCard(card_c)
print("Parser resolved to:", exp_c.parser.__name__)

results_c = ScannerModel(expcard=exp_c).run()
print("First trial response:", results_c[0][0]["pred_resp"])

Mode D — Direct Pydantic class¶

The cleanest mode when working in code: pass the class object itself.
No string lookup, no JSON, no ambiguity.

In [ ]:

Copied!





from psychscanner.parsers import DefaultLiteralVivid15

card_d = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralVivid15,  # ← Mode D — class object
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeD",
    tunnel_status="0",
)
exp_d = ExpCard(card_d)
print("Parser:", exp_d.parser)

results_d = ScannerModel(expcard=exp_d).run()
print("First trial response:", results_d[0][0]["pred_resp"])
from psychscanner.parsers import DefaultLiteralVivid15

card_d = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralVivid15,  # ← Mode D — class object
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeD",
    tunnel_status="0",
)
exp_d = ExpCard(card_d)
print("Parser:", exp_d.parser)

results_d = ScannerModel(expcard=exp_d).run()
print("First trial response:", results_d[0][0]["pred_resp"])

Mode E — Callable dispatch (replaces old `"dynamic"`)¶

Migration note: The old parser='dynamic' sentinel is gone in ps-parser.
Replace it with a callable that maps trcode → parser class.

When parser is a callable (not a class), TaskRunner calls it once per trial with the trial's trcode string and uses the returned class for that trial's structured output. This gives you arbitrary per-trial routing without any hardcoded LangGraph nodes.

Reality-monitoring example — two parsers depending on trial phase:

In [ ]:

Copied!





from psychscanner.parsers import Response_part_1_rm, Response_part_2_rm

# Form B callable: maps trcode → parser class
# encoding trials (imagined_*, perceived_*)  → Response_part_1_rm  (Word_2 + Rating)
# test trials     (test:imagined_*, …)        → Response_part_2_rm  (Judgment + Confidence)
def rm_dispatch(trcode: str):
    if "test" in trcode:
        return Response_part_2_rm
    return Response_part_1_rm

# Or as a one-liner lambda:
rm_dispatch_lambda = lambda trcode: Response_part_2_rm if "test" in trcode else Response_part_1_rm

# Verify routing:
for tc in ["imagined_1", "perceived_3", "test:imagined_1", "test:perceived_3"]:
    chosen = rm_dispatch(tc)
    print(f"  {tc:<25} → {chosen.__name__}")
from psychscanner.parsers import Response_part_1_rm, Response_part_2_rm

# Form B callable: maps trcode → parser class
# encoding trials (imagined_*, perceived_*)  → Response_part_1_rm  (Word_2 + Rating)
# test trials     (test:imagined_*, …)        → Response_part_2_rm  (Judgment + Confidence)
def rm_dispatch(trcode: str):
    if "test" in trcode:
        return Response_part_2_rm
    return Response_part_1_rm

# Or as a one-liner lambda:
rm_dispatch_lambda = lambda trcode: Response_part_2_rm if "test" in trcode else Response_part_1_rm

# Verify routing:
for tc in ["imagined_1", "perceived_3", "test:imagined_1", "test:perceived_3"]:
    chosen = rm_dispatch(tc)
    print(f"  {tc:<25} → {chosen.__name__}")

In [ ]:

Copied!





# Build a minimal RM task on the fly to demo callable dispatch
rm_task_json = {
    "tasktype" : "survey",
    "taskname" : "rm_callable_dispatch_demo",
    "instructions": {
        "definition": [
            "Encoding: report the second word and rate its relatedness (0–100).",
            "Test: judge whether the word was internal/external and rate confidence (1–6)."
        ]
    },
    "contexts"       : ["encode", "test"],
    "contexts_id"    : ["E", "T"],
    "context_present": True,
    "chain_type"     : "item",
    "parser"         : "0",    # fallback at card level — callable handles each trial
    "items": {
        "encode_1": [{"trcode": "encode_1",   "stimulus": "Word 1: APPLE | Word 2: FRUIT"}],
        "encode_2": [{"trcode": "encode_2",   "stimulus": "Word 1: TABLE | Word 2: ____ (imagine)"}],
        "test_1"  : [{"trcode": "test_1",     "stimulus": "APPLE — was the second word internal or external?"}],
        "test_2"  : [{"trcode": "test_2",     "stimulus": "TABLE — was the second word internal or external?"}],
    },
}

rm_task_path = PROJECT_DIR / "rm_dispatch_demo.json"
rm_task_path.write_text(json.dumps(rm_task_json, indent=2))

card_e = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=rm_task_path,
    parser  =rm_dispatch,     # ← Mode E — callable
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeE_callable",
    tunnel_status="0",
)
exp_e = ExpCard(card_e)
results_e = ScannerModel(expcard=exp_e).run()

print("Results:")
for trial in results_e[0]:
    print(f"  {trial['trcode']:<20} → {trial['pred_resp']}")
# Build a minimal RM task on the fly to demo callable dispatch
rm_task_json = {
    "tasktype" : "survey",
    "taskname" : "rm_callable_dispatch_demo",
    "instructions": {
        "definition": [
            "Encoding: report the second word and rate its relatedness (0–100).",
            "Test: judge whether the word was internal/external and rate confidence (1–6)."
        ]
    },
    "contexts"       : ["encode", "test"],
    "contexts_id"    : ["E", "T"],
    "context_present": True,
    "chain_type"     : "item",
    "parser"         : "0",    # fallback at card level — callable handles each trial
    "items": {
        "encode_1": [{"trcode": "encode_1",   "stimulus": "Word 1: APPLE | Word 2: FRUIT"}],
        "encode_2": [{"trcode": "encode_2",   "stimulus": "Word 1: TABLE | Word 2: ____ (imagine)"}],
        "test_1"  : [{"trcode": "test_1",     "stimulus": "APPLE — was the second word internal or external?"}],
        "test_2"  : [{"trcode": "test_2",     "stimulus": "TABLE — was the second word internal or external?"}],
    },
}

rm_task_path = PROJECT_DIR / "rm_dispatch_demo.json"
rm_task_path.write_text(json.dumps(rm_task_json, indent=2))

card_e = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=rm_task_path,
    parser  =rm_dispatch,     # ← Mode E — callable
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="modeE_callable",
    tunnel_status="0",
)
exp_e = ExpCard(card_e)
results_e = ScannerModel(expcard=exp_e).run()

print("Results:")
for trial in results_e[0]:
    print(f"  {trial['trcode']:<20} → {trial['pred_resp']}")

5. Custom Parsers¶

Any pydantic.BaseModel subclass works as a parser — just pass it as Mode D.
The class docstring becomes the system-level instruction; each Field description guides the model on how to fill that field.

Design tips:

Use Literal[...] for categorical fields (rating scales, source judgments).
Use float with ge= / le= constraints for continuous scales.
Keep description= strings concise but complete — they appear verbatim in the structured-output schema.

In [ ]:

Copied!





from pydantic import BaseModel, Field
from typing import Literal


class SentimentRating(BaseModel):
    """Rate the sentiment of the given statement and your confidence in that rating."""

    sentiment: Literal["positive", "neutral", "negative"] = Field(
        ...,
        description="The overall sentiment of the statement.",
    )
    confidence: Literal[1, 2, 3, 4, 5] = Field(
        ...,
        description=(
            "How confident are you in the sentiment rating? "
            "1 = not at all confident, 5 = very confident."
        ),
    )


# Inspect the schema that will be sent to the model
print(json.dumps(SentimentRating.model_json_schema(), indent=2))
from pydantic import BaseModel, Field
from typing import Literal


class SentimentRating(BaseModel):
    """Rate the sentiment of the given statement and your confidence in that rating."""

    sentiment: Literal["positive", "neutral", "negative"] = Field(
        ...,
        description="The overall sentiment of the statement.",
    )
    confidence: Literal[1, 2, 3, 4, 5] = Field(
        ...,
        description=(
            "How confident are you in the sentiment rating? "
            "1 = not at all confident, 5 = very confident."
        ),
    )


# Inspect the schema that will be sent to the model
print(json.dumps(SentimentRating.model_json_schema(), indent=2))

In [ ]:

Copied!





# A minimal survey task that uses the custom parser
sentiment_task = {
    "tasktype": "survey",
    "taskname": "sentiment_demo",
    "instructions": {"definition": "Rate the sentiment of each statement."},
    "contexts"  : ["stmt"],
    "contexts_id": ["S"],
    "context_present": False,
    "chain_type": "item",
    "parser"    : "0",
    "items": {
        "s1": [{"trcode": "s1", "stimulus": "The experiment results exceeded all expectations."}],
        "s2": [{"trcode": "s2", "stimulus": "The data collection took longer than planned."}],
        "s3": [{"trcode": "s3", "stimulus": "The model failed to converge on most runs."}],
    },
}

sentiment_path = PROJECT_DIR / "sentiment_demo.json"
sentiment_path.write_text(json.dumps(sentiment_task, indent=2))

card_custom = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=sentiment_path,
    parser  =SentimentRating,   # ← custom Pydantic class
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="custom_parser",
    tunnel_status="0",
)
exp_custom = ExpCard(card_custom)
results_custom = ScannerModel(expcard=exp_custom).run()

for trial in results_custom[0]:
    print(f"{trial['trcode']}: {trial['pred_resp']}")
# A minimal survey task that uses the custom parser
sentiment_task = {
    "tasktype": "survey",
    "taskname": "sentiment_demo",
    "instructions": {"definition": "Rate the sentiment of each statement."},
    "contexts"  : ["stmt"],
    "contexts_id": ["S"],
    "context_present": False,
    "chain_type": "item",
    "parser"    : "0",
    "items": {
        "s1": [{"trcode": "s1", "stimulus": "The experiment results exceeded all expectations."}],
        "s2": [{"trcode": "s2", "stimulus": "The data collection took longer than planned."}],
        "s3": [{"trcode": "s3", "stimulus": "The model failed to converge on most runs."}],
    },
}

sentiment_path = PROJECT_DIR / "sentiment_demo.json"
sentiment_path.write_text(json.dumps(sentiment_task, indent=2))

card_custom = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=sentiment_path,
    parser  =SentimentRating,   # ← custom Pydantic class
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="custom_parser",
    tunnel_status="0",
)
exp_custom = ExpCard(card_custom)
results_custom = ScannerModel(expcard=exp_custom).run()

for trial in results_custom[0]:
    print(f"{trial['trcode']}: {trial['pred_resp']}")

6. Per-trial Routing — Form A vs Form B¶

When different trials in the same task need different parsers you have two options.

Form A — `"parser"` key in each trial dict (task JSON)¶

Add a "parser" field to individual trial dicts.
The value is a registered parser class name (string).

"items": {
    "encode_1": [{
        "trcode":  "encode_1",
        "parser": "Response_part_1_rm",
        "stimulus": {...}
    }],
    "test_1": [{
        "trcode":  "test:encode_1",
        "parser": "Response_part_2_rm",
        "stimulus": {...}
    }]
}

Set card.parser = "0" (no card-level default) — the per-trial "parser" field handles routing.

Form B — callable at the card level (§4 Mode E)¶

card.parser = lambda trcode: Response_part_2_rm if "test" in trcode else Response_part_1_rm

Keep the task JSON "parser" field absent or "0". The callable runs on every trial.

Priority rule: per-trial JSON "parser" (Form A) takes precedence over the card-level callable (Form B).
You can mix: use the callable as a fallback and override individual trials via the JSON field.

In [ ]:

Copied!





# Build a Form-A task: each trial carries its own parser name
form_a_task = {
    "tasktype": "survey",
    "taskname": "form_a_demo",
    "instructions": {"definition": "Two-phase RM demo using per-trial parser field."},
    "contexts"  : ["encode", "test"],
    "contexts_id": ["E", "T"],
    "context_present": True,
    "chain_type": "item",
    "parser"    : "0",    # card-level fallback = no parser
    "items": {
        "encode_1": [{
            "trcode"  : "encode_1",
            "parser"  : "Response_part_1_rm",   # ← Form A: per-trial
            "stimulus": {"Word_Pair": {"word_1": "ocean", "word_2": "________"}},
        }],
        "test_1": [{
            "trcode"  : "test:encode_1",
            "parser"  : "Response_part_2_rm",   # ← Form A: per-trial
            "corrAns" : "internal",
            "stimulus": {"Test_Word": "ocean"},
        }],
    },
}

form_a_path = PROJECT_DIR / "form_a_demo.json"
form_a_path.write_text(json.dumps(form_a_task, indent=2))

# Show the routing table
print(f"{'trcode':<22} {'per-trial parser'}")
print("-" * 45)
for key, trials in form_a_task["items"].items():
    t = trials[0]
    print(f"{t['trcode']:<22} {t.get('parser', '(none — uses card default)')}")
# Build a Form-A task: each trial carries its own parser name
form_a_task = {
    "tasktype": "survey",
    "taskname": "form_a_demo",
    "instructions": {"definition": "Two-phase RM demo using per-trial parser field."},
    "contexts"  : ["encode", "test"],
    "contexts_id": ["E", "T"],
    "context_present": True,
    "chain_type": "item",
    "parser"    : "0",    # card-level fallback = no parser
    "items": {
        "encode_1": [{
            "trcode"  : "encode_1",
            "parser"  : "Response_part_1_rm",   # ← Form A: per-trial
            "stimulus": {"Word_Pair": {"word_1": "ocean", "word_2": "________"}},
        }],
        "test_1": [{
            "trcode"  : "test:encode_1",
            "parser"  : "Response_part_2_rm",   # ← Form A: per-trial
            "corrAns" : "internal",
            "stimulus": {"Test_Word": "ocean"},
        }],
    },
}

form_a_path = PROJECT_DIR / "form_a_demo.json"
form_a_path.write_text(json.dumps(form_a_task, indent=2))

# Show the routing table
print(f"{'trcode':<22} {'per-trial parser'}")
print("-" * 45)
for key, trials in form_a_task["items"].items():
    t = trials[0]
    print(f"{t['trcode']:<22} {t.get('parser', '(none — uses card default)')}")

In [ ]:

Copied!





card_fa = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=form_a_path,
    parser  ="0",              # ← card-level: no default; per-trial JSON handles it
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="formA",
    tunnel_status="0",
)
results_fa = ScannerModel(expcard=ExpCard(card_fa)).run()

for trial in results_fa[0]:
    print(f"{trial['trcode']:<22} → {trial['pred_resp']}")
card_fa = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=form_a_path,
    parser  ="0",              # ← card-level: no default; per-trial JSON handles it
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="formA",
    tunnel_status="0",
)
results_fa = ScannerModel(expcard=ExpCard(card_fa)).run()

for trial in results_fa[0]:
    print(f"{trial['trcode']:<22} → {trial['pred_resp']}")

7. Diagnostic Options: `parser_raw` and `parser_config`¶

`parser_raw=True` — keep the raw AIMessage¶

When True, the trial result carries both the parsed dict and the original AIMessage.
Useful for debugging: see exactly what the model emitted before schema coercion.

In [ ]:

Copied!





from psychscanner.parsers import DefaultLiteralAgree

card_raw = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralAgree,
    parser_raw=True,           # ← keep raw AIMessage alongside parsed output
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="parser_raw",
    tunnel_status="0",
)
results_raw = ScannerModel(expcard=ExpCard(card_raw)).run()

t0 = results_raw[0][0]
print("pred_resp keys:", list(t0["pred_resp"].keys()) if isinstance(t0["pred_resp"], dict) else type(t0["pred_resp"]).__name__)
print()
pprint(t0["pred_resp"])
from psychscanner.parsers import DefaultLiteralAgree

card_raw = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralAgree,
    parser_raw=True,           # ← keep raw AIMessage alongside parsed output
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="parser_raw",
    tunnel_status="0",
)
results_raw = ScannerModel(expcard=ExpCard(card_raw)).run()

t0 = results_raw[0][0]
print("pred_resp keys:", list(t0["pred_resp"].keys()) if isinstance(t0["pred_resp"], dict) else type(t0["pred_resp"]).__name__)
print()
pprint(t0["pred_resp"])

`parser_config` — control LangChain's `with_structured_output()`¶

parser_config is forwarded as **kwargs to model.with_structured_output(...).
The default is {"method": "json_schema"}.
Other options: "function_calling", "json_mode" — availability depends on the provider.

Change this when your model/provider doesn't support JSON schema mode.

In [ ]:

Copied!





card_cfg = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralAgree,
    parser_config={"method": "json_schema"},   # ← default; try "function_calling" if supported
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="parser_config",
    tunnel_status="0",
)
results_cfg = ScannerModel(expcard=ExpCard(card_cfg)).run()
print("First trial response:", results_cfg[0][0]["pred_resp"])
card_cfg = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralAgree,
    parser_config={"method": "json_schema"},   # ← default; try "function_calling" if supported
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="parser_config",
    tunnel_status="0",
)
results_cfg = ScannerModel(expcard=ExpCard(card_cfg)).run()
print("First trial response:", results_cfg[0][0]["pred_resp"])

8. Feedback API¶

Trial-level corrective feedback lets the model see a JSON message after each trial before the next one.
The ps-parser branch introduces two improvements:

card.feedback = True (bool instead of the old string '1'; '1' still accepted for backward compat)
card.feedback_fn = MyHandler — pass a class (not an instance); psychscanner creates one instance per participant simulation so cross-trial state stays safe inside self.

Minimal example¶

In [ ]:

Copied!





from psychscanner import FeedbackBase
import json as _json   # avoid shadowing the top-level import


class RatingFeedback(FeedbackBase):
    """Give corrective feedback when the rating is outside the expected range."""

    def __init__(self):
        self.trial_count = 0          # cross-trial state lives safely here

    def on_response(self, trial: dict, response: dict) -> str | None:
        self.trial_count += 1
        rating = response.get("rating") or response.get("Vividness")

        if rating is None:
            return None    # unstructured output — skip feedback

        if int(rating) >= 4:
            feedback = {"feedback": f"Trial {self.trial_count}: rating {rating} — noted (high vividness)."}
        else:
            feedback = {"feedback": f"Trial {self.trial_count}: rating {rating} — remember to use the full scale."}

        return _json.dumps(feedback)


# Verify on a synthetic response
handler = RatingFeedback()
print(handler.on_response({"trcode": "item_1"}, {"Vividness": 2}))
print(handler.on_response({"trcode": "item_2"}, {"Vividness": 5}))
print(f"Trials processed so far: {handler.trial_count}")
from psychscanner import FeedbackBase
import json as _json   # avoid shadowing the top-level import


class RatingFeedback(FeedbackBase):
    """Give corrective feedback when the rating is outside the expected range."""

    def __init__(self):
        self.trial_count = 0          # cross-trial state lives safely here

    def on_response(self, trial: dict, response: dict) -> str | None:
        self.trial_count += 1
        rating = response.get("rating") or response.get("Vividness")

        if rating is None:
            return None    # unstructured output — skip feedback

        if int(rating) >= 4:
            feedback = {"feedback": f"Trial {self.trial_count}: rating {rating} — noted (high vividness)."}
        else:
            feedback = {"feedback": f"Trial {self.trial_count}: rating {rating} — remember to use the full scale."}

        return _json.dumps(feedback)


# Verify on a synthetic response
handler = RatingFeedback()
print(handler.on_response({"trcode": "item_1"}, {"Vividness": 2}))
print(handler.on_response({"trcode": "item_2"}, {"Vividness": 5}))
print(f"Trials processed so far: {handler.trial_count}")

In [ ]:

Copied!





# Wire it into an ExpCard
card_fb = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralAgree,
    feedback   =True,           # ← bool (was '1' in old API)
    feedback_fn=RatingFeedback, # ← the class, not an instance
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="feedback_demo",
    tunnel_status="0",
)
results_fb = ScannerModel(expcard=ExpCard(card_fb)).run()

for trial in results_fb[0][:3]:
    print(f"{trial['trcode']:<20} response={trial['pred_resp']}")
    if trial.get("fb_response"):
        fb = _json.loads(trial["fb_response"])
        print(f"  └─ feedback: {fb.get('feedback')}")
# Wire it into an ExpCard
card_fb = ExpCardInit(
    model   =MODEL_NAME,
    family  =MODEL_FAMILY,
    parameters={"temperature": 0},
    task_file=SURVEY_TASK,
    parser  =DefaultLiteralAgree,
    feedback   =True,           # ← bool (was '1' in old API)
    feedback_fn=RatingFeedback, # ← the class, not an instance
    cogtype ="no",
    nsim    =1,
    proj_dir=PROJECT_DIR,
    projectname="feedback_demo",
    tunnel_status="0",
)
results_fb = ScannerModel(expcard=ExpCard(card_fb)).run()

for trial in results_fb[0][:3]:
    print(f"{trial['trcode']:<20} response={trial['pred_resp']}")
    if trial.get("fb_response"):
        fb = _json.loads(trial["fb_response"])
        print(f"  └─ feedback: {fb.get('feedback')}")

For a full Reality Monitoring + feedback example with per-trial parser routing see 07_rm_feedback_task.ipynb.

FeedbackBase also provides an inject_feedback(input_dict, fb_str) hook you can override to control how the feedback is merged into the next trial's input.

9. Summary¶

Parser mode quick-reference¶

from psychscanner.parsers import DefaultLiteralVivid15, Response_part_1_rm, Response_part_2_rm

# Mode A — no structured output (works with mock-llm)
card.parser = None          # or "0"

# Mode B — resolve class name from task JSON 'parser' field
card.parser = "1"

# Mode C — string name direct registry lookup
card.parser = "DefaultLiteralVivid15"

# Mode D — Pydantic class (recommended when working in code)
card.parser = DefaultLiteralVivid15

# Mode E — callable dispatch; replaces old parser='dynamic'
card.parser = lambda trcode: Response_part_2_rm if "test" in trcode else Response_part_1_rm

# Cross-cutting flags
card.parser_raw    = True                         # keep AIMessage alongside parsed output
card.parser_config = {"method": "json_schema"}    # forwarded to with_structured_output()

Feedback quick-reference¶

from psychscanner import FeedbackBase

class MyFeedback(FeedbackBase):
    def on_response(self, trial, response):
        # return a JSON string or None
        ...

card.feedback    = True          # bool (old '1' still accepted)
card.feedback_fn = MyFeedback    # the class — psychscanner instantiates once per participant

Registry quick-reference¶

from psychscanner.parsers import list_parsers, get_parser, resolve_parser

list_parsers()                          # → sorted list of all registered names
get_parser("DefaultLiteralVivid15")     # → class  (KeyError with suggestions on typo)
resolve_parser("1", task_parser_name=…) # → class  (any spec → class | None)

Module import paths (all equivalent)¶

from psychscanner.parsers import DefaultLiteralVivid15                       # ← recommended
from psychscanner.datasets.prompts.parser_tasks import DefaultLiteralVivid15  # by paradigm
from psychscanner.datasets.prompts.parser import DefaultLiteralVivid15        # legacy shim

Notebook	What it covers
03_parsers.ipynb	Original four-mode walkthrough (updated for ps-parser)
04_parser_modules.ipynb	Module reorganization + registry internals
05_rm_task.ipynb	Full RM word-pair task (no feedback)
06_feedback_api.ipynb	`FeedbackBase` API in depth
07_rm_feedback_task.ipynb	RM + per-trial feedback, full study

PsychScanner Parser System — ps-parser Branch Guide¶

What changed from the archive (old) API¶

Outline¶

1. Setup¶

2. Parser Registry¶

2a. Splitting parsers by paradigm¶

3. Module Layout — Three Import Paths¶

4. Parser Modes in ExpCard¶

Mode A — No parser (None / "0")¶

Mode B — Resolve from task JSON ("1")¶

Mode C — String name ("DefaultLiteralVivid15")¶

Mode D — Direct Pydantic class¶

Mode E — Callable dispatch (replaces old "dynamic")¶

5. Custom Parsers¶

6. Per-trial Routing — Form A vs Form B¶

Form A — "parser" key in each trial dict (task JSON)¶

Form B — callable at the card level (§4 Mode E)¶

7. Diagnostic Options: parser_raw and parser_config¶

parser_raw=True — keep the raw AIMessage¶

parser_config — control LangChain's with_structured_output()¶

8. Feedback API¶

Minimal example¶

9. Summary¶

Parser mode quick-reference¶

Feedback quick-reference¶

Registry quick-reference¶

Module import paths (all equivalent)¶

Related notebooks¶

4. Parser Modes in `ExpCard`¶

Mode A — No parser (`None` / `"0"`)¶

Mode B — Resolve from task JSON (`"1"`)¶

Mode C — String name (`"DefaultLiteralVivid15"`)¶

Mode E — Callable dispatch (replaces old `"dynamic"`)¶

Form A — `"parser"` key in each trial dict (task JSON)¶

7. Diagnostic Options: `parser_raw` and `parser_config`¶

`parser_raw=True` — keep the raw AIMessage¶

`parser_config` — control LangChain's `with_structured_output()`¶