Multi-agent AutoGen teams for market research automation

AutoGen agent team market research automation lets you replace manual analyst work with a reproducible pipeline of specialized agents that plan, search, extract, and synthesize. This tutorial walks through building a four-agent team — planner, researcher, extractor, synthesizer — that produces a structured competitive landscape report from a single prompt. You’ll get runnable code, a verification checklist, and patterns that scale to production workloads.

Step 1: Set up the environment and dependencies

Create a fresh virtual environment and install the minimal set of packages. AutoGen 0.2+ uses a different import path than the legacy 0.1 series, so pin accordingly.

python -m venv .venv
source .venv/bin/activate
pip install "autogen-agentchat>=0.2" "autogen-ext[openai]>=0.2" \
    "httpx>=0.27" "beautifulsoup4>=4.12" "lxml>=4.9" \
    "pydantic>=2.7" "tenacity>=8.2" "python-dotenv>=1.0"

Create a .env file with your API keys. If you route through a gateway that normalizes provider interfaces, you only need one base URL and key.

# .env
OPENAI_API_KEY=sk-...
OPENAI_BASE_URL=https://api.openai.com/v1
# Optional: if you use a gateway like n4n.ai that forwards cache-control hints
# OPENAI_BASE_URL=https://api.n4n.ai/v1

Verify the install:

# verify_install.py
import autogen
from autogen_agentchat.agents import AssistantAgent
from autogen_ext.models.openai import OpenAIChatCompletionClient

client = OpenAIChatCompletionClient(model="gpt-4o-mini")
agent = AssistantAgent(name="test", model_client=client)
print("AutoGen version:", autogen.__version__)
print("Client ready:", client.model_info)

Run python verify_install.py — you should see version info and model metadata without errors.

Step 2: Define the agent roles and capabilities

Each agent gets a focused system message and a typed tool set. Use Pydantic models for tool arguments and return values so the orchestrator can validate handoffs.

# agents.py
from pydantic import BaseModel, Field
from typing import List, Optional
from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.tools import Tool
from autogen_ext.models.openai import OpenAIChatCompletionClient
import httpx
from bs4 import BeautifulSoup
import json
import re

model_client = OpenAIChatCompletionClient(model="gpt-4o-mini")

class SearchQuery(BaseModel):
    query: str = Field(description="Search query string")
    num_results: int = Field(default=10, ge=1, le=20)

class SearchResult(BaseModel):
    url: str
    title: str
    snippet: str

class CompanyProfile(BaseModel):
    name: str
    website: str
    pricing_model: Optional[str] = None
    key_features: List[str] = []
    target_market: Optional[str] = None
    funding_stage: Optional[str] = None
    employee_count: Optional[str] = None

class ResearchPlan(BaseModel):
    competitors: List[str]
    search_queries: List[str]
    focus_areas: List[str]

# --- Tools ---

async def web_search(query: SearchQuery) -> List[SearchResult]:
    """DuckDuckGo HTML scrape — replace with SerpAPI/Brave in production."""
    url = "https://html.duckduckgo.com/html/"
    params = {"q": query.query, "kl": "us-en"}
    headers = {"User-Agent": "Mozilla/5.0 (compatible; MarketResearchBot/1.0)"}
    async with httpx.AsyncClient(timeout=30.0) as client:
        resp = await client.post(url, data=params, headers=headers)
    soup = BeautifulSoup(resp.text, "lxml")
    results = []
    for result in soup.select(".result__snippet")[:query.num_results]:
        link = result.find_previous("a", class_="result__url")
        title_el = result.find_previous("a", class_="result__snippet")
        title = title_el.get_text(strip=True) if title_el else ""
        snippet = result.get_text(strip=True)
        href = link.get("href") if link else ""
        if href and snippet:
            results.append(SearchResult(url=href, title=title, snippet=snippet))
    return results

async def fetch_page(url: str) -> str:
    """Fetch and extract main text content from a page."""
    async with httpx.AsyncClient(timeout=30.0, follow_redirects=True) as client:
        resp = await client.get(url, headers={"User-Agent": "Mozilla/5.0"})
    soup = BeautifulSoup(resp.text, "lxml")
    # Remove nav, footer, scripts
    for tag in soup(["nav", "footer", "script", "style", "aside"]):
        tag.decompose()
    text = soup.get_text(separator="\n", strip=True)
    return text[:15000]  # truncate for context window

