refactor: align codebase with seperated agent logics

- feat: centralized prompt management in src/utils/prompts.py (no magic strings)
- refactor: update Research, Analyst, and Writer agents to use explicit return types
- test: updated integration tests to match strict data shapes
- fix: enable in-memory checkpointing for robust integration testing

CLAUDE.md

@@ -0,0 +1,77 @@
+# CLAUDE.md - The Ultrathink Constitution
+
+> **"We're not here to write code. We're here to make a dent in the universe."**
+
+This file is the single source of truth for the philosophy, architecture, and standards of the **Agentic Market Research Orchestrator**. It is not a suggestion. It is the law.
+
+## I. The Philosophy (Ultrathink)
+
+1.  ✅ **Think Different**: Question every assumption. If the code looks "standard," it's probably wrong. Look for the elegant angle capable of 80x improvements.
+2.  ✅ **Craft, Don't Code**: Variable names explain. Functions sing. The code is literature for the next engineer.
+3.  ✅ **Simplify Ruthlessly**: Complexity is the enemy. If a feature doesn't add exponential value, delete it.
+4.  ⭐ **Iterate Relentlessly**: The first draft is for the trash. The second is for the critic. The third is for the user.
+
+## II. The Architecture
+
+The system is a **Symbiotic Triad** orchestrated by `LangGraph`.
+
+```mermaid
+graph TD
+    User((User)) -->|Topic| Coordinator{LangGraph}
+    Coordinator -->|Explore| Researcher[Researcher Agent]
+    Coordinator -->|Synthesize| Analyst[Analyst Agent]
+    Coordinator -->|Narrate| Writer[Writer Agent]
+    
+    Researcher -->|Raw Data| Analyst
+    Analyst -->|Insights| Writer
+    Writer -->|Report| User
+```
+
+### The Agents
+*All agents must inherit from `src.agents.base.BaseAgent`.*
+
+1.  **The Researcher (`src/agents/researcher.py`)**
+    *   **Role**: The Hunter.
+    *   **Behavior**: Deep, recursive searching. Never satisfied with the first search result. Verifies sources.
+    *   **Output**: Verified, raw data points with citations.
+2.  **The Analyst (`src/agents/analyst.py`)**
+    *   **Role**: The Strategist.
+    *   **Behavior**: Ruthless synthesis. Looks for patterns, gaps, and SWOT elements. Does NOT just summarize; creates *meaning*.
+    *   **Output**: Structured insights, contradictions, and strategic opportunities.
+3.  **The Writer (`src/agents/writer.py`)**
+    *   **Role**: The Storyteller.
+    *   **Behavior**: Clear, professional, executive-level prose. No fluff.
+    *   **Output**: The final Markdown report that "wows" the user.
+
+## III. Coding Standards
+
+### Python (The Core)
+*   ✅ **Version**: 3.12+
+*   ✅ **Style**: Strict `ruff` compliance.
+*   ✅ **Typing**: Static typing is mandatory. No `Any` unless legally unavoidable. Use `src.workflows.types` for shared models.
+*   ✅ **Async**: The world is asynchronous. Use `greenlet` / `asyncio` patterns via FastAPI.
+
+### Patterns
+*   ✅ **State Management**: Use `TypedDict` for LangGraph state (see `src/workflows/types.py`).
+*   ✅ **Configuration**: All prompts and model configs live in `src/utils/prompts/` or environment variables. No magic strings in code.
+*   ⭐ **Error Handling**: Fail gracefully. Agents should report "Intelligence Gaps" rather than crashing.
+
+## IV. Commands & Workflows
+
+### Setup
+```bash
+python -m venv venv
+source venv/bin/activate
+pip install -r requirements.txt
+```
+
+### Testing (The Gauntlet)
+Run the full suite. If this fails, you do not push.
+```bash
+./scripts/run_all_tests.sh
+```
+
+### Docker
+```bash
+docker-compose up --build
+```

src/agents/analyst.py

@@ -1,10 +1,18 @@
 """Analysis Agent for competitive intelligence and SWOT analysis."""
 
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from src.agents.base import BaseAgent
 from src.utils.cost_tracker import CostTracker
 from src.utils.logging import setup_logger
+from src.utils.prompts import (
+    ANALYST_COMPETITIVE_MATRIX,
+    ANALYST_POSITIONING,
+    ANALYST_RECOMMENDATIONS,
+    ANALYST_SWOT,
+    ANALYST_SYSTEM,
+)
+from src.workflows.types import AnalysisOutput, ResearchOutput
 
 logger = setup_logger(__name__)
 
@@ -43,29 +51,12 @@ class AnalysisAgent(BaseAgent):
 
     def get_system_prompt(self) -> str:
         """Get system prompt for analysis agent."""
