Deep Research Workflows

pip install perplexityai

from perplexity import Perplexity

client = Perplexity()

response = client.responses.create(
    preset="deep-research",
    input=(
        "Provide a comprehensive analysis of the current state of nuclear fusion research. "
        "Cover the main approaches (tokamak, stellarator, inertial confinement, laser-driven), "
        "key milestones achieved in the past 2 years, major private companies involved, "
        "and realistic timelines for commercial fusion power."
    ),
)

print(f"Model: {response.model}")
print(f"\n{response.output_text}")

from perplexity import Perplexity

client = Perplexity()


def deep_research(query: str) -> dict:
    """Run a deep research query and extract structured results."""
    print(f"Researching: {query[:80]}...")

    response = client.responses.create(
        preset="deep-research",
        input=query,
    )

    content = response.output_text
    usage = response.usage

    return {
        "content": content,
        "model": response.model,
        "tokens": {
            "input": usage.input_tokens if usage else 0,
            "output": usage.output_tokens if usage else 0,
        },
        "word_count": len(content.split()),
    }


if __name__ == "__main__":
    output = deep_research(
        "What is the current state of solid-state battery technology? "
        "Cover the leading companies, technical challenges remaining, "
        "and expected timeline for mass production in EVs."
    )
    print(f"\nModel: {output['model']}")
    print(f"Words: {output['word_count']}")
    print(f"Tokens: {output['tokens']['input']} in, {output['tokens']['output']} out")
    print(f"\n{'='*60}\n")
    print(output["content"][:2000])

from perplexity import Perplexity

client = Perplexity()

# Deep research restricted to government and academic sources
response = client.responses.create(
    model="openai/gpt-5.2",
    input=(
        "Analyze the current regulatory landscape for AI in healthcare. "
        "Cover FDA guidance, EU AI Act implications, and recent enforcement actions."
    ),
    tools=[{
        "type": "web_search",
        "filters": {
            "search_domain_filter": [".gov", ".europa.eu", "who.int", "nature.com", ".edu"],
        },
    }],
    instructions=(
        "Conduct thorough research using only government and academic sources. "
        "Provide specific regulatory references, dates, and policy details."
    ),
)

print(response.output_text)

import asyncio
import time
from perplexity import AsyncPerplexity


async def single_research(client: AsyncPerplexity, query: str) -> dict:
    """Run a single deep research query."""
    start = time.time()
    try:
        response = await client.responses.create(
            preset="deep-research",
            input=query,
        )
        return {
            "query": query,
            "content": response.output_text,
            "model": response.model,
            "elapsed": time.time() - start,
        }
    except Exception as e:
        return {"query": query, "error": str(e), "elapsed": time.time() - start}


async def batch_research(queries: list[str], max_concurrent: int = 3) -> list[dict]:
    """Run multiple deep research queries with concurrency limits."""
    semaphore = asyncio.Semaphore(max_concurrent)

    async def limited_research(client, query):
        async with semaphore:
            return await single_research(client, query)

    async with AsyncPerplexity() as client:
        tasks = [limited_research(client, q) for q in queries]
        return await asyncio.gather(*tasks)


if __name__ == "__main__":
    queries = [
        "What are the latest advances in room-temperature superconductors?",
        "What is the current state of quantum error correction?",
        "What are the most promising approaches to carbon capture and storage?",
    ]

    print(f"Starting batch research: {len(queries)} queries\n")
    results = asyncio.run(batch_research(queries, max_concurrent=3))

    for r in results:
        status = "OK" if "content" in r else f"FAILED ({r.get('error')})"
        word_count = len(r.get("content", "").split()) if "content" in r else 0
        print(f"  [{r['elapsed']:.0f}s] {r['query'][:60]}... → {status} ({word_count} words)")