“

Series: S13 AI Engineering · Article 7 | Level: Advanced | Reading Time: 18 min

Enterprise Prompt Engineering Practices

#TL;DR

Enterprise prompt engineering is far more complex than chat scenarios. This article covers prompt template management, version control, A/B testing, output format constraints (JSON Schema enforcement), token cost optimization, multilingual prompt strategies, and governing prompts as platform resources in coomia-dip.

#1. Background & Challenges

#1.1 Enterprise AI Realities

In enterprise AI engineering practice, technology selection is just the starting point. The real challenge is seamlessly integrating AI capabilities into existing data platforms and business processes. As an ontology-driven intelligent decision platform, coomia-dip has a natural architectural advantage for fusing structured knowledge (Ontology) with unstructured AI capabilities (LLM/vector retrieval).

The core problem this article addresses: Enterprise prompt engineering is far more complex than chat scenarios.

#1.2 Industry Landscape

Enterprise AI deployment faces common challenges:

Fragmentation: AI capabilities scattered across systems and teams, lacking unified governance
Poor observability: AI decision processes are opaque, difficult to troubleshoot
Cost escalation: LLM API costs scale linearly, lacking optimization mechanisms
Security compliance: Enterprise data flows through AI systems with blurred security boundaries

coomia-dip's AI engineering framework aims to systematically address these issues.

#2. Architecture Design

#2.1 Overall Architecture

In coomia-dip's Layered architecture, AI capabilities are primarily distributed across Reasoning & Decision Layer (Reasoning & Decision) and Agent Runtime Layer (Agent Runtime):

Python

class AIArchitecture:
    """coomia-dip AI Engineering Architecture"""

    # Reasoning & Decision Layer: Reasoning & Decision
    reasoning_engine: ReasoningEngine
    vector_store: VectorStore
    embedding_pipeline: EmbeddingPipeline
    llm_router: LLMRouter

    # Agent Runtime Layer: Agent Runtime
    agent_runtime: AgentRuntime
    tool_registry: ToolRegistry
    workflow_engine: TemporalClient
    memory_store: MemoryStore

#2.2 Core Data Flow

The request processing pipeline flows from user input through intent recognition, routing decisions, parallel execution across OQL queries, LLM reasoning, and workflow triggers, followed by result fusion, output formatting, safety filtering, and user response delivery.

#3. Core Implementation

#3.1 Configuration

Python

from ontology_sdk import OntoPlatform
from ontology_sdk.ai import AIConfig

platform = OntoPlatform(base_url="http://localhost:8080", token="admin-token")

ai_config = AIConfig(
    embedding_model="text-embedding-3-small",
    embedding_dimension=768,
    llm_models={
        "default": {"provider": "openai", "model": "gpt-4o", "temperature": 0.1},
        "fast": {"provider": "openai", "model": "gpt-4o-mini", "temperature": 0.0},
        "local": {"provider": "ollama", "model": "llama3:8b", "endpoint": "http://ollama:11434"},
    },
    vector_store={"type": "doris", "index_type": "hnsw", "metric": "cosine"},
    safety={"pii_detection": True, "content_filter": True},
)
platform.ai.configure(ai_config)

#3.2 Core Features

Python

from ontology_sdk.ai import EmbeddingPipeline, SemanticSearch, LLMChain

# Vector pipeline
pipeline = EmbeddingPipeline(model=ai_config.embedding_model, batch_size=100)
documents = platform.objects.list("Document", limit=1000)
for doc in documents:
    chunks = pipeline.chunk(doc.content, chunk_size=512, overlap=50)
    embeddings = pipeline.embed(chunks)
    for i, (chunk, emb) in enumerate(zip(chunks, embeddings)):
        platform.objects.create("KnowledgeChunk", {
            "documentId": doc.documentId, "chunkIndex": i,
            "content": chunk, "embedding": emb,
        })

