Module 6: Code Quality Patterns for AI
5. Dependency Injection for LLM Providers
Description
Dependency Injection (DI) is the pattern that decouples your business logic from any specific LLM implementation. Without DI, switching from OpenAI to Anthropic means rewriting business code. With DI, it's creating a new Protocol implementation and changing one line in the configuration. This capsule implements complete DI: the Protocol, the implementations (OpenAI, Mock, Fallback), and the injection in FastAPI.
The problem without DI: deep coupling
# ❌ Without DI: the business is coupled to OpenAI
# src/domain/sentiment_service.py
from openai import OpenAI # ← Infrastructure import in domain
import os
def analyze_sentiment(text: str) -> dict:
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY")) # ← Internal creation
response = client.chat.completions.create(
model="gpt-4o-mini", # ← Hardcoded
messages=[...],
temperature=0.7 # ← Magic number
)
return parse_result(response.choices[0].message.content)
# Problems:
#
# 1. To test:
# → You need patch("openai.OpenAI") or patch("openai.chat.completions.create")
# → If OpenAI changes its API internally, your mock breaks
# → The test is tied to OpenAI's implementation
#
# 2. To switch providers:
# → Search all `from openai import OpenAI` imports across the whole codebase
# → Rewrite the call logic (which differs between OpenAI and Anthropic)
# → Verify nothing broke (no guarantee if you don't have tests)
#
# 3. To add a fallback:
# → Wrap the call with try/except and duplicate the logic for another provider
# → Hard to testThe Protocol: defining the interface
# src/infrastructure/llm_provider.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class LLMProvider(Protocol):
"""
Interface for LLM providers.
@runtime_checkable allows using isinstance() at runtime:
>>> isinstance(my_provider, LLMProvider)
True
Any class that has the complete() method with this signature
satisfies the Protocol WITHOUT explicit inheritance.
This is "duck typing" with type safety:
- OpenAIProvider doesn't need `class OpenAIProvider(LLMProvider)`
- It just needs to implement complete() with the correct signature
"""
def complete(self, messages: list[dict], **kwargs) -> str:
"""
Send messages to the LLM and return the text content of the response.
Args:
messages: List of messages in standard format
[{"role": "system", "content": "..."},
{"role": "user", "content": "..."}]
**kwargs: Optional provider-specific parameters
Returns:
The content of the response message as a string
Raises:
LLMProviderError: If the provider cannot complete the request
(rate limit, timeout, invalid response, etc.)
"""
...
class LLMProviderError(Exception):
"""
Generic LLM provider error.
Wraps provider-specific errors (OpenAIError, etc.)
in a generic error that the domain can handle without knowing
which provider is being used.
"""
def __init__(self, message: str, original_error: Exception = None):
super().__init__(message)
self.original_error = original_errorOpenAI implementation
# src/infrastructure/openai_provider.py
import time
import structlog
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
from src.logging_config import calculate_cost
log = structlog.get_logger()
class OpenAIProvider:
"""
LLMProvider implementation for the OpenAI API.
Encapsulates EVERYTHING OpenAI-specific:
- Request format (messages, model, temperature, etc.)
- Response format (choices[0].message.content)
- Usage and cost (usage.prompt_tokens, usage.completion_tokens)
- Handling of OpenAI-specific errors
- Logging of tokens, cost, latency
"""
def __init__(self, client, model: str, temperature: float, max_tokens: int,
seed: int = None):
self._client = client
self._model = model
self._temperature = temperature
self._max_tokens = max_tokens
self._seed = seed
def complete(self, messages: list[dict], **kwargs) -> str:
"""Make a call to the OpenAI API."""
start = time.time()
request_kwargs = {
"model": self._model,
"messages": messages,
"temperature": self._temperature,
"max_tokens": self._max_tokens,
}
if self._seed is not None:
request_kwargs["seed"] = self._seed
request_kwargs.update(kwargs)
try:
response = self._client.chat.completions.create(**request_kwargs)
duration_ms = (time.time() - start) * 1000
cost_usd = calculate_cost(
self._model,
response.usage.prompt_tokens,
response.usage.completion_tokens
)
log.info(
"llm_call_completed",
model=self._model,
input_tokens=response.usage.prompt_tokens,
output_tokens=response.usage.completion_tokens,
cost_usd=cost_usd,
duration_ms=round(duration_ms, 1),
finish_reason=response.choices[0].finish_reason
)
return response.choices[0].message.content
except Exception as e:
log.error(
"llm_call_failed",
model=self._model,
error_type=type(e).__name__,
duration_ms=round((time.time() - start) * 1000, 1)
)
raise LLMProviderError(
f"OpenAI call failed: {type(e).__name__}: {str(e)}",
original_error=e
)
@classmethod
def from_settings(cls, settings) -> "OpenAIProvider":
"""Factory to create from Settings."""
return cls(
client=settings.create_openai_client(),
model=settings.model,
temperature=settings.temperature,
max_tokens=settings.max_tokens,
seed=settings.seed
)Mock implementation for tests
# src/infrastructure/mock_provider.py
from src.infrastructure.llm_provider import LLMProvider
from typing import Union, Callable, Optional
class MockProvider:
"""
LLMProvider mock for tests and development.
Supports three modes:
1. fixed response: always returns the same string
2. callable: returns whatever the function returns
3. sequence: returns different responses on each call
Records all calls for verification in tests.
"""
def __init__(
self,
response: Union[str, Callable[[list], str]] = None,
responses: list[str] = None,
raise_error: Exception = None
):
"""
Args:
response: Fixed string or callable(messages) -> str
responses: List of strings, used in order (circular)
raise_error: If specified, raise this error in complete()
"""
self._response = response or '{"sentiment": "positive", "score": 0.8, "confidence": 0.9}'
self._responses = responses
self._raise_error = raise_error
self._call_count = 0
self.calls: list[list[dict]] = [] # Record of all calls
def complete(self, messages: list[dict], **kwargs) -> str:
"""Return the mock response without calling any API."""
self.calls.append(messages)
self._call_count += 1
if self._raise_error:
raise self._raise_error
if self._responses:
# Return responses in sequence (circular)
idx = (self._call_count - 1) % len(self._responses)
return self._responses[idx]
if callable(self._response):
return self._response(messages)
return self._response
@property
def call_count(self) -> int:
return self._call_count
def was_called_with_system_message(self, content: str) -> bool:
"""Check whether it was called with a specific system message."""
for call_messages in self.calls:
for msg in call_messages:
if msg.get("role") == "system" and content in msg.get("content", ""):
return True
return False
def get_last_user_message(self) -> Optional[str]:
"""Return the last user message."""
if not self.calls:
return None
for msg in reversed(self.calls[-1]):
if msg.get("role") == "user":
return msg.get("content")
return NoneFallbackProvider implementation
# src/infrastructure/fallback_provider.py
import structlog
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
log = structlog.get_logger()
class FallbackProvider:
"""
Provider that tries multiple providers in order.
If the first one fails, it tries the next.
Useful for:
- High availability: OpenAI → Anthropic → local Ollama
- Cost reduction: gpt-4o-mini (fails) → gpt-3.5-turbo
- Development: real API (fails with rate limit) → local mock
"""
def __init__(self, providers: list[LLMProvider], name: str = "fallback"):
if len(providers) < 2:
raise ValueError("FallbackProvider requires at least 2 providers")
self._providers = providers
self._name = name
def complete(self, messages: list[dict], **kwargs) -> str:
"""Try each provider in order until one works."""
last_error = None
for i, provider in enumerate(self._providers):
try:
result = provider.complete(messages, **kwargs)
if i > 0:
log.warning(
"fallback_provider_used",
fallback_index=i,
provider_type=type(provider).__name__
)
return result
except LLMProviderError as e:
last_error = e
log.warning(
"provider_failed_trying_next",
provider_index=i,
provider_type=type(provider).__name__,
error=str(e)[:100],
has_next=i < len(self._providers) - 1
)
# All providers failed
raise LLMProviderError(
f"All {len(self._providers)} providers failed. "
f"Last error: {last_error}",
original_error=last_error
)Dependency injection in FastAPI
# src/app/dependencies.py
from functools import lru_cache
from src.config import get_settings
from src.infrastructure.llm_provider import LLMProvider
from src.infrastructure.openai_provider import OpenAIProvider
from src.infrastructure.mock_provider import MockProvider
def get_llm_provider() -> LLMProvider:
"""
FastAPI dependency that creates and returns the correct LLM provider
based on the current configuration.
This is the only place where you decide which implementation to use.
The rest of the code (domain, endpoints) only knows LLMProvider.
"""
settings = get_settings()
if settings.use_mock_llm:
return MockProvider(response=settings.mock_response)
return OpenAIProvider.from_settings(settings)
# For singleton (the same provider for the whole app):
@lru_cache(maxsize=1)
def get_cached_provider() -> LLMProvider:
"""Singleton provider — created once and reused."""
return get_llm_provider()
# src/app/routers/sentiment.py
from fastapi import APIRouter, Depends
from src.domain.sentiment_service import analyze_sentiment
from src.infrastructure.llm_provider import LLMProvider
from src.app.dependencies import get_llm_provider
router = APIRouter()
@router.post("/analyze")
async def analyze_endpoint(
body: AnalyzeRequest,
provider: LLMProvider = Depends(get_llm_provider) # Injected by FastAPI
):
"""
The endpoint doesn't know whether it's using OpenAI, Anthropic, or Mock.
It only knows it has an LLMProvider.
"""
result = analyze_sentiment(text=body.text, provider=provider)
return AnalyzeResponse(**result, request_id=get_request_id())Simple tests thanks to DI
# tests/unit/test_sentiment_service.py
import pytest
from src.domain.sentiment_service import analyze_sentiment, LowConfidenceError
from src.infrastructure.mock_provider import MockProvider
class TestAnalyzeSentiment:
"""
Domain tests using MockProvider.
There's no:
- patch() of any module
- OpenAI API keys
- Internet connection
- Sleep or rate limits
Each test is deterministic and fast (<10ms).
"""
def test_positive_sentiment(self):
mock = MockProvider('{"sentiment": "positive", "score": 0.8, "confidence": 0.9}')
result = analyze_sentiment("This is great!", mock)
assert result["sentiment"] == "positive"
assert result["score"] == 0.8
assert result["confidence"] == 0.9
def test_negative_sentiment(self):
mock = MockProvider('{"sentiment": "negative", "score": -0.7, "confidence": 0.85}')
result = analyze_sentiment("This is terrible", mock)
assert result["sentiment"] == "negative"
def test_low_confidence_raises_error(self):
mock = MockProvider('{"sentiment": "mixed", "score": 0.1, "confidence": 0.1}')
with pytest.raises(LowConfidenceError):
analyze_sentiment("ambiguous text", mock)
def test_invalid_json_from_llm_returns_unknown(self):
mock = MockProvider("I'm sorry, I cannot analyze this text.")
result = analyze_sentiment("test", mock)
assert result["sentiment"] == "unknown"
assert result["score"] == 0.0
def test_provider_called_once(self):
"""Verify that the provider is called exactly once."""
mock = MockProvider('{"sentiment": "positive", "score": 0.9, "confidence": 0.95}')
analyze_sentiment("test text", mock)
assert mock.call_count == 1
def test_prompt_contains_input_text(self):
"""Verify that the user's text is in the prompt sent to the LLM."""
mock = MockProvider('{"sentiment": "positive", "score": 0.9, "confidence": 0.95}')
test_text = "unique_test_string_12345"
analyze_sentiment(test_text, mock)
last_message = mock.get_last_user_message()
assert test_text in last_message, \
f"Expected '{test_text}' in the prompt, got: {last_message}"
def test_fallback_provider_tries_secondary_on_failure(self):
"""FallbackProvider uses the second one if the first fails."""
from src.infrastructure.llm_provider import LLMProviderError
from src.infrastructure.fallback_provider import FallbackProvider
failing_provider = MockProvider(
raise_error=LLMProviderError("Rate limit exceeded")
)
working_provider = MockProvider(
'{"sentiment": "positive", "score": 0.9, "confidence": 0.95}'
)
fallback = FallbackProvider([failing_provider, working_provider])
result = analyze_sentiment("test", fallback)
assert result["sentiment"] == "positive"
assert failing_provider.call_count == 1 # Tried the first
assert working_provider.call_count == 1 # Used the secondFull comparison: without vs with DI
Without DI:
TESTING analyze_sentiment():
→ patch("openai.OpenAI") to mock the client
→ patch("openai.chat.completions.create") for the response
→ Configure the mock with the exact OpenAI API structure
→ If OpenAI changes its API, the mock breaks even if your code is fine
→ Setup time: 15-20 lines of code per test
With DI:
TESTING analyze_sentiment():
→ mock = MockProvider(response='{"sentiment": "positive", ...}')
→ result = analyze_sentiment("text", mock)
→ Setup time: 1 line of code per test
─────────────────────────────────────────────
Without DI:
SWITCHING OpenAI → Anthropic:
→ Search all `from openai import` in the project
→ Search all `client.chat.completions.create()`
→ Rewrite with the Anthropic API (different format)
→ Test everything manually
→ Time: 1-2 days
With DI:
SWITCHING OpenAI → Anthropic:
→ Create AnthropicProvider(LLMProvider) with its complete()
→ In dependencies.py: return AnthropicProvider() instead of OpenAIProvider()
→ All the domain tests keep passing without changes
→ Time: 2-4 hours
Exercises
Exercise 1: Implement AnthropicProvider
Write the skeleton of AnthropicProvider that implements LLMProvider:
See solution
# src/infrastructure/anthropic_provider.py
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
class AnthropicProvider:
"""LLMProvider implementation for Anthropic Claude."""
def __init__(self, client, model: str = "claude-3-haiku-20240307",
max_tokens: int = 500):
self._client = client
self._model = model
self._max_tokens = max_tokens
def complete(self, messages: list[dict], **kwargs) -> str:
# Anthropic has a different format — separate system message
system = ""
user_messages = []
for msg in messages:
if msg["role"] == "system":
system = msg["content"]
else:
user_messages.append(msg)
try:
response = self._client.messages.create(
model=self._model,
max_tokens=self._max_tokens,
system=system,
messages=user_messages
)
return response.content[0].text
except Exception as e:
raise LLMProviderError(f"Anthropic call failed: {e}", original_error=e)Exercise 2: FallbackProvider test
Write a test that verifies that if the first provider raises LLMProviderError 3 times in a row, the FallbackProvider uses the second one:
See solution
def test_fallback_after_multiple_failures():
from src.infrastructure.llm_provider import LLMProviderError
from src.infrastructure.fallback_provider import FallbackProvider
primary = MockProvider(raise_error=LLMProviderError("Primary unavailable"))
secondary = MockProvider('{"sentiment": "neutral", "score": 0.0, "confidence": 0.8}')
fallback = FallbackProvider([primary, secondary])
# Three separate calls
for _ in range(3):
result = fallback.complete([{"role": "user", "content": "test"}])
data = json.loads(result)
assert data["sentiment"] == "neutral"
assert primary.call_count == 3 # Tried 3 times
assert secondary.call_count == 3 # Used as fallback 3 timesSummary
- Protocol defines the interface that any LLM provider must implement — without explicit inheritance
- OpenAIProvider encapsulates all OpenAI-specific logic (format, logging, errors)
- MockProvider configurable for tests: fixed response, callable, sequence, or error
- FallbackProvider implements high availability: if A fails, it tries B
Depends(get_llm_provider)in FastAPI: the endpoint receives an LLMProvider without knowing which one- Tests without patch: with DI, the domain's unit tests are trivial and don't break with API changes
Additional resources
- typing.Protocol Python Docs — Official documentation
- FastAPI Dependency Injection — DI in FastAPI
- Dependency Injection (Martin Fowler) — The original article
- Hexagonal Architecture (Alistair Cockburn) — Ports and Adapters
- Anthropic Python SDK — For implementing AnthropicProvider