# --- Agents ---

planner = AssistantAgent(
    name="planner",
    model_client=model_client,
    system_message="""You are a market research planner. Given a market category (e.g., "AI code review tools"),
    output a ResearchPlan JSON with:
    - competitors: 5-8 known or likely competitor names
    - search_queries: 6-10 specific queries to uncover pricing, features, positioning
    - focus_areas: 4-6 dimensions to compare (pricing, integrations, target segment, etc.)
    Be specific. Avoid generic queries.""",
    tools=[],
)

researcher = AssistantAgent(
    name="researcher",
    model_client=model_client,
    system_message="""You execute search queries and return raw SearchResult lists.
    Call web_search for each query. Do not summarize — just return results.""",
    tools=[web_search],
)

extractor = AssistantAgent(
    name="extractor",
    model_client=model_client,
    system_message="""You receive a company name and a list of SearchResult objects.
    Fetch each URL with fetch_page, then extract a CompanyProfile.
    If a field is not found, omit it. Return valid JSON only.""",
    tools=[fetch_page],
)

synthesizer = AssistantAgent(
    name="synthesizer",
    model_client=model_client,
    system_message="""You receive a list of CompanyProfile objects and the original market category.
    Produce a final markdown report with:
    1. Executive summary (3-4 bullets)
    2. Comparison table (markdown) across focus areas
    3. Per-company deep dive (2-3 paragraphs each)
    4. Gaps and opportunities
    5. Sources appendix with URLs
    Be specific, cite sources inline with [url], and flag uncertain data.""",
    tools=[],
)

Step 3: Build the research workflow

AutoGen 0.2 uses a team-based API. Define a RoundRobinGroupChat with a termination condition that fires when the synthesizer emits a final report.

# workflow.py
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_agentchat.conditions import TextMentionTermination
from autogen_agentchat.messages import TextMessage
from agents import planner, researcher, extractor, synthesizer, ResearchPlan, CompanyProfile
import json
import asyncio

termination = TextMentionTermination("FINAL_REPORT_READY")

team = RoundRobinGroupChat(
    participants=[planner, researcher, extractor, synthesizer],
    termination_condition=termination,
    max_turns=25,
)

async def run_market_research(category: str) -> str:
    """Execute the full pipeline and return the final markdown report."""
    task = f"""Market category: {category}

    Planner: Create a ResearchPlan for this category. Output ONLY valid JSON matching the ResearchPlan schema.
    Researcher: For each query in the plan, call web_search. Return all results.
    Extractor: For each competitor in the plan, use the search results to build a CompanyProfile.
    Fetch pages as needed. Output a JSON list of CompanyProfile objects.
    Synthesizer: Write the final report. End with the exact string FINAL_REPORT_READY."""
    
    result = await team.run(task=task)
    # The last message from synthesizer contains the report
    for msg in result.messages:
        if msg.source == "synthesizer" and "FINAL_REPORT_READY" in msg.content:
            return msg.content.replace("FINAL_REPORT_READY", "").strip()
    return "No report generated"

if __name__ == "__main__":
    report = asyncio.run(run_market_research("AI code review tools"))
    print(report)

Step 4: Add caching and cost controls

LLM calls and web fetches are the two cost drivers. Wrap both with a disk cache keyed by request hash. This also makes re-runs instant during development.

# caching.py
import hashlib
import pickle
from pathlib import Path
from functools import wraps
from typing import Callable, Any

CACHE_DIR = Path(".cache")
CACHE_DIR.mkdir(exist_ok=True)

def disk_cache(key_prefix: str = "") -> Callable:
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        async def wrapper(*args, **kwargs) -> Any:
            # Create a deterministic key from args/kwargs
            key_data = f"{key_prefix}:{func.__name__}:{args}:{sorted(kwargs.items())}"
            key_hash = hashlib.sha256(key_data.encode()).hexdigest()[:16]
            cache_path = CACHE_DIR / f"{key_hash}.pkl"
            
            if cache_path.exists():
                with open(cache_path, "rb") as f:
                    return pickle.load(f)
            
            result = await func(*args, **kwargs)
            
            with open(cache_path, "wb") as f:
                pickle.dump(result, f)
            return result
        return wrapper
    return decorator

