Skip to content

RFC-032: Anthropic Provider Implementation (Revised)

  • Status: Completed (v2.4.0)
  • Revision: 3
  • Date: 2026-02-04
  • Authors:
  • Stakeholders: Maintainers, users wanting Anthropic API integration, developers implementing providers
  • Related PRDs:
  • docs/prd/PRD-009-anthropic-provider-integration.md
  • Related RFCs:
  • docs/rfc/RFC-013-openai-provider-implementation.md (reference - unified provider pattern)
  • docs/rfc/RFC-021-modularization-refactoring-plan.md (architecture foundation)
  • docs/rfc/RFC-017-prompt-management.md (prompt system)

Abstract

Design and implement Anthropic Claude API as a unified provider for speaker detection and summarization capabilities. This RFC builds on the existing modularization architecture (RFC-021) and follows the unified provider pattern established by OpenAI (RFC-013), where a single provider class implements multiple protocols. Anthropic does NOT support audio transcription.

Architecture Alignment: Anthropic provider follows the exact same unified provider pattern as OpenAIProvider, implementing two protocols (SpeakerDetector, SummarizationProvider) in a single class and integrating via the existing factory pattern with support for both Config-based and experiment-based modes.

Problem Statement

Users want the option to use Anthropic Claude API as an alternative to OpenAI for:

  1. Speaker Detection: Entity extraction using Claude models
  2. Summarization: High-quality summaries using Claude models

Note: Transcription is NOT supported by Anthropic (no audio API).

Requirements:

  • No changes to end-user experience or workflow when using defaults
  • Secure API key management (environment variables, never in source code)
  • Per-capability provider selection (can mix local, OpenAI, and Anthropic)
  • Build on existing modularization and provider architecture
  • Handle capability gaps gracefully (transcription not supported)
  • Use Anthropic-specific prompts (prompts are provider-specific)
  • Support both Config-based and experiment-based factory modes

Constraints & Assumptions

Constraints:

  • Prerequisite: Modularization refactoring (RFC-021) Completed
  • Prerequisite: OpenAI provider implementation (RFC-013) Completed
  • Backward Compatibility: Default providers (local) must remain unchanged
  • API Key Security: API keys must never be in source code or committed files
  • Capability Gap: Anthropic does not support audio transcription
  • Rate Limits: Must respect Anthropic API rate limits and implement retry logic
  • Must follow unified provider pattern (like OpenAI)

Assumptions:

  • Anthropic API is stable and well-documented
  • Anthropic Python SDK follows similar patterns to OpenAI SDK
  • Prompts need to be optimized for Claude (different from GPT prompts)
  • Users understand the capability matrix (what each provider supports)

Design & Implementation

0. Anthropic API Overview

Anthropic's API differs from OpenAI in several ways:

Feature OpenAI Anthropic
Chat Endpoint /v1/chat/completions /v1/messages
Audio API Yes — Whisper API No — Not available
Context Window 128k tokens 200k tokens
Temperature Range 0.0 - 2.0 0.0 - 1.0
System Message In messages array Separate system parameter
Provider Pattern Unified (OpenAIProvider) Unified (AnthropicProvider)

1. Architecture Overview

Unified Provider Pattern (following OpenAI):

src/podcast_scraper/
├── providers/
│   └── anthropic/                    # NEW: Unified Anthropic provider
│       ├── __init__.py
│       └── anthropic_provider.py     # Single class implementing 2 protocols
├── prompts/
│   └── anthropic/                    # NEW: Anthropic-specific prompts
│       ├── ner/
│       │   ├── system_ner_v1.j2
│       │   └── guest_host_v1.j2
│       └── summarization/
│           ├── system_v1.j2
│           └── long_v1.j2
├── speaker_detectors/
│   └── factory.py                    # Updated: Add "anthropic" option
├── summarization/
│   └── factory.py                    # Updated: Add "anthropic" option
└── config.py                         # Updated: Add Anthropic fields

Key Architectural Decision: Use unified provider pattern (single AnthropicProvider class) matching OpenAIProvider, not separate files per capability.

2. Configuration

Add to config.py following OpenAI pattern exactly:

from typing import Literal, Optional

# Provider Selection (updated)
speaker_detector_provider: Literal["spacy", "openai", "anthropic"] = Field(
    default="spacy",
    description="Speaker detection provider"
)

