Prompt Libraries: Design, Discoverability, & Governance — Mục tiêu: Catalog prompts, metadata, search, deprecation policy

Bài viết này sẽ đi sâu vào kiến trúc và chiến lược quản lý thư viện prompt cho các hệ thống AI, tập trung vào cách thiết kế, tìm kiếm và duy trì prompt ở quy mô lớn.

Mở đầu: Vì sao cần Prompt Libraries?

Khi tôi bắt đầu làm việc với các hệ thống AI vài năm trước, prompt chỉ đơn giản là những câu lệnh text đặt trong code. Nhưng khi dự án mở rộng, chúng tôi gặp phải vấn đề: prompt bị phân tán khắp nơi – trong config files, database, thậm chí hardcoded trong source code.

Một ngày nọ, team phải maintain một chatbot với 47 prompt khác nhau, mỗi prompt ở một vị trí khác nhau. Khi cần thay đổi một prompt cho toàn hệ thống, chúng tôi mất cả tuần để tìm và update tất cả các vị trí.

Đó là lúc chúng tôi nhận ra cần một hệ thống quản lý prompt bài bản.

1. Kiến trúc Prompt Library

1.1 Các thành phần cơ bản

Một prompt library tốt cần có các thành phần sau:

Prompt Library Architecture
┌─────────────────────────────────────────────────┐
│           Prompt Catalog Service               │
├─────────────────────────────────────────────────┤
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│  │  Metadata   │ │  Versioning │ │  Search      │ │
│  │  Store      │ │  Engine     │ │  Engine      │ │
│  └─────────────┘ └─────────────┘ └─────────────┘ │
├─────────────────────────────────────────────────┤
│           Prompt Management API                │
├─────────────────────────────────────────────────┤
│  ┌─────────────┐ ┌─────────────┐ ┌─────────────┐ │
│  │  CRUD Ops   │ │  Validation │ │  Audit Log   │ │
│  │             │ │             │ │             │ │
│  └─────────────┘ └─────────────┘ └─────────────┘ │
└─────────────────────────────────────────────────┘

1.2 Data Model cho Prompt

# Prompt metadata schema
class Prompt:
    id: str                    # UUID, ví dụ: "prompt-2024-001"
    title: str                 # Tiêu đề dễ hiểu
    description: str           # Mô tả chi tiết
    version: str               # Semantic versioning, ví dụ: "1.2.0"
    tags: List[str]            # Tags phân loại, ví dụ: ["chat", "support", "vietnamese"]
    author: str                # Người tạo
    created_at: datetime
    updated_at: datetime
    content: str               # Nội dung prompt
    parameters: Dict[str, Any] # Các tham số cần thiết
    examples: List[str]         # Ví dụ sử dụng
    deprecation_date: Optional[datetime]  # Ngày deprecated
    replacement_id: Optional[str]         # Prompt thay thế

2. Discoverability: Làm sao để tìm prompt nhanh chóng?

2.1 Chiến lược tìm kiếm

Khi thư viện prompt phát triển, việc tìm kiếm trở thành vấn đề then chốt. Chúng tôi đã thử nghiệm nhiều phương pháp:

a) Full-text search

# Elasticsearch query cho prompt search
query = {
    "query": {
        "multi_match": {
            "query": search_term,
            "fields": ["title^3", "description", "content", "tags"],
            "fuzziness": "AUTO"
        }
    },
    "size": 10,
    "from": 0
}

b) Semantic search với embeddings

# Sử dụng sentence-transformers cho semantic search
from sentence_transformers import SentenceTransformer
import numpy as np

model = SentenceTransformer('all-MiniLM-L6-v2')

# Index prompts
prompt_embeddings = model.encode([p.content for p in prompts])

# Search
query_embedding = model.encode([query_text])
similarities = np.dot(prompt_embeddings, query_embedding.T)
top_indices = similarities.argsort()[-5:][::-1]

2.2 Tagging strategy

Chúng tôi áp dụng hệ thống tagging có cấu trúc:

# Ví dụ về tagging system
prompt_tags = {
    "domain": ["customer_service", "sales", "technical_support"],
    "language": ["english", "vietnamese", "multi_language"],
    "complexity": ["simple", "medium", "complex"],
    "use_case": ["qa", "summarization", "classification"],
    "provider": ["openai", "anthropic", "local_model"]
}

3. Governance: Quản lý vòng đời prompt

3.1 Versioning strategy

Chúng tôi áp dụng semantic versioning cho prompt:

# Versioning scheme
MAJOR.MINOR.PATCH
# MAJOR: Thay đổi lớn ảnh hưởng compatibility
# MINOR: Thêm tính năng mới, không break
# PATCH: Fix bug, optimize

Version comparison logic:

