Skip to main content
Enable checkpoints that pause deep research at key decision points, allowing users to review and guide the research process.
For conceptual overview, checkpoint response shapes, and lifecycle details, see the HITL Guide. This page focuses on Python SDK usage.
HITL is only available for individual deep research tasks. It is not available for batch requests.

​
Quick start

from valyu import Valyu
import time

client = Valyu()

# Create a task with all checkpoints enabled
task = client.deepresearch.create(
    query="Analyze the competitive landscape of AI chip manufacturers",
    mode="heavy",
    hitl={
        "planning_questions": True,
        "plan_review": True,
        "source_review": True,
        "outline_review": True,
    },
)

task_id = task.deepresearch_id
print(f"Task created: {task_id}")

​
Enabling HITL

Pass a hitl dict or HitlConfig object to create(). All fields are optional booleans (default False): 
from valyu import Valyu, HitlConfig

client = Valyu()

# Using a dict
task = client.deepresearch.create(
    query="...",
    mode="heavy",
    hitl={
        "planning_questions": True,
        "source_review": True,
    },
)

# Using HitlConfig
task = client.deepresearch.create(
    query="...",
    mode="heavy",
    hitl=HitlConfig(
        planning_questions=True,
        plan_review=True,
        source_review=True,
        outline_review=True,
    ),
)

​
Available checkpoints

CheckpointPhaseWhen it fires
planning_questionsPre-researchBefore research — clarifying questions for the user
plan_reviewPre-researchAfter planning — user reviews the research plan
source_reviewPost-researchAfter research — user filters sources by domain
outline_reviewPost-researchAfter source review — user reviews the report outline
Checkpoints fire in order. Only enabled checkpoints fire.

​
Polling for checkpoints

Poll with status() and check for awaiting_input or paused: 
import time

while True:
    status = client.deepresearch.status(task_id)

    if status.status in ("awaiting_input", "paused"):
        # A checkpoint is waiting for your response
        interaction = status.interaction
        print(f"Checkpoint: {interaction.type}")
        print(f"Data: {interaction.data}")

        # Build and send response (see sections below)
        response = build_response(interaction)
        client.deepresearch.respond(
            task_id,
            interaction_id=interaction.interaction_id,
            response=response,
        )

    elif status.status in ("completed", "failed", "cancelled"):
        break

    time.sleep(5)

​
Status values

StatusMeaning
awaiting_inputCheckpoint active, container holding capacity, fast resume
pausedCheckpoint timed out (5 min), state saved, respond anytime to resume
runningResearch or writing in progress
queuedRe-enqueued after responding to a paused task

​
Responding to checkpoints

​
respond()

result = client.deepresearch.respond(
    task_id,
    interaction_id="int_abc123",
    response={"approved": True},
)
print(result.status)  # "running" or "queued"
Parameters:
ParameterTypeDescription
task_idstrThe deep research task ID
interaction_idstrFrom status.interaction.interaction_id
responsedictResponse data matching the checkpoint type (see below)
Returns: DeepResearchRespondResponse with success, status, deepresearch_id, error.

​
Planning questions response

# interaction.data contains:
# {
#     "questions": [
#         {"question": "What regions?", "context": "..."},
#         {"question": "Specific competitors?"}
#     ]
# }

client.deepresearch.respond(
    task_id,
    interaction_id=interaction.interaction_id,
    response={
        "answers": [
            {"question": "What regions?", "answer": "North America and EU"},
            {"question": "Specific competitors?", "answer": "NVIDIA, AMD, Intel"},
        ]
    },
)

​
Plan review response

# interaction.data contains:
# {
#     "plan": "I'll research this topic by...",
#     "estimated_steps": 15,
#     "research_areas": ["Market size", "Competitive landscape", ...]
# }

# Approve
client.deepresearch.respond(
    task_id,
    interaction_id=interaction.interaction_id,
    response={"approved": True},
)

# Request modifications
client.deepresearch.respond(
    task_id,
    interaction_id=interaction.interaction_id,
    response={
        "approved": False,
        "modifications": "Focus more on supply chain analysis",
    },
)

