Metadata-Version: 2.4
Name: personasurvey
Version: 0.2.23
Summary: Reusable async persona survey simulations
License-Expression: MIT
Project-URL: Homepage, https://github.com/Japulgarin/personasurvey
Project-URL: Repository, https://github.com/Japulgarin/personasurvey
Project-URL: Issues, https://github.com/Japulgarin/personasurvey/issues
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Science/Research
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: pandas>=2.0
Requires-Dist: httpx>=0.25
Requires-Dist: openai>=1.0
Requires-Dist: tqdm>=4.0
Provides-Extra: gemini
Requires-Dist: google-genai>=1.0; extra == "gemini"
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.40; extra == "anthropic"
Provides-Extra: dev
Requires-Dist: pytest>=8.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23; extra == "dev"
Provides-Extra: analytics
Requires-Dist: duckdb>=1.0; extra == "analytics"
Provides-Extra: runpod
Requires-Dist: runpod>=1.7; extra == "runpod"
Dynamic: license-file

# personasurvey

Run persona surveys from a Pandas DataFrame. It creates prompts, sends async requests, saves JSONL checkpoints, resumes safely, and compares prompt experiments.

## Optional Telegram progress updates

Create a bot in Telegram with **@BotFather** using `/newbot`, follow its naming
instructions, and save the bot token. Open your new bot and press **Start**.
To find your chat ID, run this once locally after setting the token:

```python
import os
import httpx

os.environ["TELEGRAM_BOT_TOKEN"] = "your-bot-token"
response = httpx.get(
    f"https://api.telegram.org/bot{os.environ['TELEGRAM_BOT_TOKEN']}/getUpdates",
    timeout=10,
)
response.raise_for_status()
for update in response.json()["result"]:
    if "message" in update:
        print(update["message"]["chat"]["id"])
```

Set `TELEGRAM_CHAT_ID` to your chat's printed ID. If the result is empty, send
your bot another message and run the lookup again. Keep the real token out of
shared notebooks and source control; environment variables can also be set in
your shell or notebook's secret storage.

```python
os.environ["TELEGRAM_CHAT_ID"] = "your-chat-id"

run = run_simulation(
    df=df,
    prompt=prompt,
    provider=provider,
    checkpoint="experiment.jsonl",
    telegram_interval_minutes=5,
)
```

The same option works with `continue_simulation(...)`. Omit it or pass `None`
to disable notifications. Intervals must be finite positive numbers; credentials
are required only when enabled. No additional dependency or server is needed.

The bot sends a start message with the checkpoint name, the planned number of
LLM calls, and the configured endpoints. It sends a fresh progress message every
configured interval, then a final result message with Candidate A/B averages,
winner counts, and ties. The start and final formats are:

```
START
Simulation: usa_persona
LLM calls: 5,000
Endpoints: https://pod-1.example/v1, https://pod-2.example/v1

RESULTS
Candidate A average: 50
Candidate B average: 50
Candidate A wins: 2,400
Candidate B wins: 2,400
Ties: 200
```

Progress updates include processed and successful people, errors, resumed
successes, elapsed time, and approximate ETA. ETA uses work processed in the current run.
Notifications work with `show_progress=False` and contain no prompts or responses.
Telegram delivery failures only produce a sanitized warning; requests have a
10-second timeout. Start and final delivery can each add that much latency;
periodic delivery runs in the background. Abrupt process termination or loss of
connectivity can prevent a final message.

