Implementation Guide: Generate product specifications, quality control procedures, and supplier rfqs

az login
az account set --subscription "<CLIENT_SUBSCRIPTION_ID>"
az group create --name rg-mfg-ai-content --location eastus2
az cognitiveservices account create --name oai-mfg-content-prod --resource-group rg-mfg-ai-content --kind OpenAI --sku S0 --location eastus2 --yes
az cognitiveservices account deployment create --name oai-mfg-content-prod --resource-group rg-mfg-ai-content --deployment-name gpt-5.4-mfg --model-name gpt-5.4 --model-version 2024-11-20 --model-format OpenAI --sku-capacity 30 --sku-name GlobalStandard
az cognitiveservices account deployment create --name oai-mfg-content-prod --resource-group rg-mfg-ai-content --deployment-name gpt-5.4-mini-mfg --model-name gpt-5.4-mini --model-version 2024-07-18 --model-format OpenAI --sku-capacity 60 --sku-name GlobalStandard
az cognitiveservices account deployment create --name oai-mfg-content-prod --resource-group rg-mfg-ai-content --deployment-name text-embedding-3-large --model-name text-embedding-3-large --model-version 1 --model-format OpenAI --sku-capacity 50 --sku-name Standard
az cognitiveservices account keys list --name oai-mfg-content-prod --resource-group rg-mfg-ai-content

az keyvault create --name kv-mfg-ai-content --resource-group rg-mfg-ai-content --location eastus2 --sku standard
AOAI_KEY=$(az cognitiveservices account keys list --name oai-mfg-content-prod --resource-group rg-mfg-ai-content --query key1 -o tsv)
az keyvault secret set --vault-name kv-mfg-ai-content --name azure-openai-key --value "$AOAI_KEY"
az keyvault secret set --vault-name kv-mfg-ai-content --name azure-openai-endpoint --value "https://oai-mfg-content-prod.openai.azure.com/"
az keyvault secret set --vault-name kv-mfg-ai-content --name pinecone-api-key --value "<PINECONE_API_KEY>"

pip install pinecone-client
python3 -c "
import pinecone
from pinecone import Pinecone, ServerlessSpec

pc = Pinecone(api_key='<PINECONE_API_KEY>')

pc.create_index(
    name='mfg-knowledge-base',
    dimension=3072,  # text-embedding-3-large dimension
    metric='cosine',
    spec=ServerlessSpec(
        cloud='aws',
        region='us-east-1'
    )
)
print('Index created successfully')
print(pc.describe_index('mfg-knowledge-base'))
"

mkdir mfg-ai-content-gen && cd mfg-ai-content-gen
python3 -m venv .venv && source .venv/bin/activate
pip install langchain langchain-openai langchain-community pinecone-client python-docx openpyxl fastapi uvicorn pydantic azure-identity azure-keyvault-secrets tiktoken unstructured pdf2image pytesseract
pip freeze > requirements.txt
mkdir -p src/{ingestion,generation,templates,api} tests/ docs/ prompts/
touch src/__init__.py src/ingestion/__init__.py src/generation/__init__.py src/templates/__init__.py src/api/__init__.py
git init && echo '.venv/\n__pycache__/\n.env\n*.pyc' > .gitignore && git add -A && git commit -m 'Initial project structure'

