Using Research Plans

The Workflows API behavior changes based on whether you provide a research_plan in your template. This gives you control over how structured or exploratory your research workflow should be.

How Research Plans Affect Execution

Include `research_plan`?	Behavior
No	Agent dynamically determines research steps based on your prompt
Yes	Agent follows your predefined steps for structured, predictable output

Without a Research Plan

When no research plan is provided, the agent operates dynamically:

Agent analyzes the prompt and creates a research strategy on the fly
Multiple research iterations may occur as the agent explores the topic
Agent decides when sufficient information has been gathered
More exploratory, suitable for open-ended questions

Example: Dynamic Research

import requests
import json
from dotenv import load_dotenv
import os

load_dotenv()
my_api_key = os.getenv('BIGDATA_API_KEY')

# No research_plan - agent determines steps dynamically
payload = {
    "template": {
        "name": "Open-ended Analysis",
        "prompt": "What are the key factors affecting {{ company_id }}'s competitive position in the AI chip market?",
        "expected_input": {
            "company_id": {"type": "rp_entity_id"}
        }
        # No research_plan provided
    },
    "input": {
        "company_id": "D8442A"  # NVIDIA
    },
    "time_range": "last_90_days"
}

headers = {
    "X-API-KEY": my_api_key,
    "Content-Type": "application/json"
}

with requests.post(
    "https://agents.bigdata.com/v1/workflow/execute",
    headers=headers,
    json=payload,
    stream=True,
    timeout=180
) as r:
    r.raise_for_status()
    for raw_line in r.iter_lines(decode_unicode=True):
        if not raw_line or not raw_line.startswith("data: "):
            continue

        event = json.loads(raw_line[6:])
        delta = event.get("delta", {})
        msg_type = delta.get("type")

        if msg_type == "THINKING":
            print(f"[Thinking] {delta.get('content', '')[:100]}...")
        elif msg_type == "ACTION":
            print(f"[Action] {delta.get('tool_name')}")
        elif msg_type == "ANSWER":
            print(delta.get("content", ""), end="", flush=True)
        elif msg_type == "COMPLETE":
            print("\n\n[Complete]")

With a Research Plan

When a research plan is provided, the agent follows your predefined structure:

Agent follows the predefined steps in order
Structured, predictable execution
Progress is tracked per step (NOT_STARTED, IN_PROGRESS, COMPLETED, SKIPPED, FAILED)
Better for standardized reports and consistent outputs

Example: Structured Research

import requests
import json
from dotenv import load_dotenv
import os

load_dotenv()
my_api_key = os.getenv('BIGDATA_API_KEY')

# With research_plan - agent follows predefined steps
payload = {
    "template": {
        "name": "Structured Credit Analysis",
        "prompt": "Perform a credit analysis of {{ company_id }} following the research plan.",
        "expected_input": {
            "company_id": {"type": "rp_entity_id"}
        },
        "research_plan": {
            "title": "Credit Analysis Framework",
            "steps": [
                {"description": "Analyze business model and market position"},
                {"description": "Review financial statements and leverage metrics"},
                {"description": "Assess liquidity and debt maturity profile"},
                {"description": "Evaluate credit ratings and outlook"},
                {"description": "Identify key risks and mitigants"},
                {"description": "Formulate credit recommendation"}
            ]
        }
    },
    "input": {
        "company_id": "4E4980"  # Apple
    },
    "time_range": "last_180_days"
}

headers = {
    "X-API-KEY": my_api_key,
    "Content-Type": "application/json"
}

current_step = None

with requests.post(
    "https://agents.bigdata.com/v1/workflow/execute",
    headers=headers,
    json=payload,
    stream=True,
    timeout=180
) as r:
    r.raise_for_status()
    for raw_line in r.iter_lines(decode_unicode=True):
        if not raw_line or not raw_line.startswith("data: "):
            continue

        event = json.loads(raw_line[6:])
        delta = event.get("delta", {})
        msg_type = delta.get("type")

        if msg_type == "PLANNING":
            plan = delta.get("plan", {})
            print(f"\n=== {plan.get('title')} ===")
            for i, step in enumerate(plan.get("steps", [])):
                status = step.get("status", "NOT_STARTED")
                icon = {"NOT_STARTED": " ", "IN_PROGRESS": ">", "COMPLETED": "x", "SKIPPED": "-", "FAILED": "!"}.get(status, " ")
                print(f"  [{icon}] {step.get('description')}")
                if status == "IN_PROGRESS" and current_step != i:
                    current_step = i
                    print(f"\n--- Working on: {step.get('description')} ---")

        elif msg_type == "ANSWER":
            print(delta.get("content", ""), end="", flush=True)

        elif msg_type == "COMPLETE":
            print("\n\n=== Analysis Complete ===")

