Build a Multi-Agent Research System with CrewAI

python3 -m venv .venv && source .venv/bin/activate
pip install "crewai[anthropic,tools]" "pydantic-ai-slim[anthropic]>=2" \
  haystack-ai sentence-transformers-haystack pypdf markdown-it-py mdit-plain
export ANTHROPIC_API_KEY=...     # from your secret manager
export SERPER_API_KEY=...        # serper.dev key for SerperDevTool

# kb.py
from pathlib import Path
from haystack import Pipeline
from haystack.components.converters import MarkdownToDocument, PyPDFToDocument
from haystack_integrations.components.embedders.sentence_transformers import (
    SentenceTransformersDocumentEmbedder,
    SentenceTransformersTextEmbedder,
)
from haystack.components.preprocessors import DocumentSplitter
from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
from haystack.components.writers import DocumentWriter
from haystack.document_stores.in_memory import InMemoryDocumentStore

MODEL = "BAAI/bge-small-en-v1.5"
document_store = InMemoryDocumentStore()


def index_documents(docs_dir: str) -> None:
    docs = []
    for pattern, converter in (("*.md", MarkdownToDocument()), ("*.pdf", PyPDFToDocument())):
        paths = list(Path(docs_dir).rglob(pattern))
        if paths:
            docs += converter.run(sources=paths)["documents"]

    indexing = Pipeline()
    indexing.add_component("splitter", DocumentSplitter(split_by="word", split_length=200, split_overlap=30))
    indexing.add_component("embedder", SentenceTransformersDocumentEmbedder(model=MODEL))
    indexing.add_component("writer", DocumentWriter(document_store=document_store))
    indexing.connect("splitter.documents", "embedder.documents")
    indexing.connect("embedder.documents", "writer.documents")
    indexing.run({"splitter": {"documents": docs}})


def search_knowledge_base(query: str, top_k: int = 5) -> list[str]:
    search = Pipeline()
    search.add_component("embedder", SentenceTransformersTextEmbedder(model=MODEL))
    search.add_component("retriever", InMemoryEmbeddingRetriever(document_store=document_store, top_k=top_k))
    search.connect("embedder.embedding", "retriever.query_embedding")
    hits = search.run({"embedder": {"text": query}})["retriever"]["documents"]
    return [f"[{d.meta.get('file_path', 'unknown')}] {d.content}" for d in hits]


if __name__ == "__main__":
    index_documents("docs")
    print(search_knowledge_base("enterprise private deployment"))

# models.py
from pydantic import BaseModel, Field
from pydantic_ai import Agent


class CompetitorProfile(BaseModel):
    name: str
    website: str
    positioning: str = Field(description="One-sentence market positioning")
    key_products: list[str]
    target_market: str
    pricing_model: str
    strengths: list[str] = Field(min_length=2, max_length=5)
    weaknesses: list[str] = Field(min_length=2, max_length=5)
    recent_moves: list[str] = Field(description="Notable actions in the last 6 months")
    sources: list[str] = Field(min_length=1, description="URLs the facts came from")


extractor = Agent(
    "anthropic:claude-sonnet-5-5",
    output_type=CompetitorProfile,
    instructions=(
        "Extract a competitor profile from the research notes. Use only facts present "
        "in the notes; if a field is not stated, say 'not found' instead of guessing."
    ),
)


def extract_profile(notes: str) -> CompetitorProfile:
    return extractor.run_sync(notes).output

# crew.py
from crewai import Agent, Crew, Process, Task
from crewai.tools import tool
from crewai_tools import ScrapeWebsiteTool, SerperDevTool

from kb import index_documents, search_knowledge_base
from models import extract_profile

LLM = "anthropic/claude-sonnet-5-5"


@tool("Internal knowledge base search")
def kb_search(query: str) -> str:
    """Search past reports and strategy documents. Returns the best matching passages with their source file."""
    return "\n\n".join(search_knowledge_base(query)) or "No matching internal documents."


web_researcher = Agent(
    role="Competitive Intelligence Researcher",
    goal="Find current, sourced facts about competitors in the {market} space",
    backstory="A careful analyst who checks every number against at least two sources and "
    "writes 'not found' when a fact cannot be sourced.",
    tools=[SerperDevTool(), ScrapeWebsiteTool()],
    llm=LLM,
    max_iter=8,
)
kb_analyst = Agent(
    role="Internal Knowledge Analyst",
    goal="Surface what our past reports and strategy documents say about {market}",
    backstory="Owns the company's competitive-intelligence archive and quotes it exactly.",
    tools=[kb_search],
    llm=LLM,
)
strategist = Agent(
    role="Chief Strategy Analyst",
    goal="Turn research into prioritized recommendations for our position in {market}",
    backstory="Thinks in moats, positioning and asymmetric advantages; ranks everything by impact.",
    llm=LLM,
)
report_writer = Agent(
    role="Executive Report Writer",
    goal="Write a board-ready competitive intelligence report",
    backstory="Writes for executives: short sentences, every claim tied to a source.",
    llm=LLM,
)

research_task = Task(
    description="Research these competitors in {market}: {competitors}. For each: funding and team "
    "size, products and launches in the last 6 months, pricing and target customer, strengths, "
    "weaknesses, strategic moves. Cite a URL for every fact; write 'not found' otherwise.",
    expected_output="A research dossier, one section per competitor, with source URLs",
    agent=web_researcher,
)
kb_task = Task(
    description="Search the internal knowledge base for past assessments of {competitors}, "
    "market sizing for {market}, and recommendations that still apply.",
    expected_output="Relevant internal findings with the source file for each",
    agent=kb_analyst,
)
analysis_task = Task(
    description="Combine the web research and internal findings into: market trends (3-5, with "
    "evidence), a SWOT for our position in {market}, prioritized recommendations (P0/P1/P2) and a "
    "90-day plan.",
    expected_output="A structured strategy analysis",
    agent=strategist,
    context=[research_task, kb_task],
)
report_task = Task(
    description="Write the report: executive summary (250 words max), landscape, trends, "
    "recommendations, 90-day roadmap, risks. Cite sources for every claim.",
    expected_output="A markdown report of about 3000 words",
    agent=report_writer,
    context=[research_task, kb_task, analysis_task],
    output_file="output/competitive_intelligence_report.md",
)

crew = Crew(
    agents=[web_researcher, kb_analyst, strategist, report_writer],
    tasks=[research_task, kb_task, analysis_task, report_task],
    process=Process.sequential,
    verbose=True,
)

if __name__ == "__main__":
    index_documents("knowledge")  # past reports, strategy docs (.md / .pdf)
    competitors = ["Cursor", "Tabnine", "Replit"]
    result = crew.kickoff(inputs={"market": "AI developer tools", "competitors": ", ".join(competitors)})

    # Validate the dossier into typed profiles (PydanticAI), one per competitor
    dossier = result.tasks_output[0].raw
    profiles = [extract_profile(f"Competitor: {c}\n\n{dossier}") for c in competitors]
    print(f"{len(profiles)} validated profiles; report in output/competitive_intelligence_report.md")