cat > prompts/product_specification.json << 'JSONEOF'
{
  "template_id": "PROD_SPEC_V1",
  "template_version": "1.0",
  "document_type": "product_specification",
  "model": "gpt-5.4",
  "max_tokens": 4000,
  "temperature": 0.2,
  "system_prompt": "You are a senior manufacturing engineer creating formal product specifications for {{company_name}}. You write precise, unambiguous technical documentation that conforms to the company's document control standards. All specifications must be complete, measurable, and testable. Use SI units with imperial equivalents in parentheses where applicable. Reference industry standards (ASTM, ISO, MIL-STD, SAE) when relevant. Never fabricate test results or certifications.",
  "user_prompt_template": "Generate a complete Product Specification document with the following structure:\n\n## Document Header\n- Document Number: {{doc_number}}\n- Revision: {{revision}}\n- Date: {{date}}\n- Product Name: {{product_name}}\n- Part Number: {{part_number}}\n\n## Sections Required:\n1. **Scope and Purpose** — Define what this specification covers\n2. **Applicable Documents and Standards** — List referenced standards and specs\n3. **Requirements**\n   3.1 Material Requirements (reference BOM data below)\n   3.2 Dimensional Requirements (include tolerances)\n   3.3 Performance Requirements\n   3.4 Environmental Requirements (operating temp, humidity, etc.)\n   3.5 Reliability Requirements\n4. **Quality Assurance Provisions**\n   4.1 Inspection Requirements\n   4.2 Test Methods\n   4.3 Acceptance Criteria\n5. **Packaging and Marking**\n6. **Revision History**\n\n## Input Data:\n- Product Description: {{product_description}}\n- BOM/Material Data: {{bom_data}}\n- Key Performance Parameters: {{performance_params}}\n- Target Application: {{application}}\n- Regulatory Requirements: {{regulatory_reqs}}\n\n## Reference Context (from existing company specifications):\n{{rag_context}}\n\nGenerate the complete specification following the section structure above. Be specific with numerical values, tolerances, and test methods. Where exact values are not provided in the input data, provide realistic placeholder values clearly marked with [TBD - VERIFY] tags."
}
JSONEOF
cat > prompts/qc_procedure.json << 'JSONEOF'
{
  "template_id": "QC_PROC_V1",
  "template_version": "1.0",
  "document_type": "qc_procedure",
  "model": "gpt-5.4",
  "max_tokens": 5000,
  "temperature": 0.15,
  "system_prompt": "You are a Quality Assurance Manager creating formal quality control procedures and Standard Operating Procedures (SOPs) for {{company_name}}, a manufacturing facility certified to {{certifications}}. All procedures must follow the company's QMS document format, include clear step-by-step instructions that a trained operator can follow, specify required equipment and calibration requirements, define acceptance criteria with measurable values, and include proper safety warnings. Reference applicable standards and regulations. Never skip safety-critical steps.",
  "user_prompt_template": "Generate a complete Quality Control Procedure / SOP with the following structure:\n\n## Document Header\n- Document Number: {{doc_number}}\n- Revision: {{revision}}\n- Effective Date: {{date}}\n- Procedure Title: {{procedure_title}}\n- Department: {{department}}\n- Process Area: {{process_area}}\n\n## Sections Required:\n1. **Purpose** — Why this procedure exists\n2. **Scope** — What processes/products it covers\n3. **References** — Standards, regulations, related SOPs\n4. **Definitions** — Technical terms used\n5. **Responsibilities** — Who does what (roles, not names)\n6. **Equipment and Materials Required** — Include calibration requirements\n7. **Safety Requirements** — PPE, hazards, emergency procedures\n8. **Procedure Steps** — Numbered, detailed, with decision points and acceptance criteria\n   - Each step: Action, parameters, acceptance criteria, what to do if out-of-spec\n9. **Recording and Documentation** — What forms to fill, what data to record\n10. **Non-Conformance Handling** — Escalation path for failures\n11. **Revision History**\n\n## Input Data:\n- Process Description: {{process_description}}\n- Product/Part Affected: {{product_info}}\n- Critical Quality Parameters: {{quality_params}}\n- Equipment Available: {{equipment_list}}\n- Applicable Standards: {{standards}}\n- Known Failure Modes: {{failure_modes}}\n\n## Reference Context (from existing company procedures):\n{{rag_context}}\n\nGenerate the complete procedure. Every step must be actionable and measurable. Mark any values that need client verification with [TBD - VERIFY]."
}
JSONEOF
cat > prompts/supplier_rfq.json << 'JSONEOF'
{
  "template_id": "SUPPLIER_RFQ_V1",
  "template_version": "1.0",
  "document_type": "supplier_rfq",
  "model": "gpt-5.4-mini",
  "max_tokens": 3000,
  "temperature": 0.25,
  "system_prompt": "You are a Procurement Specialist at {{company_name}} creating formal Requests for Quotation to send to suppliers. RFQs must be professional, comprehensive, and unambiguous. Include all information a supplier needs to provide an accurate quote: exact specifications, quantities, delivery requirements, quality requirements, and commercial terms. Reference specific part numbers, material specs, and standards. The tone should be professional and direct.",
  "user_prompt_template": "Generate a complete Supplier Request for Quotation (RFQ) with the following structure:\n\n## RFQ Header\n- RFQ Number: {{rfq_number}}\n- Date Issued: {{date}}\n- Response Deadline: {{deadline}}\n- Buyer Contact: {{buyer_name}}, {{buyer_email}}, {{buyer_phone}}\n- Company: {{company_name}}\n\n## Sections Required:\n1. **Introduction** — Brief company intro and purpose of RFQ\n2. **Items Requested** — Table format with:\n   - Line Item #, Part Number, Description, Specification Reference, Material, Quantity, Unit of Measure\n3. **Technical Requirements**\n   - Material specifications and certifications required\n   - Dimensional tolerances\n   - Surface finish / treatment requirements\n   - Testing and inspection requirements\n   - Applicable standards\n4. **Quality Requirements**\n   - QMS certification required (ISO 9001, AS9100, etc.)\n   - Certificate of Conformance requirements\n   - First Article Inspection requirements\n   - Material traceability requirements\n5. **Commercial Terms**\n   - Requested delivery date / lead time\n   - Shipping terms (FOB point)\n   - Payment terms\n   - Pricing validity period\n   - Minimum order quantities\n6. **Quotation Response Format** — What info supplier must include\n7. **Terms and Conditions** — Standard T&Cs reference\n\n## Input Data:\n- Items to Quote: {{items_data}}\n- Quantities Needed: {{quantities}}\n- Required Delivery Date: {{delivery_date}}\n- Special Requirements: {{special_requirements}}\n- Approved Suppliers (if sending to specific): {{supplier_info}}\n\n## Reference Context (from existing company RFQs and specs):\n{{rag_context}}\n\nGenerate the complete RFQ document. Use the company's standard formatting where available from the reference context."
}
JSONEOF
git add prompts/ && git commit -m 'Add manufacturing prompt templates v1.0'

