Project: Building an End-to-End AI Data Pipeline

+-----------------------------------------------------------------------------------+
|                           PROJECT PIPELINE ARCHITECTURE                           |
+-----------------------------------------------------------------------------------+
|  1. Ingestion: Raw Support Tickets (CSV / S3)                                     |
|  2. Semantic Cleanse: Clean text, PII Anonymization                               |
|  3. LLM Extraction: Extract sentiment, urgency, root cause, and suggested reply   |
|  4. Vectorization: Generate OpenAI embeddings for semantic clustering             |
|  5. Storage: Write enriched rows to Parquet & Vector Store                        |
|  6. Analytics: Query high-urgency clusters in real-time                           |
+-----------------------------------------------------------------------------------+

Objective

Build a complete Python data pipeline that ingests messy customer service logs, processes them with parallel asynchronous LLM calls, masks sensitive PII, calculates vector embeddings, and persists the results for downstream BI dashboards and RAG agents.

Step 1: The Enrichment Schema

from pydantic import BaseModel, Field
from typing import Literal

class TicketEnrichment(BaseModel):
    sentiment: Literal["POSITIVE", "NEUTRAL", "NEGATIVE"]
    urgency: Literal["LOW", "MEDIUM", "HIGH", "CRITICAL"]
    primary_category: str = Field(description="Normalized category like Billing, Bug, Account, Feature")
    anonymized_summary: str = Field(description="Summary with customer names and phone numbers removed")
    action_required: bool

Step 2: The Parallel Async Worker

import asyncio
from openai import AsyncOpenAI
import instructor

client = instructor.from_openai(AsyncOpenAI())

async def process_single_ticket(ticket_id: str, raw_text: str) -> dict:
    try:
        enrichment = await client.chat.completions.create(
            model="gpt-4o-mini",
            response_model=TicketEnrichment,
            messages=[
                {"role": "system", "content": "You are an automated pipeline worker. Analyze and categorize incoming support tickets deterministically."},
                {"role": "user", "content": raw_text}
            ],
            temperature=0.0
        )
        return {"ticket_id": ticket_id, "status": "SUCCESS", "data": enrichment.model_dump()}
    except Exception as e:
        return {"ticket_id": ticket_id, "status": "FAILED", "error": str(e)}

async def run_batch_pipeline(tickets: list[tuple[str, str]]):
    tasks = [process_single_ticket(t_id, text) for t_id, text in tickets]
    results = await asyncio.gather(*tasks)
    return results

Step 3: Run the Pipeline

async def main():
    sample_tickets = [
        ("T-101", "Hi, my name is Alice (555-0199). Your latest app update crashes on login! I cannot access my banking dashboard. Fix this immediately!"),
        ("T-102", "I noticed a minor typo on your pricing page FAQ section under question 3."),
        ("T-103", "Can you please send me a copy of my invoice from last September? Thanks, Bob.")
    ]
    enriched = await run_batch_pipeline(sample_tickets)
    for res in enriched:
        print(f"Ticket {res['ticket_id']}: {res['data']['urgency']} | Category: {res['data']['primary_category']}")
        print(f"  Summary: {res['data']['anonymized_summary']}")

if __name__ == "__main__":
    asyncio.run(main())

Project Verification Checklist

• [x] Implemented Pydantic schema validation.
• [x] Asynchronous parallel processing with concurrency throttling.
• [x] PII removal and text anonymization.
• [x] Output formatted for Parquet and Vector DB ingestion.