summary_provider: Literal["transformers", "openai", "anthropic"] = Field(
    default="transformers",
    description="Summarization provider"
)

# Anthropic API Configuration (following OpenAI pattern)
anthropic_api_key: Optional[str] = Field(
    default=None,
    alias="anthropic_api_key",
    description="Anthropic API key (prefer ANTHROPIC_API_KEY env var or .env file)"
)

anthropic_api_base: Optional[str] = Field(
    default=None,
    alias="anthropic_api_base",
    description="Anthropic API base URL (for E2E testing with mock servers)"
)

# Anthropic Model Selection (environment-based defaults, like OpenAI)
anthropic_speaker_model: str = Field(
    default_factory=_get_default_anthropic_speaker_model,
    alias="anthropic_speaker_model",
    description="Anthropic model for speaker detection (default: environment-based)"
)

anthropic_summary_model: str = Field(
    default_factory=_get_default_anthropic_summary_model,
    alias="anthropic_summary_model",
    description="Anthropic model for summarization (default: environment-based)"
)

# Shared settings (like OpenAI)
anthropic_temperature: float = Field(
    default=0.3,
    alias="anthropic_temperature",
    description="Temperature for Anthropic generation (0.0-1.0, lower = more deterministic)"
)

anthropic_max_tokens: Optional[int] = Field(
    default=None,
    alias="anthropic_max_tokens",
    description="Max tokens for Anthropic generation (None = model default)"
)

# Anthropic Prompt Configuration (following OpenAI pattern)
anthropic_speaker_system_prompt: Optional[str] = Field(
    default=None,
    alias="anthropic_speaker_system_prompt",
    description="Anthropic system prompt for speaker detection (default: anthropic/ner/system_ner_v1)"
)

anthropic_speaker_user_prompt: str = Field(
    default="anthropic/ner/guest_host_v1",
    alias="anthropic_speaker_user_prompt",
    description="Anthropic user prompt for speaker detection"
)

anthropic_summary_system_prompt: Optional[str] = Field(
    default=None,
    alias="anthropic_summary_system_prompt",
    description="Anthropic system prompt for summarization (default: anthropic/summarization/system_v1)"
)

anthropic_summary_user_prompt: str = Field(
    default="anthropic/summarization/long_v1",
    alias="anthropic_summary_user_prompt",
    description="Anthropic user prompt for summarization"
)

Environment-based defaults (like OpenAI):

# In config_constants.py
TEST_DEFAULT_ANTHROPIC_SPEAKER_MODEL = "claude-haiku-4-5"
PROD_DEFAULT_ANTHROPIC_SPEAKER_MODEL = "claude-3-5-sonnet-latest"

TEST_DEFAULT_ANTHROPIC_SUMMARY_MODEL = "claude-haiku-4-5"
PROD_DEFAULT_ANTHROPIC_SUMMARY_MODEL = "claude-3-5-sonnet-latest"

# In config.py
def _get_default_anthropic_speaker_model() -> str:
    """Get default Anthropic speaker detection model based on environment."""
    if _is_test_environment():
        return TEST_DEFAULT_ANTHROPIC_SPEAKER_MODEL
    return PROD_DEFAULT_ANTHROPIC_SPEAKER_MODEL

def _get_default_anthropic_summary_model() -> str:
    """Get default Anthropic summarization model based on environment."""
    if _is_test_environment():
        return TEST_DEFAULT_ANTHROPIC_SUMMARY_MODEL
    return PROD_DEFAULT_ANTHROPIC_SUMMARY_MODEL

3. API Key Management

Follow OpenAI pattern exactly:

# In config.py

@field_validator("anthropic_api_key", mode="before")
@classmethod
def _load_anthropic_api_key_from_env(cls, value: Any) -> Optional[str]:
    """Load Anthropic API key from environment variable if not provided."""
    if value is not None:
        return value
    env_key = os.getenv("ANTHROPIC_API_KEY")
    if env_key:
        return env_key
    return None

@field_validator("anthropic_api_base", mode="before")
@classmethod
def _load_anthropic_api_base_from_env(cls, value: Any) -> Optional[str]:
    """Load Anthropic API base URL from environment variable if not provided."""
    if value is not None:
        return value
    env_base = os.getenv("ANTHROPIC_API_BASE")
    if env_base:
        return env_base
    return None