Apply it to the expensive operations in agents.py:

# In agents.py, update the tool definitions:
from caching import disk_cache

@disk_cache("web_search")
async def web_search(query: SearchQuery) -> List[SearchResult]:
    ...

@disk_cache("fetch_page")
async def fetch_page(url: str) -> str:
    ...

Add a token budget guard to the model client. AutoGen’s OpenAIChatCompletionClient accepts a model_info dict where you can set limits, but a simpler approach is a wrapper that tracks cumulative usage.

# cost_control.py
from dataclasses import dataclass, field
from autogen_ext.models.openai import OpenAIChatCompletionClient
from typing import Any

@dataclass
class BudgetTracker:
    max_tokens: int = 500_000
    used_tokens: int = 0
    
    def check_budget(self, estimated: int) -> bool:
        if self.used_tokens + estimated > self.max_tokens:
            raise RuntimeError(f"Token budget exceeded: {self.used_tokens}/{self.max_tokens}")
        return True
    
    def record_usage(self, prompt_tokens: int, completion_tokens: int):
        self.used_tokens += prompt_tokens + completion_tokens

budget = BudgetTracker(max_tokens=200_000)  # Adjust per run

class BudgetedClient(OpenAIChatCompletionClient):
    async def create(self, messages, **kwargs) -> Any:
        # Rough estimation: 4 chars per token
        estimated = sum(len(str(m)) for m in messages) // 4 + 1000
        budget.check_budget(estimated)
        response = await super().create(messages, **kwargs)
        if hasattr(response, "usage"):
            budget.record_usage(response.usage.prompt_tokens, response.usage.completion_tokens)
        return response

# In agents.py, replace model_client:
model_client = BudgetedClient(model="gpt-4o-mini")

Step 5: Implement structured output validation

The planner and extractor must emit valid JSON. Use Pydantic’s model_validate_json with a retry loop — this is more reliable than prompting for “JSON only.”

# validation.py
from pydantic import ValidationError
from agents import ResearchPlan, CompanyProfile
from typing import Type, TypeVar, List
import asyncio

T = TypeVar("T", ResearchPlan, List[CompanyProfile])

async def validated_json_output(agent: AssistantAgent, task: str, output_type: Type[T], max_retries: int = 3) -> T:
    """Run agent until it produces valid JSON matching output_type."""
    for attempt in range(max_retries):
        result = await agent.run(task=task)
        content = result.messages[-1].content
        try:
            return output_type.model_validate_json(content)
        except ValidationError as e:
            task = f"""Previous output failed validation: {e}
            Original task: {task}
            Respond with ONLY valid JSON matching the schema."""
            await asyncio.sleep(1)
    raise RuntimeError(f"Failed to get valid {output_type.__name__} after {max_retries} attempts")

Update workflow.py to use validated outputs for the planner and extractor steps:

# workflow.py (updated run_market_research)
async def run_market_research(category: str) -> str:
    # Step 1: Planner with validation
    plan = await validated_json_output(
        planner,
        f"Create a ResearchPlan for market category: {category}. Output ONLY valid JSON.",
        ResearchPlan
    )
    
    # Step 2: Researcher - run searches in parallel
    search_tasks = [web_search(SearchQuery(query=q)) for q in plan.search_queries]
    all_results = await asyncio.gather(*search_tasks)
    flat_results = [r for batch in all_results for r in batch]
    
    # Step 3: Extractor with validation - process each competitor
    profiles = []
    for competitor in plan.competitors:
        # Filter results relevant to this competitor
        relevant = [r for r in flat_results if competitor.lower() in r.title.lower() or competitor.lower() in r.snippet.lower()]
        if not relevant:
            continue
        extractor_task = f"""Company: {competitor}
        Search results: {json.dumps([r.model_dump() for r in relevant], default=str)}
        Extract a CompanyProfile. Output ONLY valid JSON."""
        profile = await validated_json_output(extractor, extractor_task, CompanyProfile)
        profiles.append(profile)
    
    # Step 4: Synthesizer
    synth_task = f"""Market category: {category}
    Company profiles: {json.dumps([p.model_dump() for p in profiles], default=str)}
    Focus areas: {plan.focus_areas}
    Write the final report. End with FINAL_REPORT_READY."""
    
    result = await team.run(task=synth_task)
    for msg in result.messages:
        if msg.source == "synthesizer" and "FINAL_REPORT_READY" in msg.content:
            return msg.content.replace("FINAL_REPORT_READY", "").strip()
    return "No report generated"