Telegram reference: [Bot API](https://core.telegram.org/bots/api#getupdates).

## Install

```bash
pip install personasurvey
pip install "personasurvey[analytics]"
```

## 1. Load people

```python
import pandas as pd

df = pd.read_csv("people.csv")
df["row_id"] = range(1, len(df) + 1)
df.head()
```

## 2. Create a prompt

```python
from personasurvey import PersonaPrompt, ResponseSchema

prompt = PersonaPrompt(
    name="usa_persona",
    demographic_cols=["state", "city", "age", "education_level", "occupation"],
    persona_detail_cols=["persona"],
    scenario="""Candidate A supports local health clinics.
Candidate B supports lower prescription prices.
How would this person divide support?""",
    response_schema=ResponseSchema.probabilities(["Candidate A", "Candidate B"]),
)

prompt.validate(df)
print(prompt.preview(df.iloc[0]))
```

For a vLLM prefix-cache experiment, create the same prompt with `persona_at_end=True`. All person-specific fields (`DEMOGRAPHICS` and `PERSONA DETAILS`) move after the static context and JSON instructions.

## 3. Choose a provider

```python
from personasurvey import RunPodProvider

provider = RunPodProvider(
    urls=[
        "https://your-pod-1.proxy.runpod.net/",
        "https://your-pod-2.proxy.runpod.net/",
    ],
    model="openai/gpt-oss-20b",
    api_key="EMPTY",
)
```

### Four or more RunPod pods

`more_than_2_pods` is the only extra option. It is `False` by default, so the
normal one- or two-pod call remains unchanged. With three or more URLs, set it
to `True` and the package selects its tested native-batch profile internally.

```python
POD_URLS = ["pod-1-url", "pod-2-url", "pod-3-url", "pod-4-url"]

provider = RunPodProvider(
    urls=POD_URLS,
    model="openai/gpt-oss-20b",
    api_key="EMPTY",
)

run = run_simulation(
    df=df,
    prompt=prompt,
    provider=provider,
    checkpoint="usa_persona.jsonl",
    max_tokens=800,
    reasoning_effort="low",
    temperature=0,
    json_mode=True,
    more_than_2_pods=True,
    retries=3,
    timeout=180,
)
```

Internally this uses batches of four, 24 batch requests per pod, a shared
queue, and a final retry batch containing only invalid people. For one or two
pods, omit `more_than_2_pods` entirely. When it is `True`, the profile also
overrides an old `concurrency=...` value so the call can otherwise stay intact.

```python
from personasurvey import OpenAIProvider, OpenRouterProvider, GeminiProvider, AnthropicProvider

local = OpenAIProvider(base_url="http://127.0.0.1:1234/v1", model="google/gemma-4-e2b", api_key="lm-studio")
openrouter = OpenRouterProvider(model="openai/gpt-oss-20b")
openai = OpenAIProvider(model="gpt-4.1-mini")
gemini = GeminiProvider(model="gemini-2.5-flash")
claude = AnthropicProvider(model="claude-sonnet-4-5")
```

## 4. Preflight, then run

Build a `Simulation` once with your config, call `.test(...)` on a small
sample to check success rate and project full-run time/cost, then `.run(...)`
the same config for the full, checkpointed experiment.

```python
from personasurvey import Simulation, GPUPricing

sim = Simulation(
    df=df,
    prompt=prompt,
    provider=provider,
    id_col="row_id",
    retries=3,
    timeout=180,
    stall_timeout=60,
)

preflight = sim.test(sample_n=100, pricing={"RunPod": GPUPricing(default_per_hour=1.19)})
print(preflight.stats)  # success rate, projected time/tokens, and cost per pricing option

run = sim.run(checkpoint="usa_persona.jsonl", n=5_000)
```

`.test(...)` never writes a checkpoint: it samples `df`, runs the sample,
and scales the result to `len(df)`. Any argument `run_simulation`/
`test_simulation` accept can be overridden per call, e.g. `sim.run(checkpoint=..., concurrency=800)`.

`run_simulation(...)` and `test_simulation(...)` remain available directly
if you don't need to share one configuration between a preflight and the
full run:

```python
from personasurvey import run_simulation

run = run_simulation(
    df=df,
    prompt=prompt,
    provider=provider,
    id_col="row_id",
    checkpoint="usa_persona.jsonl",
    n=5_000,
    # These are the tuned RunPod defaults: 400 direct requests per pod,
    # shared queue, max_tokens=800, reasoning_effort="low", temperature=0,
    # and JSON mode. Pass an argument only to override a default.
    retries=3,
    timeout=180,
    stall_timeout=60,
)

print(run.summary)
display(run.results.head())
```

With two endpoints, the RunPod defaults use 400 direct requests per pod (800
total). The shared queue lets either healthy Pod take the next person, while
the progress bar reports global requests per second and an ETA.

## 5. Read one result

```python
row = run.results.iloc[0]

print(row["input"])
print(row["output"])
print(row[["Candidate A", "Candidate B", "reason", "status", "in_tok", "out_tok"]])
```

## Run output and console monitoring

`run_simulation(...)` returns a `SimulationResult`. Its three main attributes are:

| Attribute | Description |
|---|---|
| `run.results` | One row per person with the model response, parsed fields, status, endpoint, tokens, latency, and any error. |
| `run.merged` | The original DataFrame with the simulation result columns attached, ready for analysis or export. |
| `run.summary` | Global totals for successful and failed requests, elapsed time, RPS, average latency, token counts, and optional cost. |

By default, `run_simulation(...)` displays one `tqdm` progress bar and one
final summary. It never writes periodic per-pod lines, so the bar stays clean
during long RunPod requests. The bar shows global RPS and completed persons per
pod; HTTP client request logs are hidden by default. Set `show_progress=False`
for fully silent runs, or `quiet_http_logs=False` if you need HTTP debug logs.

## Stop selected RunPod Pods after experiments

Install the optional control dependency with `pip install "personasurvey[runpod]"`.
Set `RUNPOD_API_KEY` and `RUNPOD_POD_IDS` before starting Jupyter. The latter is
a comma-separated list of the exact Pod IDs used for the experiment; it is not
derived from proxy URLs, and the package never stops every Pod in your account.

```python
from personasurvey import RunPodPodController

# Call this only after every planned run_simulation(...) call has succeeded.
RunPodPodController.from_environment().stop()
```

This uses the official RunPod Python SDK, whose Pod-control functions already
wrap RunPod's GraphQL API. GraphQL is appropriate for the infrequent lifecycle
operation; the high-volume model requests continue through the Pods' vLLM HTTP
endpoints.

## 6. Continue

```python
from personasurvey import continue_simulation

run = continue_simulation(
    df=df,
    prompt=prompt,
    provider=provider,
    id_col="row_id",
    checkpoint="usa_persona.jsonl",
)
```

Successful IDs are skipped. Errors and pending rows are retried.

## Checkpoints on Google Drive (Colab)

Mounted Google Drive only uploads a file after it is closed, so a long run
would otherwise stay on the Colab VM until it finishes. Pass `colab=True` and
put the checkpoint inside the mounted Drive folder:

```python
from google.colab import drive
drive.mount("/content/drive")
WORKDIR = "/content/drive/MyDrive/thesis"

run = sim.run(checkpoint=f"{WORKDIR}/usa_persona.jsonl", colab=True)
```

The checkpoint is fsynced and closed every 30 seconds, so Drive keeps up with
the run. A warning is printed if the path is outside `/content/drive/`. The same
option works with `run_simulation(...)` and `continue_simulation(...)`.

## Compare prompts

```python
from personasurvey import compare

comparison = compare(
    df,
    id_col="row_id",
    cultural="usa_cultural.jsonl",
    persona="usa_persona.jsonl",
)

display(comparison.data.head())
display(comparison.summary())
display(comparison.by("education_level"))
display(comparison.flips())
display(comparison.change_of_mind(n=10))
```

`compare()` detects numeric candidate fields, joins original population columns, creates winner columns, and never sends a model request.

```python
comparison.sql("""
    SELECT state,
           COUNT(*) AS people,
           AVG("Candidate A__persona" - "Candidate A__cultural") AS prompt_effect
    FROM comparison
    GROUP BY state
    ORDER BY prompt_effect DESC
""")
```

## Functions

| Function | Use |
|---|---|
| `PersonaPrompt(...)` | Define columns, scenario, and JSON response. |
| `prompt.preview(row)` | Print one exact prompt before running. |
| `Simulation(df=..., prompt=..., provider=..., ...)` | Store shared config once for `.test(...)` and `.run(...)`. |
| `sim.test(sample_n=...)` / `test_simulation(...)` | Dry-run a sample; project time, tokens, and cost. |
| `sim.run(checkpoint=...)` / `run_simulation(...)` | Start a checkpointed experiment. |
| `continue_simulation(...)` | Resume the same JSONL checkpoint. |
| `compare(df, cultural=..., persona=...)` | Compare existing prompt experiments. |
| `comparison.by("state")` | Results by any population column. |
| `comparison.flips()` | People whose predicted winner changed. |
| `comparison.change_of_mind()` | Person, probabilities, and both model reasons. |
| `comparison.sql(...)` | Fast DuckDB query over the comparison. |

Examples: `examples/walkthrough.ipynb` and `examples/fictional_personas_100.csv`.