# HTTP POST body for Power Automate HTTP action. Method: POST, URI:
# https://ca-mfg-ai-api.<REGION>.azurecontainerapps.io/api/v1/generate,
# Headers: { 'Content-Type': 'application/json', 'X-Api-Key':
# '<MFG_API_KEY>' }. Triggered when a new item is created in a SharePoint
# 'Spec Requests' list. Save full flow definition as product-spec-flow.json
# and import via Power Automate UI.

{
  "document_type": "product_specification",
  "variables": {
    "company_name": "<CLIENT_NAME>",
    "doc_number": "triggerOutputs()?['body/DocNumber']",
    "revision": "A",
    "date": "utcNow('yyyy-MM-dd')",
    "product_name": "triggerOutputs()?['body/ProductName']",
    "part_number": "triggerOutputs()?['body/PartNumber']",
    "product_description": "triggerOutputs()?['body/Description']",
    "bom_data": "triggerOutputs()?['body/BOMData']",
    "performance_params": "triggerOutputs()?['body/PerformanceParams']",
    "application": "triggerOutputs()?['body/Application']",
    "regulatory_reqs": "triggerOutputs()?['body/RegulatoryReqs']"
  },
  "requested_by": "triggerOutputs()?['body/Author/Email']"
}

cat > src/integration/erp_connector.py << 'PYEOF'
import requests
from typing import Dict, Any, Optional
from abc import ABC, abstractmethod


class ERPConnector(ABC):
    @abstractmethod
    def get_part_info(self, part_number: str) -> Dict[str, Any]:
        pass

    @abstractmethod
    def get_bom(self, part_number: str) -> Dict[str, Any]:
        pass

    @abstractmethod
    def get_approved_suppliers(self, part_number: str) -> list:
        pass