def compare_versions(v1, v2):
    v1_parts = list(map(int, v1.split('.')))
    v2_parts = list(map(int, v2.split('.')))

    for i in range(3):
        if v1_parts[i] > v2_parts[i]:
            return 1
        elif v1_parts[i] < v2_parts[i]:
            return -1
    return 0

3.2 Deprecation policy

Chúng tôi thiết lập chính sách deprecation rõ ràng:

# Deprecation workflow
class DeprecationWorkflow:
    def __init__(self, prompt_id, deprecation_date, replacement_id):
        self.prompt_id = prompt_id
        self.deprecation_date = deprecation_date
        self.replacement_id = replacement_id
        self.notification_sent = False

    def check_and_notify(self):
        if not self.notification_sent and datetime.now() > self.deprecation_date - timedelta(days=30):
            # Gửi notification cho các service đang sử dụng
            notify_services(self.prompt_id)
            self.notification_sent = True

    def is_active(self):
        return datetime.now() < self.deprecation_date

4. Performance & Scalability

4.1 Caching strategy

# Caching layer cho prompt lookup
from functools import lru_cache

@lru_cache(maxsize=1000)
def get_prompt(prompt_id: str) -> Prompt:
    # Lấy từ database hoặc external service
    return prompt_repository.get(prompt_id)

# TTL-based caching với Redis
import redis

redis_client = redis.Redis(host='localhost', port=6379, db=0)

def get_prompt_with_cache(prompt_id: str, ttl: int = 3600) -> Prompt:
    cache_key = f"prompt:{prompt_id}"
    cached = redis_client.get(cache_key)

    if cached:
        return Prompt.from_json(cached)

    prompt = prompt_repository.get(prompt_id)
    redis_client.setex(cache_key, ttl, prompt.to_json())
    return prompt

4.2 Database schema optimization

-- PostgreSQL schema cho prompt catalog
CREATE TABLE prompts (
    id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    title VARCHAR(255) NOT NULL,
    description TEXT,
    content TEXT NOT NULL,
    version VARCHAR(20) NOT NULL,
    tags JSONB,
    author VARCHAR(100),
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    deprecation_date TIMESTAMP WITH TIME ZONE,
    replacement_id UUID REFERENCES prompts(id)
);

-- Index cho search performance
CREATE INDEX idx_prompts_search ON prompts USING GIN (to_tsvector('english', title || ' ' || description));
CREATE INDEX idx_prompts_tags ON prompts USING GIN (tags);
CREATE INDEX idx_prompts_version ON prompts (version);

5. Security & Access Control

5.1 Role-based access

# RBAC cho prompt management
class PromptPermission(Enum):
    VIEW = "view"
    CREATE = "create"
    UPDATE = "update"
    DELETE = "delete"
    ADMIN = "admin"

class PromptAccessControl:
    def __init__(self, user_role):
        self.permissions = self._get_permissions_for_role(user_role)

    def _get_permissions_for_role(self, role):
        permissions_map = {
            "viewer": [PromptPermission.VIEW],
            "developer": [PromptPermission.VIEW, PromptPermission.CREATE, PromptPermission.UPDATE],
            "admin": [PromptPermission.VIEW, PromptPermission.CREATE, PromptPermission.UPDATE, PromptPermission.DELETE],
            "super_admin": [PromptPermission.ADMIN]
        }
        return permissions_map.get(role, [])

    def can(self, action: PromptPermission):
        return action in self.permissions

5.2 Audit logging

# Audit log system
class PromptAuditLog:
    def __init__(self, prompt_id, action, user_id, details=None):
        self.prompt_id = prompt_id
        self.action = action
        self.user_id = user_id
        self.details = details
        self.timestamp = datetime.now()
        self.ip_address = self._get_client_ip()

    def log_to_database(self):
        # Insert vào audit table
        pass

    def send_to_logstash(self):
        # Gửi đến ELK stack
        pass

6. Integration Patterns

6.1 Microservice integration

# REST API cho prompt service
from fastapi import FastAPI, HTTPException
from typing import List

app = FastAPI()

@app.get("/prompts/{prompt_id}")
async def get_prompt(prompt_id: str):
    try:
        prompt = await prompt_service.get(prompt_id)
        return prompt
    except PromptNotFoundError:
        raise HTTPException(status_code=404, detail="Prompt not found")

@app.post("/prompts/search")
async def search_prompts(query: PromptSearchQuery):
    results = await prompt_service.search(query)
    return {"results": results, "total": len(results)}

6.2 SDK integration