-        return """You are a strategic business analyst specializing in competitive intelligence and market analysis.
-
-Your role is to analyze research data and provide actionable strategic insights.
-
-When performing analysis, you should:
-1. Use structured frameworks (SWOT, competitive matrices, positioning maps)
-2. Identify clear patterns and trends
-3. Provide specific, actionable recommendations
-4. Support conclusions with evidence from the research
-5. Consider multiple perspectives (competitors, customers, market forces)
-
-Your analysis should be:
-- Objective and data-driven
-- Structured and easy to scan
-- Focused on business impact
-- Actionable for decision-makers
-
-Use bullet points, clear headings, and strategic language."""
+        return ANALYST_SYSTEM
 
     async def run(  # type: ignore[override]
         self,
-        research_data: Dict[str, Any],
-    ) -> Dict[str, Any]:
+        research_data: ResearchOutput,
+    ) -> AnalysisOutput:
         """
         Perform comprehensive analysis on research data.
 
@@ -82,10 +73,10 @@ Use bullet points, clear headings, and strategic language."""
                 - positioning: Market positioning analysis
                 - strategic_recommendations: Action items
         """
-        company_name = research_data.get("company_name", "Unknown Company")
+        company_name = research_data["company_name"]
         logger.info(f"Starting analysis for: {company_name}")
 
-        results = {
+        results: AnalysisOutput = {
             "company_name": company_name,
             "swot": "",
             "competitive_matrix": "",
@@ -120,194 +111,49 @@ Use bullet points, clear headings, and strategic language."""
 
     async def _perform_swot_analysis(
         self,
-        research_data: Dict[str, Any],
+        research_data: ResearchOutput,
     ) -> str:
-        """
-        Generate SWOT analysis from research data.
-
-        Args:
-            research_data: Research results
-
-        Returns:
-            SWOT analysis text
-        """
-        company_name = research_data.get("company_name")
-        company_overview = research_data.get("company_overview", "")
-        competitors = research_data.get("competitors", "")
-        market_trends = research_data.get("market_trends", "")
-
-        user_message = f"""Based on the research data, perform a comprehensive SWOT analysis for {company_name}.
-
-Research Data:
-
-COMPANY OVERVIEW:
-{company_overview}
-
-COMPETITORS:
-{competitors}
-
-MARKET TRENDS:
-{market_trends}
-
-Provide a detailed SWOT analysis with:
-
-STRENGTHS (internal positive factors):
-- List 4-6 key strengths
-- Focus on competitive advantages, resources, capabilities
-
-WEAKNESSES (internal negative factors):
-- List 4-6 key weaknesses
-- Include operational limits, resource constraints, vulnerabilities
-
-OPPORTUNITIES (external positive factors):
-- List 4-6 market opportunities
-- Consider market trends, gaps, emerging needs
-
-THREATS (external negative factors):
-- List 4-6 threats
-- Include competitive threats, market risks, industry challenges
-
-Use bullet points and be specific with evidence."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Generate SWOT analysis from research data."""
+        user_message = ANALYST_SWOT.format(
+            company_name=research_data.get("company_name"),
+            company_overview=research_data.get("company_overview", ""),
+            competitors=research_data.get("competitors", ""),
+            market_trends=research_data.get("market_trends", ""),
+        )
+        return await self._invoke_llm(self._create_messages(user_message))
 
     async def _create_competitive_matrix(
         self,
-        research_data: Dict[str, Any],
+        research_data: ResearchOutput,
     ) -> str:
-        """
-        Create competitive comparison matrix.
-
-        Args:
-            research_data: Research results
-
-        Returns:
-            Competitive matrix as formatted text
-        """
-        company_name = research_data.get("company_name")
-        competitors_info = research_data.get("competitors", "")
-
-        user_message = f"""Based on the competitor research, create a competitive matrix comparing {company_name} with its main competitors.
-
-Competitor Research:
-{competitors_info}
-
-Create a comparison matrix with these dimensions:
-1. Market Share/Size
-2. Product Range
-3. Pricing Strategy
-4. Technology/Innovation
-5. Customer Segments
-6. Strengths
-7. Weaknesses
-
-Format as a clear table or structured comparison.
-Include 3-5 main competitors plus {company_name}.
-Use "High/Medium/Low" or specific data points where available."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Create competitive comparison matrix."""
+        user_message = ANALYST_COMPETITIVE_MATRIX.format(
+            company_name=research_data.get("company_name"),
+            competitors_info=research_data.get("competitors", ""),
+        )
+        return await self._invoke_llm(self._create_messages(user_message))
 
     async def _analyze_market_positioning(
         self,
-        research_data: Dict[str, Any],
+        research_data: ResearchOutput,
     ) -> str:
-        """
-        Analyze market positioning strategy.
-
-        Args:
-            research_data: Research results
-
-        Returns:
-            Positioning analysis
-        """
-        company_name = research_data.get("company_name")
-        company_overview = research_data.get("company_overview", "")
-        competitors = research_data.get("competitors", "")
-
-        user_message = f"""Analyze the market positioning of {company_name}.
-
-Company Overview:
-{company_overview}
-
-Competitive Landscape:
-{competitors}
-
-Provide analysis covering:
-
-1. CURRENT POSITIONING
-   - How is {company_name} currently positioned in the market?
-   - What is their value proposition?
-   - What customer segments do they target?
-
-2. COMPETITIVE DIFFERENTIATION
-   - What makes {company_name} different from competitors?
-   - What is their unique selling proposition (USP)?
-   - Where do they fit in the competitive landscape?
-
-3. POSITIONING GAPS
-   - Are there market segments they're missing?
-   - Are there positioning opportunities?
-   - How could they strengthen their position?
-
-Be specific and strategic."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Analyze market positioning strategy."""
+        user_message = ANALYST_POSITIONING.format(
+            company_name=research_data.get("company_name"),
+            company_overview=research_data.get("company_overview", ""),
+            competitors=research_data.get("competitors", ""),
+        )
+        return await self._invoke_llm(self._create_messages(user_message))
 
     async def _generate_recommendations(
         self,
-        research_data: Dict[str, Any],
+        research_data: ResearchOutput,
         swot: str,
     ) -> str:
-        """
-        Generate strategic recommendations.
-
-        Args:
-            research_data: Research results
-            swot: SWOT analysis
-
-        Returns:
-            Strategic recommendations
-        """
-        company_name = research_data.get("company_name")
-        market_trends = research_data.get("market_trends", "")
-
-        user_message = f"""Based on the SWOT analysis and market trends, provide strategic recommendations for {company_name}.
-
-SWOT ANALYSIS:
-{swot}
-
-MARKET TRENDS:
-{market_trends}
-
-Provide 5-7 actionable strategic recommendations organized by priority:
-
-HIGH PRIORITY (immediate action needed):
-- Recommendation 1 (with rationale)
-- Recommendation 2 (with rationale)
-
-MEDIUM PRIORITY (next 6-12 months):
-- Recommendation 3 (with rationale)
-- Recommendation 4 (with rationale)
-
-LONG-TERM (strategic initiatives):
-- Recommendation 5 (with rationale)
-
-Each recommendation should:
-- Be specific and actionable
-- Address a key opportunity or threat
-- Leverage strengths or address weaknesses
-- Include expected impact"""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Generate strategic recommendations."""
+        user_message = ANALYST_RECOMMENDATIONS.format(
+            company_name=research_data.get("company_name"),
+            swot=swot,
+            market_trends=research_data.get("market_trends", ""),
+        )
+        return await self._invoke_llm(self._create_messages(user_message))

src/agents/researcher.py

@@ -1,11 +1,18 @@
 """Research Agent for gathering market intelligence data."""
 
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from src.agents.base import BaseAgent
 from src.tools.search import TavilySearchTool
 from src.utils.cost_tracker import CostTracker
 from src.utils.logging import setup_logger
+from src.utils.prompts import (
+    RESEARCHER_ANALYZE_COMPANY,
+    RESEARCHER_ANALYZE_COMPETITORS,
+    RESEARCHER_ANALYZE_TRENDS,
+    RESEARCHER_SYSTEM,
+)
+from src.workflows.types import ResearchOutput
 
 logger = setup_logger(__name__)
 
@@ -46,26 +53,14 @@ class ResearchAgent(BaseAgent):
 
     def get_system_prompt(self) -> str:
         """Get system prompt for research agent."""
-        return """You are a professional business research analyst specializing in competitive intelligence.
-
-Your role is to gather and synthesize information about companies, markets, and competitors from web sources.
-
-When analyzing search results, you should:
-1. Focus on factual, verifiable information
-2. Identify key data points: revenue, employees, products, market position
-3. Note dates and sources for important claims
-4. Distinguish between facts and opinions
-5. Highlight competitive advantages and weaknesses
-
-Provide your analysis in a structured format with clear sections and bullet points.
-Always cite sources when making specific claims."""
+        return RESEARCHER_SYSTEM
 
     async def run(  # type: ignore[override]
         self,
         company_name: str,
         industry: Optional[str] = None,
         research_depth: str = "comprehensive",
-    ) -> Dict[str, Any]:
+    ) -> ResearchOutput:
         """
         Gather research data about a company.
 
@@ -83,7 +78,7 @@ Always cite sources when making specific claims."""
         """
         logger.info(f"Starting research for: {company_name}")
 
