Architektur einer Prompt-Registry

Warum eine Prompt-Registry?

Kernfelder eines Prompt-Artefakts

Datenbankschema-Entwurf

-- prompts table: one row per unique prompt identity
CREATE TABLE prompts (
  prompt_id   VARCHAR(100) PRIMARY KEY,
  description TEXT,
  created_at  TIMESTAMP DEFAULT NOW()
);

-- prompt_versions table: one row per versioned artifact
CREATE TABLE prompt_versions (
  id          SERIAL PRIMARY KEY,
  prompt_id   VARCHAR(100) REFERENCES prompts(prompt_id),
  version     VARCHAR(20)  NOT NULL,
  template    TEXT         NOT NULL,
  author      VARCHAR(100),
  tags        TEXT[],
  model       VARCHAR(50),
  is_active   BOOLEAN DEFAULT FALSE,
  created_at  TIMESTAMP DEFAULT NOW(),
  UNIQUE(prompt_id, version)
);

Dateibasierter Registry-Entwurf

# Directory structure
prompts/
  summarize-article/
    1.0.0.yaml
    1.1.0.yaml
    latest -> 1.1.0.yaml  # symlink
  classify-sentiment/
    1.0.0.yaml

# Example: summarize-article/1.1.0.yaml
prompt_id: summarize-article
version: '1.1.0'
model: gpt-4o-mini
author: alice@company.com
tags: [summarization, articles, english]
created_at: '2024-06-01T10:00:00Z'
template: |
  Summarize the following article in {num_sentences} sentences.
  Focus on: {focus_area}.

  Article:
  {article_text}

Python-Klasse PromptRegistry

import psycopg2
import json
from datetime import datetime

class PromptRegistry:
    def __init__(self, dsn):
        self.conn = psycopg2.connect(dsn)

    def register(self, prompt_id, version, template, author, tags, model):
        with self.conn.cursor() as cur:
            # Ensure prompt identity exists
            cur.execute(
                'INSERT INTO prompts (prompt_id) VALUES (%s) ON CONFLICT DO NOTHING',
                (prompt_id,)
            )
            cur.execute(
                '''INSERT INTO prompt_versions
                   (prompt_id, version, template, author, tags, model)
                   VALUES (%s, %s, %s, %s, %s, %s)''',
                (prompt_id, version, template, author, tags, model)
            )
        self.conn.commit()
        print(f'Registered {prompt_id}@{version}')

    def get_active(self, prompt_id):
        with self.conn.cursor() as cur:
            cur.execute(
                'SELECT template, version FROM prompt_versions '
                'WHERE prompt_id=%s AND is_active=TRUE LIMIT 1',
                (prompt_id,)
            )
            row = cur.fetchone()
        if not row:
            raise ValueError(f'No active version for {prompt_id}')
        return {'template': row[0], 'version': row[1]}

Tiefgehende Betrachtung des Metadatenschemas

# Extended metadata example
prompt_metadata = {
    'prompt_id': 'extract-key-dates',
    'version': '2.0.0',
    'author': 'bob@company.com',
    'tags': ['extraction', 'dates', 'contracts', 'production'],
    'model': 'gpt-4o',
    'changelog': 'Added support for relative dates (next quarter, end of year)',
    'test_suite': 's3://company-evals/extract-key-dates/v2-testset.jsonl',
    'created_at': '2024-07-15T09:30:00Z',
    'is_active': True
}

Template-Rendering mit Variablen

class PromptRegistry:
    # ... (previous methods)

    def render(self, prompt_id, variables: dict) -> str:
        artifact = self.get_active(prompt_id)
        template = artifact['template']
        try:
            rendered = template.format_map(variables)
        except KeyError as e:
            raise ValueError(f'Missing variable {e} for prompt {prompt_id}')
        return rendered