# Semantic search
search = SemanticSearch(platform)
results = search.query("How to configure multi-tenancy?", object_type="KnowledgeChunk", top_k=5)

# RAG chain
rag = LLMChain(
    model="default",
    system_prompt="You are a technical assistant for coomia-dip. Answer based on provided context.",
    context_builder=lambda q: search.query(q, top_k=5),
)
response = rag.invoke("How to configure multi-tenancy?")

#4. Advanced Scenarios

#4.1 Natural Language Data Exploration

Python

async def nl_query(question: str) -> dict:
    oql = await platform.ai.text_to_oql(question=question,
        context_object_types=["Order", "Customer", "Product"])
    result = platform.oql.execute(oql)
    summary = await platform.ai.summarize(data=result.rows, question=question)
    return {"oql": oql, "data": result.rows, "summary": summary}

#4.2 AI-Powered Anomaly Detection

Python

async def detect_anomalies(metric: str, hours: int = 24):
    data = platform.metrics.query(metric=metric, duration=f"{hours}h", granularity="5m")
    analysis = await platform.ai.analyze_timeseries(data=data.values, sensitivity="medium")
    for a in analysis.anomalies:
        print(f"Anomaly at {a.timestamp}: {a.description} ({a.severity})")

#4.3 Knowledge Graph Enrichment

Python

async def extract_knowledge(document: str):
    extraction = await platform.ai.extract_entities(
        text=document,
        entity_types=["Person", "Organization", "Technology"],
        relation_types=["worksAt", "uses", "relatedTo"],
    )
    for entity in extraction.entities:
        platform.objects.create(entity.type, entity.properties)
    for rel in extraction.relations:
        platform.links.create(rel.source_rid, rel.relation_type, rel.target_rid)

#5. Performance Optimization

Area	Technique	Expected Impact
Vector retrieval latency	HNSW tuning + warmup	P99 < 50ms
LLM cost	Model routing + caching	40-60% cost reduction
Embedding throughput	Batch processing + GPU	10x improvement
End-to-end latency	Streaming + parallel retrieval	First token < 500ms

#Semantic Caching

Python

class SemanticCache:
    def __init__(self, threshold: float = 0.95):
        self.threshold = threshold

    async def get(self, query: str) -> str | None:
        embedding = await embed(query)
        similar = await vector_search(embedding, top_k=1)
        if similar and similar[0].score >= self.threshold:
            return similar[0].cached_response
        return None

#6. Monitoring & Observability

Python

metrics = platform.ai.get_metrics(duration="1h")
print(f"LLM calls: {metrics.llm.total_calls}, cost: ${metrics.llm.estimated_cost:.2f}")
print(f"Vector queries: {metrics.vector.total_queries}, cache hit: {metrics.vector.cache_hit_rate:.1%}")
print(f"Safety: PII={metrics.safety.pii_detections}, blocks={metrics.safety.filter_blocks}")

#7. Best Practices

AI augments, not replaces: AI enhances Ontology operations; critical decisions need human confirmation
Observability first: Every AI call should be logged and traceable
Progressive adoption: Start with low-risk scenarios (search), expand to high-risk (decisions)
Cost awareness: Monitor token consumption, set budget caps per AI feature
Safety by default: All AI outputs must pass safety filters before reaching end systems

#Common Pitfalls

Pitfall	Consequence	Solution
Always using largest model	Cost explosion	Route by task complexity
No prompt versioning	Irreproducible issues	Version control prompts
Unlimited output length	Token waste	Set max_tokens + format constraints
No PII filtering	Data leak risk	Bidirectional PII detection

#Summary

This article explored enterprise prompt engineering practices in coomia-dip, covering architecture design, core implementation, advanced scenarios, performance optimization, monitoring, and lessons learned. AI engineering is not about stacking individual technologies but requires platform-level systematic design -- coomia-dip provides exactly this infrastructure.

Next: [S13-08] Previous: [S13-06]