[VIDEO-014] Building Specialist Agents

Track: 3 - Multi-Agent Systems Module: 4 Duration: 12 minutes Level requirement: 39 XP reward: 400 XP

---

Scene 1: The Power of Specialization (0:00-1:30)

[Visual]: Generalist vs specialist comparison [Animation]: Specialist agents excelling in their domains

[Audio/Script]:

"A jack of all trades is a master of none.

The same applies to AI agents.

A generalist agent might:

- Research... okay

- Code... okay

- Write... okay

A specialist agent:

- Research... exceptional

When you combine specialists, you get exceptional everything.

Let's build some specialists."

[Lower third]: "Track 3: Multi-Agent Systems | Level 39"

---

Scene 2: Anatomy of a Specialist (1:30-3:30)

[Visual]: Specialist agent structure [Animation]: Components of a specialist agent

[Audio/Script]:

"A specialist agent has three key elements:

1. Domain Knowledge - Deep understanding of its specialty

2. Specialized Tools - Custom tools for its domain

3. Focused Prompting - System prompts optimized for the task"

[Demo - Specialist Base]:

from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import List, Dict, Any
@dataclass
class SpecialistConfig:
    """Configuration for specialist agents"""
    name: str
    specialty: str
    system_prompt: str
    tools: List[str]
    max_tokens: int = 4096
    temperature: float = 0.7
class SpecialistAgent(ABC):
    """Base class for specialist agents"""
    def __init__(self, config: SpecialistConfig):
        self.config = config
        self.conversation_history: List[Dict] = []
    @property
    @abstractmethod
    def capabilities(self) -> List[str]:
        """What this specialist can do"""
        pass
    @abstractmethod
    async def execute(self, task: str, context: Dict[str, Any]) -> Dict[str, Any]:
        """Execute a task"""
        pass
    def _build_prompt(self, task: str, context: Dict[str, Any]) -> str:
        """Build the full prompt with system + context + task"""
        parts = [
            f"# System\n{self.config.system_prompt}",
            f"\n# Context\n{self._format_context(context)}" if context else "",
            f"\n# Task\n{task}"
        ]
        return "\n".join(parts)
    def _format_context(self, context: Dict[str, Any]) -> str:
        """Format context for the prompt"""
        lines = []
        for key, value in context.items():
            if isinstance(value, dict):
                lines.append(f"## {key}")
                for k, v in value.items():
                    lines.append(f"- {k}: {v}")
            else:
                lines.append(f"- {key}: {value}")
        return "\n".join(lines)

---

Scene 3: Research Specialist (3:30-5:30)

[Visual]: Research agent with web tools [Animation]: Agent gathering and organizing information

[Audio/Script]:

"Let's build our first specialist: The Research Agent."

[Demo - Research Agent]:

class ResearchAgent(SpecialistAgent):
    """Specialist in information gathering and synthesis"""
    SYSTEM_PROMPT = """You are an expert research analyst.
Your specialty is gathering, verifying, and synthesizing information.
Core capabilities:
1. Web search and source discovery
2. Information extraction and summarization
3. Source credibility assessment
4. Fact verification and cross-referencing
Always:
Cite your sources
Note confidence levels
Flag contradictions
Distinguish fact from opinion

Output format: Structured research report with sources."""
    def __init__(self):
        config = SpecialistConfig(
            name="research_specialist",
            specialty="information_gathering",
            system_prompt=self.SYSTEM_PROMPT,
            tools=["web_search", "web_fetch", "summarize"],
            temperature=0.3  # Lower for factual accuracy
        )
        super().__init__(config)
    @property
    def capabilities(self) -> List[str]:
        return [
            "web_research",
            "source_analysis",
            "fact_checking",
            "literature_review",
            "competitive_analysis"
        ]
    async def execute(self, task: str, context: Dict[str, Any]) -> Dict[str, Any]:
        """Execute research task"""
        # Determine research strategy
        strategy = self._determine_strategy(task)
        # Gather information
        sources = await self._gather_sources(task, strategy)
        # Analyze and synthesize
        analysis = await self._analyze_sources(sources)
        # Compile report
        report = self._compile_report(task, sources, analysis)
        return {
            "status": "complete",
            "type": "research_report",
            "summary": report["summary"],
            "sources": report["sources"],
            "confidence": report["confidence"],
            "full_report": report["full"]
        }
    def _determine_strategy(self, task: str) -> str:
        """Determine the best research strategy"""
        task_lower = task.lower()
        if "compare" in task_lower or "vs" in task_lower:
            return "comparative"
        if "trend" in task_lower or "history" in task_lower:
            return "longitudinal"
        if "how" in task_lower or "why" in task_lower:
            return "explanatory"
        return "exploratory"
    async def _gather_sources(self, task: str, strategy: str) -> List[Dict]:
        """Gather sources based on strategy"""
        # Implementation would use actual web search
        return []
    async def _analyze_sources(self, sources: List[Dict]) -> Dict:
        """Analyze gathered sources"""
        return {"key_findings": [], "contradictions": [], "gaps": []}
    def _compile_report(self, task: str, sources: List, analysis: Dict) -> Dict:
        """Compile final research report"""
        return {
            "summary": "",
            "sources": sources,
            "confidence": 0.8,
            "full": ""
        }

