> ## Documentation Index > Fetch the complete documentation index at: https://docs.bigdata.com/llms.txt > Use this file to discover all available pages before exploring further. # Using Research Plans > Learn how research plans control workflow execution 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 ```python theme={null} 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 ```python theme={null} 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 ```python theme={null} 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`: ```python theme={null} 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 | ## Next Steps Learn template anatomy and best practices Ready-to-use financial analysis templates