Module 6: Code Quality Patterns for AI
3. Separation of Concerns
Description
Separation of concerns is the principle that eliminates god functions — those 60-line functions that do prompt construction, API call, JSON parsing, validation, logging, and business logic, all in a single block. This capsule shows how to identify the responsibilities in AI code, how to extract them, and how to decide when NOT to separate further.
The god function: anatomy of the problem
# This real function from an AI app before Module 6.
# Count how many responsibilities it has:
def analyze_text(text: str) -> dict:
# Responsibility 1: INPUT VALIDATION
if not text or len(text) < 5:
raise ValueError("Text too short")
if len(text) > 10000:
text = text[:10000] # Truncate silently (hide the fact)
# Responsibility 2: PROMPT CONSTRUCTION
system_prompt = "You are a sentiment analysis expert."
user_prompt = f"""
Analyze the sentiment of the following text.
Return a JSON with these exact fields:
- sentiment: "positive", "negative", "neutral", or "mixed"
- score: float from -1.0 (very negative) to 1.0 (very positive)
- confidence: float from 0.0 to 1.0
Text to analyze: {text}
"""
# Responsibility 3: LLM CALL (infrastructure)
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
response = client.chat.completions.create(
model=os.getenv("MODEL", "gpt-4o-mini"),
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=float(os.getenv("TEMPERATURE", "0.7")),
max_tokens=int(os.getenv("MAX_TOKENS", "500"))
)
# Responsibility 4: RESPONSE PARSING
raw = response.choices[0].message.content
try:
result = json.loads(raw)
except json.JSONDecodeError:
# Try to extract JSON with regex
match = re.search(r'\{.*\}', raw, re.DOTALL)
if match:
result = json.loads(match.group())
else:
result = {"sentiment": "unknown", "score": 0.0, "confidence": 0.0}
# Responsibility 5: BUSINESS VALIDATION
if "sentiment" not in result:
result["sentiment"] = "unknown"
if result.get("score", 0) > 1.0:
result["score"] = 1.0
if result.get("score", 0) < -1.0:
result["score"] = -1.0
# Responsibility 6: LOGGING
print(f"[{datetime.now()}] Analyzed text ({len(text)} chars): {result['sentiment']}")
return result
# Count: 6 responsibilities in ~55 lines
# Problems:
# - To test the parser, you need the LLM call
# - To test the LLM call, you need the hardcoded prompt
# - To change the prompt, you edit this function
# - To switch from OpenAI to Anthropic, you edit this function
# - To test the validation, you need to mock everything above
# - To reuse the parser in another endpoint, you copy codeThe refactoring step by step
Step 1: Extract the prompt to configuration
# prompts/sentiment/v1.yaml
---
version: "v1"
author: "team"
created: "2024-01-15"
description: "Sentiment analysis prompt - first version"
system: "You are a sentiment analysis expert. Return only valid JSON."
template: |
Analyze the sentiment of the following text.
Return a JSON object with these exact fields:
- "sentiment": one of "positive", "negative", "neutral", "mixed"
- "score": float from -1.0 (very negative) to 1.0 (very positive)
- "confidence": float from 0.0 to 1.0
Text to analyze:
{text}
# Required variables: [text]
# Optional variables: [language, max_words]# src/prompts/loader.py
import yaml
from pathlib import Path
from dataclasses import dataclass
from typing import Optional
PROMPTS_DIR = Path(__file__).parent.parent.parent / "prompts"
@dataclass
class PromptTemplate:
version: str
system: str
template: str
description: Optional[str] = None
def render(self, **kwargs) -> str:
"""Render the template with the given variables."""
try:
return self.template.format(**kwargs)
except KeyError as e:
raise ValueError(f"Missing variable in prompt template: {e}")
def load_prompt(name: str) -> PromptTemplate:
"""
Load a prompt from a YAML file.
Args:
name: path relative to the prompts/ directory, without extension
Example: "sentiment/v1" loads prompts/sentiment/v1.yaml
Returns:
PromptTemplate with the loaded template
Raises:
FileNotFoundError: if the file doesn't exist
"""
path = PROMPTS_DIR / f"{name}.yaml"
if not path.exists():
raise FileNotFoundError(f"Prompt not found: {path}")
with open(path) as f:
data = yaml.safe_load(f)
return PromptTemplate(
version=data.get("version", "v1"),
system=data.get("system", ""),
template=data.get("template", ""),
description=data.get("description")
)Step 2: Extract the parser to output processing
# src/processing/sentiment_parser.py
import json
import re
from typing import Optional
from pydantic import BaseModel, field_validator
class SentimentOutput(BaseModel):
"""Output schema for sentiment analysis."""
sentiment: str
score: float
confidence: float = 1.0
@field_validator("sentiment")
@classmethod
def valid_sentiment(cls, v: str) -> str:
allowed = {"positive", "negative", "neutral", "mixed", "unknown"}
v_lower = v.lower().strip()
if v_lower not in allowed:
return "unknown"
return v_lower
@field_validator("score")
@classmethod
def clamp_score(cls, v: float) -> float:
return max(-1.0, min(1.0, v))
@field_validator("confidence")
@classmethod
def clamp_confidence(cls, v: float) -> float:
return max(0.0, min(1.0, v))
def extract_json(raw: str) -> Optional[str]:
"""Try to extract a JSON object from a string that may contain extra text."""
# Try to parse directly
stripped = raw.strip()
if stripped.startswith("{"):
return stripped
# Look for JSON embedded in a markdown code block
code_block = re.search(r"```(?:json)?\s*(\{.*?\})\s*```", raw, re.DOTALL)
if code_block:
return code_block.group(1)
# Look for JSON anywhere in the string
json_match = re.search(r"\{[^{}]*\}", raw, re.DOTALL)
if json_match:
return json_match.group()
return None
def parse_sentiment_output(raw: str) -> dict:
"""
Parse the LLM output and return a validated dict.
Strategy:
1. Try to parse JSON directly
2. If that fails, try to extract JSON with regex
3. If that fails, return a fallback output
"""
json_str = extract_json(raw)
if json_str is None:
# Fallback: we couldn't extract JSON
return SentimentOutput(
sentiment="unknown",
score=0.0,
confidence=0.0
).model_dump()
try:
data = json.loads(json_str)
output = SentimentOutput(**data)
return output.model_dump()
except (json.JSONDecodeError, ValueError):
return SentimentOutput(
sentiment="unknown",
score=0.0,
confidence=0.0
).model_dump()Step 3: Extract the LLM call to infrastructure
# src/infrastructure/llm_provider.py
from typing import Protocol, runtime_checkable
@runtime_checkable
class LLMProvider(Protocol):
"""
Interface for LLM providers.
Any class that implements this Protocol can be used
as a provider without explicit inheritance.
"""
def complete(self, messages: list[dict], **kwargs) -> str:
"""
Send messages to the LLM and return the content of the response.
Args:
messages: List of messages in OpenAI format
[{"role": "system", "content": "..."}, ...]
Returns:
The content of the LLM's response message
Raises:
LLMProviderError: If the provider cannot complete the request
"""
...
class LLMProviderError(Exception):
"""Generic LLM provider error."""
pass
# src/infrastructure/openai_provider.py
import structlog
import time
from src.infrastructure.llm_provider import LLMProvider, LLMProviderError
log = structlog.get_logger()
class OpenAIProvider:
"""
LLMProvider implementation for OpenAI.
Encapsulates all OpenAI-specific logic.
"""
def __init__(self, client, model: str, temperature: float, max_tokens: int):
self._client = client
self._model = model
self._temperature = temperature
self._max_tokens = max_tokens
def complete(self, messages: list[dict], **kwargs) -> str:
"""Make a call to the OpenAI API."""
start = time.time()
try:
response = self._client.chat.completions.create(
model=self._model,
messages=messages,
temperature=self._temperature,
max_tokens=self._max_tokens,
**kwargs
)
duration_ms = (time.time() - start) * 1000
log.info(
"llm_call_completed",
model=self._model,
input_tokens=response.usage.prompt_tokens,
output_tokens=response.usage.completion_tokens,
duration_ms=round(duration_ms, 1)
)
return response.choices[0].message.content
except Exception as e:
log.error(
"llm_call_failed",
error_type=type(e).__name__,
model=self._model
)
raise LLMProviderError(f"OpenAI call failed: {e}") from eStep 4: Clean business logic
# src/domain/sentiment_service.py
"""
Business logic for sentiment analysis.
This module does NOT import:
- openai, anthropic, or any specific provider
- json, re, or parsing logic
- logging, structlog
It only orchestrates the high-level flow.
"""
from typing import Optional
from src.infrastructure.llm_provider import LLMProvider
from src.processing.sentiment_parser import parse_sentiment_output
from src.prompts.loader import load_prompt
# Business threshold: if confidence < MINIMUM_CONFIDENCE, mark as low_confidence
MINIMUM_CONFIDENCE = 0.3
class LowConfidenceError(Exception):
"""The LLM couldn't analyze with enough confidence."""
pass
def analyze_sentiment(text: str, provider: LLMProvider) -> dict:
"""
Analyze the sentiment of a text.
Business rules:
- If confidence < 0.3, raises LowConfidenceError
- The text must have passed the guardrails before reaching here
Args:
text: Text to analyze (already sanitized by guardrails)
provider: injected LLM provider
Returns:
dict with: sentiment, score, confidence
"""
prompt_template = load_prompt("sentiment/v1")
messages = [
{"role": "system", "content": prompt_template.system},
{"role": "user", "content": prompt_template.render(text=text)}
]
raw_response = provider.complete(messages)
result = parse_sentiment_output(raw_response)
# Business rule: minimum confidence
if result["confidence"] < MINIMUM_CONFIDENCE:
raise LowConfidenceError(
f"Analysis confidence too low: {result['confidence']:.2f} "
f"(minimum: {MINIMUM_CONFIDENCE})"
)
return resultComparison: before vs after
BEFORE (55-line god function):
┌─────────────────────────────────────────────────────────────┐
│ analyze_text(text: str) -> dict │
│ ├── Validate input (should be a guardrail) │
│ ├── Build prompt (should be prompts/) │
│ ├── Call OpenAI directly (should be infra) │
│ ├── Parse JSON with try/except + regex (should be │
│ │ processing) │
│ ├── Validate and normalize values (should be Pydantic) │
│ └── Log with print() (should be an infra wrapper) │
└─────────────────────────────────────────────────────────────┘
AFTER (4 modules, each with one responsibility):
┌─────────────────────────────────────────────────────────────┐
│ prompts/sentiment/v1.yaml (8 lines) │
│ └── The prompt template, versioned, editable │
├─────────────────────────────────────────────────────────────┤
│ src/processing/sentiment_parser.py (45 lines) │
│ └── extract_json() + parse_sentiment_output() │
│ Testable completely independently │
├─────────────────────────────────────────────────────────────┤
│ src/infrastructure/openai_provider.py (35 lines) │
│ └── complete() → API call + logging │
│ Swappable with MockProvider │
├─────────────────────────────────────────────────────────────┤
│ src/domain/sentiment_service.py (20 lines) │
│ └── analyze_sentiment() — orchestrates the 3 above │
│ Fully testable with MockProvider │
└─────────────────────────────────────────────────────────────┘
Total: similar number of lines, but 4 clear responsibilities
When NOT to separate further
# Separation has a limit. Over-separating creates "over-engineering"
# ❌ Over-engineering: creating abstractions with no real value
class PromptRenderer:
class TemplateLoader:
class FileSystemAdapter:
class PathResolver:
def resolve_path(self, name: str) -> Path: ...
# For a function that just does:
PROMPTS_DIR / f"{name}.yaml"
# This is 4 abstraction layers for 1 line of code
# ✅ Practical rule: separate if...
# 1. You want to test each part separately
# → extract_json() has tests, parse_sentiment_output() has tests
# 2. You want to reuse it elsewhere
# → parse_sentiment_output() is used in /analyze AND in /batch
# 3. A change affects one thing, not the others
# → Changing the parser must not affect the LLM call
# 4. The function does more than one thing
# → "does this AND does that" = separate
# ❌ Do NOT separate if...
# - The abstraction only exists for its own sake (YAGNI)
# - You navigate 5 files to understand a 3-line operation
# - The "separation" duplicates code without reducing couplingTests after the separation
# Now each part has its own independent tests
# tests/unit/test_sentiment_parser.py
import pytest
from src.processing.sentiment_parser import parse_sentiment_output, extract_json
class TestExtractJson:
def test_plain_json(self):
assert extract_json('{"a": 1}') == '{"a": 1}'
def test_json_in_markdown(self):
raw = '```json\n{"a": 1}\n```'
assert extract_json(raw) == '{"a": 1}'
def test_json_with_surrounding_text(self):
raw = 'Sure! Here is the result: {"sentiment": "positive"} Hope this helps!'
result = extract_json(raw)
assert '{"sentiment": "positive"}' in result
def test_no_json_returns_none(self):
assert extract_json("no json here") is None
class TestParseSentimentOutput:
def test_valid_output(self):
raw = '{"sentiment": "positive", "score": 0.8, "confidence": 0.9}'
result = parse_sentiment_output(raw)
assert result["sentiment"] == "positive"
assert result["score"] == 0.8
def test_score_clamped(self):
raw = '{"sentiment": "positive", "score": 1.5}'
result = parse_sentiment_output(raw)
assert result["score"] == 1.0 # Clamped
def test_invalid_json_returns_unknown(self):
result = parse_sentiment_output("not valid json")
assert result["sentiment"] == "unknown"
assert result["score"] == 0.0
def test_invalid_sentiment_becomes_unknown(self):
raw = '{"sentiment": "very_positive", "score": 0.9}'
result = parse_sentiment_output(raw)
assert result["sentiment"] == "unknown" # Normalized
# tests/unit/test_sentiment_service.py
import pytest
from src.domain.sentiment_service import analyze_sentiment, LowConfidenceError
class MockProvider:
def __init__(self, response: str):
self._response = response
def complete(self, messages: list) -> str:
return self._response
class TestAnalyzeSentiment:
def test_positive_result(self):
mock = MockProvider('{"sentiment": "positive", "score": 0.9, "confidence": 0.95}')
result = analyze_sentiment("Great product!", mock)
assert result["sentiment"] == "positive"
def test_low_confidence_raises(self):
mock = MockProvider('{"sentiment": "mixed", "score": 0.1, "confidence": 0.1}')
with pytest.raises(LowConfidenceError):
analyze_sentiment("ambiguous text", mock)
def test_parser_called_with_provider_output(self):
"""Domain delegates parsing to the parser, doesn't do it inline."""
mock = MockProvider("```json\n{\"sentiment\": \"negative\", \"score\": -0.5, \"confidence\": 0.8}\n```")
result = analyze_sentiment("This is bad", mock)
# The parser must handle the markdown code block
assert result["sentiment"] == "negative"Exercises
Exercise 1: Identify responsibilities
Read this function and list each responsibility (at least 4):
def generate_summary(article: str, max_words: int = 200) -> dict:
if len(article) > 20000:
article = article[:20000]
prompt = f"Summarize in {max_words} words: {article}"
client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
resp = client.chat.completions.create(model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}])
raw = resp.choices[0].message.content
try:
result = json.loads(raw)
except:
result = {"summary": raw, "word_count": len(raw.split())}
if "summary" not in result:
result["summary"] = raw
print(f"Summary generated: {len(result['summary'])} chars")
return resultSee solution
- Input validation/sanitization:
if len(article) > 20000→ should be a guardrail or in the entrypoint - Prompt construction:
f"Summarize in {max_words} words..."→ should be prompts/summarization/v1.yaml - LLM call:
client.chat.completions.create(...)→ should beOpenAIProvider.complete() - Output parsing:
json.loads(raw)→ should beparse_summary_output()in processing/ - Output validation:
if "summary" not in result→ should be aSummaryOutputPydantic model - Logging:
print(...)→ should be structlog in the provider wrapper
Exercise 2: Incremental refactoring
Propose the order of steps to refactor the function above without breaking anything:
See guide
Safe order (red-green-refactor):
- Add tests for the original function (if they don't exist) → safety net
- Extract the parser to
processing/summary_parser.py+ parser tests - Extract the prompt to
prompts/summarization/v1.yaml+ loader - Create MockProvider for tests
- Extract the LLM call to
OpenAIProvider.complete() - Clean up the main function so it only orchestrates
- Run all the tests → they must pass
At each step: run tests before and after.
Summary
- God functions are the enemy: a function that does 6 things can't be tested, changed, or reused
- Incremental refactoring: extract one responsibility at a time, checking tests at each step
- The limit of separation: separate when it makes testing, reuse, or maintenance easier — not out of dogma
- Tests as a demonstration of value: with the parser separated, its tests are trivial and give high confidence
Additional resources
- Single Responsibility Principle (SRP) — The foundational principle
- Refactoring: Improving the Design of Existing Code (Fowler) — The reference book
- Working Effectively with Legacy Code (Feathers) — For refactoring code without tests
- YAGNI — You Ain't Gonna Need It — The principle of not over-engineering