# Usage
registry = PromptRegistry(dsn='postgresql://...')
prompt = registry.render(
    'summarize-article',
    {
        'num_sentences': 3,
        'focus_area': 'financial impact',
        'article_text': 'Apple reported record revenue of $119B...'
    }
)
print(prompt)
# Output: Summarize the following article in 3 sentences.
# Focus on: financial impact. ...

Eine Version aktivieren

def activate_version(self, prompt_id, version):
    with self.conn.cursor() as cur:
        # Deactivate all current versions
        cur.execute(
            'UPDATE prompt_versions SET is_active=FALSE '
            'WHERE prompt_id=%s AND is_active=TRUE',
            (prompt_id,)
        )
        # Activate target version
        cur.execute(
            'UPDATE prompt_versions SET is_active=TRUE '
            'WHERE prompt_id=%s AND version=%s',
            (prompt_id, version)
        )
        if cur.rowcount == 0:
            self.conn.rollback()
            raise ValueError(f'Version {version} not found for {prompt_id}')
    self.conn.commit()
    print(f'Activated {prompt_id}@{version}')

Prompts auflisten und suchen

def list_versions(self, prompt_id):
    with self.conn.cursor() as cur:
        cur.execute(
            'SELECT version, author, is_active, created_at '
            'FROM prompt_versions WHERE prompt_id=%s '
            'ORDER BY created_at DESC',
            (prompt_id,)
        )
        return cur.fetchall()

def search_by_tag(self, tag):
    with self.conn.cursor() as cur:
        cur.execute(
            'SELECT prompt_id, version, tags FROM prompt_versions '
            'WHERE %s = ANY(tags)',
            (tag,)
        )
        return cur.fetchall()

# Usage
for v in registry.list_versions('summarize-article'):
    print(v)  # ('1.1.0', 'alice', True, datetime(...))

for p in registry.search_by_tag('production'):
    print(p)  # ('summarize-article', '1.1.0', ['production', 'summarization'])

Registry-API-Endpunkte

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI()
registry = PromptRegistry(dsn='postgresql://user:pass@localhost/prompts')

class VersionPayload(BaseModel):
    version: str
    template: str
    author: str
    tags: list
    model: str

@app.post('/prompts/{prompt_id}/versions')
def register_version(prompt_id: str, payload: VersionPayload):
    registry.register(
        prompt_id, payload.version, payload.template,
        payload.author, payload.tags, payload.model
    )
    return {'status': 'registered'}

@app.get('/prompts/{prompt_id}/active')
def get_active(prompt_id: str):
    try:
        return registry.get_active(prompt_id)
    except ValueError as e:
        raise HTTPException(404, str(e))

@app.put('/prompts/{prompt_id}/activate/{version}')
def activate(prompt_id: str, version: str):
    registry.activate_version(prompt_id, version)
    return {'status': 'activated'}

Prüfprotokoll und Änderungshistorie

CREATE TABLE prompt_audit_log (
  id          SERIAL PRIMARY KEY,
  prompt_id   VARCHAR(100),
  version     VARCHAR(20),
  action      VARCHAR(50),  -- 'registered', 'activated', 'deactivated'
  actor       VARCHAR(100), -- user or service that performed the action
  reason      TEXT,
  created_at  TIMESTAMP DEFAULT NOW()
);

-- Trigger to auto-log activations
CREATE OR REPLACE FUNCTION log_activation()
RETURNS TRIGGER AS $func$
BEGIN
  IF NEW.is_active != OLD.is_active THEN
    INSERT INTO prompt_audit_log (prompt_id, version, action)
    VALUES (NEW.prompt_id, NEW.version,
            CASE WHEN NEW.is_active THEN 'activated' ELSE 'deactivated' END);
  END IF;
  RETURN NEW;
END;
$func$ LANGUAGE plpgsql;

CREATE TRIGGER trg_activation
AFTER UPDATE ON prompt_versions
FOR EACH ROW EXECUTE FUNCTION log_activation();

Kurzcheck

Zusammenfassung der Registry-Architektur