---

Scene 4: Code Specialist (5:30-7:30)

[Visual]: Code agent with IDE tools [Animation]: Agent writing and refactoring code

[Audio/Script]:

"Next: The Code Agent - an expert software developer."

[Demo - Code Agent]:

class CodeAgent(SpecialistAgent):
    """Specialist in software development"""
    SYSTEM_PROMPT = """You are an expert software engineer.
Your specialty is writing clean, efficient, well-tested code.
Core capabilities:
1. Code generation in multiple languages
2. Code review and refactoring
3. Bug fixing and debugging
4. Test writing and TDD
Always:
Follow best practices for the language
Write self-documenting code
Include error handling
Consider edge cases
Add appropriate tests

Output: Working code with explanations."""
    def __init__(self, primary_language: str = "python"):
        config = SpecialistConfig(
            name="code_specialist",
            specialty="software_development",
            system_prompt=self.SYSTEM_PROMPT,
            tools=["file_read", "file_write", "bash", "grep"],
            temperature=0.2  # Very low for precise code
        )
        super().__init__(config)
        self.primary_language = primary_language
    @property
    def capabilities(self) -> List[str]:
        return [
            "code_generation",
            "code_review",
            "bug_fixing",
            "refactoring",
            "test_writing",
            "documentation"
        ]
    async def execute(self, task: str, context: Dict[str, Any]) -> Dict[str, Any]:
        """Execute coding task"""
        # Classify the task type
        task_type = self._classify_task(task)
        if task_type == "generate":
            result = await self._generate_code(task, context)
        elif task_type == "review":
            result = await self._review_code(context.get("code", ""))
        elif task_type == "fix":
            result = await self._fix_bug(task, context)
        elif task_type == "refactor":
            result = await self._refactor_code(task, context)
        elif task_type == "test":
            result = await self._write_tests(context.get("code", ""))
        else:
            result = await self._general_coding(task, context)
        return {
            "status": "complete",
            "type": task_type,
            "code": result.get("code", ""),
            "explanation": result.get("explanation", ""),
            "tests": result.get("tests", ""),
            "files_modified": result.get("files", [])
        }
    def _classify_task(self, task: str) -> str:
        """Classify the type of coding task"""
        task_lower = task.lower()
        if any(w in task_lower for w in ["write", "create", "implement", "build"]):
            return "generate"
        if any(w in task_lower for w in ["review", "check", "analyze"]):
            return "review"
        if any(w in task_lower for w in ["fix", "bug", "error", "debug"]):
            return "fix"
        if any(w in task_lower for w in ["refactor", "improve", "clean"]):
            return "refactor"
        if any(w in task_lower for w in ["test", "spec", "coverage"]):
            return "test"
        return "general"
    async def _generate_code(self, task: str, context: Dict) -> Dict:
        """Generate new code"""
        # Implementation would use Claude to generate code
        return {"code": "", "explanation": "", "files": []}
    async def _review_code(self, code: str) -> Dict:
        """Review existing code"""
        return {"issues": [], "suggestions": [], "score": 0}
    async def _fix_bug(self, task: str, context: Dict) -> Dict:
        """Fix a bug in code"""
        return {"code": "", "explanation": "", "root_cause": ""}
    async def _refactor_code(self, task: str, context: Dict) -> Dict:
        """Refactor code"""
        return {"code": "", "changes": [], "improvements": []}
    async def _write_tests(self, code: str) -> Dict:
        """Write tests for code"""
        return {"tests": "", "coverage_estimate": 0}
    async def _general_coding(self, task: str, context: Dict) -> Dict:
        """Handle general coding tasks"""
        return {"code": "", "explanation": ""}

