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 Workflows with and without a research plan emit the same sequence of typed SSE
events. The full reference for every public message type — including a canonical
Python handler that dispatches on type — lives in the Concepts section:
Streaming responses All 11 message types, their fields, when each is emitted, and a copy-paste handler.
Grounding and citations How GROUNDING references map to spans of the answer text.
Workflows wrap each event’s typed payload in a delta field, so dispatch on
event["delta"]["type"]. The Research Agent uses an analogous message field.
Everything else about the message types is identical between the two services.
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 performancepro - 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
Next Steps
Creating Templates Learn template anatomy and best practices
Example Templates Ready-to-use financial analysis templates