@model_validator(mode="after")
def _validate_anthropic_provider_requirements(self) -> "Config":
    """Validate that Anthropic API key is provided when Anthropic providers are selected."""
    anthropic_providers_used = []
    if self.speaker_detector_provider == "anthropic":
        anthropic_providers_used.append("speaker_detection")
    if self.summary_provider == "anthropic":
        anthropic_providers_used.append("summarization")

    if anthropic_providers_used and not self.anthropic_api_key:
        providers_str = ", ".join(anthropic_providers_used)
        raise ValueError(
            f"Anthropic API key required for Anthropic providers: {providers_str}. "
            "Set ANTHROPIC_API_KEY environment variable or anthropic_api_key in config."
        )

    return self

4. Provider Capability Validation

Add validation to prevent using Anthropic for transcription:

# config.py - Capability Validation

@model_validator(mode="after")
def _validate_provider_capabilities(self) -> "Config":
    """Validate provider supports requested capability."""
    if self.transcription_provider == "anthropic":
        raise ValueError(
            "Anthropic provider does not support transcription. "
            "Use 'whisper' (local) or 'openai' instead. "
            "See provider capability matrix in documentation."
        )
    return self

5. Unified Provider Implementation

File: src/podcast_scraper/providers/anthropic/anthropic_provider.py

Follow OpenAIProvider pattern exactly:

"""Unified Anthropic provider for speaker detection and summarization.

This module provides a single AnthropicProvider class that implements two protocols:
- SpeakerDetector (using Claude API)
- SummarizationProvider (using Claude API)

This unified approach matches the pattern of OpenAI providers, where a single
provider type handles multiple capabilities using shared API client.

Note: Anthropic does NOT support transcription (no audio API).
"""

from __future__ import annotations

import json
import logging
import os
from typing import Any, Dict, Optional, Set, Tuple

try:
    from anthropic import Anthropic
except ImportError:
    Anthropic = None  # type: ignore

from ... import config, models
from ...workflow import metrics

logger = logging.getLogger(__name__)

# Default speaker names when detection fails
DEFAULT_SPEAKER_NAMES = ["Host", "Guest"]

# Anthropic API pricing constants (for cost estimation)
# Source: https://www.anthropic.com/pricing
# Last updated: 2026-02
ANTHROPIC_HAIKU_INPUT_COST_PER_1M_TOKENS = 0.25
ANTHROPIC_HAIKU_OUTPUT_COST_PER_1M_TOKENS = 1.25
ANTHROPIC_SONNET_INPUT_COST_PER_1M_TOKENS = 3.00
ANTHROPIC_SONNET_OUTPUT_COST_PER_1M_TOKENS = 15.00


