Build a Multi-Agent Research Team (Planner, Searcher, Writer)

pip install langchain langchain-openai langchain-community tavily-python python-dotenv autogen-agentchat

import os
import json
import asyncio
from typing import Dict, List, Any, Optional
from dataclasses import dataclass, field
from dotenv import load_dotenv

from langchain_openai import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema import HumanMessage, SystemMessage
from langchain_community.tools.tavily_search import TavilySearchResults

load_dotenv()

# Model configs — different models for different roles
PLANNER_MODEL = ChatOpenAI(model="gpt-4o", temperature=0.1)
SEARCHER_MODEL = ChatOpenAI(model="gpt-4o-mini", temperature=0.0)
WRITER_MODEL = ChatOpenAI(model="gpt-4o", temperature=0.3)

# Search tool
search_tool = TavilySearchResults(
    max_results=5,
    api_key=os.environ["TAVILY_API_KEY"]
)

@dataclass
class ResearchState:
    topic: str
    research_plan: Optional[Dict] = None
    search_results: Dict[str, List[Dict]] = field(default_factory=dict)
    final_report: Optional[str] = None
    errors: List[str] = field(default_factory=list)
    metadata: Dict[str, Any] = field(default_factory=dict)
    
    def add_search_result(self, sub_question: str, results: List[Dict]):
        if sub_question not in self.search_results:
            self.search_results[sub_question] = []
        self.search_results[sub_question].extend(results)
    
    def to_summary(self) -> str:
        """Compact summary for passing to writer."""
        summary = f"Research Topic: {self.topic}\n\n"
        for question, results in self.search_results.items():
            summary += f"## {question}\n"
            for r in results[:3]:  # Top 3 per question
                summary += f"- **Source:** {r.get('url', 'Unknown')}\n"
                summary += f"  {r.get('content', '')[:300]}...\n\n"
        return summary

class PlannerAgent:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.system_prompt = """You are a research planning specialist. 
        
Your job is to analyze a research topic and create a structured research plan.

Output a JSON object with this exact structure:
{
    "topic": "the research topic",
    "objective": "what we want to learn",
    "sub_questions": [
        {
            "id": 1,
            "question": "specific sub-question",
            "priority": "high|medium|low",
            "search_queries": ["query1", "query2"]
        }
    ],
    "output_format": "description of expected report structure"
}

Rules:
- Generate 4-6 sub-questions that together cover the full topic
- Each sub-question should have 2-3 specific search queries
- Questions should move from foundational to specific
- Do NOT answer the questions — just plan how to research them
"""
    
    def plan(self, topic: str) -> Dict:
        messages = [
            SystemMessage(content=self.system_prompt),
            HumanMessage(content=f"Create a research plan for: {topic}")
        ]
        
        response = self.llm.invoke(messages)
        
        # Extract JSON from response
        content = response.content
        # Handle markdown code blocks if present
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0].strip()
        elif "```" in content:
            content = content.split("```")[1].split("```")[0].strip()
        
        plan = json.loads(content)
        print(f"[Planner] Generated plan with {len(plan['sub_questions'])} sub-questions")
        return plan

class SearcherAgent:
    def __init__(self, llm: ChatOpenAI, search_tool: TavilySearchResults):
        self.llm = llm
        self.search_tool = search_tool
        self.system_prompt = """You are a research search specialist.

You receive search results and must:
1. Extract the most relevant information for the given question
2. Note the source URL and date if available
3. Flag any conflicting information between sources
4. Return a structured summary

Output format:
{
    "question": "the sub-question",
    "key_findings": ["finding 1", "finding 2"],
    "sources": [{"url": "...", "relevance": "high|medium"}],
    "conflicts": ["any conflicting claims"],
    "confidence": "high|medium|low"
}"""
    
    def _search_with_query(self, query: str) -> List[Dict]:
        """Execute a single search query."""
        try:
            results = self.search_tool.invoke(query)
            return results if isinstance(results, list) else []
        except Exception as e:
            print(f"[Searcher] Search failed for '{query}': {e}")
            return []
    
    def _deduplicate_results(self, results: List[Dict]) -> List[Dict]:
        """Remove duplicate URLs from search results."""
        seen_urls = set()
        unique = []
        for r in results:
            url = r.get("url", "")
            if url and url not in seen_urls:
                seen_urls.add(url)
                unique.append(r)
        return unique
    
    def _synthesize_findings(self, question: str, raw_results: List[Dict]) -> Dict:
        """Use LLM to synthesize raw search results into structured findings."""
        results_text = "\n\n".join([
            f"Source: {r.get('url', 'unknown')}\nContent: {r.get('content', '')[:500]}"
            for r in raw_results[:5]
        ])
        
        messages = [
            SystemMessage(content=self.system_prompt),
            HumanMessage(content=f"Question: {question}\n\nSearch Results:\n{results_text}")
        ]
        
        response = self.llm.invoke(messages)
        content = response.content
        
        if "```json" in content:
            content = content.split("```json")[1].split("```")[0].strip()
        
        try:
            return json.loads(content)
        except json.JSONDecodeError:
            return {"question": question, "key_findings": [content], "sources": [], "confidence": "low"}
    
    def search(self, research_plan: Dict, state: ResearchState) -> ResearchState:
        """Execute searches for all sub-questions in the plan."""
        for sub_q in research_plan["sub_questions"]:
            question = sub_q["question"]
            queries = sub_q["search_queries"]
            priority = sub_q.get("priority", "medium")
            
            print(f"[Searcher] Searching: {question} (priority: {priority})")
            
            # Execute all queries for this sub-question
            all_raw_results = []
            for query in queries:
                raw = self._search_with_query(query)
                all_raw_results.extend(raw)
            
            # Deduplicate
            unique_results = self._deduplicate_results(all_raw_results)
            
            # Synthesize into structured findings
            findings = self._synthesize_findings(question, unique_results)
            
            state.add_search_result(question, [findings])
            print(f"[Searcher] Found {len(unique_results)} unique sources for: {question}")
        
        return state

