NLP in Production: Text, Retrieval, and Bilingual Systems

from langchain.text_splitter import RecursiveCharacterTextSplitter

# Wrong: character-based chunking
bad_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=100
)

# Better: semantic boundaries with metadata preservation
from llama_index.core.node_parser import SemanticSplitterNodeParser
from llama_index.embeddings.openai import OpenAIEmbedding

def chunk_document(document: str, metadata: dict) -> list[dict]:
    """
    Chunk at semantic boundaries, not character limits.
    Preserve metadata for filtering and citation.
    """
    splitter = SemanticSplitterNodeParser(
        buffer_size=1,
        breakpoint_percentile_threshold=95,
        embed_model=OpenAIEmbedding()
    )

    nodes = splitter.get_nodes_from_documents([document])

    chunks = []
    for node in nodes:
        chunks.append({
            "text": node.text,
            "metadata": {
                **metadata,
                "chunk_index": node.metadata.get("chunk_index"),
                "section": extract_section_header(node.text)
            }
        })

    return chunks

from openai import OpenAI
import psycopg2
import numpy as np

client = OpenAI()

def embed_and_store(chunks: list[dict], collection: str):
    """Batch embed and store with metadata."""
    texts = [chunk["text"] for chunk in chunks]

    # Batch embedding (more efficient than one at a time)
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=texts
    )

    embeddings = [item.embedding for item in response.data]

    # Store in pgvector
    with psycopg2.connect(DATABASE_URL) as conn:
        with conn.cursor() as cur:
            for chunk, embedding in zip(chunks, embeddings):
                cur.execute("""
                    INSERT INTO embeddings (collection, text, embedding, metadata)
                    VALUES (%s, %s, %s, %s)
                """, (
                    collection,
                    chunk["text"],
                    np.array(embedding),
                    json.dumps(chunk["metadata"])
                ))

import cohere
from openai import OpenAI

co = cohere.Client(COHERE_API_KEY)
openai_client = OpenAI()

def retrieve_with_rerank(
    query: str,
    collection: str,
    top_k: int = 10,
    top_n: int = 3
) -> list[dict]:
    """
    Two-stage retrieval: broad embedding search → precise reranking.
    """
    # Stage 1: Embedding-based retrieval (top_k candidates)
    query_embedding = openai_client.embeddings.create(
        model="text-embedding-3-small",
        input=query
    ).data[0].embedding

    candidates = vector_search(query_embedding, collection, k=top_k)

    # Stage 2: Rerank for semantic precision
    rerank_result = co.rerank(
        query=query,
        documents=[c["text"] for c in candidates],
        top_n=top_n,
        model="rerank-english-v3.0"
    )

    reranked = [
        {
            **candidates[r.index],
            "relevance_score": r.relevance_score
        }
        for r in rerank_result.results
    ]

    return reranked

def generate_rag_response(query: str, context_chunks: list[dict]) -> str:
    context = "\n\n---\n\n".join([
        f"Source: {c['metadata'].get('section', 'Unknown')}\n{c['text']}"
        for c in context_chunks
    ])

    prompt = f"""Answer the user's question using ONLY the information in the provided context.

If the context doesn't contain enough information to answer confidently, say so explicitly.
Do not extrapolate or add information not present in the context.

Context:
{context}

Question: {query}

Answer:"""

    # ... LLM call

Query embedding: "surprised projects over budget"
Top-10 candidates include: Pre-Mortem, Planning Fallacy, Sunk Cost,
  Inversion, Optimism Bias, Scope Creep, Resource Management...

After Cohere Rerank:
  1. Pre-Mortem (relevance: 0.89)
  2. Planning Fallacy (relevance: 0.82)
  3. Second-Order Thinking (relevance: 0.74)

import re

# Telugu-specific normalization for mixed content
NORMALIZATIONS = [
    # Standardize common code-switched words
    (r'\bcomputer\b', 'కంప్యూటర్'),
    (r'\bsoftware\b', 'సాఫ్ట్‌వేర్'),
    # Number normalization
    (r'(\d+)\s*%', r'\1 శాతం'),
    # Remove excessive punctuation from OCR errors
    (r'[।।]{2,}', '।'),
]

def normalize_telugu(text: str) -> str:
    """Normalize Telugu text for better model performance."""
    for pattern, replacement in NORMALIZATIONS:
        text = re.sub(pattern, replacement, text, flags=re.IGNORECASE)
    return text.strip()

def chunk_bilingual_document(text: str) -> list[str]:
    """
    Chunk Telugu+English mixed content at natural language boundaries.
    Telugu sentences end with ।  (Devanagari danda) or \n
    """
    # Split on Telugu sentence boundaries
    sentences = re.split(r'[।\n]+', text)

    # Regroup into chunks of 3-5 sentences for better context
    chunks = []
    current_chunk = []

    for sentence in sentences:
        current_chunk.append(sentence.strip())
        if len(current_chunk) >= 4:
            chunks.append(' '.join(current_chunk))
            current_chunk = []

    if current_chunk:
        chunks.append(' '.join(current_chunk))

    return [c for c in chunks if len(c) > 20]  # Filter tiny fragments

def automated_eval(outputs: list[dict], test_cases: list[dict]) -> dict:
    results = {}

    for test, output in zip(test_cases, outputs):
        # Check if key facts are present (for RAG)
        results["fact_coverage"] = check_facts_covered(
            output["text"], test["required_facts"]
        )

        # Check if hallucinations are absent
        results["hallucination_rate"] = check_hallucination(
            output["text"], output["context"]
        )

        # Check format compliance
        results["format_ok"] = validate_output_format(output["text"])

    return results

def llm_judge(query: str, response: str, context: str) -> dict:
    judgment = claude.messages.create(
        model="claude-sonnet-4-6",
        messages=[{
            "role": "user",
            "content": f"""Rate this RAG response on a scale of 1-5 for each criterion:

Query: {query}
Context provided: {context}
Response: {response}

Rate:
1. Faithfulness (1-5): Is the response grounded in the context?
2. Relevance (1-5): Does it answer the query?
3. Completeness (1-5): Are all key points covered?

Return JSON with keys: faithfulness (N), relevance (N), completeness (N), reasoning (string)"""
        }]
    )
    return json.loads(judgment.content[0].text)