-        results: Dict[str, Any] = {
+        results: ResearchOutput = {
             "company_name": company_name,
             "industry": industry,
             "company_overview": "",
@@ -99,8 +94,7 @@ Always cite sources when making specific claims."""
                 max_results=10 if research_depth == "comprehensive" else 5,
             )
 
-            if isinstance(results["raw_sources"], list):
-                results["raw_sources"].extend(company_data.get("results", []))
+            results["raw_sources"].extend(company_data.get("results", []))
 
             # Analyze company data with LLM
             company_context = self.search_tool.format_results_for_llm(company_data)
@@ -116,8 +110,7 @@ Always cite sources when making specific claims."""
                 max_results=10 if research_depth == "comprehensive" else 5,
             )
 
-            if isinstance(results["raw_sources"], list):
-                results["raw_sources"].extend(competitor_data.get("results", []))
+            results["raw_sources"].extend(competitor_data.get("results", []))
 
             competitor_context = self.search_tool.format_results_for_llm(
                 competitor_data
@@ -134,8 +127,7 @@ Always cite sources when making specific claims."""
                     max_results=8 if research_depth == "comprehensive" else 4,
                 )
 
-                if isinstance(results["raw_sources"], list):
-                    results["raw_sources"].extend(trend_data.get("results", []))
+                results["raw_sources"].extend(trend_data.get("results", []))
 
                 trend_context = self.search_tool.format_results_for_llm(trend_data)
                 trend_analysis = await self._analyze_trends(industry, trend_context)
@@ -157,97 +149,30 @@ Always cite sources when making specific claims."""
         company_name: str,
         search_context: str,
     ) -> str:
-        """
-        Analyze company information from search results.
-
-        Args:
-            company_name: Company name
-            search_context: Formatted search results
-
-        Returns:
-            Structured company analysis
-        """
-        user_message = f"""Analyze the following search results about {company_name}.
-
-Provide a structured analysis covering:
-1. Company Overview (founded, headquarters, size)
-2. Products & Services (main offerings)
-3. Business Model (how they make money)
-4. Market Position (market share, ranking)
-5. Key Metrics (revenue, employees, growth)
-
-Search Results:
-{search_context}
-
-Provide your analysis in clear sections with bullet points. Cite sources for specific claims."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Analyze company information from search results."""
+        user_message = RESEARCHER_ANALYZE_COMPANY.format(
+            company_name=company_name, search_context=search_context
+        )
+        return await self._invoke_llm(self._create_messages(user_message))
 
     async def _analyze_competitors(
         self,
         company_name: str,
         search_context: str,
     ) -> str:
-        """
-        Analyze competitor landscape.
-
-        Args:
-            company_name: Target company
-            search_context: Formatted search results
-
-        Returns:
-            Competitor analysis
-        """
-        user_message = f"""Analyze the competitive landscape for {company_name}.
-
-Based on the search results, identify:
-1. Main Competitors (list 3-5 key competitors)
-2. Competitive Positioning (how each differs)
-3. Market Dynamics (who leads, who follows)
-4. Differentiation Factors (what makes each unique)
-
-Search Results:
-{search_context}
-
-Format as a structured list with clear comparisons."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Analyze competitor landscape."""
+        user_message = RESEARCHER_ANALYZE_COMPETITORS.format(
+            company_name=company_name, search_context=search_context
+        )
+        return await self._invoke_llm(self._create_messages(user_message))
 
     async def _analyze_trends(
         self,
         industry: str,
         search_context: str,
     ) -> str:
-        """
-        Analyze market trends.
-
-        Args:
-            industry: Industry name
-            search_context: Formatted search results
-
-        Returns:
-            Trend analysis
-        """
-        user_message = f"""Analyze market trends for the {industry} industry.
-
-Identify:
-1. Key Trends (major shifts in the market)
-2. Growth Drivers (what's fueling growth)
-3. Challenges (obstacles facing the industry)
-4. Future Outlook (predictions for next 1-2 years)
-
-Search Results:
-{search_context}
-
-Provide analysis with clear trends and supporting evidence."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Analyze market trends."""
+        user_message = RESEARCHER_ANALYZE_TRENDS.format(
+            industry=industry, search_context=search_context
+        )
+        return await self._invoke_llm(self._create_messages(user_message))

src/agents/writer.py

@@ -1,11 +1,17 @@
 """Writer Agent for generating professional market intelligence reports."""
 
 from datetime import datetime
-from typing import Any, Dict, Optional
+from typing import Optional
 
 from src.agents.base import BaseAgent
 from src.utils.cost_tracker import CostTracker
 from src.utils.logging import setup_logger
+from src.utils.prompts import (
+    WRITER_EXECUTIVE_SUMMARY,
+    WRITER_FULL_REPORT,
+    WRITER_SYSTEM,
+)
+from src.workflows.types import AnalysisOutput, ReportOutput, ResearchOutput
 
 logger = setup_logger(__name__)
 
@@ -43,33 +49,13 @@ class WriterAgent(BaseAgent):
 
     def get_system_prompt(self) -> str:
         """Get system prompt for writer agent."""
-        return """You are a professional business report writer specializing in market intelligence and competitive analysis.
-
-Your role is to transform research and analysis into polished, executive-ready reports.
-
-When writing reports, you should:
-1. Use clear, professional business language
-2. Structure content logically with proper headings
-3. Include executive summaries for busy stakeholders
-4. Use bullet points and tables for scannability
-5. Cite sources properly
-6. Make insights actionable
-
-Report format guidelines:
-- Use markdown formatting
-- Include clear section headers (#, ##, ###)
-- Use tables for competitive comparisons
-- Include bullet points for lists
-- Add citations [source]
-- Keep executive summary to 200-300 words
-
-Write for senior executives and decision-makers."""
+        return WRITER_SYSTEM
 
     async def run(  # type: ignore[override]
         self,
-        research_data: Dict[str, Any],
-        analysis_data: Dict[str, Any],
-    ) -> Dict[str, Any]:
+        research_data: ResearchOutput,
+        analysis_data: AnalysisOutput,
+    ) -> ReportOutput:
         """
         Generate comprehensive market intelligence report.
 
@@ -83,7 +69,7 @@ Write for senior executives and decision-makers."""
                 - full_report: Complete markdown report
                 - metadata: Report metadata (date, sources count, etc.)
         """
-        company_name = research_data.get("company_name", "Unknown Company")
+        company_name = research_data.get("company_name")
         logger.info(f"Starting report generation for: {company_name}")
 
         try:
@@ -119,133 +105,41 @@ Write for senior executives and decision-makers."""
 
     async def _write_executive_summary(
         self,
-        research_data: Dict[str, Any],
-        analysis_data: Dict[str, Any],
+        research_data: ResearchOutput,
+        analysis_data: AnalysisOutput,
     ) -> str:
-        """
-        Write executive summary (200-300 words).
-
-        Args:
-            research_data: Research results
-            analysis_data: Analysis results
-
-        Returns:
-            Executive summary text
-        """
-        company_name = research_data.get("company_name")
-
-        user_message = f"""Write a concise executive summary for a market intelligence report on {company_name}.
-
-Use this information:
-
-COMPANY OVERVIEW:
-{research_data.get("company_overview", "")}
-
-KEY INSIGHTS FROM SWOT:
-{analysis_data.get("swot", "")}
-
-STRATEGIC RECOMMENDATIONS:
-{analysis_data.get("strategic_recommendations", "")}
-
-Requirements:
-- 200-300 words
-- Cover: company overview, market position, key findings, main recommendations
-- Written for senior executives (clear, actionable)
-- Professional business tone
-
-Start directly with content (no "Executive Summary" heading)."""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        """Write executive summary (200-300 words)."""
+        user_message = WRITER_EXECUTIVE_SUMMARY.format(
+            company_name=research_data.get("company_name"),
+            company_overview=research_data.get("company_overview", ""),
+            swot=analysis_data.get("swot", ""),
+            strategic_recommendations=analysis_data.get(
+                "strategic_recommendations", ""
+            ),
+        )
+        return await self._invoke_llm(self._create_messages(user_message))
 
     async def _write_full_report(
         self,
-        research_data: Dict[str, Any],
-        analysis_data: Dict[str, Any],
+        research_data: ResearchOutput,
+        analysis_data: AnalysisOutput,
         exec_summary: str,
     ) -> str:
-        """
-        Write complete markdown report.
-
-        Args:
-            research_data: Research results
-            analysis_data: Analysis results
-            exec_summary: Executive summary
-
-        Returns:
-            Full report in markdown format
-        """
+        """Write complete markdown report."""
         company_name = research_data.get("company_name")
-        industry = research_data.get("industry") or "Market"
-
-        # Build comprehensive context
-        context = f"""
-COMPANY: {company_name}
-INDUSTRY: {industry}
-
-RESEARCH DATA:
-Company Overview: {research_data.get("company_overview", "")}
-Competitors: {research_data.get("competitors", "")}
-Market Trends: {research_data.get("market_trends", "")}
-
-ANALYSIS DATA:
-SWOT: {analysis_data.get("swot", "")}
-Competitive Matrix: {analysis_data.get("competitive_matrix", "")}
-Market Positioning: {analysis_data.get("positioning", "")}
-Strategic Recommendations: {analysis_data.get("strategic_recommendations", "")}
-
-EXECUTIVE SUMMARY:
-{exec_summary}
-"""
-
-        user_message = f"""Create a comprehensive market intelligence report for {company_name} in markdown format.
-
-Use all the provided research and analysis data.
-
-Structure the report as follows:
-
-# Market Intelligence Report: {company_name}
-
-## Executive Summary
-[Insert the provided executive summary]
 
-## 1. Company Overview
-[Detailed company information from research]
-
-## 2. Competitive Landscape
-[Competitor analysis and competitive matrix]
-
-## 3. SWOT Analysis
-[Detailed SWOT with clear sections: Strengths, Weaknesses, Opportunities, Threats]
-
-## 4. Market Positioning
-[Positioning analysis and differentiation]
-
-## 5. Market Trends & Insights
-[Industry trends and market dynamics]
-
-## 6. Strategic Recommendations
-[Prioritized recommendations with rationale]
-
-## 7. Sources
-[List key sources used]
-
----
-Report generated: {datetime.now().strftime("%B %d, %Y")}
-
-Data to use:
-{context}
-
-Format requirements:
-- Use proper markdown (headers, bullets, tables)
-- Make it professional and polished
-- Include all relevant details
-- Cite sources where appropriate
-- Make it actionable for executives"""
-
-        messages = self._create_messages(user_message)
-        response = await self._invoke_llm(messages)
-
-        return response
+        user_message = WRITER_FULL_REPORT.format(
+            company_name=company_name,
+            exec_summary=exec_summary,
+            company_overview=research_data.get("company_overview", ""),
+            competitors=research_data.get("competitors", ""),
+            competitive_matrix=analysis_data.get("competitive_matrix", ""),
+            swot=analysis_data.get("swot", ""),
+            positioning=analysis_data.get("positioning", ""),
+            market_trends=research_data.get("market_trends", ""),
+            strategic_recommendations=analysis_data.get(
+                "strategic_recommendations", ""
+            ),
+            date=datetime.now().strftime("%B %d, %Y"),
+        )
+        return await self._invoke_llm(self._create_messages(user_message))

src/utils/prompts.py

