--- name: ai-research-workflow description: > Deep research and AI workflow orchestration patterns for Octane. Use when working on brand guide generation, market analysis, competitive analysis, content brief, or any multi-step AI research pipeline. Covers the WorkflowBuilder/WorkflowStep orchestration system, Anthropic dual-model routing (Opus for research, Sonnet for generation), and the ai_tasks persistence pattern. compatibility: > Best in the Octane Rust repo with Anthropic workflow code and network access for external model or API verification. metadata: opensite-category: ai opensite-scope: octane opensite-visibility: public --- # AI Research Workflow Skill ## Skill Resources - Activation and cross-agent notes: [references/activation.md](references/activation.md) - Use `ultrathink` or the deepest available reasoning mode before changing architecture, security, migration, or performance-critical paths. - Example: [examples/workflow-brief.example.md](examples/workflow-brief.example.md) - Helper: `scripts/validate_workflow_brief.py` - Template: [templates/workflow-brief.md](templates/workflow-brief.md) ## Task Focus for $ARGUMENTS When this skill is invoked explicitly, treat `$ARGUMENTS` as the primary scope to optimize around: a repo path, component name, incident id, rollout target, or other concrete task boundary. You are building or modifying multi-step AI research workflows in Octane (`github.com/Toastability/octane`). These workflows power the core AI services: brand guide, market analysis, competitive analysis, content brief, and SEO analysis. ## Orchestration Architecture Workflows use the `WorkflowBuilder` in `src/services/orchestration/` and are executed by Anthropic-backed agents. Two models, two phases: - **Phase 1 (Research)**: Opus 4.6 + web search (`context-1m-2025-08-07`, `web-search-2025-03-05`) - **Phase 2 (Generation)**: Sonnet 4.6 + schema tools (structured output, report generation) ```rust // Example: Market Analysis workflow use crate::services::orchestration::{WorkflowBuilder, WorkflowStep, Agent, MemoryStore}; use std::sync::Arc; use tokio::sync::RwLock; pub async fn run_market_analysis( anthropic: Arc, input: MarketAnalysisInput, ) -> anyhow::Result { let memory = Arc::new(RwLock::new(MemoryStore::default())); let research_agent = Agent::new( "Research Agent", anthropic.clone(), AgentConfig { model: "claude-opus-4-6".into(), max_tokens: 32000, tools: vec!["web_search".into()], beta_features: vec![ "context-1m-2025-08-07".into(), "web-search-2025-03-05".into(), ], }, ); let generation_agent = Agent::new( "Generation Agent", anthropic.clone(), AgentConfig { model: "claude-sonnet-4-6".into(), max_tokens: 8192, tools: vec!["schema_output".into()], beta_features: vec!["context-1m-2025-08-07".into()], }, ); let workflow = WorkflowBuilder::new("Market Analysis") .with_memory(memory.clone()) .add_step(WorkflowStep::new( "Industry Research", research_agent.clone(), format_industry_research_prompt(&input), )) .add_parallel_steps(vec![ WorkflowStep::new("Competitor Analysis", research_agent.clone(), format_competitor_prompt(&input)), WorkflowStep::new("Market Sizing", research_agent.clone(), format_market_sizing_prompt(&input)), ]) .add_step(WorkflowStep::new( "Report Generation", generation_agent.clone(), REPORT_GENERATION_PROMPT.into(), )) .build(); let result = workflow.execute().await?; // Persist result to ai_tasks table let report: MarketAnalysisReport = serde_json::from_value( result.final_output .ok_or_else(|| anyhow::anyhow!("Workflow produced no output"))? )?; Ok(report) } ``` ## WorkflowBuilder API ```rust // src/services/orchestration/workflow.rs pub struct WorkflowBuilder { name: String, steps: Vec, memory: Option>>, max_retries: u32, retry_delay_ms: u64, } impl WorkflowBuilder { pub fn new(name: impl Into) -> Self pub fn with_memory(self, memory: Arc>) -> Self pub fn add_step(self, step: WorkflowStep) -> Self pub fn add_parallel_steps(self, steps: Vec) -> Self // Run concurrently pub fn with_max_retries(self, n: u32) -> Self pub fn build(self) -> Workflow // Workflow execution // .execute() → WorkflowResult // .step_results: Vec // .final_output: Option // .memory: MemoryStore (exported) // .errors: Vec // .total_tokens: i64 } pub struct WorkflowStep { pub name: String, pub agent: Arc, pub prompt: String, pub depends_on: Vec, // Step names this step reads from memory } ``` ## Memory Store Pattern Steps communicate via the shared `MemoryStore`. Each step writes its output; subsequent steps read it: ```rust // src/services/orchestration/memory.rs #[derive(Default)] pub struct MemoryStore { entries: HashMap, } impl MemoryStore { pub fn set(&mut self, key: impl Into, value: serde_json::Value) pub fn get(&self, key: &str) -> Option<&serde_json::Value> pub fn get_as(&self, key: &str) -> anyhow::Result pub fn to_context_string(&self) -> String // Formats memory for prompt injection } // In a prompt — inject memory from previous steps: const REPORT_GENERATION_PROMPT: &str = r#" Based on the research completed in previous steps: {memory_context} Generate a comprehensive market analysis report following the output schema. "#; // The orchestration layer replaces {memory_context} with memory.to_context_string() ``` ## Anthropic Service — Dual Model Routing The `AnthropicService` in Octane already implements the dual-model architecture. Key methods: ```rust // Opus 4.6 — use for research (web search + extended thinking) anthropic.create_research_message(ResearchRequest { prompt: research_prompt, thinking_budget: Some(16000), web_search: true, max_tokens: 32000, }).await?; // Sonnet 4.6 — use for generation (structured output + schema tools) anthropic.create_generation_message(GenerationRequest { prompt: generation_prompt, output_schema: Some(report_schema), max_tokens: 8192, }).await?; // Direct message — use when you need explicit model control anthropic.create_message(CreateMessageRequest { model: "claude-sonnet-4-6".into(), messages: vec![/* ... */], max_tokens: 4096, tools: None, beta_features: vec![], }).await?; ``` ## ai_tasks Persistence Pattern Every AI workflow writes its state to the `ai_tasks` table. Never skip this: ```rust // Pattern for persisting workflow state pub struct AiTaskRecord { pub id: uuid::Uuid, pub task_type: String, // "market_analysis" | "brand_guide" | etc. pub account_id: uuid::Uuid, pub status: AiTaskStatus, // Pending | Running | Completed | Failed pub input: serde_json::Value, // The original request (no PHI) pub output: Option, pub error: Option, pub tokens_used: i64, pub duration_ms: i64, pub created_at: chrono::DateTime, pub completed_at: Option>, } pub async fn run_workflow_with_persistence( db: &DatabaseService, task_type: &str, account_id: uuid::Uuid, run: impl Future>, ) -> anyhow::Result { let task_id = db.create_ai_task(task_type, account_id).await?; let start = std::time::Instant::now(); match run.await { Ok(output) => { db.complete_ai_task(task_id, &output, start.elapsed()).await?; Ok(output) } Err(e) => { db.fail_ai_task(task_id, &e.to_string(), start.elapsed()).await?; Err(e) } } } ``` ## Workflow Types Reference | Workflow | Handler File | Steps | Primary Model | |---------|-------------|-------|---------------| | Brand Guide | brand_guide.rs | Research → Tone → Brand → Report | Opus → Sonnet | | Market Analysis | handlers/seo.rs (market section) | Industry → Competitors → Market → Report | Opus → Sonnet | | Competitive Analysis | competitive_analysis.rs | Competitors → Pricing → Positioning → Report | Opus → Sonnet | | Content Brief | content_brief.rs | Keywords → Topics → Structure → Brief | Opus → Sonnet | | SEO Report | seo.rs | Crawl → Keywords → Backlinks → Report | Sonnet only | | Video Insight | (Phase 1 launch) | Transcribe → Analyze → Summarize | Sonnet (vLLM) | ## Parallel Step Rules 1. Steps can only run in parallel if they don't depend on each other's memory outputs 2. Maximum 3 parallel steps per batch (to avoid context window conflicts) 3. Each parallel step gets its own memory namespace to prevent write conflicts 4. Parallel steps share the read-only input context but have isolated write namespaces ```rust // Safe: these steps don't depend on each other .add_parallel_steps(vec![ WorkflowStep::new("Competitor Research", agent.clone(), competitor_prompt), WorkflowStep::new("Market Sizing", agent.clone(), sizing_prompt), WorkflowStep::new("Industry Trends", agent.clone(), trends_prompt), ]) // Unsafe: if Report Generation needs competitor data, it can't run in parallel // It must be a sequential step after the parallel batch completes .add_step(WorkflowStep::new("Report Generation", gen_agent, report_prompt)) ``` ## Token Budget Management ```rust // Estimate token cost before running expensive workflows pub fn estimate_workflow_tokens(workflow_type: &str) -> TokenBudget { match workflow_type { "brand_guide" => TokenBudget { research: 50_000, generation: 8_000 }, "market_analysis" => TokenBudget { research: 80_000, generation: 12_000 }, "competitive_analysis" => TokenBudget { research: 60_000, generation: 10_000 }, "content_brief" => TokenBudget { research: 20_000, generation: 4_000 }, _ => TokenBudget { research: 32_000, generation: 4_096 }, } } // Check account token balance before starting pub async fn check_token_balance( db: &DatabaseService, account_id: uuid::Uuid, required: i64, ) -> anyhow::Result<()> { let balance = db.get_token_balance(account_id).await?; if balance < required { return Err(anyhow::anyhow!( "Insufficient token balance: {} available, {} required", balance, required )); } Ok(()) } ```