class EpicorKineticConnector(ERPConnector):
    """Connector for Epicor Kinetic ERP REST API v2."""

    def __init__(self, base_url: str, api_key: str, company: str):
        self.base_url = base_url.rstrip('/')
        self.headers = {
            'Authorization': f'Bearer {api_key}',
            'X-API-Key': api_key,
            'Content-Type': 'application/json',
            'Company': company
        }

    def get_part_info(self, part_number: str) -> Dict[str, Any]:
        url = f'{self.base_url}/api/v2/odata/Erp.BO.PartSvc/Parts'
        params = {'$filter': f"PartNum eq '{part_number}'",
                  '$select': 'PartNum,PartDescription,ClassID,TypeCode,UOMClassID,NetWeight,NetWeightUOM'}
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        data = resp.json()
        return data.get('value', [{}])[0] if data.get('value') else {}

    def get_bom(self, part_number: str) -> Dict[str, Any]:
        url = f'{self.base_url}/api/v2/odata/Erp.BO.BillOfMaterialSvc/BillOfMaterials'
        params = {'$filter': f"PartNum eq '{part_number}'",
                  '$expand': 'BOMDetails'}
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        return resp.json()

    def get_approved_suppliers(self, part_number: str) -> list:
        url = f'{self.base_url}/api/v2/odata/Erp.BO.AprvVendSvc/AprvVends'
        params = {'$filter': f"PartNum eq '{part_number}'"}
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        return resp.json().get('value', [])


class AcumaticaConnector(ERPConnector):
    """Connector for Acumatica Cloud ERP REST API."""

    def __init__(self, base_url: str, username: str, password: str, company: str, branch: str):
        self.base_url = base_url.rstrip('/')
        self.session = requests.Session()
        # Authenticate
        login_url = f'{self.base_url}/entity/auth/login'
        self.session.post(login_url, json={
            'name': username, 'password': password,
            'company': company, 'branch': branch
        })

    def get_part_info(self, part_number: str) -> Dict[str, Any]:
        url = f'{self.base_url}/entity/Default/24.200.001/StockItem/{part_number}'
        resp = self.session.get(url)
        resp.raise_for_status()
        return resp.json()

    def get_bom(self, part_number: str) -> Dict[str, Any]:
        url = f'{self.base_url}/entity/Manufacturing/24.200.001/BillOfMaterial'
        params = {'$filter': f"InventoryID eq '{part_number}'"}
        resp = self.session.get(url, params=params)
        resp.raise_for_status()
        return resp.json()

    def get_approved_suppliers(self, part_number: str) -> list:
        url = f'{self.base_url}/entity/Default/24.200.001/StockItem/{part_number}'
        params = {'$expand': 'VendorDetails'}
        resp = self.session.get(url, params=params)
        resp.raise_for_status()
        data = resp.json()
        return data.get('VendorDetails', [])


class DynamicsBCConnector(ERPConnector):
    """Connector for Dynamics 365 Business Central API."""

    def __init__(self, tenant_id: str, environment: str, company_id: str, access_token: str):
        self.base_url = f'https://api.businesscentral.dynamics.com/v2.0/{tenant_id}/{environment}/api/v2.0/companies({company_id})'
        self.headers = {
            'Authorization': f'Bearer {access_token}',
            'Content-Type': 'application/json'
        }

    def get_part_info(self, part_number: str) -> Dict[str, Any]:
        url = f'{self.base_url}/items'
        params = {'$filter': f"number eq '{part_number}'"}
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        data = resp.json()
        return data.get('value', [{}])[0] if data.get('value') else {}

    def get_bom(self, part_number: str) -> Dict[str, Any]:
        url = f'{self.base_url}/productionBOMs'
        params = {'$filter': f"number eq '{part_number}'"}
        resp = requests.get(url, headers=self.headers, params=params)
        resp.raise_for_status()
        return resp.json()

    def get_approved_suppliers(self, part_number: str) -> list:
        url = f'{self.base_url}/vendors'
        resp = requests.get(url, headers=self.headers)
        resp.raise_for_status()
        return resp.json().get('value', [])