​
Source review response

# interaction.data contains:
# {
#     "domains": [
#         {
#             "domain": "sec.gov",
#             "source_count": 8,
#             "avg_relevance_score": 0.87,
#             "sources": [{"source_id": 1, "title": "...", "url": "...", "relevance_score": 0.92}],
#             "ai_recommendation": "include"
#         },
#         ...
#     ],
#     "total_sources": 42
# }

client.deepresearch.respond(
    task_id,
    interaction_id=interaction.interaction_id,
    response={
        "included_domains": ["sec.gov", "plos.org"],
        "excluded_domains": ["example.com"],
    },
)
Domains not listed fall back to the AI recommendation. Pass empty arrays [] to accept all AI recommendations.

​
Outline review response

# interaction.data contains:
# {
#     "outline": "1. Introduction\n2. Market Analysis\n...",
#     "sections": [
#         {"title": "Introduction", "description": "...", "estimated_length": "short"},
#         {"title": "Market Analysis", "description": "...", "estimated_length": "long"},
#         ...
#     ]
# }

# Approve
client.deepresearch.respond(
    task_id,
    interaction_id=interaction.interaction_id,
    response={"approved": True},
)

# Request modifications
client.deepresearch.respond(
    task_id,
    interaction_id=interaction.interaction_id,
    response={
        "approved": False,
        "modifications": "Add a regulatory risks section after Market Analysis",
    },
)

​
Complete example

from valyu import Valyu, HitlConfig
import time

client = Valyu()

# Create task with selective checkpoints
task = client.deepresearch.create(
    query="Due diligence report on renewable energy company SunPower",
    mode="heavy",
    hitl=HitlConfig(
        plan_review=True,
        source_review=True,
    ),
)

task_id = task.deepresearch_id
print(f"Task: {task_id}")

while True:
    status = client.deepresearch.status(task_id)
    print(f"Status: {status.status}")

    if status.status in ("awaiting_input", "paused"):
        interaction = status.interaction

        if interaction.type == "plan_review":
            print(f"\nResearch plan:\n{interaction.data['plan']}")
            print(f"Areas: {interaction.data.get('research_areas', [])}")

            # Auto-approve with a tweak
            client.deepresearch.respond(
                task_id,
                interaction_id=interaction.interaction_id,
                response={
                    "approved": False,
                    "modifications": "Include analysis of their patent portfolio",
                },
            )

        elif interaction.type == "source_review":
            print(f"\nFound {interaction.data['total_sources']} sources:")
            for domain in interaction.data["domains"]:
                rec = domain["ai_recommendation"]
                print(f"  {domain['domain']}: {domain['source_count']} sources (AI: {rec})")

            # Accept AI recommendations
            client.deepresearch.respond(
                task_id,
                interaction_id=interaction.interaction_id,
                response={
                    "included_domains": [],
                    "excluded_domains": [],
                },
            )

    elif status.status == "completed":
        print(f"\nReport complete! Cost: ${status.cost}")
        print(status.output[:500])
        break
    elif status.status in ("failed", "cancelled"):
        print(f"Task {status.status}: {status.error}")
        break

    time.sleep(5)

​
HITL history

After checkpoints complete, the hitl_history field on the status response contains a record of each interaction: 
status = client.deepresearch.status(task_id)

if status.hitl_history:
    for entry in status.hitl_history:
        print(f"Checkpoint: {entry.type}")
        print(f"  Auto-continued: {entry.auto_continued}")
        if entry.responded_at:
            response_time = (entry.responded_at - entry.created_at) / 1000
            print(f"  Response time: {response_time:.1f}s")

​
Types reference

TypeDescription
HitlConfigConfiguration with 4 optional boolean fields
InteractionTypeEnum: planning_questions, plan_review, source_review, outline_review
InteractionCheckpoint payload with interaction_id, type, data, created_at, timeout_ms
InteractionHistoryEntryCompleted checkpoint record with auto_continued, responded_at, response
DeepResearchRespondResponseResponse from respond() with success, status, deepresearch_id