@@ -0,0 +1,282 @@
+"""
+Centralized prompt management for the Agentic Market Research system.
+
+"Words are the source of misunderstandings." - Antoine de Saint-Exupéry
+But here, they are the source of intelligence.
+"""
+
+# ==============================================================================
+# RESEARCH AGENT PROMPTS
+# ==============================================================================
+
+RESEARCHER_SYSTEM = """You are a professional business research analyst specializing in competitive intelligence.
+
+Your role is to gather and synthesize information about companies, markets, and competitors from web sources.
+
+When analyzing search results, you should:
+1. Focus on factual, verifiable information
+2. Identify key data points: revenue, employees, products, market position
+3. Note dates and sources for important claims
+4. Distinguish between facts and opinions
+5. Highlight competitive advantages and weaknesses
+
+Provide your analysis in a structured format with clear sections and bullet points.
+Always cite sources when making specific claims."""
+
+RESEARCHER_ANALYZE_COMPANY = """Analyze the following search results about {company_name}.
+
+Provide a structured analysis covering:
+1. Company Overview (founded, headquarters, size)
+2. Products & Services (main offerings)
+3. Business Model (how they make money)
+4. Market Position (market share, ranking)
+5. Key Metrics (revenue, employees, growth)
+
+Search Results:
+{search_context}
+
+Provide your analysis in clear sections with bullet points. Cite sources for specific claims."""
+
+RESEARCHER_ANALYZE_COMPETITORS = """Analyze the competitive landscape for {company_name}.
+
+Based on the search results, identify:
+1. Main Competitors (list 3-5 key competitors)
+2. Competitive Positioning (how each differs)
+3. Market Dynamics (who leads, who follows)
+4. Differentiation Factors (what makes each unique)
+
+Search Results:
+{search_context}
+
+Format as a structured list with clear comparisons."""
+
+RESEARCHER_ANALYZE_TRENDS = """Analyze market trends for the {industry} industry.
+
+Identify:
+1. Key Trends (major shifts in the market)
+2. Growth Drivers (what's fueling growth)
+3. Challenges (obstacles facing the industry)
+4. Future Outlook (predictions for next 1-2 years)
+
+Search Results:
+{search_context}
+
+Provide analysis with clear trends and supporting evidence."""
+
+# ==============================================================================
+# ANALYST AGENT PROMPTS
+# ==============================================================================
+
+ANALYST_SYSTEM = """You are a strategic business analyst specializing in competitive intelligence and market analysis.
+
+Your role is to analyze research data and provide actionable strategic insights.
+
+When performing analysis, you should:
+1. Use structured frameworks (SWOT, competitive matrices, positioning maps)
+2. Identify clear patterns and trends
+3. Provide specific, actionable recommendations
+4. Support conclusions with evidence from the research
+5. Consider multiple perspectives (competitors, customers, market forces)
+
+Your analysis should be:
+- Objective and data-driven
+- Structured and easy to scan
+- Focused on business impact
+- Actionable for decision-makers
+
+Use bullet points, clear headings, and strategic language."""
+
+ANALYST_SWOT = """Based on the research data, perform a comprehensive SWOT analysis for {company_name}.
+
+Research Data:
+
+COMPANY OVERVIEW:
+{company_overview}
+
+COMPETITORS:
+{competitors}
+
+MARKET TRENDS:
+{market_trends}
+
+Provide a detailed SWOT analysis with:
+
+STRENGTHS (internal positive factors):
+- List 4-6 key strengths
+- Focus on competitive advantages, resources, capabilities
+
+WEAKNESSES (internal negative factors):
+- List 4-6 key weaknesses
+- Include operational limits, resource constraints, vulnerabilities
+
+OPPORTUNITIES (external positive factors):
+- List 4-6 market opportunities
+- Consider market trends, gaps, emerging needs
+
+THREATS (external negative factors):
+- List 4-6 threats
+- Include competitive threats, market risks, industry challenges
+
+Use bullet points and be specific with evidence."""
+
+ANALYST_COMPETITIVE_MATRIX = """Based on the competitor research, create a competitive matrix comparing {company_name} with its main competitors.
+
+Competitor Research:
+{competitors_info}
+
+Create a comparison matrix with these dimensions:
+1. Market Share/Size
+2. Product Range
+3. Pricing Strategy
+4. Technology/Innovation
+5. Customer Segments
+6. Strengths
+7. Weaknesses
+
+Format as a clear table or structured comparison.
+Include 3-5 main competitors plus {company_name}.
+Use "High/Medium/Low" or specific data points where available."""
+
+ANALYST_POSITIONING = """Analyze the market positioning of {company_name}.
+
+Company Overview:
+{company_overview}
+
+Competitive Landscape:
+{competitors}
+
+Provide analysis covering:
+
+1. CURRENT POSITIONING
+   - How is {company_name} currently positioned in the market?
+   - What is their value proposition?
+   - What customer segments do they target?
+
+2. COMPETITIVE DIFFERENTIATION
+   - What makes {company_name} different from competitors?
+   - What is their unique selling proposition (USP)?
+   - Where do they fit in the competitive landscape?
+
+3. POSITIONING GAPS
+   - Are there market segments they're missing?
+   - Are there positioning opportunities?
+   - How could they strengthen their position?
+
+Be specific and strategic."""
+
+ANALYST_RECOMMENDATIONS = """Based on the SWOT analysis and market trends, provide strategic recommendations for {company_name}.
+
+SWOT ANALYSIS:
+{swot}
+
+MARKET TRENDS:
+{market_trends}
+
+Provide 5-7 actionable strategic recommendations organized by priority:
+
+HIGH PRIORITY (immediate action needed):
+- Recommendation 1 (with rationale)
+- Recommendation 2 (with rationale)
+
+MEDIUM PRIORITY (next 6-12 months):
+- Recommendation 3 (with rationale)
+- Recommendation 4 (with rationale)
+
+LONG-TERM (strategic initiatives):
+- Recommendation 5 (with rationale)
+
+Each recommendation should:
+- Be specific and actionable
+- Address a key opportunity or threat
+- Leverage strengths or address weaknesses
+- Include expected impact"""
+
+# ==============================================================================
+# WRITER AGENT PROMPTS
+# ==============================================================================
+
+WRITER_SYSTEM = """You are a professional business report writer specializing in market intelligence and competitive analysis.
+
+Your role is to transform research and analysis into polished, executive-ready reports.
+
+When writing reports, you should:
+1. Use clear, professional business language
+2. Structure content logically with proper headings
+3. Include executive summaries for busy stakeholders
+4. Use bullet points and tables for scannability
+5. Cite sources properly
+6. Make insights actionable
+
+Report format guidelines:
+- Use markdown formatting
+- Include clear section headers (#, ##, ###)
+- Use tables for competitive comparisons
+- Include bullet points for lists
+- Add citations [source]
+- Keep executive summary to 200-300 words
+
+Write for senior executives and decision-makers."""
+
+WRITER_EXECUTIVE_SUMMARY = """Write a concise executive summary for a market intelligence report on {company_name}.
+
+Use this information:
+
+COMPANY OVERVIEW:
+{company_overview}
+
+KEY INSIGHTS FROM SWOT:
+{swot}
+
+STRATEGIC RECOMMENDATIONS:
+{strategic_recommendations}
+
+Requirements:
+- 200-300 words
+- Cover: company overview, market position, key findings, main recommendations
+- Written for senior executives (clear, actionable)
+- Professional business tone
+
+Start directly with content (no "Executive Summary" heading)."""
+
+WRITER_FULL_REPORT = """Create a comprehensive market intelligence report for {company_name} in markdown format.
+
+Use all the provided research and analysis data.
+
+Structure the report as follows:
+
+# Market Intelligence Report: {company_name}
+
+## Executive Summary
+{exec_summary}
+
+## 1. Company Overview
+{company_overview}
+
+## 2. Competitive Landscape
+{competitors}
+{competitive_matrix}
+
+## 3. SWOT Analysis
+{swot}
+
+## 4. Market Positioning
+{positioning}
+
+## 5. Market Trends & Insights
+{market_trends}
+
+## 6. Strategic Recommendations
+{strategic_recommendations}
+
+## 7. Sources
+[List key sources used]
+
+---
+Report generated: {date}
+
+Format requirements:
+- Use proper markdown (headers, bullets, tables)
+- Make it professional and polished
+- Include all relevant details
+- Cite sources where appropriate
+- Make it actionable for executives"""