Processing Streaming Responses

Both approaches return the same message types via streaming:

Message Types Reference

Type	Description	Key Fields
`THINKING`	Agent reasoning	`content`
`PLANNING`	Plan status updates	`plan.title`, `plan.steps[]`
`ACTION`	Tool invocations	`tool_name`, `tool_arguments`
`ANSWER`	Research output	`content`
`GROUNDING`	Source references	`references[]` with `start`, `end`, `source`
`AUDIT`	Search traces	`audit_traces[]` with query and results
`COMPLETE`	Finish signal	`consumption[]` with token counts
`ERROR`	Error occurred	`error` message

Full Streaming Handler

def process_workflow_stream(response):
    """Process all message types from a workflow stream."""
    answer_text = ""
    sources = []

    for raw_line in response.iter_lines(decode_unicode=True):
        if not raw_line or not raw_line.startswith("data: "):
            continue

        try:
            event = json.loads(raw_line[6:])
        except json.JSONDecodeError:
            continue

        delta = event.get("delta", {})
        msg_type = delta.get("type")

        if msg_type == "THINKING":
            # Optional: show agent reasoning
            pass

        elif msg_type == "PLANNING":
            plan = delta.get("plan", {})
            print(f"Plan: {plan.get('title')}")
            for step in plan.get("steps", []):
                status = step.get("status", "NOT_STARTED")
                if status == "IN_PROGRESS":
                    print(f"  Working on: {step.get('description')}")

        elif msg_type == "ACTION":
            print(f"  Calling: {delta.get('tool_name')}")

        elif msg_type == "ANSWER":
            content = delta.get("content", "")
            answer_text += content
            print(content, end="", flush=True)

        elif msg_type == "GROUNDING":
            # Collect source references
            for ref in delta.get("references", []):
                if ref.get("source"):
                    sources.append(ref["source"])

        elif msg_type == "AUDIT":
            # Optional: process search traces
            for trace in delta.get("audit_traces", []):
                if trace.get("audit_type") == "SearchAuditV1":
                    query = trace.get("query", {}).get("text", "")
                    print(f"  Searched: {query}")

        elif msg_type == "COMPLETE":
            consumption = delta.get("consumption", [])
            print("\n\n--- Complete ---")
            for c in consumption:
                if c["type"] in ("base", "pro"):
                    print(f"  {c['type']}: {c['input_tokens']} in, {c['output_tokens']} out")

        elif msg_type == "ERROR":
            print(f"Error: {delta.get('error')}")

    return answer_text, sources

Model Selection

Both approaches support model selection via model_name:

payload = {
    "template": template_id,
    "input": {"company_id": "D8442A"},
    "model_name": "pro"  # Use enhanced model
}

Available models:

base - Default, balanced performance
pro - Enhanced reasoning

When to Use Research Plans

Use Case	Recommendation
Exploratory research	No plan - let the agent decide
Standardized reports	Use a plan for consistency
Batch processing	Use a plan for predictable output
Complex investigations	No plan - allows flexibility
Consistent output format	Use a plan

How to guides

Research Service

Search Service

Chat Service

Proprietary Content

Knowledge Graph

How Research Plans Affect Execution

Without a Research Plan

Example: Dynamic Research

With a Research Plan

Example: Structured Research

Processing Streaming Responses

Message Types Reference

Full Streaming Handler

Model Selection

When to Use Research Plans

Next Steps

Creating Templates

Example Templates

How to guides

Research Service

Search Service

Chat Service

Proprietary Content

Knowledge Graph

Documentation Index

​How Research Plans Affect Execution

​Without a Research Plan

​Example: Dynamic Research

​With a Research Plan

​Example: Structured Research

​Processing Streaming Responses

​Message Types Reference

​Full Streaming Handler

​Model Selection

​When to Use Research Plans

​Next Steps

Creating Templates

Example Templates

How Research Plans Affect Execution

Without a Research Plan

Example: Dynamic Research

With a Research Plan

Example: Structured Research

Processing Streaming Responses

Message Types Reference

Full Streaming Handler

Model Selection

When to Use Research Plans

Next Steps