---

Scene 5: Writer Specialist (7:30-9:00)

[Visual]: Writer agent with content tools [Animation]: Agent crafting and polishing content

[Audio/Script]:

"The Writer Agent - expert in content creation."

[Demo - Writer Agent]:

class WriterAgent(SpecialistAgent):
    """Specialist in content creation"""
    SYSTEM_PROMPT = """You are an expert content writer.
Your specialty is creating engaging, clear, and effective written content.
Core capabilities:
1. Long-form content (articles, reports, documentation)
2. Short-form content (summaries, abstracts, social posts)
3. Technical writing (documentation, guides, tutorials)
4. Creative writing (stories, scripts, marketing copy)
Always:
Match tone to audience
Structure content logically
Use clear, concise language
Include relevant examples
Optimize for readability

Output: Polished content ready for publication."""
    def __init__(self, style: str = "professional"):
        config = SpecialistConfig(
            name="writer_specialist",
            specialty="content_creation",
            system_prompt=self.SYSTEM_PROMPT,
            tools=["grammar_check", "readability_score", "seo_analyze"],
            temperature=0.7  # Higher for creativity
        )
        super().__init__(config)
        self.style = style
    @property
    def capabilities(self) -> List[str]:
        return [
            "article_writing",
            "documentation",
            "summarization",
            "editing",
            "copywriting",
            "technical_writing"
        ]
    async def execute(self, task: str, context: Dict[str, Any]) -> Dict[str, Any]:
        """Execute writing task"""
        # Determine content type
        content_type = self._determine_content_type(task)
        # Get source material
        source_material = context.get("research_data") or context.get("input", "")
        # Generate content
        content = await self._create_content(task, content_type, source_material)
        # Polish content
        polished = await self._polish_content(content, content_type)
        return {
            "status": "complete",
            "type": content_type,
            "content": polished["content"],
            "word_count": polished["word_count"],
            "readability_score": polished["readability"],
            "metadata": polished["metadata"]
        }
    def _determine_content_type(self, task: str) -> str:
        """Determine the type of content needed"""
        task_lower = task.lower()
        if any(w in task_lower for w in ["article", "blog", "post"]):
            return "article"
        if any(w in task_lower for w in ["document", "docs", "guide"]):
            return "documentation"
        if any(w in task_lower for w in ["summary", "summarize", "brief"]):
            return "summary"
        if any(w in task_lower for w in ["report", "analysis"]):
            return "report"
        return "general"
    async def _create_content(self, task: str, content_type: str, source: str) -> str:
        """Create the initial content"""
        # Implementation would use Claude to generate content
        return ""
    async def _polish_content(self, content: str, content_type: str) -> Dict:
        """Polish and finalize content"""
        return {
            "content": content,
            "word_count": len(content.split()),
            "readability": 0.8,
            "metadata": {}
        }

---

Scene 6: Creating Custom Specialists (9:00-10:30)

[Visual]: Specialist configuration [Animation]: Building a new specialist

[Audio/Script]:

"You can create specialists for any domain. Here's the pattern:"

[Demo - Custom Specialist]:

def create_specialist(
    name: str,
    specialty: str,
    system_prompt: str,
    capabilities: List[str],
    tools: List[str] = None,
    temperature: float = 0.5
) -> SpecialistAgent:
    """Factory function to create custom specialists"""
    class CustomSpecialist(SpecialistAgent):
        _capabilities = capabilities
        @property
        def capabilities(self) -> List[str]:
            return self._capabilities
        async def execute(self, task: str, context: Dict[str, Any]) -> Dict[str, Any]:
            prompt = self._build_prompt(task, context)
            # Execute with Claude
            result = await self._call_claude(prompt)
            return {"status": "complete", "output": result}
        async def _call_claude(self, prompt: str) -> str:
            # Implementation would call Claude API
            return ""
    config = SpecialistConfig(
        name=name,
        specialty=specialty,
        system_prompt=system_prompt,
        tools=tools or [],
        temperature=temperature
    )
    return CustomSpecialist(config)