src/workflows/market_analysis.py

@@ -2,6 +2,7 @@
 
 from langgraph.graph import StateGraph, END
 from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
+from langgraph.checkpoint.memory import MemorySaver
 
 from src.workflows.types import IntelligenceState
 from src.agents.researcher import ResearchAgent
@@ -108,8 +109,8 @@ class MarketIntelligenceWorkflow:
             return {
                 "current_agent": "research",
                 "research_data": research_results,
-                "competitors": research_results.get("competitors", []),
-                "market_trends": research_results.get("market_trends", {}),
+                "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,
             }
@@ -137,11 +138,11 @@ class MarketIntelligenceWorkflow:
             # Update state
             return {
                 "current_agent": "analysis",
-                "swot": analysis_results.get("swot", {}),
-                "competitive_matrix": analysis_results.get("competitive_matrix", {}),
-                "positioning": analysis_results.get("positioning", {}),
+                "swot": analysis_results.get("swot", ""),
+                "competitive_matrix": analysis_results.get("competitive_matrix", ""),
+                "positioning": analysis_results.get("positioning", ""),
                 "strategic_recommendations": analysis_results.get(
-                    "strategic_recommendations", {}
+                    "strategic_recommendations", ""
                 ),
             }
 
@@ -167,11 +168,12 @@ class MarketIntelligenceWorkflow:
             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", {}),
+                    "company_name": state["company_name"],
+                    "swot": state.get("swot", ""),
+                    "competitive_matrix": state.get("competitive_matrix", ""),
+                    "positioning": state.get("positioning", ""),
                     "strategic_recommendations": state.get(
-                        "strategic_recommendations", {}
+                        "strategic_recommendations", ""
                     ),
                 },
             )
@@ -265,14 +267,21 @@ class MarketIntelligenceWorkflow:
             "company_name": company_name,
             "industry": industry,
             "research_depth": research_depth,
-            "research_data": {},
-            "competitors": [],
-            "market_trends": {},
+            "research_data": {
+                "company_name": company_name,
+                "industry": industry,
+                "company_overview": "",
+                "competitors": "",
+                "market_trends": "",
+                "raw_sources": [],
+            },
+            "competitors": "",
+            "market_trends": "",
             "raw_sources": [],
-            "swot": {},
-            "competitive_matrix": {},
-            "positioning": {},
-            "strategic_recommendations": {},
+            "swot": "",
+            "competitive_matrix": "",
+            "positioning": "",
+            "strategic_recommendations": "",
             "executive_summary": "",
             "full_report": "",
             "report_metadata": {},
@@ -290,11 +299,16 @@ class MarketIntelligenceWorkflow:
         config = {"configurable": {"thread_id": thread_id or "default"}}
 
         try:
-            async with AsyncSqliteSaver.from_conn_string(
-                self.checkpoint_path
-            ) as checkpointer:
-                workflow = self.graph_builder.compile(checkpointer=checkpointer)
+            if self.checkpoint_path == ":memory:":
+                memory_checkpointer = MemorySaver()
+                workflow = self.graph_builder.compile(checkpointer=memory_checkpointer)
                 final_state = await workflow.ainvoke(initial_state, config)  # type: ignore[arg-type]
