DeepResearch Guide

from valyu import Valyu

valyu = Valyu()

# Create a research task
task = valyu.deepresearch.create(
    input="What are the key differences between RAG and fine-tuning for LLMs?",
    model="lite"
)

print(f"Task created: {task.deepresearch_id}")
print(f"Status: {task.status}")

# Wait for the task to complete
result = valyu.deepresearch.wait(
    task.deepresearch_id,
    poll_interval=5,      # Check every 5 seconds
    max_wait_time=900     # Timeout after 15 minutes (lite mode)
)

if result.status == "completed":
    print("Research completed!")
    print(result.output)
    
    # Access sources used
    for source in result.sources:
        print(f"- {source.title}: {source.url}")
    
    # Check costs
    print(f"Total cost: ${result.usage.total_cost:.4f}")

task = valyu.deepresearch.create(input="Research query")

if task.status == "queued":
    print(f"Task queued: {task.message}")
    # Task will start automatically - just wait for it
    
result = valyu.deepresearch.wait(task.deepresearch_id)

# Use heavy mode for complex research
task = valyu.deepresearch.create(
    input="Analyze the competitive landscape of the cloud computing market in 2024",
    model="heavy"
)

task = valyu.deepresearch.create(
    input="Explain quantum computing advancements",
    output_formats=["markdown"]
)

task = valyu.deepresearch.create(
    input="Write a report on renewable energy trends",
    output_formats=["markdown", "pdf"]
)

# After completion, access the PDF URL
result = valyu.deepresearch.wait(task.deepresearch_id)
if result.pdf_url:
    print(f"PDF available at: {result.pdf_url}")

task = valyu.deepresearch.create(
    input="Research competitor pricing in the SaaS market",
    output_formats=[{
        "type": "object",
        "properties": {
            "competitors": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "pricing_model": {"type": "string"},
                        "price_range": {"type": "string"},
                        "key_features": {
                            "type": "array",
                            "items": {"type": "string"}
                        }
                    },
                    "required": ["name", "pricing_model"]
                }
            },
            "market_summary": {"type": "string"},
            "recommendations": {
                "type": "array",
                "items": {"type": "string"}
            }
        },
        "required": ["competitors", "market_summary"]
    }]
)

task = valyu.deepresearch.create(
    input="Latest AI research in healthcare diagnostics",
    model="heavy",
    search={
        "search_type": "all",  # "all", "web", or "proprietary"
        "included_sources": ["pubmed", "arxiv", "nature.com"],
        "start_date": "2024-01-01",
        "end_date": "2025-01-01"
    }
)

import base64

# Read and encode a PDF
with open("report.pdf", "rb") as f:
    pdf_data = base64.b64encode(f.read()).decode()

task = valyu.deepresearch.create(
    input="Summarize the key findings from this report and compare with current market trends",
    model="heavy",
    files=[{
        "data": f"data:application/pdf;base64,{pdf_data}",
        "filename": "report.pdf",
        "mediaType": "application/pdf",
        "context": "Q4 2024 financial report"
    }]
)

task = valyu.deepresearch.create(
    input="Compare the approaches described in these articles",
    urls=[
        "https://example.com/article-1",
        "https://example.com/article-2"
    ]
)

status = valyu.deepresearch.status(task_id)

print(f"Status: {status.status}")
if status.progress:
    print(f"Progress: {status.progress.current_step}/{status.progress.total_steps}")

valyu.deepresearch.update(
    task_id,
    instruction="Focus more on peer-reviewed sources"
)

valyu.deepresearch.cancel(task_id)

valyu.deepresearch.delete(task_id)

tasks = valyu.deepresearch.list(api_key_id="your-api-key-id", limit=50)

for task in tasks.data:
    print(f"{task['query']} - {task['status']}")

task = valyu.deepresearch.create(
    input="Research market trends",
    webhook_url="https://your-app.com/webhooks/deepresearch"
)

# IMPORTANT: Save the secret immediately - it's only returned once
webhook_secret = task.webhook_secret
print(f"Store this secret securely: {webhook_secret}")

import hmac
import hashlib

def verify_webhook(
    payload_body: str,
    signature_header: str,
    timestamp_header: str,
    secret: str
) -> bool:
    """Verify the webhook signature is valid."""
    # Reconstruct the signed payload: timestamp.payload
    signed_payload = f"{timestamp_header}.{payload_body}"
    
    # Generate expected signature
    expected_signature = "sha256=" + hmac.new(
        secret.encode(),
        signed_payload.encode(),
        hashlib.sha256
    ).hexdigest()
    
    # Use constant-time comparison to prevent timing attacks
    return hmac.compare_digest(expected_signature, signature_header)


# Example Flask endpoint
from flask import Flask, request, jsonify

app = Flask(__name__)
WEBHOOK_SECRET = "your-stored-secret"  # Load from secure storage

@app.route("/webhooks/deepresearch", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("X-Webhook-Signature")
    timestamp = request.headers.get("X-Webhook-Timestamp")
    payload = request.get_data(as_text=True)
    
    if not verify_webhook(payload, signature, timestamp, WEBHOOK_SECRET):
        return jsonify({"error": "Invalid signature"}), 401
    
    data = request.json
    
    if data["status"] == "completed":
        print(f"Research completed: {data['deepresearch_id']}")
        print(f"Output: {data['output'][:200]}...")
    elif data["status"] == "failed":
        print(f"Research failed: {data['error']}")
    
    return jsonify({"received": True}), 200

def on_progress(status):
    if status.progress:
        print(f"Step {status.progress.current_step}/{status.progress.total_steps}")
    print(f"Status: {status.status}")

result = valyu.deepresearch.wait(
    task_id,
    poll_interval=5,
    on_progress=on_progress
)

task = valyu.deepresearch.create(input="Research query")

if not task.success:
    print(f"Failed to create task: {task.error}")
else:
    try:
        result = valyu.deepresearch.wait(task.deepresearch_id)
    except TimeoutError:
        print("Task took too long")
        valyu.deepresearch.cancel(task.deepresearch_id)
    except ValueError as e:
        print(f"Task failed: {e}")