Database

Runtime Policy

Единый JSONB объект для хранения промптов, лимитов, правил. policy = DB, не hardcode.

8+
Namespaces
8
Template Vars
50+
Keys
service_settings
Table

Связи и зависимости

Предыдущие шаги: Agent Tables

Следующие шаги: Configuration, Platform Constitution

Необходимые навыки: JSON/JSONB, PostgreSQL, Template engines

Policy B — Centralized Truth

No Hardcode

Все промпты, лимиты, правила живут в DB. Не в коде.

Runtime Update

Изменения вступают в силу без пересборки. UPDATE → готово.

Version History

updated_at timestamp для аудита изменений.

Структура runtime_policy_json

service_settings
└── key = 'runtime_policy_json'
    └── value = JSONB {
        "runtime": {
          "copilot_style_appendix": "Ты ассистент в Telegram...",
          "promode_style_appendix": "Ты экспертный режим..."
        },
        "runtime_utils": {
          "soft_command_alias_map": {...},
          "web_intent_markers": [...],
          "web_process_complex_markers": [...]
        },
        "process_runner": {
          "prompt_templates": {
            "default": "Контекст: {context}...",
            "web_search": "Найди в интернете..."
          },
          "timeout_fallback_templates": {...}
        },
        "group": {
          "summary_system_prompt": "Ты summarizer...",
          "summary_user_prompt_template": "Summarize: {messages}",
          "summary_num_predict": 900,
          "summary_temperature": 0.2
        },
        "callbacks": {
          "feedback_expand_system_prompt": "...",
          "feedback_redo_user_template": "...",
          "feedback_min_num_predict": 128
        },
        "proactive": {
          "context_limit": 8,
          "decision_max_tokens": 280,
          "decision_system_prompt": "..."
        },
        "operator": {
          "classify_prompt": "You are an operator task classifier...",
          "safety_keywords": ["trading", "wallet", "payment"]
        },
        "agent_selection": {
          "default_agent": "assistant",
          "intent_routing": {...}
        }
      }

Namespaces

runtime

Стилевые appendix для copilot и promode режимов

copilot_style_appendixpromode_style_appendix

runtime_utils

Утилиты для обработки команд и интентов

soft_command_alias_mapweb_intent_markersweb_process_complex_markers

process_runner

Шаблоны промптов для процессов

prompt_templatestimeout_fallback_templateserror_templates

group

Настройки для group chat summary

summary_system_promptsummary_user_prompt_templatesummary_num_predictsummary_temperature

callbacks

Обработка feedback (expand, redo, deep-redo)

feedback_expand_system_promptfeedback_redo_user_templatefeedback_deep_redo_*

proactive

Proactive messaging

context_limitdecision_max_tokensdecision_system_prompt

operator

Operator plane configuration

classify_promptsearch_templatessafety_keywords

agent_selection

Agent selection policy

default_agentintent_routingmodel_preferences

Чтение policy в Python

# services/runtime_policy.py
async def get_runtime_policy(conn, namespace: str = None) -> dict:
    """Загрузить runtime policy из DB.
    
    Args:
        conn: AsyncPG connection
        namespace: Опционально, вернуть только этот namespace
        
    Returns:
        dict: Весь policy или конкретный namespace
    """
    row = await conn.fetchrow("""
        SELECT value::jsonb 
        FROM clowbot.service_settings 
        WHERE key = 'runtime_policy_json'
    """)
    
    if not row:
        return {}
        
    policy = json.loads(row['value'])
    
    if namespace:
        return policy.get(namespace, {})
    return policy

# Использование
runtime = await get_runtime_policy(conn, 'runtime')
copilot_appendix = runtime.get('copilot_style_appendix', '')

# Получить конкретное значение
group_config = await get_runtime_policy(conn, 'group')
max_tokens = group_config.get('summary_num_predict', 900)

Обновление policy

Обновить конкретный ключ:

UPDATE clowbot.service_settings
SET value = jsonb_set(
  value::jsonb,
  '{runtime,copilot_style_appendix}',
  '"Новый текст appendix"'::jsonb
)::text,
updated_at = now()
WHERE key = 'runtime_policy_json';

Добавить новый ключ:

UPDATE clowbot.service_settings
SET value = jsonb_set(
  value::jsonb,
  '{runtime,new_key}',
  '"value"'::jsonb
)::text,
updated_at = now()
WHERE key = 'runtime_policy_json';

Обновить весь namespace:

UPDATE clowbot.service_settings
SET value = jsonb_set(
  value::jsonb,
  '{group}',
  '{
    "summary_system_prompt": "Новый промпт",
    "summary_num_predict": 1200,
    "summary_temperature": 0.3
  }'::jsonb
)::text,
updated_at = now()
WHERE key = 'runtime_policy_json';

Template Variables

ПеременнаяОписаниеПример
{input_compact}Сжатый ввод пользователянапомни позвонить
{context_short_or_none}Краткий контекст или 'нет данных'Previous: meeting at 3pm
{chat_db_id}ID чата в БДuuid-1234
{runtime_facts}Текущие runtime фактыDate: 2024-03-08, Time: 14:30
{source_query}Исходный запрос пользователяНапомни мне позвонить маме
{assistant_text}Предыдущий ответ ассистентаХорошо, я напомню...
{user_name}Имя пользователяAlex
{model_name}Текущая модельphi4-mini

Soft Command Aliases

Маппинг "мягких" команд в настоящие /команды для естественного языка:

"soft_command_alias_map": {
  "web": "/web",
  "web_search": "/web_search",
  "веб": "/web",
  "поиск": "/web",
  "checkin": "/checkin",
  "чекин": "/checkin",
  "remind": "/remind",
  "напомни": "/remind",
  "напомнить": "/remind",
  "reminder": "/remind",
  "summary": "/summary",
  "саммари": "/summary",
  "summarize": "/summary"
}

// Пользователь пишет: "напомни позвонить в 15:00"
// Система мапит в: "/remind позвонить в 15:00"

// Пользователь пишет: "сделай веб поиск"
// Система мапит в: "/web поиск"

Web Intent Markers

Маркеры для определения web-search intent:

"web_intent_markers": [
  "в интернете", "в сети", "найди", "погугл",
  "google", "ссылк", "официальный сайт", 
  "источник", "source", "search", "lookup",
  "что такое", "какой", "кто такой"
]

// Если сообщение содержит любой маркер → web search intent
// "найди информацию о Python" → web_search
// "что такое machine learning" → web_search

Caching Strategy

Без кеширования

# Каждый запрос читает из DB
policy = await get_runtime_policy(conn)
# + Гарантия актуальности
# - Нагрузка на DB

С кешированием

# TTL кеш 5 минут
@cached(ttl=300, key="runtime_policy")
async def get_runtime_policy_cached():
    return await get_runtime_policy(conn)
# + Меньше нагрузка
# - Задержка обновлений

Рекомендация: Используй кеш с TTL 60-300 сек для production. Для критичных изменений используй cache invalidation через UPDATE trigger.

Troubleshooting

Policy not found

Возвращается пустой dict или default значение.

# Проверить что ключ существует
SELECT key FROM clowbot.service_settings 
WHERE key = 'runtime_policy_json';

# Если нет - создать
INSERT INTO clowbot.service_settings (key, value)
VALUES ('runtime_policy_json', '{}'::jsonb);

Invalid JSON

JSON синтаксическая ошибка в value.

# Валидация JSON
SELECT value::jsonb FROM clowbot.service_settings 
WHERE key = 'runtime_policy_json';

# Если ошибка - исправить JSON
UPDATE clowbot.service_settings
SET value = '{"runtime": {}}'::jsonb::text
WHERE key = 'runtime_policy_json';

Template variable not replaced

Переменная в промпте не подставилась.

# Проверить что переменная передана
template = policy['process_runner']['prompt_templates']['default']
# template = "Context: {context_short_or_none}"

# Замена
result = template.replace("{context_short_or_none}", context)
# или использовать .format()

DO

  • Используй default значения при чтении
  • Документируй все namespace и ключи
  • Валидируй JSON перед UPDATE
  • Используй кеш с TTL для частых запросов

DON'T

  • Хардкодить промпты в Python коде
  • Изменять policy без бэкапа
  • Использовать бесконечный кеш
  • Удалять ключи без проверки использования