# Python SDK cho prompt library
class PromptClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.session = requests.Session()
        self.session.headers.update({"Authorization": f"Bearer {api_key}"})

    def get_prompt(self, prompt_id: str) -> Prompt:
        response = self.session.get(f"{self.base_url}/prompts/{prompt_id}")
        response.raise_for_status()
        return Prompt.from_json(response.json())

    def search(self, query: str, tags: List[str] = None) -> List[Prompt]:
        params = {"query": query}
        if tags:
            params["tags"] = ",".join(tags)

        response = self.session.post(f"{self.base_url}/prompts/search", json=params)
        response.raise_for_status()
        return [Prompt.from_json(p) for p in response.json()["results"]]

7. Monitoring & Metrics

7.1 Key metrics

# Prometheus metrics cho prompt service
from prometheus_client import Counter, Histogram, Gauge

# Counters
PROMPT_REQUESTS_TOTAL = Counter('prompt_requests_total', 'Total prompt requests', ['status', 'endpoint'])
PROMPT_SEARCHES_TOTAL = Counter('prompt_searches_total', 'Total prompt searches', ['status'])

# Histograms
PROMPT_LATENCY_HISTOGRAM = Histogram('prompt_latency_histogram', 'Prompt request latency', ['endpoint'])
SEARCH_LATENCY_HISTOGRAM = Histogram('search_latency_histogram', 'Search request latency')

# Gauges
ACTIVE_PROMPTS_GAUGE = Gauge('active_prompts_count', 'Number of active prompts')
DEPRECATED_PROMPTS_GAUGE = Gauge('deprecated_prompts_count', 'Number of deprecated prompts')

7.2 Alerting rules

# Prometheus alerting rules
groups:
- name: prompt-library.rules
  rules:
  - alert: HighPromptErrorRate
    expr: rate(prompt_requests_total{status="error"}[5m]) > 0.1
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "High prompt error rate"
      description: "Error rate for prompt service is {{ $value }}"

  - alert: HighSearchLatency
    expr: histogram_quantile(0.95, rate(search_latency_histogram_bucket[5m])) > 500
    for: 2m
    labels:
      severity: warning
    annotations:
      summary: "High search latency"
      description: "95th percentile search latency is {{ $value }}ms"

8. Migration strategy

8.1 From ad-hoc to managed system

# Migration script từ hệ thống cũ
class PromptMigration:
    def __init__(self, old_prompts: List[Dict], new_catalog: PromptCatalog):
        self.old_prompts = old_prompts
        self.new_catalog = new_catalog

    def migrate(self):
        for old_prompt in self.old_prompts:
            new_prompt = self._transform_prompt(old_prompt)
            self.new_catalog.create(new_prompt)

        self._update_references()
        self._cleanup_old_system()

    def _transform_prompt(self, old_prompt):
        # Transform từ format cũ sang format mới
        pass

9. Cost optimization

9.1 Storage optimization

# Compression cho prompt content
import zlib

class PromptCompressor:
    @staticmethod
    def compress(content: str) -> bytes:
        return zlib.compress(content.encode('utf-8'))

    @staticmethod
    def decompress(compressed: bytes) -> str:
        return zlib.decompress(compressed).decode('utf-8')

9.2 Caching strategy optimization

# Multi-layer caching
class PromptCache:
    def __init__(self):
        self.local_cache = {}
        self.redis_cache = redis.Redis()
        self.ttl = 3600

    def get(self, prompt_id: str) -> Optional[Prompt]:
        # Try local cache first
        if prompt_id in self.local_cache:
            return self.local_cache[prompt_id]

        # Try Redis
        cached = self.redis_cache.get(f"prompt:{prompt_id}")
        if cached:
            prompt = Prompt.from_json(cached)
            self.local_cache[prompt_id] = prompt  # Populate local cache
            return prompt

        return None

    def set(self, prompt_id: str, prompt: Prompt):
        serialized = prompt.to_json()
        self.redis_cache.setex(f"prompt:{prompt_id}", self.ttl, serialized)
        self.local_cache[prompt_id] = prompt

Kết luận

Sau khi triển khai hệ thống prompt library có cấu trúc, team chúng tôi thấy:

• Giảm 80% thời gian tìm kiếm prompt
• Giảm 60% lỗi do duplicate prompt
• Dễ dàng deprecate và migrate prompt cũ

Ba điểm cốt lõi cần nhớ:

1. Cấu trúc metadata rõ ràng là nền tảng cho discoverability
2. Versioning và deprecation policy giúp maintain lâu dài
3. Caching và monitoring đảm bảo performance ở quy mô lớn

Anh em đã từng gặp vấn đề với việc quản lý prompt trong dự án của mình chưa? Giải quyết thế nào?

Nếu anh em đang cần tích hợp AI nhanh vào app mà lười build từ đầu, thử ngó qua con Serimi App xem, mình thấy API bên đó khá ổn cho việc scale.