Example: Create a Data Analyst specialist
data_analyst = create_specialist(
    name="data_analyst",
    specialty="data_analysis",
    system_prompt="""You are an expert data analyst.
You excel at:
Statistical analysis
Data visualization recommendations
Pattern recognition
Insight extraction

Always provide data-driven conclusions with confidence intervals.""",
    capabilities=["statistical_analysis", "visualization", "pattern_detection"],
    tools=["python_exec", "plot_generator"],
    temperature=0.3
)
Example: Create a Security specialist
security_expert = create_specialist(
    name="security_expert",
    specialty="cybersecurity",
    system_prompt="""You are a cybersecurity expert.
You specialize in:
Vulnerability assessment
Security best practices
Threat modeling
Code security review

Always prioritize security over convenience.""",
    capabilities=["vulnerability_scan", "threat_model", "security_review"],
    tools=["code_scanner", "dependency_check"],
    temperature=0.2
)

---

Scene 7: Specialist Teams (10:30-11:30)

[Visual]: Team of specialists working together [Animation]: Specialists collaborating

[Audio/Script]:

"The real power comes from combining specialists into teams."

[Demo - Specialist Team]:

class SpecialistTeam:
    """A team of specialist agents"""
    def __init__(self, name: str):
        self.name = name
        self.specialists: Dict[str, SpecialistAgent] = {}
    def add_specialist(self, role: str, agent: SpecialistAgent):
        """Add a specialist to the team"""
        self.specialists[role] = agent
    def get_capabilities(self) -> Dict[str, List[str]]:
        """Get all team capabilities"""
        return {
            role: agent.capabilities
            for role, agent in self.specialists.items()
        }
    async def execute_workflow(self, task: str, workflow: List[Dict]) -> Dict:
        """Execute a multi-step workflow"""
        results = {}
        for step in workflow:
            role = step["role"]
            step_task = step["task"]
            if role not in self.specialists:
                raise ValueError(f"No specialist for role: {role}")
            # Build context from previous results
            context = {
                "previous_results": results,
                "original_task": task
            }
            # Execute step
            result = await self.specialists[role].execute(step_task, context)
            results[step["name"]] = result
        return results
Example: Content creation team
content_team = SpecialistTeam("content_creation")
content_team.add_specialist("researcher", ResearchAgent())
content_team.add_specialist("writer", WriterAgent())
content_team.add_specialist("coder", CodeAgent())
Define workflow
workflow = [
    {"name": "research", "role": "researcher", "task": "Research the topic"},
    {"name": "outline", "role": "writer", "task": "Create content outline"},
    {"name": "draft", "role": "writer", "task": "Write first draft"},
    {"name": "examples", "role": "coder", "task": "Create code examples"},
    {"name": "final", "role": "writer", "task": "Finalize with examples"}
]
Execute
result = await content_team.execute_workflow(
    "Create a tutorial on async Python",
    workflow
)

---

Scene 8: Challenge Time (11:30-12:00)

[Visual]: Challenge specification [Animation]: XP reward display

[Audio/Script]:

"Your challenge: Build a team of 3 specialists.

Create:

1. ResearchAgent - information gathering

2. CodeAgent - software development

3. WriterAgent - content creation

Then combine them in a SpecialistTeam and execute a workflow.

Test task: 'Create a technical blog post about MCP tools with working examples'

Complete this for 800 XP and the 'Team Builder' badge.

Next: Track 3 Capstone - build an autonomous multi-agent system."

---

Post-Video Challenge

Challenge ID: TRACK3_004_CHALLENGE Type: Code + Integration Instructions:

Task 1: Create base classes

claude "Create specialist_base.py with:
1. SpecialistConfig dataclass
2. SpecialistAgent abstract base class
3. capabilities property and execute method"

Task 2: Build three specialists

claude "Create three specialist agents:
1. research_agent.py - ResearchAgent with web research
2. code_agent.py - CodeAgent for development
3. writer_agent.py - WriterAgent for content
Each with proper system prompts and capabilities."

Task 3: Create specialist team

claude "Create specialist_team.py with:
1. SpecialistTeam class
2. add_specialist and get_capabilities methods
3. execute_workflow for multi-step tasks"

Task 4: Integration test

claude "Test the team with this workflow:
1. Research 'Python async patterns'
2. Create code examples
3. Write a tutorial combining both
The final output should be a complete tutorial."

Rewards:

• XP: 800 (400 base + 400 challenge)
• Achievement: "Team Builder"

SEO Metadata

Alt-text: Building specialist AI agents - research, code, and writer specialists. Create focused, expert agents for multi-agent systems.

Tags: specialist agents, AI specialists, research agent, code agent, writer agent, agent teams, focused AI

Keywords: specialist ai agents, build expert agents, research agent ai, code generation agent, ai content writer, agent team patterns