Database
Runtime Policy
Единый JSONB объект для хранения промптов, лимитов, правил. policy = DB, не hardcode.
Связи и зависимости
Предыдущие шаги: Agent Tables
Следующие шаги: Configuration, Platform Constitution
Необходимые навыки: JSON/JSONB, PostgreSQL, Template engines
Policy B — Centralized Truth
Все промпты, лимиты, правила живут в DB. Не в коде.
Изменения вступают в силу без пересборки. UPDATE → готово.
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 режимов
runtime_utils
Утилиты для обработки команд и интентов
process_runner
Шаблоны промптов для процессов
group
Настройки для group chat summary
callbacks
Обработка feedback (expand, redo, deep-redo)
proactive
Proactive messaging
operator
Operator plane configuration
agent_selection
Agent selection policy
Чтение 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 без бэкапа
- ✗Использовать бесконечный кеш
- ✗Удалять ключи без проверки использования