Phase 3: LangGraph Workflow Orchestration (#2)

* Add LangGraph workflow state and orchestration

Implemented core workflow infrastructure:
- State schema with TypedDict for all workflow phases
- StateGraph with research, analysis, writing, human review nodes
- Conditional routing between agents
- SQLite checkpointing for state persistence
- Cost tracking integration across all nodes
- Error handling and recovery logic
- Budget enforcement

* Add SQLite checkpointing for production persistenceAdded langgraph-checkpoint-sqlite for production-grade state management:- Workflows persist across restarts and crashes- Enables audit trails for compliance- Supports long-running multi-day analyses - Critical for human-in-the-loop approval workflows

requirements.txt

@@ -39,3 +39,8 @@ mypy==1.18.2
 python-dotenv==1.2.1
 tenacity==9.1.2
 httpx==0.28.1
+langgraph==1.0.4
+langgraph-checkpoint==3.0.1
+langgraph-checkpoint-sqlite==3.0.0
+langgraph-prebuilt==1.0.5
+langgraph-sdk==0.2.10

src/workflows/intelligence.py

@@ -0,0 +1,288 @@
+"""Main LangGraph workflow for market intelligence."""
+
+from langgraph.graph import StateGraph, END
+from langgraph.checkpoint.sqlite import SqliteSaver
+
+from src.workflows.state import IntelligenceState
+from src.agents.researcher import ResearchAgent
+from src.agents.analyst import AnalysisAgent
+from src.agents.writer import WriterAgent
+from src.utils.cost_tracker import CostTracker, BudgetExceededError
+from src.utils.logging import setup_logger
+
+logger = setup_logger(__name__)
+
+
+class MarketIntelligenceWorkflow:
+    """
+    LangGraph workflow orchestrating research, analysis, and writing agents.
+
+    Features:
+    - Multi-agent coordination
+    - State persistence with checkpointing
+    - Cost tracking and budget enforcement
+    - Human-in-the-loop approval
+    - Error recovery
+    """
+
+    def __init__(
+        self, checkpoint_path: str = "./checkpoints.db", max_budget: float = 2.0
+    ):
+        """
+        Initialize workflow.
+
+        Args:
+            checkpoint_path: Path to SQLite checkpoint database
+            max_budget: Maximum cost per run in USD
+        """
+        self.max_budget = max_budget
+        self.cost_tracker = CostTracker()
+        self.checkpoint_path = checkpoint_path
+
+        # Initialize agents (shared cost tracker)
+        self.research_agent = ResearchAgent(cost_tracker=self.cost_tracker)
+        self.analysis_agent = AnalysisAgent(cost_tracker=self.cost_tracker)
+        self.writer_agent = WriterAgent(cost_tracker=self.cost_tracker)
+
+        # Build workflow graph
+        self.workflow = self._build_graph()
+
+        logger.info("Market Intelligence Workflow initialized")
+
+    def _build_graph(self) -> StateGraph:
+        """Build LangGraph workflow."""
+        # Initialize graph
+        graph = StateGraph(IntelligenceState)
+
+        # Add nodes (agent wrappers)
+        graph.add_node("research", self._research_node)
+        graph.add_node("analysis", self._analysis_node)
+        graph.add_node("writing", self._writing_node)
+        graph.add_node("human_review", self._human_review_node)
+
+        # Set entry point
+        graph.set_entry_point("research")
+
+        # Add edges
+        graph.add_conditional_edges(
+            "research",
+            self._should_continue_to_analysis,
+            {"analysis": "analysis", "end": END},
+        )
+
+        graph.add_edge("analysis", "writing")
+        graph.add_edge("writing", "human_review")
+
+        graph.add_conditional_edges(
+            "human_review",
+            self._check_approval,
+            {"approved": END, "revise": "research", "max_revisions": END},
+        )
+
+        # Compile with SQLite checkpointing for production persistence
+        checkpointer = SqliteSaver.from_conn_string(self.checkpoint_path)
+        return graph.compile(checkpointer=checkpointer)
+
+    async def _research_node(self, state: IntelligenceState) -> dict:
+        """Research agent node."""
+        logger.info(f"Research node: {state['company_name']}")
+
+        try:
+            # Run research agent
+            research_results = await self.research_agent.run(
+                company_name=state["company_name"],
+                industry=state.get("industry"),
+                research_depth="comprehensive",
+            )
+
+            # Update state
+            return {
+                "current_agent": "research",
+                "research_data": research_results,
+                "competitors": research_results.get("competitors", []),
+                "market_trends": research_results.get("market_trends", {}),
+                "raw_sources": research_results.get("raw_sources", []),
+                "iteration": state.get("iteration", 0) + 1,
+            }
+
+        except Exception as e:
+            logger.error(f"Research node failed: {e}")
+            return {
+                "errors": [f"Research failed: {str(e)}"],
+                "current_agent": "research",
+            }
+
+    async def _analysis_node(self, state: IntelligenceState) -> dict:
+        """Analysis agent node."""
+        logger.info(f"Analysis node: {state['company_name']}")
+
+        try:
+            # Check budget before expensive analysis
+            self.cost_tracker.check_budget(self.max_budget)
+
+            # Run analysis agent
+            analysis_results = await self.analysis_agent.run(
+                research_data=state["research_data"]
+            )
+
+            # Update state
+            return {
+                "current_agent": "analysis",
+                "swot": analysis_results.get("swot", {}),
+                "competitive_matrix": analysis_results.get("competitive_matrix", {}),
+                "positioning": analysis_results.get("positioning", {}),
+                "strategic_recommendations": analysis_results.get(
+                    "strategic_recommendations", {}
+                ),
+            }
+
+        except BudgetExceededError as e:
+            logger.error(f"Budget exceeded: {e}")
+            return {
+                "errors": [f"Budget exceeded: {str(e)}"],
+                "current_agent": "analysis",
+            }
+        except Exception as e:
+            logger.error(f"Analysis node failed: {e}")
+            return {
+                "errors": [f"Analysis failed: {str(e)}"],
+                "current_agent": "analysis",
+            }
+
+    async def _writing_node(self, state: IntelligenceState) -> dict:
+        """Writer agent node."""
+        logger.info(f"Writing node: {state['company_name']}")
+
+        try:
+            # Run writer agent
+            report_results = await self.writer_agent.run(
+                research_data=state["research_data"],
+                analysis_data={
+                    "swot": state.get("swot", {}),
+                    "competitive_matrix": state.get("competitive_matrix", {}),
+                    "positioning": state.get("positioning", {}),
+                    "strategic_recommendations": state.get(
+                        "strategic_recommendations", {}
+                    ),
+                },
+            )
+
+            # Get cost summary
+            cost_summary = self.cost_tracker.get_summary()
+
+            # Update state
+            return {
+                "current_agent": "writing",
+                "executive_summary": report_results.get("executive_summary", ""),
+                "full_report": report_results.get("full_report", ""),
+                "report_metadata": report_results.get("metadata", {}),
+                "total_cost": cost_summary["total_cost"],
+                "total_tokens": cost_summary["total_tokens"],
+            }
+
+        except Exception as e:
+            logger.error(f"Writing node failed: {e}")
+            return {
+                "errors": [f"Writing failed: {str(e)}"],
+                "current_agent": "writing",
+            }
+
+    async def _human_review_node(self, state: IntelligenceState) -> dict:
+        """Human review node (placeholder for now)."""
+        logger.info(f"Human review node: {state['company_name']}")
+
+        # For now, auto-approve
+        # In Phase 5, this will connect to the Gradio UI
+        return {
+            "current_agent": "human_review",
+            "approved": True,  # Auto-approve for testing
+            "human_feedback": None,
+        }
+
+    def _should_continue_to_analysis(self, state: IntelligenceState) -> str:
+        """Decide whether to continue to analysis or end."""
+        # Check if research was successful
+        if state.get("errors") and state["errors"]:
+            logger.warning("Research had errors, ending workflow")
+            return "end"
+
+        if not state.get("research_data"):
+            logger.warning("No research data, ending workflow")
+            return "end"
+
+        return "analysis"
+
+    def _check_approval(self, state: IntelligenceState) -> str:
+        """Check if report is approved or needs revision."""
+        # Check max revisions
+        revision_count = state.get("revision_count", 0)
+        if revision_count >= 2:
+            logger.warning("Max revisions reached")
+            return "max_revisions"
+
+        # Check approval
+        if state.get("approved"):
+            return "approved"
+
+        # Revision requested
+        if state.get("human_feedback"):
+            return "revise"
+
+        # Default to approved
+        return "approved"
+
+    async def run(
+        self,
+        company_name: str,
+        industry: str | None = None,
+        thread_id: str | None = None,
+    ) -> dict:
+        """
+        Run the complete workflow.
+
+        Args:
+            company_name: Target company name
+            industry: Optional industry context
+            thread_id: Optional thread ID for checkpointing
+
+        Returns:
+            Final state dictionary
+        """
+        logger.info(f"Starting workflow for: {company_name}")
+
+        # Initial state
+        initial_state = {
+            "company_name": company_name,
+            "industry": industry,
+            "research_data": {},
+            "competitors": [],
+            "market_trends": {},
+            "raw_sources": [],
+            "swot": {},
+            "competitive_matrix": {},
+            "positioning": {},
+            "strategic_recommendations": {},
+            "executive_summary": "",
+            "full_report": "",
+            "report_metadata": {},
+            "current_agent": "research",
+            "iteration": 0,
+            "total_cost": 0.0,
+            "total_tokens": 0,
+            "errors": [],
+            "human_feedback": None,
+            "approved": False,
+            "revision_count": 0,
+        }
+
+        # Run workflow
+        config = {"configurable": {"thread_id": thread_id or "default"}}
+
+        try:
+            final_state = await self.workflow.ainvoke(initial_state, config)
+            logger.info(f"Workflow complete. Cost: ${final_state['total_cost']:.4f}")
+            return final_state
+
+        except Exception as e:
+            logger.error(f"Workflow failed: {e}")
+            raise

src/workflows/state.py

@@ -0,0 +1,45 @@
+"""State definitions for LangGraph workflow."""
+
+from typing import Annotated, Literal, TypedDict
+import operator
+
+
+class IntelligenceState(TypedDict):
+    """
+    State for market intelligence workflow.
+
+    This state is passed between agents and persisted across checkpoints.
+    """
+
+    # Input
+    company_name: str
+    industry: str | None
+
+    # Research phase outputs
+    research_data: dict
+    competitors: list[dict]
+    market_trends: dict
+    raw_sources: list[dict]
+
+    # Analysis phase outputs
+    swot: dict
+    competitive_matrix: dict
+    positioning: dict
+    strategic_recommendations: dict
+
+    # Writing phase outputs
+    executive_summary: str
+    full_report: str
+    report_metadata: dict
+
+    # Workflow metadata
+    current_agent: Literal["research", "analysis", "writing", "human_review", "done"]
+    iteration: int
+    total_cost: float
+    total_tokens: int
+    errors: Annotated[list, operator.add]  # Accumulate errors across nodes
+
+    # Human-in-the-loop
+    human_feedback: str | None
+    approved: bool
+    revision_count: int

test_workflow.py

@@ -0,0 +1,85 @@
+"""
+Test the complete LangGraph workflow.
+
+Replaces test_agents.py with proper workflow orchestration.
+"""
+
+import asyncio
+
+from src.workflows.intelligence import MarketIntelligenceWorkflow
+
+
+async def test_workflow():
+    """Test complete workflow with LangGraph orchestration."""
+
+    print("=" * 80)
+    print("TESTING LANGGRAPH WORKFLOW")
+    print("=" * 80)
+    print()
+
+    # Initialize workflow
+    workflow = MarketIntelligenceWorkflow(
+        checkpoint_path="./test_checkpoints.db", max_budget=2.0
+    )
+
+    # Run on Instagram viral query
+    company_name = "going viral on Instagram using AI without showing face"
+    industry = "Social Media Marketing"
+
+    print(f"Query: {company_name}")
+    print(f"Industry: {industry}")
+    print()
+    print("Running workflow...")
+    print("-" * 80)
+
+    try:
+        final_state = await workflow.run(
+            company_name=company_name, industry=industry, thread_id="test-run-1"
+        )
+
+        print("\nWorkflow Complete!")
+        print("=" * 80)
+
+        # Display results
+        print(f"\nCompany: {final_state['company_name']}")
+        print(f"Final Agent: {final_state['current_agent']}")
+        print(f"Iterations: {final_state['iteration']}")
+        print(f"Approved: {final_state['approved']}")
+        print(f"Errors: {len(final_state['errors'])}")
+
+        print(f"\nCost: ${final_state['total_cost']:.4f}")
+        print(f"Tokens: {final_state['total_tokens']:,}")
+        print(f"Sources: {len(final_state.get('raw_sources', []))}")
+
+        print("\n" + "=" * 80)
+        print("EXECUTIVE SUMMARY")
+        print("=" * 80)
+        print(final_state["executive_summary"])
+
+        print("\n" + "=" * 80)
+        print("FULL REPORT (first 1000 chars)")
+        print("=" * 80)
+        print(final_state["full_report"][:1000] + "...")
+
+        # Save report
+        with open("workflow_test_report.md", "w") as f:
+            f.write(f"# Market Intelligence Report\n\n")
+            f.write(f"Generated via LangGraph Workflow\n\n")
+            f.write(f"---\n\n")
+            f.write(final_state["full_report"])
+            f.write(f"\n\n---\n\n")
+            f.write(f"**Cost:** ${final_state['total_cost']:.4f}\n")
+            f.write(f"**Tokens:** {final_state['total_tokens']:,}\n")
+
+        print(f"\n\nReport saved to: workflow_test_report.md")
+        print("\nTest PASSED!")
+
+    except Exception as e:
+        print(f"\nTest FAILED: {e}")
+        import traceback
+
+        traceback.print_exc()
+
+
+if __name__ == "__main__":
+    asyncio.run(test_workflow())

workflow_test_report.md

@@ -0,0 +1,156 @@
+# Market Intelligence Report
+
+Generated via LangGraph Workflow
+
+---
+
+# Market Intelligence Report: Going Viral on Instagram Using AI Without Showing Face
+
+## Executive Summary
+**Executive Summary: Leveraging AI for Faceless Viral Growth on Instagram**
+
+The faceless Instagram virality market is led by niche AI platforms including Invideo AI (text-to-Reel generation), Virlo (visuals and shop setup), Zebracat (viral video tools), and TailorTalk (automation), with limited corporate details available (e.g., no founding dates or revenues) [1-3,5]. These SaaS/freemium tools enable rapid content creation, tying into user monetization via affiliates, shops, and digital products, mirroring successful AI influencers like Lil Miquela (3M+ followers) [4,9].
+
+**Market Position**: Positioned as efficiency leaders in a $32.55B influencer market (57.1% Instagram share), these tools deliver 3-5% engagement rates vs. 2% average, driven by scalable, anonymous Reels in niches like AI art and pets [3,9]. Amid 93% marketer video spend growth to $140B by 2025, they exploit short-form trends but face saturation from competitors like BigMotion AI [3,4].
+
+**Key Findings**:
+- **Strengths**: 20-minute Reel production, 24/7 scalability, privacy advantages [1,3].
+- **Weaknesses**: Tool dependencies, quality limits, transparency gaps [1-3].
+- **Opportunities**: Social commerce ($2.9T by 2026), AI personalization [2,4].
+- **Threats**: Algorithm volatility, authenticity fatigue, bot crackdowns [1,4].
+
+**Main Recommendations** (Prioritized for 2-3x growth in 30 days):
+- Post 5-7 daily Reels in proven niches using Invideo/Zebracat; integrate voiceovers/visuals for UGC-style hooks [1,3].
+- Automate via TailorTalk/Madgicx; launch Shops with 20% commerce content, cross-posting to TikTok [3,5].
+- Diversify tools and build Discord communities for retention [3,7].
+
+Actionable path: Deploy high-priority tactics immediately to capture 10-20% conversion uplift, mitigating risks for sustained 20% YoY growth. Deeper diligence (e.g., Crunchbase) advised [1-10].
+
+(248 words)
+
+## 1. Company Overview
+The "Going Viral on Instagram Using AI Without Showing Face" represents an emerging strategy and ecosystem in the social media marketing industry, focused on AI-powered, faceless content creation for Instagram Reels, posts, and automation. It aggregates specialized SaaS tools rather than a single corporate entity, with limited foundational details available (e.g., no founding dates, headquarters, or employee counts specified in public data). Key platforms driving this strategy include:
+
+- **Invideo AI**: AI video creation platform generating faceless Instagram Reels from text prompts in ~20 minutes; supports monetization via digital products and affiliates [1].
+- **Virlo (virlo.ai)**: AI tool for faceless Instagram content ideas, visuals (via Leonardo.ai integration), voice-overs (Eleven Labs), and video transformation (Flarecut); enables Instagram Shop setup for merchandise [2].
+- **Zebracat (zebracat.ai)**: AI video generator targeting viral faceless content in niches like pet videos and AI art, achieving up to 5% engagement rates (vs. 2% average) [3].
+- **TailorTalk (tailortalk.ai)**: AI platform for Instagram business automation, including DMs, comments, leads, and product recommendations via natural language processing [5].
+- **Supporting Tools**: Madgicx (AI for trend forecasting and automated funnels) [6]; Juicer (AI for content ideas, captions, and social listening) [7]; ancillary tools like Copy.ai (copywriting) and Descript (video editing) [8].
+
+**Business Model**: Primarily SaaS/subscription-based with freemium access, enabling user monetization through affiliates, digital products (e.g., spreadsheets, courses), sponsored posts, Instagram Shops, and brand deals. AI influencers (e.g., Aitana Lopez) exemplify scalability, generating revenue via 24/7 operations and 3% higher engagement [4,9].
+
+**Key Metrics**: No direct revenue or employee data; indirect indicators include 5% engagement for AI art accounts and virtual influencers with 3M+ followers (e.g., Lil Miquela) [3,9]. Growth is implied through viral potential and rapid workflows.
+
+This ecosystem emphasizes time-saving AI for anonymous, scalable growth, with data gaps suggesting need for sources like Crunchbase [1-10].
+
+## 2. Competitive Landscape
+The competitive landscape for faceless AI-driven Instagram virality is emerging and fragmented, dominated by specialized tools for short-form video generation, automation, and monetization. Leaders focus on niches like pets and AI art, with automation challengers gaining traction.
+
+### Main Competitors
+- **Zebracat**: AI video generator for faceless niches (pets, AI art); high engagement (5% vs. 2% average); manual creation with virality tips (e.g., 509K-follower accounts) [1].
+- **BigMotion AI**: Free faceless AI video generator with auto-posting to Instagram Reels, TikTok, and YouTube Shorts; <5-minute production, real-time algorithm adaptation [3].
+- **Faceless.video**: Automates faceless video ads and cross-platform posting; e-commerce focus [4].
+- **Crayo AI**: Short-form video generator for Reels; backed by faceless YouTube experts (millions in revenue) [6].
+
+### Competitive Matrix
+
+| Dimension              | Going Viral AI (Faceless IG Strategy) | Zebracat                          | BigMotion AI                      | Faceless.video                    | Crayo AI                          |
+|------------------------|---------------------------------------|-----------------------------------|-----------------------------------|-----------------------------------|-----------------------------------|
+| **Market Share/Size** | Low (Emerging; no metrics)           | **High** (Leader; 509K examples, 5% engagement) | Medium-High (Free entrant, multi-platform) | Low (Ad-focused niche)           | Medium (Expert-backed)           |
+| **Product Range**     | Low (IG Reels-specific)              | Medium (Niche videos from text)  | **High** (Short videos + auto-post) | Low (Video ads only)             | Medium (Short-form Reels/Shorts) |
+| **Pricing Strategy**  | N/A (Assumed subscription)           | N/A (Assumed paid)               | **Free tier** (Aggressive)        | N/A (Assumed paid)               | N/A (Assumed paid)               |
+| **Technology/Innovation** | Medium (IG viral focus)            | Medium (Text-to-video + tips)    | **High** (Algo adaptation, auto-publish) | Medium (Ad auto-posting)       | **High** (Ideation by experts)   |
+| **Customer Segments** | IG creators (viral growth)           | Niche creators (pets/AI art)     | Passive income seekers            | E-com/marketers                  | Viral short-form creators        |
+
+**Strategic Insights**: Zebracat leads creatively; BigMotion excels in automation. Opportunity for "Going Viral AI" to integrate both for IG-specific dominance [1,3,4,6].
+
+## 3. SWOT Analysis
+
+#### Strengths
+- High engagement (5% in niches vs. 2% average); 3% uplift for virtual influencers [3,9].
+- Rapid scalability (Reels in 20 mins); 24/7 automation [1,3].
+- Privacy/low risk; diverse monetization (affiliates, shops) [1,2,4].
+- Niche expertise (e.g., 509K-follower proofs); efficiency via integrations [2,3,5].
+
+#### Weaknesses
+- Limited transparency (no HQ/revenue data) [1-3,5].
+- Third-party dependencies; content limits (<90s videos) [2,3,8].
+- Engagement automation gaps; skill barriers [3,5,6].
+- Freemium churn risks [1,2].
+
+#### Opportunities
+- Short-form video boom ($140B market); AI personalization [3,4].
+- Social commerce ($2.9T by 2026); influencer surge ($32.55B) [2,4].
+- Niche/authenticity demand; multi-platform expansion [1,3,7].
+
+#### Threats
+- Competition (e.g., BigMotion free tier) [1,3,4,6].
+- Consumer fatigue (50% limiting use); algorithm volatility [1,2,4].
+- Authenticity pressure; platform risks [1,4,7].
+
+## 4. Market Positioning
+Faceless AI Instagram virality is positioned as a **privacy-first, scalable viral enabler** in short-form video AI, delivering 3-5% engagement via anonymous Reels [3,9].
+
+**Value Proposition**:
+| Core Element       | Description                          | Evidence [Source]              |
+|--------------------|--------------------------------------|-------------------------------|
+| Speed & Scalability| Text-to-video <20 mins; 24/7 output | Invideo/BigMotion [1,3]      |
+| Privacy            | Faceless niches (pets/AI art)        | 5% engagement [3]             |
+| Monetization       | Affiliates/shops/brands              | Aitana Lopez model [2,9]      |
+| Engagement Edge    | Algorithm-optimized                  | PLS-SEM study [3,10]          |
+
+**Target Segments**: Aspiring creators (primary); e-com brands (secondary) [1-5].
+
+**Differentiation**: Bridges creative (Zebracat) and automation (BigMotion) gaps with IG-specific hooks [1,3].
+
+**Opportunities**: Freemium MVP; niche expansion (e.g., virtual influencers) [2,9].
+
+## 5. Market Trends & Insights
+**Key Trends**:
+- Short-form video dominance (93% marketers increasing spend; $140B market) [3,4].
+- AI personalization (50% leaders maximizing tools) [2,4,5].
+- Authenticity/UGC shift; social commerce boom [1,3,7].
+
+**Growth Drivers**:
+- Ad/influencer spend ($276.7B social ads; $32.55B influencers) [2,4].
+- Video efficacy (99% product understanding) [4].
+
+**Challenges**:
+- Fatigue (50% limiting use); algorithm volatility [1,2].
+
+**Outlook**: $2.9T social commerce by 2026; AI + authenticity key to 20% YoY growth [2,4].
+
+## 6. Strategic Recommendations
+
+#### High Priority (Weeks 1-4)
+- **Post 5-7 daily Reels in niches (AI art/pets)**: Use Invideo/BigMotion for hooks; 2-3x growth expected [1,3].
+- **Integrate voiceovers/visuals + auto-scheduling**: 20-30% higher shares [2,3].
+
+#### Medium Priority (6-12 Months)
+- **Automate interactions (TailorTalk)**: Double replies; 15-25% retention [3,5].
+- **Launch Shops + cross-post**: 10-20% conversions; $5K+ monthly [2,3].
+
+#### Long-Term (12+ Months)
+- **Diversify tools**: 30-50% downtime reduction [3,8].
+- **Build communities (Discord)**: 20-40% LTV uplift [3,7].
+
+## 7. Sources
+- [1] Invideo AI platform and faceless Reels data.
+- [2] Virlo.ai tool features and Instagram Shop integration.
+- [3] Zebracat.ai; BigMotion AI; engagement metrics (5% vs. 2%).
+- [4] Faceless.video; AI influencers (Aitana Lopez); market projections.
+- [5] TailorTalk.ai automation details.
+- [6] Madgicx; Crayo AI expert backing.
+- [7] Juicer AI tools; social listening trends.
+- [8] Copy.ai; Descript; ancillary workflows.
+- [9] Virtual influencers (Lil Miquela, Shudu Gram); 3% engagement edge.
+- [10] PLS-SEM study (n=391) on AI influencer attributes.
+
+---
+
+Report generated: November 28, 2025
+
+---
+
+**Cost:** $0.0000
+**Tokens:** 0