# 🏛️ Software Architecture Specification: Automated Structured Content Generator (ASCGenerator) > **Status:** Active Standard | **Version:** 1.0.0 | **Scope:** Microservice Architecture & Technical Invariants for `4_ASCGenerator` --- ## 1. Executive Overview & System Topology `4_ASCGenerator` is a production-grade microservice designed to ingest unstructured, noisy news payloads (raw text, cleaned HTML snippets, RSS/Atom feed items) and transform them into strictly typed, schema-validated JSON records. The architecture enforces a **two-tiered resilience pipeline**, **Pydantic V2 validation firewall**, and **defensive prompt isolation** to guarantee zero schema drift and protect downstream consumption pipelines from non-deterministic LLM failures or prompt injection attacks. ```text ┌──────────────────────────────────────────────────────────────────────────────────┐ │ INGESTION LAYER │ │ FastAPI Microservice (POST /v1/extract, /v1/extract/batch) │ └────────────────────────────────────────┬─────────────────────────────────────────┘ │ ▼ ┌───────────────────┐ │ Config & Schemas │ │ (config.py, │ │ schemas.py) │ └─────────┬─────────┘ │ ┌──────────────────────┼──────────────────────┐ │ │ │ ▼ ▼ ▼ ┌──────────────────┐ ┌──────────────────┐ ┌──────────────────┐ │ Defensive Prompt │ │ Deduplication │ │ Structlog JSON │ │ Isolation Layer │ │ Cache (SHA256) │ │ FinOps Logger │ │ (security.py) │ └────────┬─────────┘ │ (logger.py) │ └────────┬─────────┘ │ └────────┬─────────┘ │ │ Hit │ │ ├───────────────────────┤ ▼ │ Miss │ ┌───────────────────────────────┴───────────────────────┐ │ LLM ENGINE & TWO-TIERED RESILIENCE │ │ (engine.ExtractionEngine) │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Tier 1: Tenacity Exponential Backoff Retry │ │ │ │ (Network, Rate Limits 429, 5xx) │ │ │ └────────────────────────┬────────────────────────┘ │ │ │ │ │ ▼ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Instructor SDK Constrained LLM Inference │ │ │ │ (AsyncOpenAI / Anthropic / Mistral, temp=0.0) │ │ │ └────────────────────────┬────────────────────────┘ │ │ │ Raw Output │ │ ▼ │ │ ┌─────────────────────────────────────────────────┐ │ │ │ Tier 2: Pydantic V2 Validation Firewall & │ │ │ │ Schema Self-Correction Loop (Max 2) │ │ │ └────────────────────────┬────────────────────────┘ │ └───────────────────────────┼───────────────────────────┘ │ Validated Record ▼ ┌─────────────────────────┐ │ ArticleExtractionRecord │ │ (JSON Response / DB) │ └─────────────────────────┘ ``` --- ## 2. AI Architecture Enforcement Rules > ⚠️ **STRICT COMPLIANCE REQUIRED:** Any diff or code modification violating these architectural invariants MUST be rejected. ### Rule 1: Layered Dependency Flow (Strict Directional Imports) Dependencies flow strictly top-down. Leaf modules containing schemas and configurations MUST NOT import higher-level components or clients: ```text main.py (FastAPI App & Endpoints) └──> engine.py (Extraction Engine & Resilience Pipeline) ├──> security.py (Prompt Sanitization & Delimiters) ├──> config.py (Pydantic BaseSettings) ├──> logger.py (Structlog JSON Logger) └──> schemas.py (Pydantic V2 Models & DTOs) ``` * **Invariant:** `schemas.py`, `security.py`, and `config.py` are leaf/pure modules and MUST NOT import `engine.py` or `main.py`. * **Violation:** Circular dependencies or importing `FastAPI` inside domain schemas. ### Rule 2: Exception Shielding ("Zero Naked Crash") * No raw third-party exceptions (`httpx.HTTPError`, `openai.APIError`, `pydantic.ValidationError`) may bubble unhandled outside the microservice boundary. * All transport and validation errors MUST be trapped inside `engine.py` or handled via FastAPI exception handlers. * Production runtime MUST respond with structured HTTP status codes (e.g., `422 Unprocessable Entity` for unrepairable schema mismatches, `500 Internal Server Error` with masked details for unhandled runtime failures) and structured JSON logs. **Raw stack traces exposed to API clients are forbidden.** ### Rule 3: Deterministic Data Contracts (Pydantic V2 & Grammar Decoding) * Data entering or leaving any component MUST be strictly typed with Pydantic V2 models. Raw, untyped `dict` objects are forbidden for core domain logic. * LLM inference parameters MUST enforce `temperature = 0.0` for deterministic extraction behavior. * Schema bounds MUST be enforced during model decoding using provider-native constrained output modes (e.g., `Instructor` patch on Async API clients). ### Rule 4: FinOps Observability & Metadata Auditability Every LLM extraction execution path MUST capture and log FinOps metrics via `structlog`: 1. `prompt_tokens` & `completion_tokens` (exact API usage counts). 2. `execution_time_seconds` (measured via `time.perf_counter()`). 3. `estimated_cost_usd` (computed via model token pricing matrix). 4. `is_cached` (boolean indicator for deduplication hits). 5. `schema_correction_attempts` (number of Tier-2 self-repair cycles used). ### Rule 5: Pure Logic & Side-Effect Isolation * Text sanitization (`prepare_user_payload`), XML tag stripping, and Pydantic validation rules MUST remain pure functions with zero I/O side effects. * Network calls (Async OpenAI API requests, HTTP transport) MUST be isolated inside `ExtractionEngine` in `engine.py`. ### Rule 6: Guardrails & Defensive Security Isolation * Untrusted input MUST be sanitized to remove closing delimiter tags (`` replaced with `[TAG_REMOVED]`) and wrapped inside `` tags. * Developer authority is established in the `SYSTEM_PROMPT`, strictly prohibiting the model from obeying instructions embedded in the payload text. * API keys and secrets MUST be retrieved via typed environment settings (`Pydantic BaseSettings` in `config.py`). Hardcoding secrets is strictly prohibited. --- ## 3. Core Contract Specification (Pydantic V2) The data models defined in [`src/schemas.py`](file:///home/michael/Code/ai-engineering/projets/4_ASCGenerator/src/schemas.py) form the strict wire contract for the microservice: ```python from datetime import datetime from enum import Enum from typing import List, Literal, Optional from uuid import UUID, uuid4 from pydantic import BaseModel, EmailStr, Field, HttpUrl, field_validator class ArticleCategory(str, Enum): POLITICS = "politics" TECHNOLOGY = "technology" BUSINESS = "business" SCIENCE = "science" HEALTH = "health" SPORTS = "sports" ENTERTAINMENT = "entertainment" OTHER = "other" class SentimentLabel(str, Enum): POSITIVE = "positive" NEUTRAL = "neutral" NEGATIVE = "negative" class ImpactLevel(str, Enum): LOW = "LOW" MEDIUM = "MEDIUM" HIGH = "HIGH" CRITICAL = "CRITICAL" class ArticleAuthor(BaseModel): name: str = Field(..., min_length=1, max_length=100, description="Author's full name") email: Optional[EmailStr] = Field(default=None, description="Verified contact email if explicitly provided") organization: Optional[str] = Field(default=None, max_length=100, description="Associated news outlet or agency") class NamedEntity(BaseModel): name: str = Field(..., description="Canonical name of the extracted entity") category: Literal["ORGANIZATION", "PERSON", "LOCATION", "PRODUCT"] = Field( ..., description="Entity domain classification" ) sentiment: SentimentLabel = Field(..., description="Entity-specific sentiment orientation") class FinancialMetric(BaseModel): metric_name: str = Field(..., description="Name of the metric (e.g., Revenue, Operating Margin, YoY Growth)") value: float = Field(..., description="Raw numerical value extracted from text") unit: str = Field(..., description="Unit of measure (e.g., USD, EUR, Percentage, Subscribers)") time_period: Optional[str] = Field(default=None, description="Applicable timeframe (e.g., Q2 2026, FY2025)") class ArticleExtractionRecord(BaseModel): schema_version: Literal["1.0"] = "1.0" article_id: UUID = Field(default_factory=uuid4, description="Unique internal identifier") # Core Content title: str = Field(..., min_length=5, max_length=300, description="Cleaned, standardized article title") summary: str = Field(..., min_length=10, max_length=500, description="Concise factual narrative summary") # Metadata source_name: str = Field(..., max_length=100, description="Publishing organization or domain") source_url: Optional[HttpUrl] = Field(default=None, description="Canonical article link") published_at: Optional[datetime] = Field(default=None, description="Original publication timestamp in ISO 8601") extracted_at: datetime = Field(default_factory=datetime.utcnow, description="System extraction timestamp") # Classification primary_category: ArticleCategory = Field(..., description="Primary domain topic") overall_sentiment: SentimentLabel = Field(..., description="Dominant sentiment classification") impact_assessment: ImpactLevel = Field(..., description="Estimated operational or market impact rating") # Extracted Lists authors: List[ArticleAuthor] = Field(default_factory=list, description="List of identified authors") entities: List[NamedEntity] = Field(default_factory=list, description="Extracted named entities") financial_metrics: List[FinancialMetric] = Field(default_factory=list, description="Extracted financial/numerical data") # System Metrics confidence_score: float = Field(..., description="Model extraction confidence strictly bounded [0.0, 1.0]") @field_validator("confidence_score") @classmethod def validate_confidence_range(cls, value: float) -> float: if not (0.0 <= value <= 1.0): raise ValueError("confidence_score must be bounded between 0.0 and 1.0") return value ``` --- ## 4. Real-Time Payload Processing vs. Vector Store & RAG * **Architecture Scope:** `4_ASCGenerator` operates as a real-time extraction microservice designed for low-latency payload extraction directly at the API ingestion boundary. Therefore, persistent vector databases (e.g., Chroma, Qdrant) are intentionally omitted from this component to maintain high throughput and low infrastructure complexity. * **Pre-Processing Pipeline:** Prior to LLM ingestion, raw input text undergo sanitization: - HTML tags and script elements are cleaned using boilerplate strippers (`trafilatura` or `newspaper3k`). - Whitespace and newline characters are normalized. - Potential XML delimiter injection sequences (``) are neutralized. --- ## 5. Resilience Engine & Caching Specification ### Two-Tiered Resilience Architecture The system decouples network/transport failures from semantic/schema validation failures: ```text [ Incoming Request ] │ ▼ [ Tier 1: Tenacity Transport Retry ] - Intercepts API 429, 5xx, Timeouts - Applies Exponential Backoff (mult=1, min=2s, max=10s) - Max 3 Attempts │ ▼ [ LLM Provider API Execution ] │ ▼ [ Tier 2: Instructor Schema Self-Correction Loop ] - Validates JSON output against Pydantic V2 ArticleExtractionRecord - If ValidationError occurs: 1. Captures Pydantic validation failure error traceback 2. Appends error feedback to conversation turns 3. Invokes model for self-repair (Max 2 Attempts) │ ▼ [ Validated Payload / HTTP 200 ] ``` ### Idempotency & Deduplication Cache Strategy To optimize token cost and latency, identical news payloads are deduplicated prior to LLM invocation: - **Cache Key:** `SHA256(raw_text.strip() + ":" + model_name)` - **Hit Strategy:** Returns pre-validated `ArticleExtractionRecord` with `is_cached=True` metadata. --- ## 6. Tech Stack & Infrastructure Constraints | Component | Technology | Specification / Constraint | | --- | --- | --- | | **Language Runtime** | Python 3.11+ | Enforced via `mypy --strict` static type checking | | **API Framework** | FastAPI + Uvicorn | Async ASGI microservice with CORS & OpenAPI `/docs` | | **Data Contracts** | Pydantic V2 | Immutable, strictly validated domain models | | **LLM Constrained Decoding** | Instructor SDK | Context-Free Grammar enforcement & schema retries | | **LLM Providers** | Async OpenAI / Anthropic / Mistral | Multi-provider async client bindings | | **Network Resilience** | Tenacity | Decorated backoff for network drops & rate limits | | **Structured Logging** | Structlog | JSON formatted stdout logs with FinOps metrics | | **Containerization** | Docker | Multi-stage build (`python:3.11-slim`), non-root user | | **Dependency Management** | Poetry | Locked dependencies via `pyproject.toml` | | **Code Quality & Linting** | Ruff | Enforced via pre-commit hooks and CI pipelines |