System Architecture

IO-Link Cloud RAG Pipeline v1.0

Pipeline Overview

flowchart TB subgraph Ingestion["📥 Ingestion Flow"] LS[Log Shipper CronJob
every 3 hours] -->|POST /webhook/siemens-rag-ingest| QUEUE[Ingestion Queue
Deduplication + Alerting] UI_UPLOAD[Manual Upload
Drag & Drop] --> QUEUE AUTO[Auto-Training
Scheduler 2AM UTC] --> QUEUE QUEUE --> SPLIT[Text Splitter
1000 chars / 200 overlap] SPLIT --> BATCH[Batch Processor
500 chunks/batch] BATCH -->|POST /v1/embeddings| EMB_API[Siemens LLM API
qwen3-embedding-8b] EMB_API --> VS_LOGS[(PGVector
siemens_logs)] EMB_API --> VS_KB[(PGVector
siemens_knowledge_base)] end subgraph Query["💬 Query Flow"] USER[User Question] -->|POST /chat/ask/stream| CHAT[Chat Endpoint
SSE Streaming] CHAT -->|POST /v1/embeddings| EMB_Q[Embed Query] EMB_Q --> HYBRID[Hybrid Search
Vector + Keyword] HYBRID --> VS_LOGS HYBRID --> VS_KB HYBRID --> RERANK[Reranker
POST /v1/rerank
bge-reranker-v2-m3] RERANK --> TOP5[Top 5 Documents] TOP5 --> PROMPT[Build Prompt
System + History + Context] PROMPT -->|POST /v1/chat/completions
streaming| LLM[Siemens LLM API
qwen-3.6-27b] LLM -->|SSE tokens| RESP[Stream Response
Markdown Rendered] RESP --> SAVE[Save to SQLite
Chat History + Token Usage] end subgraph Maintenance["🔧 Maintenance"] TTL[TTL Cleanup
Daily 3AM UTC] -->|Delete vectors > 7 days| VS_LOGS ALERT[Error Alerting] -->|Webhook| TEAMS[Teams/Slack] RATE[Rate Limiter
20 req/min per IP] --> CHAT end subgraph Infra["⚙️ Infrastructure"] K8S[Kubernetes EKS
aws-euce1-iolink-prod40] --> POD[RAG Pipeline Pod
2Gi mem / 1 CPU] POD --> SQLITE[(SQLite /data/
Config + Sessions + Usage)] POD --> PVC[PVC 5Gi] POD --> RDS[(AWS RDS PostgreSQL
pgvector extension)] IRSA[IRSA Service Account] -.->|IAM Auth| RDS end

📥 Ingestion Pipeline

  • •Log Shipper CronJob collects container logs every 3 hours
  • •Ingestion queue processes jobs sequentially (max 200)
  • •SHA-256 content hashing prevents duplicate embeddings
  • •Error pattern detection triggers alerts on ingestion
  • •Batched embedding (500 chunks/batch) prevents OOM
  • •Multi-collection: logs → siemens_logs, docs → siemens_knowledge_base

💬 Query Pipeline

  • •Hybrid search: vector similarity + SQL keyword ILIKE
  • •Searches both knowledge base and logs collections
  • •Reranker scores top 30 → selects best 5 for context
  • •Streaming SSE delivers tokens in real-time
  • •Markdown rendering with code highlighting
  • •Chat history preserved in SQLite per session

🔧 Maintenance

  • •TTL cleanup removes log vectors older than 7 days (configurable)
  • •Auto-training re-ingests URLs/files nightly at 2AM UTC
  • •CronJob pods auto-destroy 5 min after completion
  • •Token usage tracking for cost monitoring
  • •Rate limiting: 20 requests/min per IP on chat endpoints

⚙️ Infrastructure

  • •Single pod on EKS (2Gi memory, 1 CPU, 5Gi PVC)
  • •AWS RDS PostgreSQL with pgvector extension
  • •IRSA for secure IAM-based RDS authentication
  • •ArgoCD GitOps deployment from code.siemens.com
  • •Fully async non-blocking FastAPI (uvicorn)
  • •All settings configurable from UI without restart

API Endpoints

Method Endpoint Description
POST/chat/ask/streamRAG chat with SSE streaming
POST/chat/askRAG chat (non-streaming)
GET/chat/sessionsList chat sessions
GET/chat/sessions/{id}/exportExport session as markdown
POST/webhook/siemens-rag-ingestIngest documents (queued)
GET/webhook/queue/statusIngestion queue status
GET/api/modelsList available LLM models
GET/api/usageToken usage statistics
POST/api/maintenance/cleanupTrigger TTL vector cleanup
DELETE/api/documents/{doc_id}Delete document + vectors
GET/healthHealth check (liveness/readiness)

Siemens LLM API Endpoints Used

POST /v1/chat/completions

Chat/reasoning with streaming

Active

POST /v1/embeddings

Document & query embeddings

Active

POST /v1/rerank

Re-rank retrieved documents

Active

GET /v1/models

Dynamic model discovery for settings

Active

POST /v1/audio/transcriptions

Audio file ingestion

Planned

POST /v1/score

Evaluation scoring

Planned