Step 6: Run and verify the pipeline

Create a CLI entry point that accepts a category argument and writes the report to a timestamped file.

# main.py
import argparse
import asyncio
from datetime import datetime
from pathlib import Path
from workflow import run_market_research

def main():
    parser = argparse.ArgumentParser(description="Automated market research with AutoGen")
    parser.add_argument("category", help="Market category to research (e.g., 'AI code review tools')")
    parser.add_argument("-o", "--output", help="Output file path", default=None)
    args = parser.parse_args()
    
    report = asyncio.run(run_market_research(args.category))
    
    if args.output:
        out_path = Path(args.output)
    else:
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        safe_category = args.category.lower().replace(" ", "_")[:40]
        out_path = Path(f"reports/{safe_category}_{timestamp}.md")
    
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(report)
    print(f"Report written to {out_path}")
    print(f"Token usage: {budget.used_tokens:,} / {budget.max_tokens:,}")

if __name__ == "__main__":
    main()

Run it:

mkdir -p reports
python main.py "AI code review tools" -o reports/ai_code_review_2024.md

Verification checklist

After the run completes, confirm:

1. File exists and has content: wc -l reports/ai_code_review_2024.md should show 100+ lines.
2. Structure is correct: Open the file and verify all five sections exist (executive summary, comparison table, deep dives, gaps, sources).
3. Citations are present: Search for [http in the report — every claim should have a bracketed URL.
4. No validation errors: The run should complete without ValidationError traces in stdout.
5. Token budget respected: Final token count should be under your configured max_tokens.
6. Cache populated: .cache/ should contain .pkl files for repeated queries.

# Quick verification script
grep -c "^\|" reports/ai_code_review_2024.md  # Should show table rows
grep -c "\[" reports/ai_code_review_2024.md   # Citation count
ls -la .cache/                                 # Cache entries

Step 7: Production hardening notes

Three things separate a demo from a pipeline you can schedule nightly:

Rate limiting and retries: The tenacity dependency handles transient failures. Wrap web_search and fetch_page with @retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=2, max=10)).

Provider fallback: If you route through a gateway that supports automatic fallback (e.g., when OpenAI is degraded, fail over to Anthropic), configure the base URL once and let the gateway handle model mapping. The agent code stays unchanged.

Observability: Log each agent’s turn duration, token usage, and tool calls to structured JSON. A minimal middleware:

# observability.py
import time
import json
import logging
from autogen_agentchat.messages import BaseMessage

logging.basicConfig(level=logging.INFO, format="%(message)s")
logger = logging.getLogger("market_research")

def log_turn(agent_name: str, messages: list, response: BaseMessage, duration_ms: int):
    logger.info(json.dumps({
        "agent": agent_name,
        "duration_ms": duration_ms,
        "input_tokens": getattr(response, "usage", {}).get("prompt_tokens", 0),
        "output_tokens": getattr(response, "usage", {}).get("completion_tokens", 0),
        "tool_calls": len(getattr(response, "tool_calls", [])),
    }))

Hook it into each agent’s on_message callback or wrap the team’s run method.

Common failure modes

Next steps

• Swap DuckDuckGo for a proper search API (SerpAPI, Brave, Exa) — the tool interface stays identical.
• Add a critic agent that scores the report for completeness and forces a revision loop.
• Persist CompanyProfile objects to a database for longitudinal tracking across runs.
• Parameterize the focus areas per vertical (SaaS vs. hardware vs. services) via a config file.

The pattern — planner → parallel researchers → extractor → synthesizer — generalizes to any multi-source synthesis task: technical due diligence, literature reviews, sales intelligence. The validation and caching layers are what make it reliable enough to run unattended.