class WriterAgent:
    def __init__(self, llm: ChatOpenAI):
        self.llm = llm
        self.system_prompt = """You are a research writer who synthesizes information into clear, 
well-structured reports.

Your reports must:
1. Start with an executive summary (2-3 sentences)
2. Have clearly labeled sections for each major finding
3. Cite sources with [Source: URL] inline
4. Acknowledge uncertainty where confidence is low
5. End with key conclusions and open questions
6. Use markdown formatting throughout

Style guidelines:
- Write for an informed, professional audience
- Use specific numbers and dates when available
- Avoid vague generalizations
- Distinguish between established facts and current estimates
"""
    
    def write(self, topic: str, state: ResearchState, plan: Dict) -> str:
        research_summary = state.to_summary()
        
        messages = [
            SystemMessage(content=self.system_prompt),
            HumanMessage(content=f"""Write a comprehensive research report on: {topic}

Research Objective: {plan.get('objective', 'Not specified')}

Research Findings:
{research_summary}

Please structure the report according to these sections implied by the research plan.
Include all available sources as citations.""")
        ]
        
        response = self.llm.invoke(messages)
        print(f"[Writer] Report generated: {len(response.content)} characters")
        return response.content

class ResearchOrchestrator:
    def __init__(self):
        self.planner = PlannerAgent(PLANNER_MODEL)
        self.searcher = SearcherAgent(SEARCHER_MODEL, search_tool)
        self.writer = WriterAgent(WRITER_MODEL)
    
    def run(self, topic: str) -> ResearchState:
        print(f"\n{'='*60}")
        print(f"Starting research: {topic}")
        print('='*60)
        
        state = ResearchState(topic=topic)
        
        # Step 1: Plan
        print("\n[Phase 1: Planning]")
        state.research_plan = self.planner.plan(topic)
        
        # Step 2: Search
        print("\n[Phase 2: Searching]")
        state = self.searcher.search(state.research_plan, state)
        
        # Step 3: Write
        print("\n[Phase 3: Writing]")
        state.final_report = self.writer.write(topic, state, state.research_plan)
        
        print("\n[Research Complete]")
        return state
    
    def save_report(self, state: ResearchState, output_path: str):
        with open(output_path, "w", encoding="utf-8") as f:
            f.write(f"# Research Report: {state.topic}\n\n")
            f.write(state.final_report)
        print(f"Report saved to: {output_path}")

# Usage
if __name__ == "__main__":
    orchestrator = ResearchOrchestrator()
    
    result = orchestrator.run(
        "The current state of open-source LLM fine-tuning methods in 2026"
    )
    
    orchestrator.save_report(result, "research_report.md")
    print("\n--- FINAL REPORT PREVIEW ---")
    print(result.final_report[:500])

import autogen

llm_config = {"model": "gpt-4o", "api_key": os.environ["OPENAI_API_KEY"]}

planner = autogen.AssistantAgent(
    name="Planner",
    system_message="""Break the research topic into 4-6 specific sub-questions.
    For each question, suggest 2-3 search queries.
    Output as a numbered list. End with: 'Plan complete.'""",
    llm_config=llm_config
)

searcher = autogen.AssistantAgent(
    name="Searcher",
    system_message="""Given research questions, find relevant information.
    For each question, provide a summary of what's known with source URLs.
    Be specific. End with: 'Search complete.'""",
    llm_config=llm_config
)

writer = autogen.AssistantAgent(
    name="Writer",
    system_message="""Synthesize research findings into a structured report.
    Use headers, bullet points, and inline citations.
    End with: 'REPORT COMPLETE'""",
    llm_config=llm_config
)

user_proxy = autogen.UserProxyAgent(
    name="UserProxy",
    human_input_mode="NEVER",
    is_termination_msg=lambda m: "REPORT COMPLETE" in m.get("content", ""),
    code_execution_config=False
)

groupchat = autogen.GroupChat(
    agents=[user_proxy, planner, searcher, writer],
    messages=[],
    max_round=12,
    speaker_selection_method="auto"
)

manager = autogen.GroupChatManager(
    groupchat=groupchat,
    llm_config=llm_config,
    system_message="""Coordinate: Planner goes first, then Searcher, then Writer.
    Each agent speaks once per cycle unless revision is needed."""
)

user_proxy.initiate_chat(
    manager,
    message="Research: How are companies using AI agents in production in 2026?"
)

# Research Report: [Topic]

## Executive Summary
[2-3 sentence overview of key findings]

## Background
[Foundational context from Planner's first sub-questions]

## Current State
[Main findings with inline citations [Source: URL]]

## Key Developments
[Specific findings, statistics, examples]

## Challenges and Open Questions
[What we don't know, conflicting information]

## Conclusions
[Synthesized takeaways]

## Sources
[Full list of URLs used]