class AnthropicProvider:
    """Unified Anthropic provider implementing SpeakerDetector and SummarizationProvider.

    This provider initializes and manages:
    - Anthropic Claude API for speaker detection
    - Anthropic Claude API for summarization

    All capabilities share the same Anthropic client, similar to how OpenAI providers
    share the same OpenAI client. The client is initialized once and reused.

    Note: Transcription is NOT supported (Anthropic has no audio API).
    """

    def __init__(self, cfg: config.Config):
        """Initialize unified Anthropic provider.

        Args:
            cfg: Configuration object with settings for both capabilities

        Raises:
            ValueError: If Anthropic API key is not provided
            ImportError: If anthropic package is not installed
        """
        if Anthropic is None:
            raise ImportError(
                "anthropic package required for Anthropic provider. "
                "Install with: pip install -e '.[llm]' (anthropic is in the llm extra)"
            )

        if not cfg.anthropic_api_key:
            raise ValueError(
                "Anthropic API key required for Anthropic provider. "
                "Set ANTHROPIC_API_KEY environment variable or anthropic_api_key in config."
            )

        self.cfg = cfg

        # Suppress verbose Anthropic SDK debug logs (if needed)
        # Similar to OpenAI provider pattern

        # Support custom base_url for E2E testing with mock servers
        client_kwargs: dict[str, Any] = {"api_key": cfg.anthropic_api_key}
        if cfg.anthropic_api_base:
            client_kwargs["base_url"] = cfg.anthropic_api_base
        self.client = Anthropic(**client_kwargs)

        # Speaker detection settings
        self.speaker_model = getattr(cfg, "anthropic_speaker_model", "claude-haiku-4-5")
        self.speaker_temperature = getattr(cfg, "anthropic_temperature", 0.3)

        # Summarization settings
        self.summary_model = getattr(cfg, "anthropic_summary_model", "claude-haiku-4-5")
        self.summary_temperature = getattr(cfg, "anthropic_temperature", 0.3)
        # Claude 3.5 Sonnet supports 200k context window
        self.max_context_tokens = 200000  # Conservative estimate

        # Initialization state
        self._speaker_detection_initialized = False
        self._summarization_initialized = False

        # Mark provider as thread-safe (API clients can be shared across threads)
        self._requires_separate_instances = False

    @staticmethod
    def get_pricing(model: str, capability: str) -> Dict[str, float]:
        """Get pricing information for a specific model and capability.

        Args:
            model: Model name
            capability: Capability type ("speaker_detection", "summarization")

        Returns:
            Dictionary with pricing information
        """
        # Implementation similar to OpenAIProvider.get_pricing()
        # ...

    def initialize(self) -> None:
        """Initialize all Anthropic capabilities.

        For Anthropic API, initialization is a no-op but we track it for consistency.
        This method is idempotent and can be called multiple times safely.
        """
        # Initialize speaker detection if enabled
        if self.cfg.auto_speakers and not self._speaker_detection_initialized:
            self._initialize_speaker_detection()

        # Initialize summarization if enabled
        if self.cfg.generate_summaries and not self._summarization_initialized:
            self._initialize_summarization()

    def _initialize_speaker_detection(self) -> None:
        """Initialize speaker detection capability."""
        logger.debug("Initializing Anthropic speaker detection (model: %s)", self.speaker_model)
        self._speaker_detection_initialized = True

    def _initialize_summarization(self) -> None:
        """Initialize summarization capability."""
        logger.debug("Initializing Anthropic summarization (model: %s)", self.summary_model)
        self._summarization_initialized = True

    # ============================================================================
    # SpeakerDetector Protocol Implementation
    # ============================================================================

    def detect_hosts(
        self,
        feed_title: str | None,
        feed_description: str | None,
        feed_authors: list[str] | None = None,
    ) -> Set[str]:
        """Detect host names from feed-level metadata using Anthropic API.

        Args:
            feed_title: Feed title (can be None)
            feed_description: Optional feed description
            feed_authors: Optional list of author names from RSS feed (preferred source)

        Returns:
            Set of detected host names
        """
        if not self._speaker_detection_initialized:
            raise RuntimeError(
                "AnthropicProvider speaker detection not initialized. Call initialize() first."
            )

        # Prefer RSS author tags if available (like OpenAI)
        if feed_authors:
            return set(feed_authors)

        # Otherwise, use Anthropic API to detect hosts from feed metadata
        if not feed_title:
            return set()

        try:
            # Use detect_speakers with empty known_hosts to detect hosts
            speakers, detected_hosts, _ = self.detect_speakers(
                episode_title=feed_title,
                episode_description=feed_description,
                known_hosts=set(),
            )
            return detected_hosts
        except Exception as exc:
            logger.warning("Failed to detect hosts from feed metadata: %s", exc)
            return set()

    def detect_speakers(
        self,
        episode_title: str,
        episode_description: str | None,
        known_hosts: Set[str],
        pipeline_metrics: metrics.Metrics | None = None,
    ) -> Tuple[list[str], Set[str], bool]:
        """Detect speaker names from episode metadata using Anthropic API.

        Args:
            episode_title: Episode title
            episode_description: Optional episode description
            known_hosts: Set of known host names (for context)
            pipeline_metrics: Optional metrics tracker

        Returns:
            Tuple of:
            - List of detected speaker names (hosts + guests)
            - Set of detected host names (subset of known_hosts)
            - Success flag (True if detection succeeded)

        Raises:
            ValueError: If detection fails or API key is invalid
            RuntimeError: If provider is not initialized
        """
        # If auto_speakers is disabled, return defaults without requiring initialization
        if not self.cfg.auto_speakers:
            logger.debug("Auto-speakers disabled, detection failed")
            return DEFAULT_SPEAKER_NAMES.copy(), set(), False

        if not self._speaker_detection_initialized:
            raise RuntimeError(
                "AnthropicProvider speaker detection not initialized. Call initialize() first."
            )

        logger.debug("Detecting speakers via Anthropic API for episode: %s", episode_title[:50])

        try:
            # Build prompt using prompt_store (RFC-017)
            user_prompt = self._build_speaker_detection_prompt(
                episode_title, episode_description, known_hosts
            )

            # Get system prompt from prompt_store
            from ...prompts.store import render_prompt

            system_prompt_name = (
                self.cfg.anthropic_speaker_system_prompt or "anthropic/ner/system_ner_v1"
            )
            system_prompt = render_prompt(system_prompt_name)

            # Call Anthropic API (different from OpenAI - uses /v1/messages with separate system)
            message = self.client.messages.create(
                model=self.speaker_model,
                max_tokens=300,
                temperature=self.speaker_temperature,
                system=system_prompt,
                messages=[{"role": "user", "content": user_prompt}],
            )

            response_text = message.content[0].text if message.content else ""
            if not response_text:
                logger.warning("Anthropic API returned empty response")
                return DEFAULT_SPEAKER_NAMES.copy(), set(), False

            # Parse JSON response
            speakers, detected_hosts, success = self._parse_speakers_from_response(
                response_text, known_hosts
            )

            logger.debug(
                "Anthropic speaker detection completed: %d speakers, %d hosts, success=%s",
                len(speakers),
                len(detected_hosts),
                success,
            )

            # Track LLM call metrics if available
            if pipeline_metrics is not None and hasattr(message, "usage"):
                input_tokens = message.usage.input_tokens if message.usage else 0
                output_tokens = message.usage.output_tokens if message.usage else 0
                pipeline_metrics.record_llm_speaker_detection_call(input_tokens, output_tokens)

            return speakers, detected_hosts, success

        except json.JSONDecodeError as exc:
            logger.error("Failed to parse Anthropic API JSON response: %s", exc)
            return DEFAULT_SPEAKER_NAMES.copy(), set(), False
        except Exception as exc:
            logger.error("Anthropic API error in speaker detection: %s", exc)
            raise ValueError(f"Anthropic speaker detection failed: {exc}") from exc

    def analyze_patterns(
        self,
        episodes: list[models.Episode],
        known_hosts: Set[str],
    ) -> dict[str, object] | None:
        """Analyze patterns across multiple episodes (optional).

        For Anthropic provider, pattern analysis is not implemented.
        Returns None to use local pattern analysis logic.
        """
        return None

    def _build_speaker_detection_prompt(
        self, episode_title: str, episode_description: str | None, known_hosts: Set[str]
    ) -> str:
        """Build user prompt for speaker detection using prompt_store."""
        from ...prompts.store import render_prompt

        user_prompt_name = self.cfg.anthropic_speaker_user_prompt
        user_prompt = render_prompt(
            user_prompt_name,
            episode_title=episode_title,
            episode_description=episode_description or "",
            known_hosts=", ".join(known_hosts) if known_hosts else "",
        )
        return user_prompt

    def _parse_speakers_from_response(
        self, response_text: str, known_hosts: Set[str]
    ) -> Tuple[list[str], Set[str], bool]:
        """Parse speaker names from Anthropic API response."""
        try:
            data = json.loads(response_text)
            if isinstance(data, dict):
                speakers = data.get("speakers", [])
                hosts = set(data.get("hosts", []))
                guests = data.get("guests", [])
                all_speakers = list(hosts) + guests if not speakers else speakers
                return all_speakers, hosts, True
        except json.JSONDecodeError:
            pass

        # Fallback: parse from plain text
        speakers = []
        for line in response_text.strip().split("\n"):
            for name in line.split(","):
                name = name.strip().strip("-").strip("*").strip()
                if name and len(name) > 1:
                    speakers.append(name)
        detected_hosts = set(s for s in speakers if s in known_hosts)
        return speakers, detected_hosts, len(speakers) > 0

    # ============================================================================
    # SummarizationProvider Protocol Implementation
    # ============================================================================

    def summarize(
        self,
        text: str,
        episode_title: Optional[str] = None,
        episode_description: Optional[str] = None,
        params: Optional[Dict[str, Any]] = None,
        pipeline_metrics: metrics.Metrics | None = None,
    ) -> Dict[str, Any]:
        """Summarize text using Anthropic Claude API.

        Can handle full transcripts directly due to large context window (200k tokens).
        No chunking needed for most podcast transcripts.

        Args:
            text: Transcript text to summarize
            episode_title: Optional episode title
            episode_description: Optional episode description
            params: Optional parameters dict with max_length, min_length, etc.
            pipeline_metrics: Optional metrics tracker

        Returns:
            Dictionary with summary results:
            {
                "summary": str,
                "summary_short": Optional[str],
                "metadata": {...}
            }

        Raises:
            ValueError: If summarization fails
            RuntimeError: If provider is not initialized
        """
        if not self._summarization_initialized:
            raise RuntimeError(
                "AnthropicProvider summarization not initialized. Call initialize() first."
            )

        # Extract parameters with defaults from config
        max_length = (
            (params.get("max_length") if params else None)
            or self.cfg.summary_reduce_params.get("max_new_tokens")
            or 800
        )
        min_length = (
            (params.get("min_length") if params else None)
            or self.cfg.summary_reduce_params.get("min_new_tokens")
            or 100
        )
        custom_prompt = params.get("prompt") if params else None

        logger.debug(
            "Summarizing text via Anthropic API (model: %s, max_tokens: %d)",
            self.summary_model,
            max_length,
        )

        try:
            # Build prompts using prompt_store (RFC-017)
            (
                system_prompt,
                user_prompt,
                system_prompt_name,
                user_prompt_name,
                paragraphs_min,
                paragraphs_max,
            ) = self._build_summarization_prompts(
                text, episode_title, episode_description, max_length, min_length, custom_prompt
            )

            # Call Anthropic API (different from OpenAI - uses /v1/messages with separate system)
            message = self.client.messages.create(
                model=self.summary_model,
                max_tokens=max_length,
                temperature=self.summary_temperature,
                system=system_prompt,
                messages=[{"role": "user", "content": user_prompt}],
            )

            summary = message.content[0].text if message.content else ""
            if not summary:
                logger.warning("Anthropic API returned empty summary")
                summary = ""

            logger.debug("Anthropic summarization completed: %d characters", len(summary))

            # Track LLM call metrics if available
            if pipeline_metrics is not None and hasattr(message, "usage"):
                input_tokens = message.usage.input_tokens if message.usage else 0
                output_tokens = message.usage.output_tokens if message.usage else 0
                pipeline_metrics.record_llm_summarization_call(input_tokens, output_tokens)

            # Get prompt metadata for tracking (RFC-017)
            from ...prompts.store import get_prompt_metadata

            prompt_metadata = {}
            if system_prompt_name:
                prompt_metadata["system"] = get_prompt_metadata(system_prompt_name)
            user_params = {
                "transcript": text[:100] + "..." if len(text) > 100 else text,
                "title": episode_title or "",
                "paragraphs_min": paragraphs_min,
                "paragraphs_max": paragraphs_max,
            }
            user_params.update(self.cfg.summary_prompt_params)
            prompt_metadata["user"] = get_prompt_metadata(user_prompt_name, params=user_params)

            return {
                "summary": summary,
                "summary_short": None,  # Anthropic provider doesn't generate short summaries separately
                "metadata": {
                    "model": self.summary_model,
                    "provider": "anthropic",
                    "max_length": max_length,
                    "min_length": min_length,
                    "prompts": prompt_metadata,
                },
            }

        except Exception as exc:
            logger.error("Anthropic API error in summarization: %s", exc)
            raise ValueError(f"Anthropic summarization failed: {exc}") from exc

    def _build_summarization_prompts(
        self,
        text: str,
        episode_title: Optional[str],
        episode_description: Optional[str],
        max_length: int,
        min_length: int,
        custom_prompt: Optional[str],
    ) -> tuple[str, str, Optional[str], str, int, int]:
        """Build system and user prompts for summarization using prompt_store (RFC-017)."""
        from ...prompts.store import render_prompt

        system_prompt_name = (
            self.cfg.anthropic_summary_system_prompt or "anthropic/summarization/system_v1"
        )
        user_prompt_name = self.cfg.anthropic_summary_user_prompt

        system_prompt = render_prompt(system_prompt_name)

        paragraphs_min = max(1, min_length // 100)
        paragraphs_max = max(paragraphs_min, max_length // 100)

        if custom_prompt:
            user_prompt = custom_prompt.replace("{{ transcript }}", text)
            if episode_title:
                user_prompt = user_prompt.replace("{{ title }}", episode_title)
            user_prompt_name = "custom"
        else:
            template_params = {
                "transcript": text,
                "title": episode_title or "",
                "paragraphs_min": paragraphs_min,
                "paragraphs_max": paragraphs_max,
            }
            template_params.update(self.cfg.summary_prompt_params)
            user_prompt = render_prompt(user_prompt_name, **template_params)

        return (
            system_prompt,
            user_prompt,
            system_prompt_name,
            user_prompt_name,
            paragraphs_min,
            paragraphs_max,
        )

    # ============================================================================
    # Cleanup Methods
    # ============================================================================

    def cleanup(self) -> None:
        """Cleanup all provider resources (no-op for API provider)."""
        self._speaker_detection_initialized = False
        self._summarization_initialized = False

    def clear_cache(self) -> None:
        """Clear cache (no-op for API provider)."""
        pass

    @property
    def is_initialized(self) -> bool:
        """Check if provider is initialized (any component)."""
        return self._speaker_detection_initialized or self._summarization_initialized

6. Factory Updates

Update both factories to support both Config-based and experiment-based modes (like OpenAI):

File: src/podcast_scraper/speaker_detectors/factory.py

def create_speaker_detector(
    cfg_or_provider_type: Union[config.Config, str],
    params: Optional[Union[SpeakerDetectionParams, Dict[str, Any]]] = None,
) -> SpeakerDetector:
    # ... existing code ...

    elif provider_type == "anthropic":
        from ..providers.anthropic.anthropic_provider import AnthropicProvider

        if experiment_mode:
            from ..config import Config
            assert isinstance(params, SpeakerDetectionParams)
            cfg = Config(
                rss="",
                speaker_detector_provider="anthropic",
                anthropic_speaker_model=params.model_name if params.model_name else "claude-haiku-4-5",
                anthropic_temperature=params.temperature if params.temperature is not None else 0.3,
                anthropic_api_key=os.getenv("ANTHROPIC_API_KEY"),
            )
            return AnthropicProvider(cfg)
        else:
            return AnthropicProvider(cfg)
    else:
        raise ValueError(
            f"Unsupported speaker detector type: {provider_type}. "
            "Supported types: 'spacy', 'openai', 'anthropic'"
        )

Similar update for summarization/factory.py.

7. Dependencies

Add to pyproject.toml:

[project.optional-dependencies]
anthropic = [
    "anthropic>=0.30.0,<1.0.0",
]

8. Prompt Templates

Create Anthropic-specific prompts in src/podcast_scraper/prompts/anthropic/:

  • ner/system_ner_v1.j2 - System prompt for speaker detection
  • ner/guest_host_v1.j2 - User prompt for speaker detection
  • summarization/system_v1.j2 - System prompt for summarization
  • summarization/long_v1.j2 - User prompt for summarization

Follow OpenAI prompt patterns but optimize for Claude models.

Testing Strategy

Same pattern as OpenAI provider:

  1. Unit tests: Mock Anthropic API responses
  2. Integration tests: Use E2E mock server with Anthropic endpoints
  3. E2E tests: Full workflow with Anthropic provider

Success Criteria

  1. Anthropic supports speaker detection and summarization via unified provider
  2. Clear error when attempting transcription with Anthropic
  3. Free tier works for development (Haiku model)
  4. E2E tests pass
  5. Experiment mode supported from start
  6. Environment-based model defaults (test vs prod)
  7. Follows OpenAI provider pattern exactly

Migration Notes

  • Breaking Changes: None (new provider, backward compatible)
  • Configuration: Add ANTHROPIC_API_KEY to .env file
  • Dependencies: Install with pip install -e ".[llm]" (anthropic SDK is in the llm extra)

References

  • Related PRD: docs/prd/PRD-009-anthropic-provider-integration.md
  • Reference Implementation: src/podcast_scraper/providers/openai/openai_provider.py
  • Anthropic API Documentation: https://docs.anthropic.com/en/api
  • Anthropic Python SDK: https://github.com/anthropics/anthropic-sdk-python