feat: Beca designs, runs and analyzes Expected Parrot studies from chat (#109)

docs/subconscious/ep-research-prompt.md

@@ -0,0 +1,9 @@
+Beca is the Subconscious research assistant for surveys and experiments run with Expected Parrot's EDSL through the registered study tools. Beca is an assistant name, never the customer's company or product.
+
+Work in this order, and show each step before the next: (1) restate the decision and the research question in one line; (2) say who should answer (the audience) and what varies (the conditions); (3) build the survey with build_survey, the audience with build_agents and the conditions with build_scenarios; (4) show the design for review with show_flow and show_prompts; (5) run on the free "test" model to check the mechanics; (6) ask before any run on a paid model, after estimate_cost; (7) read results with results_review, results_tally and results_select; (8) draw charts with plot_distribution and plot_means; (9) state the finding, its limit, and the next study.
+
+Design rules from Expected Parrot's own guidance: one idea per question; closed questions get exhaustive options and an opt-out when real; scales carry labels at the ends; a question that depends on an earlier answer pipes it with {{ name.answer }} or uses targeted memory; conditions go in scenarios with {{ scenario.field }}, not into separate surveys; audiences go in agents as traits with a codebook when keys are terse. Pilot small before scaling: four agents by two scenarios is a pilot.
+
+Results from simulated respondents are simulated evidence, not human data; say so with every finding. A test-model run proves the mechanics only, never a result. Never state a number that a tool did not return. Keep every chart's numbers in the reply next to the image. Do not invent tool calls, results, files or costs. If a tool returns status "error", read its suggestion and act on it or tell the user the exact blocker.
+
+Keep replies concise: the step done, what the user should check, and the next step. No filler and no em dashes.

docs/subconscious/ep-studies.md

@@ -0,0 +1,61 @@
+# Expected Parrot studies from the Beca chat
+
+Plan and progress: [onyx #109](https://github.com/Subconscious-ai/onyx/issues/109).
+
+Beca designs, runs and analyzes surveys and experiments with the open-source
+Expected Parrot library (EDSL, MIT) through one MCP server,
+`scripts/subconscious/serve_ep.py`. It wraps EDSL and Expected Parrot's public
+chart assets (`plot_style.py` and the standard plots that ship inside the
+`edsl` package). Charts return as MCP image content, which native Onyx saves
+into the conversation. Study objects live as JSON under a workspace directory
+and are addressed by name, so a later turn can reuse what an earlier turn built.
+
+## Run the wrapper
+
+```bash
+uv venv ~/ep-venv --python 3.12
+~/ep-venv/bin/uv pip install -r scripts/subconscious/ep-requirements.txt
+EP_MCP_TOKEN=<32+ random characters> ~/ep-venv/bin/python \
+  scripts/subconscious/serve_ep.py --workspace ~/ep-workspace --host 127.0.0.1 --port 3194
+```
+
+Tests use EDSL's keyless `test` model and spend nothing:
+
+```bash
+cd scripts/subconscious && ~/ep-venv/bin/python -m pytest test_serve_ep.py -q
+```
+
+## Register it in Onyx
+
+An administrator registers the server under Actions, MCP servers: URL
+`http://<host>:3194/mcp`, authentication API token, performer admin, the
+`EP_MCP_TOKEN` value as the token. Discover tools, then attach them to a
+persona whose system prompt is `docs/subconscious/ep-research-prompt.md`.
+Onyx's outbound guard blocks loopback and private addresses unless the
+administrator's SSRF protection setting allows them; the agent-dev stack
+reaches other private tailnet hosts, so it allows private addresses.
+
+Paid models need provider keys in the wrapper's environment (`OPENAI_API_KEY`,
+`GEMINI_API_KEY`, or AWS credentials for Bedrock). Without them only the
+`test` model runs. The persona prompt makes Beca call `estimate_cost` and ask
+before any paid run.
+
+## Demos
+
+`scripts/subconscious/ep_demo.py` drives a chat through Onyx's API, one
+Expected Parrot example per demo, and saves the transcript, the tool calls and
+every returned image under `~/ep-demos/<demo>/`. That folder is the proof that
+the chat reproduced the example.
+
+```bash
+ONYX_DEMO_EMAIL=... ONYX_DEMO_PASSWORD=... ~/ep-venv/bin/python \
+  scripts/subconscious/ep_demo.py --base http://localhost:8088 --persona <id> --demo quickstart
+```
+
+## What Expected Parrot does not provide
+
+No chart library: its hosted agent writes matplotlib scripts, which the
+wrapper reproduces. No census-weighted population builder: audiences are
+trait lists, drafted personas, or imported files. Human recruiting goes
+through `humanize` (a web survey link) or Prolific, both of which need an
+Expected Parrot account.

scripts/subconscious/ep-requirements.txt

@@ -0,0 +1,6 @@
+edsl[viz]==1.0.9
+matplotlib>=3.9
+fastmcp>=4.1,<5
+pytest>=8
+requests>=2.31
+pandas>=2.2

scripts/subconscious/ep_conjoint.py

@@ -0,0 +1,307 @@
+"""Choice-based conjoint for the Expected Parrot wrapper.
+
+The method follows Expected Parrot's public ``conjoint-study`` skill: a
+balanced set of choice tasks found by random search, one multiple-choice
+question per task with static option labels, one scenario per design version,
+counting-based part-worths (``log(choice share / expected share)``,
+zero-centred within an attribute), importance as an attribute's utility range
+over the sum of ranges, and logit choice shares for a market simulation. The
+estimator is deliberately simple; it is a drafting tool, not Rehoboam's
+hierarchical Bayes.
+"""
+
+from __future__ import annotations
+
+import itertools
+import math
+import random
+from collections import Counter, defaultdict
+from typing import Any
+
+OPTION_LETTERS = "abcdefgh"
+
+
+def all_profiles(attributes: dict[str, list[str]]) -> list[dict[str, str]]:
+    names = list(attributes)
+    return [
+        dict(zip(names, combo, strict=True))
+        for combo in itertools.product(*(attributes[n] for n in names))
+    ]
+
+
+def _differences(left: dict[str, str], right: dict[str, str]) -> int:
+    return sum(left[key] != right[key] for key in left)
+
+
+def balance_score(
+    choice_sets: list[list[dict[str, str]]], attributes: dict[str, list[str]]
+) -> float:
+    """Chi-squared-like spread of levels over all shown profiles; lower is better."""
+    score = 0.0
+    for name, levels in attributes.items():
+        counts: Counter[str] = Counter(
+            profile[name] for task in choice_sets for profile in task
+        )
+        total = sum(counts.values())
+        expected = total / len(levels) if levels else 1
+        if expected:
+            score += sum(
+                (counts.get(level, 0) - expected) ** 2 / expected for level in levels
+            )
+    return score
+
+
+def one_version(
+    profiles: list[dict[str, str]],
+    attributes: dict[str, list[str]],
+    n_tasks: int,
+    per_task: int,
+    min_diff: int,
+    rng: random.Random,
+    iterations: int = 300,
+) -> tuple[list[list[dict[str, str]]], float]:
+    best: list[list[dict[str, str]]] = []
+    best_score = math.inf
+    for _ in range(iterations):
+        pool = list(profiles)
+        rng.shuffle(pool)
+        tasks = []
+        for _ in range(n_tasks):
+            if not pool:
+                pool = list(profiles)
+                rng.shuffle(pool)
+            task = [pool.pop()]
+            candidates = [p for p in profiles if p not in task]
+            rng.shuffle(candidates)
+            for candidate in candidates:
+                if len(task) >= per_task:
+                    break
+                if all(_differences(candidate, chosen) >= min_diff for chosen in task):
+                    task.append(candidate)
+            for candidate in candidates:
+                if len(task) >= per_task:
+                    break
+                if candidate not in task:
+                    task.append(candidate)
+            tasks.append(task)
+        score = balance_score(tasks, attributes)
+        if score < best_score:
+            best, best_score = tasks, score
+    return best, best_score
+
+
+def generate_design(
+    attributes: dict[str, list[str]],
+    tasks_per_version: int = 8,
+    profiles_per_task: int = 3,
+    n_versions: int = 4,
+    include_none: bool = False,
+    min_attribute_diff: int = 2,
+    seed: int = 42,
+) -> dict[str, Any]:
+    if len(attributes) < 2 or any(len(levels) < 2 for levels in attributes.values()):
+        raise ValueError(
+            "A conjoint needs at least two attributes with two levels each."
+        )
+    if profiles_per_task > len(OPTION_LETTERS):
+        raise ValueError(f"At most {len(OPTION_LETTERS)} profiles per task.")
+    profiles = all_profiles(attributes)
+    if len(profiles) < profiles_per_task:
+        raise ValueError(
+            f"Only {len(profiles)} unique profiles, but {profiles_per_task} are needed per task."
+        )
+    min_diff = min(min_attribute_diff, max(1, len(attributes) - 1))
+    versions = []
+    for index in range(n_versions):
+        rng = random.Random(seed + index)
+        tasks, score = one_version(
+            profiles, attributes, tasks_per_version, profiles_per_task, min_diff, rng
+        )
+        for task in tasks:
+            rng.shuffle(task)
+        versions.append(
+            {
+                "version": index + 1,
+                "balance_score": round(score, 4),
+                "choice_sets": tasks,
+            }
+        )
+    return {
+        "attributes": attributes,
+        "n_tasks": tasks_per_version,
+        "profiles_per_task": profiles_per_task,
+        "n_versions": n_versions,
+        "include_none": include_none,
+        "total_profiles": len(profiles),
+        "versions": versions,
+    }
+
+
+def describe(profile: dict[str, str]) -> str:
+    return ", ".join(f"{attribute}: {level}" for attribute, level in profile.items())
+
+
+def question_specs(design: dict[str, Any], intro: str) -> list[dict[str, Any]]:
+    """One multiple-choice question per task, options as static letters."""
+    letters = OPTION_LETTERS[: design["profiles_per_task"]]
+    options = [f"Option {letter.upper()}" for letter in letters]
+    if design["include_none"]:
+        options.append("None of these")
+    specs = []
+    for task in range(1, design["n_tasks"] + 1):
+        lines = [
+            f"Option {letter.upper()}: {{{{ scenario.task_{task}_opt_{letter} }}}}"
+            for letter in letters
+        ]
+        specs.append(
+            {
+                "question_type": "multiple_choice",
+                "question_name": f"choice_task_{task}",
+                "question_text": intro
+                + "\n\n"
+                + "\n".join(lines)
+                + "\n\nWhich option would you choose?",
+                "question_options": options,
+            }
+        )
+    return specs
+
+
+def scenario_rows(design: dict[str, Any]) -> list[dict[str, Any]]:
+    """One scenario per design version, with the formatted text and each level."""
+    rows = []
+    for version in design["versions"]:
+        row: dict[str, Any] = {"design_version": version["version"]}
+        for task_index, task in enumerate(version["choice_sets"], start=1):
+            for option_index, profile in enumerate(task):
+                letter = OPTION_LETTERS[option_index]
+                row[f"task_{task_index}_opt_{letter}"] = describe(profile)
+                for attribute, level in profile.items():
+                    row[f"task_{task_index}_opt_{letter}_{attribute}"] = level
+        rows.append(row)
+    return rows
+
+
+def choice_records(
+    rows: list[dict[str, Any]], design: dict[str, Any]
+) -> list[dict[str, Any]]:
+    """Map each answered task back to the chosen profile and the agent's traits."""
+    attributes = design["attributes"]
+    records = []
+    for row in rows:
+        traits = {
+            k.split(".", 1)[1]: v
+            for k, v in row.items()
+            if k.startswith("agent.") and not k.startswith("agent.agent_")
+        }
+        for task in range(1, design["n_tasks"] + 1):
+            answer = row.get(f"answer.choice_task_{task}")
+            if not isinstance(answer, str):
+                continue
+            if answer.lower().startswith("none"):
+                records.append({"task": task, "chosen": None, "traits": traits})
+                continue
+            letter = answer.strip()[-1].lower()
+            if letter not in OPTION_LETTERS[: design["profiles_per_task"]]:
+                continue
+            chosen = {
+                attribute: row.get(f"scenario.task_{task}_opt_{letter}_{attribute}")
+                for attribute in attributes
+            }
+            if all(level is not None for level in chosen.values()):
+                records.append({"task": task, "chosen": chosen, "traits": traits})
+    return records
+
+
+def part_worths(
+    records: list[dict[str, Any]], attributes: dict[str, list[str]]
+) -> dict[str, dict[str, float]]:
+    chosen_counts: dict[str, Counter[str]] = {name: Counter() for name in attributes}
+    total = 0
+    for record in records:
+        if record["chosen"] is None:
+            continue
+        total += 1
+        for name in attributes:
+            chosen_counts[name][record["chosen"][name]] += 1
+    utilities: dict[str, dict[str, float]] = {}
+    for name, levels in attributes.items():
+        expected = 1.0 / len(levels)
+        values = {}
+        for level in levels:
+            count = chosen_counts[name].get(level, 0)
+            values[level] = (
+                math.log((count / total) / expected) if total and count else -2.0
+            )
+        mean = sum(values.values()) / len(values)
+        utilities[name] = {level: value - mean for level, value in values.items()}
+    return utilities
+
+
+def importance(utilities: dict[str, dict[str, float]]) -> dict[str, float]:
+    ranges = {
+        name: (max(v.values()) - min(v.values())) if v else 0.0
+        for name, v in utilities.items()
+    }
+    total = sum(ranges.values())
+    return {name: (r / total * 100 if total else 0.0) for name, r in ranges.items()}
+
+
+def analyze(rows: list[dict[str, Any]], design: dict[str, Any]) -> dict[str, Any]:
+    records = choice_records(rows, design)
+    if not records:
+        raise ValueError("No answered choice tasks in these results.")
+    attributes = design["attributes"]
+    utilities = part_worths(records, attributes)
+    segments: dict[str, dict[str, Any]] = {}
+    for trait in sorted({key for record in records for key in record["traits"]}):
+        groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
+        for record in records:
+            groups[str(record["traits"].get(trait, "unknown"))].append(record)
+        segments[trait] = {}
+        for value, group in sorted(groups.items()):
+            group_utilities = part_worths(group, attributes)
+            segments[trait][value] = {
+                "utilities": group_utilities,
+                "importance": importance(group_utilities),
+                "n_observations": len(group),
+            }
+    none_count = sum(record["chosen"] is None for record in records)
+    return {
+        "n_observations": len(records),
+        "none_chosen": none_count,
+        "utilities": utilities,
+        "importance": importance(utilities),
+        "segments": segments,
+        "position_counts": dict(
+            Counter(
+                row.get(f"answer.choice_task_{task}")
+                for row in rows
+                for task in range(1, design["n_tasks"] + 1)
+                if isinstance(row.get(f"answer.choice_task_{task}"), str)
+            )
+        ),
+    }
+
+
+def market_shares(
+    utilities: dict[str, dict[str, float]], profiles: list[dict[str, str]]
+) -> list[dict[str, Any]]:
+    totals = [
+        sum(
+            utilities.get(attribute, {}).get(level, 0.0)
+            for attribute, level in profile.items()
+        )
+        for profile in profiles
+    ]
+    peak = max(totals) if totals else 0.0
+    weights = [math.exp(t - peak) for t in totals]
+    denominator = sum(weights) or 1.0
+    return [
+        {
+            "profile": profile,
+            "utility": round(total, 4),
+            "share": round(weight / denominator, 4),
+        }
+        for profile, total, weight in zip(profiles, totals, weights, strict=True)
+    ]