PYEOF

cd mfg-ai-content-gen && source .venv/bin/activate

cat > .env << 'ENVEOF'
AZURE_OPENAI_ENDPOINT=https://oai-mfg-content-prod.openai.azure.com/
AZURE_OPENAI_KEY=<YOUR_KEY>
PINECONE_API_KEY=<YOUR_KEY>
MFG_API_KEY=<YOUR_KEY>
ENVEOF

python3 -c "
import os
from dotenv import load_dotenv
load_dotenv()
from src.ingestion.document_processor import ManufacturingDocumentProcessor

processor = ManufacturingDocumentProcessor(
    azure_endpoint=os.getenv('AZURE_OPENAI_ENDPOINT'),
    azure_api_key=os.getenv('AZURE_OPENAI_KEY'),
    pinecone_api_key=os.getenv('PINECONE_API_KEY')
)

# Ingest each document category
for category in ['specs', 'qc-procedures', 'rfqs', 'standards', 'materials']:
    path = f'./staging/{category}'
    if os.path.exists(path):
        print(f'Processing {category}...')
        stats = processor.ingest_directory(path, namespace='default')
        print(f'  Processed: {stats["processed"]} files, {stats["chunks"]} chunks')
        if stats['errors']:
            print(f'  Errors: {len(stats["errors"])}')
            for err in stats['errors']:
                print(f'    - {err["file"]}: {err["error"]}')
"

az monitor diagnostic-settings create --resource /subscriptions/<SUB_ID>/resourceGroups/rg-mfg-ai-content/providers/Microsoft.App/containerApps/ca-mfg-ai-api --name mfg-ai-diagnostics --workspace /subscriptions/<SUB_ID>/resourceGroups/rg-mfg-ai-content/providers/Microsoft.OperationalInsights/workspaces/<WORKSPACE_NAME> --logs '[{"category":"ContainerAppConsoleLogs","enabled":true},{"category":"ContainerAppSystemLogs","enabled":true}]'

az monitor metrics alert create --name 'AI-API-Error-Rate' --resource-group rg-mfg-ai-content --scopes /subscriptions/<SUB_ID>/resourceGroups/rg-mfg-ai-content/providers/Microsoft.App/containerApps/ca-mfg-ai-api --condition 'count requests where resultCode >= 500 > 5' --window-size 15m --evaluation-frequency 5m --action /subscriptions/<SUB_ID>/resourceGroups/rg-mfg-ai-content/providers/Microsoft.Insights/actionGroups/<ACTION_GROUP>

az consumption budget create --budget-name 'AI-Content-Monthly' --resource-group rg-mfg-ai-content --amount 500 --time-grain Monthly --start-date 2025-01-01 --end-date 2026-12-31 --notification-key 80pct --notification-enabled true --notification-operator GreaterThanOrEqualTo --notification-threshold 80 --notification-contact-emails msp-alerts@yourmsp.com

# Manufacturing RAG Context Retriever Skill
# File: src/generation/rag_retriever.py

from typing import List, Dict, Any, Optional, Tuple
from langchain_openai import AzureOpenAIEmbeddings
from pinecone import Pinecone
import logging

logger = logging.getLogger('mfg-rag')