+            else:
+                async with AsyncSqliteSaver.from_conn_string(
+                    self.checkpoint_path
+                ) as checkpointer:
+                    workflow = self.graph_builder.compile(checkpointer=checkpointer)
+                    final_state = await workflow.ainvoke(initial_state, config)  # type: ignore[arg-type]
 
             logger.info(f"Workflow complete. Cost: ${final_state['total_cost']:.4f}")
             return final_state

src/workflows/types.py

@@ -1,9 +1,38 @@
 """State definitions for LangGraph workflow."""
 
-from typing import Annotated, Literal, TypedDict
+from typing import Annotated, Any, Dict, List, Literal, TypedDict, Union
 import operator
 
 
+class ResearchOutput(TypedDict):
+    """Output structure for Research Agent."""
+
+    company_name: str
+    industry: Union[str, None]
+    company_overview: str
+    competitors: str
+    market_trends: str
+    raw_sources: List[Any]
+
+
+class AnalysisOutput(TypedDict):
+    """Output structure for Analysis Agent."""
+
+    company_name: str
+    swot: str
+    competitive_matrix: str
+    positioning: str
+    strategic_recommendations: str
+
+
+class ReportOutput(TypedDict):
+    """Output structure for Writer Agent."""
+
+    executive_summary: str
+    full_report: str
+    metadata: Dict[str, Any]
+
+
 class IntelligenceState(TypedDict):
     """
     State for market intelligence workflow.
@@ -13,34 +42,34 @@ class IntelligenceState(TypedDict):
 
     # Input
     company_name: str
-    industry: str | None
+    industry: Union[str, None]
     research_depth: str  # "basic" or "comprehensive"
 
     # Research phase outputs
-    research_data: dict
-    competitors: list[dict]
-    market_trends: dict
-    raw_sources: list[dict]
+    research_data: ResearchOutput
+    competitors: str  # Markdown string from analysis
+    market_trends: str  # Markdown string from analysis
+    raw_sources: List[Any]
 
     # Analysis phase outputs
-    swot: dict
-    competitive_matrix: dict
-    positioning: dict
-    strategic_recommendations: dict
+    swot: str
+    competitive_matrix: str
+    positioning: str
+    strategic_recommendations: str
 
     # Writing phase outputs
     executive_summary: str
     full_report: str
-    report_metadata: dict
+    report_metadata: Dict[str, Any]
 
     # 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
+    errors: Annotated[List[str], operator.add]  # Accumulate errors across nodes
 
     # Human-in-the-loop
-    human_feedback: str | None
+    human_feedback: Union[str, None]
     approved: bool
     revision_count: int

tests/integration/test_workflow_integration.py

@@ -12,7 +12,7 @@ class TestWorkflowErrorRecovery:
 
     async def test_research_error_ends_workflow(self):
         """Test workflow ends gracefully when research fails."""
-        workflow = MarketIntelligenceWorkflow()
+        workflow = MarketIntelligenceWorkflow(checkpoint_path=":memory:")
 
         # Mock research to fail
         # Mock research to fail
@@ -29,7 +29,9 @@ class TestWorkflowErrorRecovery:
 
     async def test_budget_exceeded_stops_workflow(self):
         """Test workflow stops when budget is exceeded."""
-        workflow = MarketIntelligenceWorkflow(max_budget=0.001)
+        workflow = MarketIntelligenceWorkflow(
+            max_budget=0.001, checkpoint_path=":memory:"
+        )
 
         # Mock research to succeed with some cost
         # Mock research to succeed with some cost
@@ -37,9 +39,11 @@ class TestWorkflowErrorRecovery:
             workflow.cost_tracker.track_usage("openai/gpt-5-mini", 10000, 5000)
             return {
                 "company_name": "Mock Company",
-                "competitors": [],
-                "market_trends": {},
+                "competitors": "Competitor A, Competitor B",
+                "market_trends": "Market is growing",
                 "raw_sources": [],
+                "industry": "Tech",
+                "company_overview": "Overview",
             }
 
         workflow.research_agent.run = AsyncMock(side_effect=mock_run)
@@ -58,25 +62,28 @@ class TestWorkflowIntegration:
 
     async def test_workflow_with_mocked_agents(self):
         """Test complete workflow with mocked agent responses."""
-        workflow = MarketIntelligenceWorkflow()
+        workflow = MarketIntelligenceWorkflow(checkpoint_path=":memory:")
 
         # Mock all agents
         # Mock all agents
         workflow.research_agent.run = AsyncMock(
             return_value={
                 "company_name": "Test Company",
-                "competitors": [{"name": "Competitor A"}],
-                "market_trends": {"trend": "growing"},
+                "competitors": "Competitor A",
+                "market_trends": "Market growing",
                 "raw_sources": [{"url": "test.com"}],
+                "industry": "Tech",
+                "company_overview": "Overview",
             }
         )
 
         workflow.analysis_agent.run = AsyncMock(
             return_value={
-                "swot": {"strengths": ["good"]},
-                "competitive_matrix": {},
-                "positioning": {},
-                "strategic_recommendations": {},
+                "company_name": "Test Company",
+                "swot": "Strengths: Good",
+                "competitive_matrix": "Matrix data",
+                "positioning": "Leader",
+                "strategic_recommendations": "Buy low sell high",
             }
         )