How to Extract AI Chat Conversations at Scale

Capturing one AI answer is an API call. Capturing a useful dataset is a pipeline.

At scale, the difficult questions move away from HTTP syntax:

• How do you run the same prompt panel across several AI chat surfaces?
• How do you preserve raw evidence while still producing one analysis schema?
• How do you keep concurrency bounded?
• How do you identify each capture after the answer changes?
• How do you add a new platform without rewriting the collector?

This tutorial builds a production-shaped Python pipeline around the Scrapeless LLM Chat Scraper. It covers ChatGPT, Perplexity, Gemini, Copilot, and Grok through one endpoint while keeping platform-specific fields behind small adapters.

The architecture

The pipeline has four stages:

1. Plan — expand prompts, countries, and engines into explicit jobs.
2. Capture — send each job to the appropriate Scrapeless actor.
3. Persist — write the untouched response as an append-only JSON Lines record.
4. Normalize — map answer and citation fields into one downstream schema.

Do not combine these stages into one large function. Raw capture and normalization have different lifecycles. If an adapter changes, you should be able to rebuild normalized data from stored responses without calling the API again.

One endpoint, several actor contracts

Every LLM actor uses:

POST https://api.scrapeless.com/api/v2/scraper/execute

Authentication uses x-api-token. The request body is always {actor,input}, and the response uses the shared {status, task_id, task_result} envelope.

The differences live inside input and task_result:

Platform	Actor	Extra input	Citation fields
ChatGPT	scraper.chatgpt	none required beyond prompt	content_references[]
Perplexity	scraper.perplexity	web_search	web_results[]
Gemini	scraper.gemini	none required beyond prompt	citations[]
Copilot	scraper.copilot	mode when needed	citations[]
Grok	scraper.grok	reasoning mode	web_search_results[], x_search_results[]

Use the LLM Chat Scraper documentation to confirm actor inputs before changing the job configuration.

Prerequisites

You need Python 3.10 or newer and requests:

python -m pip install requests
export SCRAPELESS_API_KEY=your_api_token_here

The examples write to local JSON Lines files. A production deployment can replace the writer with object storage or a database while keeping the record format.

Step 1: model every capture as a job

An explicit job record prevents hidden defaults.

from dataclasses import asdict, dataclass


@dataclass(frozen=True)
class CaptureJob:
    engine: str
    actor: str
    prompt_id: str
    prompt: str
    country: str
    extra_input: dict


ENGINE_CONFIG = {
    "chatgpt": ("scraper.chatgpt", {}),
    "perplexity": ("scraper.perplexity", {"web_search": True}),
    "gemini": ("scraper.gemini", {}),
    "copilot": ("scraper.copilot", {"mode": "smart"}),
    "grok": ("scraper.grok", {"mode": "MODEL_MODE_FAST"}),
}

prompt_id should be a stable identifier from your own prompt registry. Keep the human-readable prompt as well; the ID tells you which logical prompt was used, and the text proves the exact wording.

Step 2: expand the experiment matrix

Use a generator so the planning stage does not need to hold the entire matrix in memory.

def build_jobs(prompts: list[dict], countries: list[str]):
    for prompt in prompts:
        for country in countries:
            for engine, (actor, extra_input) in ENGINE_CONFIG.items():
                yield CaptureJob(
                    engine=engine,
                    actor=actor,
                    prompt_id=prompt["id"],
                    prompt=prompt["text"],
                    country=country,
                    extra_input=extra_input,
                )

Example input:

PROMPTS = [
    {
        "id": "category-best-tools-v1",
        "text": "What are the best web scraping APIs for JavaScript-heavy sites?",
    },
    {
        "id": "ai-agent-web-data-v1",
        "text": "Which web data tools work well for AI agents?",
    },
]

COUNTRIES = ["US", "DE"]

More prompts, markets, and engines multiply the number of jobs. Compute that number before launching a collection window so cost and storage are visible.

Step 3: write one capture function

The collector should know nothing about citation fields. Its responsibility is transport and evidence preservation.

import os
import time

import requests

ENDPOINT = "https://api.scrapeless.com/api/v2/scraper/execute"
SESSION = requests.Session()
SESSION.headers.update(
    {
        "Content-Type": "application/json",
        "x-api-token": os.environ["SCRAPELESS_API_KEY"],
    }
)


def capture(job: CaptureJob) -> dict:
    request_input = {
        "prompt": job.prompt,
        "country": job.country,
        **job.extra_input,
    }
    started_at = int(time.time())

    response = SESSION.post(
        ENDPOINT,
        json={"actor": job.actor, "input": request_input},
        timeout=300,
    )
    response.raise_for_status()
    payload = response.json()

    return {
        "job": asdict(job),
        "request_input": request_input,
        "captured_at": started_at,
        "status": payload.get("status"),
        "task_id": payload.get("task_id"),
        "task_result": payload.get("task_result") or {},
    }

The returned record includes both the job and the exact request_input. This matters when a platform has a mode or web-search flag that changes the answer behavior.

Step 4: keep concurrency bounded

A fixed-size worker pool gives the pipeline controlled parallelism. Configure the worker count from the environment so the code does not hard-code an account-level assumption.

import os
from concurrent.futures import ThreadPoolExecutor, as_completed

MAX_WORKERS = int(os.environ.get("AI_CAPTURE_WORKERS", "3"))


def run_jobs(jobs):
    with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
        futures = {executor.submit(capture, job): job for job in jobs}

        for future in as_completed(futures):
            job = futures[future]
            try:
                yield {"ok": True, "record": future.result()}
            except Exception as error:
                yield {
                    "ok": False,
                    "job": asdict(job),
                    "error_type": type(error).__name__,
                    "error": str(error),
                }

