The IOS Architecture — A Complete Blueprint for a Domain AI Platform
This post is the synthesis. How all of these components fit together into a coherent production system for a specific domain. Not a generic architecture diagram — a real system with real tradeoffs ...
Over the past several weeks, this series has covered every major component of applied AI engineering: transformers, embeddings, tokenization, fine-tuning, the training loop, inference optimization, evaluation frameworks, RAG, agents, context management, structured outputs, prompt systems, and vector databases.
This post is the synthesis. How all of these components fit together into a coherent production system for a specific domain. Not a generic architecture diagram — a real system with real tradeoffs, explained from the perspective of someone building it.
What IOS Is Trying to Do
The thesis: finance and risk analysis is knowledge work that follows patterns. Reading a regulatory filing, identifying capital adequacy metrics, comparing them against framework requirements, flagging shortfalls, producing a structured assessment — this is a defined process with defined inputs and defined outputs. The variation is in the documents, not in the analytical process itself.
Design philosophy: every component exists to produce reliably correct structured outputs that a qualified analyst can trust, review, and act on. Reliability and traceability are first-order requirements.
The Architecture Overview
IOS Architecture Visualization — all five layers
Layer 1: Document Ingestion
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum
import asyncio
class DocumentType(Enum):
REGULATORY_FILING = "regulatory_filing"
ANNUAL_REPORT = "annual_report"
DISCLOSURE_NOTICE = "disclosure_notice"
REFERENCE_DATA = "reference_data"
@dataclass
class IngestedDocument:
doc_id: str
doc_type: DocumentType
source_url: Optional[str]
filing_entity: str
filing_period: str
regulatory_framework: str
raw_text: str
structured_metadata: Dict
ingested_at: str
processing_status: str = "pending"
class DocumentIngestionService:
def __init__(self, embedding_pipeline, regulatory_db):
self.embedding_pipeline = embedding_pipeline
self.regulatory_db = regulatory_db
async def ingest(self, document: IngestedDocument) -> str:
metadata = await self._extract_metadata(document)
document.structured_metadata.update(metadata)
processed_chunks = await self._process_by_type(document)
await self.embedding_pipeline.process_batch([{
"id": document.doc_id, "content": document.raw_text,
"metadata": document.structured_metadata
}])
if document.doc_type == DocumentType.REFERENCE_DATA:
await self.regulatory_db.update(document)
return document.doc_id
async def _process_by_type(self, document: IngestedDocument) -> List[Dict]:
if document.doc_type == DocumentType.REGULATORY_FILING:
return self._chunk_regulatory_filing(document.raw_text)
elif document.doc_type == DocumentType.ANNUAL_REPORT:
return self._chunk_annual_report(document.raw_text)
else:
return self._chunk_default(document.raw_text)
Layer 2: Retrieval
class RetrievalService:
def __init__(self, vector_db, bm25_index, reranker):
self.vector_db = vector_db
self.bm25 = bm25_index
self.reranker = reranker
def retrieve(self, query: str, k: int = 5, filters: Optional[Dict] = None, rerank: bool = True) -> List[Dict]:
semantic_results = self.vector_db.search(query, k=20, filters=filters)
bm25_results = self.bm25.search(query, k=20, filters=filters)
candidates = self._rrf_merge(semantic_results, bm25_results, k=20)
if rerank and len(candidates) > k:
return self.reranker.rerank(query, candidates, top_k=k)
return candidates[:k]
Layer 3: Analysis Engine
class AnalysisEngine:
def __init__(self, retrieval_service, prompt_registry, llm_client):
self.retrieval = retrieval_service
self.registry = prompt_registry
self.llm = llm_client
def analyze(self, request):
complexity = self._assess_complexity(request)
if complexity == "extraction":
return self._run_extraction(request)
elif complexity == "multi_step":
return self._run_chain(request)
else:
return self._run_agent(request)
def _assess_complexity(self, request) -> str:
if request.query_type in ["extract_single_metric", "classify_document"]:
return "extraction"
elif request.n_documents == 1 and request.analytical_steps_defined:
return "multi_step"
else:
return "agent"
Layer 4: Output Validation
class OutputValidationService:
def validate(self, result, source_docs: List[Dict]):
issues = []
issues.extend(self._check_citations(result, source_docs))
issues.extend(self._check_numerical_constraints(result))
issues.extend(self._check_framework_consistency(result))
confidence = result.extraction_confidence
needs_review = (len(issues) > 0 or confidence < 0.85 or
result.query_type in ["compliance_assessment", "risk_flag"])
return ValidationResult(
valid=len(issues) == 0, issues=issues,
route_to_human_review=needs_review,
automated_approval=not needs_review
)
Layer 5: Continuous Evaluation
class ContinuousEvalPipeline:
def run_on_change(self, change_type: str, version: str) -> Dict:
suite_map = {
"prompt_update": ["unit_evals"],
"model_update": ["unit_evals", "task_evals"],
"retrieval_config": ["retrieval_evals"],
"embedding_model": ["retrieval_evals", "task_evals"]
}
results = {}
for suite_name in suite_map.get(change_type, ["unit_evals"]):
result = self.suites[suite_name].run(version=version)
results[suite_name] = result
if result["regression"]:
self.alert(f"REGRESSION in {suite_name}: {result['delta_vs_baseline']:.3f}")
raise DeploymentBlocked(f"Eval regression in {suite_name}")
return results
What This Architecture Gets Right
Separation of concerns: Each layer has one job. Observability at every layer: Every component logs structured behavior for diagnosis. Eval-gated deployment: Nothing ships without passing the eval suite. Human-in-the-loop for consequential outputs: Any analysis that influences a regulatory decision is reviewed by a qualified analyst.
This is the system I’m building. The agent layer is early, the preference eval pipeline is still manual, and the embedding model hasn’t been domain-fine-tuned yet. But the architecture is the right shape.
Action for today: Draw your current AI system architecture on paper. Identify which components are missing. The missing pieces are the failure modes you haven’t encountered yet.
I am publishing lessons and my thoughts as I’m constructing IntelligenceOS — a domain AI platform, for Finance Ops and Risk — built on a family of fine-tuned models — in public, one project at a time. You’ll get the real account: what I built, what broke, what I learned, and how it compounds into a production system.