class ManufacturingRAGRetriever:
    """
    Intelligent RAG retriever for manufacturing documents.
    Implements multi-strategy retrieval based on document type.
    """

    # Define which source document types are relevant for each generation type
    RETRIEVAL_STRATEGIES = {
        'product_specification': {
            'primary_types': ['product_specification'],
            'secondary_types': ['industry_standard', 'material_data'],
            'primary_top_k': 5,
            'secondary_top_k': 3,
            'min_relevance': 0.72
        },
        'qc_procedure': {
            'primary_types': ['qc_procedure'],
            'secondary_types': ['industry_standard', 'product_specification'],
            'primary_top_k': 5,
            'secondary_top_k': 3,
            'min_relevance': 0.70
        },
        'supplier_rfq': {
            'primary_types': ['supplier_rfq', 'product_specification'],
            'secondary_types': ['material_data'],
            'primary_top_k': 4,
            'secondary_top_k': 2,
            'min_relevance': 0.68
        }
    }

    def __init__(self, azure_endpoint: str, azure_api_key: str,
                 pinecone_api_key: str, index_name: str = 'mfg-knowledge-base'):
        self.embeddings = AzureOpenAIEmbeddings(
            azure_deployment='text-embedding-3-large',
            azure_endpoint=azure_endpoint,
            api_key=azure_api_key,
            api_version='2024-06-01'
        )
        self.pc = Pinecone(api_key=pinecone_api_key)
        self.index = self.pc.Index(index_name)

    def retrieve(self, query: str, target_doc_type: str,
                 namespace: str = 'default') -> Tuple[str, List[Dict[str, Any]]]:
        """
        Retrieve relevant context using multi-strategy approach.
        Returns (formatted_context_string, source_attribution_list)
        """
        strategy = self.RETRIEVAL_STRATEGIES.get(target_doc_type, {
            'primary_types': [target_doc_type],
            'secondary_types': ['general_manufacturing'],
            'primary_top_k': 5,
            'secondary_top_k': 2,
            'min_relevance': 0.70
        })

        query_embedding = self.embeddings.embed_query(query)
        all_chunks = []
        sources = []

        # Primary retrieval
        for doc_type in strategy['primary_types']:
            results = self.index.query(
                vector=query_embedding,
                top_k=strategy['primary_top_k'],
                namespace=namespace,
                include_metadata=True,
                filter={'doc_type': {'$eq': doc_type}}
            )
            for match in results.get('matches', []):
                if match.get('score', 0) >= strategy['min_relevance']:
                    meta = match.get('metadata', {})
                    all_chunks.append({
                        'text': meta.get('text', ''),
                        'source': meta.get('source_file', 'Unknown'),
                        'doc_type': meta.get('doc_type', 'unknown'),
                        'score': match.get('score', 0),
                        'priority': 'primary'
                    })

        # Secondary retrieval
        for doc_type in strategy['secondary_types']:
            results = self.index.query(
                vector=query_embedding,
                top_k=strategy['secondary_top_k'],
                namespace=namespace,
                include_metadata=True,
                filter={'doc_type': {'$eq': doc_type}}
            )
            for match in results.get('matches', []):
                if match.get('score', 0) >= strategy['min_relevance']:
                    meta = match.get('metadata', {})
                    all_chunks.append({
                        'text': meta.get('text', ''),
                        'source': meta.get('source_file', 'Unknown'),
                        'doc_type': meta.get('doc_type', 'unknown'),
                        'score': match.get('score', 0),
                        'priority': 'secondary'
                    })

        # Sort by priority then score
        all_chunks.sort(key=lambda x: (0 if x['priority'] == 'primary' else 1, -x['score']))

        # Deduplicate by source file
        seen_sources = set()
        unique_chunks = []
        for chunk in all_chunks:
            source_key = f"{chunk['source']}_{chunk['text'][:100]}"
            if source_key not in seen_sources:
                seen_sources.add(source_key)
                unique_chunks.append(chunk)

        # Build formatted context
        context_parts = []
        for chunk in unique_chunks:
            context_parts.append(
                f"[Source: {chunk['source']} | Type: {chunk['doc_type']} | "
                f"Relevance: {chunk['score']:.2f} | Priority: {chunk['priority']}]\n"
                f"{chunk['text']}"
            )
            sources.append({
                'source_file': chunk['source'],
                'doc_type': chunk['doc_type'],
                'relevance_score': round(chunk['score'], 3),
                'priority': chunk['priority']
            })

        context_string = '\n\n---\n\n'.join(context_parts) if context_parts else \
            'No matching reference documents found. Generate based on general manufacturing best practices.'

        logger.info(f'RAG retrieval for {target_doc_type}: {len(unique_chunks)} chunks from {len(set(c["source"] for c in unique_chunks))} sources')

        return context_string, sources