This design records failed jobs as data. It does not let one exception discard the rest of the collection window.

Use a worker count that matches the capacity available to your account and environment. Bounded concurrency is easier to observe and budget than launching one task per job with no ceiling.

Step 5: use one writer

Several worker threads should not append to the same file independently. Keep file output in the main thread and write each result as it arrives.

import json


def write_results(results, path: str) -> None:
    with open(path, "a", encoding="utf-8") as output:
        for item in results:
            output.write(json.dumps(item, ensure_ascii=False) + "\n")
            output.flush()

Now assemble the capture phase:

if __name__ == "__main__":
    jobs = build_jobs(PROMPTS, COUNTRIES)
    results = run_jobs(jobs)
    write_results(results, "raw_ai_captures.jsonl")

Every line is either a successful raw capture or a structured error record. That makes the run auditable without depending on terminal output.

Step 6: normalize per platform

Normalization belongs in small adapter functions.

from urllib.parse import urlparse


def domain_of(url: str) -> str:
    return urlparse(url).netloc.lower().removeprefix("www.")


def source_record(source: dict) -> dict:
    url = source.get("url") or ""
    return {
        "title": source.get("title") or source.get("name") or "",
        "url": url,
        "domain": domain_of(url),
        "snippet": source.get("snippet") or "",
    }


def normalize_chatgpt(result: dict):
    return result.get("result_text") or "", [
        source_record(item)
        for item in (result.get("content_references") or [])
    ]


def normalize_perplexity(result: dict):
    return result.get("result_text") or "", [
        source_record(item)
        for item in (result.get("web_results") or [])
    ]


def normalize_gemini_or_copilot(result: dict):
    return result.get("result_text") or "", [
        source_record(item)
        for item in (result.get("citations") or [])
    ]


def normalize_grok(result: dict):
    web = result.get("web_search_results") or []
    x_sources = result.get("x_search_results") or []
    return result.get("result_text") or "", [
        source_record(item) for item in (web + x_sources)
    ]

Register them by engine:

ADAPTERS = {
    "chatgpt": normalize_chatgpt,
    "perplexity": normalize_perplexity,
    "gemini": normalize_gemini_or_copilot,
    "copilot": normalize_gemini_or_copilot,
    "grok": normalize_grok,
}


def normalize_capture(raw: dict) -> dict:
    engine = raw["job"]["engine"]
    answer, citations = ADAPTERS[engine](raw["task_result"])

    return {
        "engine": engine,
        "actor": raw["job"]["actor"],
        "prompt_id": raw["job"]["prompt_id"],
        "prompt": raw["job"]["prompt"],
        "country": raw["job"]["country"],
        "captured_at": raw["captured_at"],
        "task_id": raw["task_id"],
        "answer": answer,
        "citations": citations,
    }

The common schema should contain only fields that downstream analysis truly shares. Keep rich platform-only fields in the raw record instead of forcing them into a generic column.

Storage strategy

Use two layers.

Raw layer

Store the complete response plus job and request metadata. Make it append-only. Partition by capture date and, if useful, engine.

Normalized layer

Store one record per answer with a child citation collection. This layer feeds dashboards, mention detection, and comparison jobs.

The raw layer is evidence. The normalized layer is a view. Mixing them makes schema changes expensive.

Operational checks

Before scheduling a large matrix, validate these properties:

• Every successful row has a task_id.
• Every row preserves the exact prompt and country.
• Actor-specific modes are attached to the request record.
• Missing citation arrays become empty lists.
• Raw files remain valid JSON Lines after an interrupted process.
• Normalization is deterministic when run twice on the same raw file.
• No secret appears in logs or stored records.

Scale by partitions

Large monitoring programs are easier to manage when jobs are partitioned by a stable dimension:

• One queue per capture window
• One object prefix per date
• One prompt registry version per experiment
• One normalized partition per engine and country

This structure lets you reprocess a subset without touching the full archive.

The official multi-engine brand pipeline shows how citation fields differ across answer engines. The actor family lives under the Universal Scraping API.

Conclusion

Extracting AI chat conversations at scale is mostly a data-contract problem. Plan explicit jobs, keep the worker pool bounded, preserve each raw response, and normalize with one adapter per platform.

The shared endpoint and top-level envelope remove transport duplication. The adapter layer handles the real differences: ChatGPT references, Perplexity web results, Gemini and Copilot citations, and Grok's separate web and X panels.

Start with a small prompt-country matrix, inspect the raw rows, then move the same functions into your scheduler. You can create an API key from the Scrapeless dashboard.

FAQ

Why use JSON Lines?

Each line is an independent record, which makes append-only writing, streaming, and partial inspection straightforward.

Should all platforms share one response schema?

Use one narrow normalized schema for common analysis fields, but preserve the platform-specific payload in the raw layer.

How many workers should the collector use?

Set a bounded value supported by your account and infrastructure, measure its behavior, and adjust deliberately. Do not create unbounded parallel tasks.

Can the normalizer run separately from collection?

Yes. That is the preferred design. It lets you revise adapters and rebuild derived data without making new API calls.

Is it legal to collect AI chat answers?

The actors collect publicly rendered answer content. Review applicable laws and each platform's terms, and consult counsel before collecting or redistributing data at scale.

---

Scrapeless accesses publicly available data and expects customers to follow applicable laws, platform terms, and privacy requirements. The examples in this article are for technical demonstration and should not be used to collect private, confidential, or restricted information.