feat(05-01): AgentTemplate ORM model, migration 007, and system prompt builder

- Add AgentTemplate ORM model to tenant.py (global, not tenant-scoped)
- Create migration 007 with agent_templates table and 7 seed templates
- Create shared/prompts/system_prompt_builder.py with build_system_prompt()
- AI transparency clause always present (non-negotiable per Phase 1 decision)
- Unit tests pass (17 tests, all sections verified)
This commit is contained in:
2026-03-24 20:27:54 -06:00
parent bffc1f2f67
commit d1acb292a1
5 changed files with 626 additions and 0 deletions

View File

@@ -0,0 +1,306 @@
"""Agent templates: create agent_templates table with 7 seed templates
Revision ID: 007
Revises: 006
Create Date: 2026-03-25
Creates the `agent_templates` table for global (non-tenant-scoped) AI employee
templates. Templates are readable by all authenticated portal users and
deployable only by tenant admins.
Seeded with 7 professional role templates:
1. Customer Support Rep (support)
2. Sales Assistant (sales)
3. Office Manager (operations)
4. Project Coordinator (operations)
5. Financial Manager (finance)
6. Controller (finance)
7. Accountant (finance)
Design notes:
- No tenant_id column — templates are global, not RLS-protected
- konstruct_app is granted SELECT/INSERT/UPDATE/DELETE
- Deploying creates an independent Agent snapshot (no FK to templates)
"""
from __future__ import annotations
import uuid
from typing import Sequence, Union
import sqlalchemy as sa
from alembic import op
from sqlalchemy.dialects.postgresql import UUID
# Alembic migration metadata
revision: str = "007"
down_revision: Union[str, None] = "006"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
# ---------------------------------------------------------------------------
# Full system prompts for seed templates — each includes the AI transparency
# clause per Phase 1 architectural decision.
# ---------------------------------------------------------------------------
_TRANSPARENCY_CLAUSE = (
"When directly asked if you are an AI, always disclose that you are an AI assistant."
)
def _prompt(name: str, role: str, persona: str, tools: list[str], rules: list[tuple[str, str]]) -> str:
"""Assemble a system prompt for seed data (mirrors build_system_prompt logic)."""
sections = [f"You are {name}, {role}.\n\n{persona}"]
if tools:
tool_lines = "\n".join(f"- {t}" for t in tools)
sections.append(f"You have access to the following tools:\n{tool_lines}")
if rules:
rule_lines = "\n".join(f"- If {cond}: {action}" for cond, action in rules)
sections.append(f"Escalation rules:\n{rule_lines}")
sections.append(_TRANSPARENCY_CLAUSE)
return "\n\n".join(sections)
# ---------------------------------------------------------------------------
# Seed data
# ---------------------------------------------------------------------------
_TEMPLATES = [
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000001")),
"name": "Customer Support Rep",
"role": "Customer Support Representative",
"description": (
"A professional, empathetic support agent that handles customer inquiries, "
"creates and looks up support tickets, and escalates complex issues to human agents. "
"Fluent in English with a calm and solution-focused communication style."
),
"category": "support",
"persona": (
"You are professional, empathetic, and solution-oriented. You listen carefully to "
"customer concerns, acknowledge their frustration with genuine warmth, and focus on "
"resolving issues efficiently. You are calm under pressure and always maintain a "
"positive, helpful tone. You escalate to a human when the situation requires it."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search", "zendesk_ticket_lookup", "zendesk_ticket_create"],
"escalation_rules": [
{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"},
{"condition": "sentiment < -0.7", "action": "handoff_human"},
],
"sort_order": 10,
},
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000002")),
"name": "Sales Assistant",
"role": "Sales Development Representative",
"description": (
"An enthusiastic sales assistant that qualifies leads, answers product questions, "
"and books meetings with the sales team. Skilled at nurturing prospects through "
"the funnel while escalating complex pricing negotiations to senior sales staff."
),
"category": "sales",
"persona": (
"You are enthusiastic, persuasive, and customer-focused. You ask thoughtful "
"discovery questions to understand prospect needs, highlight relevant product "
"benefits without being pushy, and make it easy for prospects to take the next "
"step. You are honest about limitations and escalate pricing conversations "
"to senior staff when negotiations become complex."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search", "calendar_book"],
"escalation_rules": [
{"condition": "pricing_negotiation AND attempts > 3", "action": "handoff_human"},
],
"sort_order": 20,
},
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000003")),
"name": "Office Manager",
"role": "Office Operations Manager",
"description": (
"A highly organized operations agent that handles scheduling, facilities requests, "
"vendor coordination, and general office management tasks. Keeps the workplace "
"running smoothly and escalates HR-sensitive matters to the appropriate team."
),
"category": "operations",
"persona": (
"You are highly organized, proactive, and detail-oriented. You anticipate needs "
"before they become problems, communicate clearly and concisely, and take "
"ownership of tasks through to completion. You are diplomatic when handling "
"sensitive matters and know when to involve HR or leadership."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search", "calendar_book"],
"escalation_rules": [
{"condition": "hr_complaint", "action": "handoff_human"},
],
"sort_order": 30,
},
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000004")),
"name": "Project Coordinator",
"role": "Project Coordinator",
"description": (
"A methodical project coordinator that tracks deliverables, manages timelines, "
"coordinates cross-team dependencies, and surfaces risks early. Keeps stakeholders "
"informed and escalates missed deadlines to project leadership."
),
"category": "operations",
"persona": (
"You are methodical, communicative, and results-driven. You break complex projects "
"into clear action items, track progress diligently, and surface blockers early. "
"You communicate status updates clearly to stakeholders at all levels and remain "
"calm when priorities shift. You escalate risks and missed deadlines promptly."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search"],
"escalation_rules": [
{"condition": "deadline_missed", "action": "handoff_human"},
],
"sort_order": 40,
},
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000005")),
"name": "Financial Manager",
"role": "Financial Planning and Analysis Manager",
"description": (
"A strategic finance agent that handles budgeting, forecasting, financial reporting, "
"and analysis. Provides actionable insights from financial data and escalates "
"large or unusual transactions to senior management for approval."
),
"category": "finance",
"persona": (
"You are analytical, precise, and strategic. You translate complex financial data "
"into clear insights and recommendations. You are proactive about identifying "
"budget variances, cost-saving opportunities, and financial risks. You maintain "
"strict confidentiality and escalate any transactions that exceed approval thresholds."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search"],
"escalation_rules": [
{"condition": "large_transaction AND amount > threshold", "action": "handoff_human"},
],
"sort_order": 50,
},
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000006")),
"name": "Controller",
"role": "Financial Controller",
"description": (
"A rigorous financial controller that oversees accounting operations, ensures "
"compliance with financial regulations, manages month-end close processes, and "
"monitors budget adherence. Escalates budget overruns to leadership for action."
),
"category": "finance",
"persona": (
"You are meticulous, compliance-focused, and authoritative in financial matters. "
"You ensure financial records are accurate, processes are followed, and controls "
"are maintained. You communicate financial position clearly to leadership and "
"flag compliance risks immediately. You escalate budget overruns and control "
"failures to the appropriate decision-makers."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search"],
"escalation_rules": [
{"condition": "budget_exceeded", "action": "handoff_human"},
],
"sort_order": 60,
},
{
"id": str(uuid.UUID("00000000-0000-0000-0000-000000000007")),
"name": "Accountant",
"role": "Staff Accountant",
"description": (
"A dependable accountant that handles accounts payable/receivable, invoice "
"processing, expense reconciliation, and financial record-keeping. Ensures "
"accuracy in all transactions and escalates invoice discrepancies for review."
),
"category": "finance",
"persona": (
"You are accurate, reliable, and methodical. You process financial transactions "
"with care, maintain organized records, and flag discrepancies promptly. You "
"communicate clearly when information is missing or inconsistent and follow "
"established accounting procedures diligently. You escalate significant invoice "
"discrepancies to the controller or finance manager."
),
"model_preference": "quality",
"tool_assignments": ["knowledge_base_search"],
"escalation_rules": [
{"condition": "invoice_discrepancy AND amount > threshold", "action": "handoff_human"},
],
"sort_order": 70,
},
]
def upgrade() -> None:
import json
# Create agent_templates table
op.create_table(
"agent_templates",
sa.Column("id", UUID(as_uuid=True), primary_key=True, nullable=False),
sa.Column("name", sa.String(255), nullable=False),
sa.Column("role", sa.String(255), nullable=False),
sa.Column("description", sa.Text, nullable=False, server_default=""),
sa.Column("category", sa.String(100), nullable=False, server_default="general"),
sa.Column("persona", sa.Text, nullable=False, server_default=""),
sa.Column("system_prompt", sa.Text, nullable=False, server_default=""),
sa.Column("model_preference", sa.String(50), nullable=False, server_default="quality"),
sa.Column("tool_assignments", sa.JSON, nullable=False, server_default="[]"),
sa.Column("escalation_rules", sa.JSON, nullable=False, server_default="[]"),
sa.Column("is_active", sa.Boolean, nullable=False, server_default=sa.text("true")),
sa.Column("sort_order", sa.Integer, nullable=False, server_default="0"),
sa.Column(
"created_at",
sa.DateTime(timezone=True),
nullable=False,
server_default=sa.text("now()"),
),
)
# Grant permissions to app role
op.execute("GRANT SELECT, INSERT, UPDATE, DELETE ON agent_templates TO konstruct_app")
# Seed 7 templates using parameterized INSERT with CAST for jsonb columns
# (same pattern as existing migrations — CAST(:col AS jsonb) for asyncpg jsonb params)
conn = op.get_bind()
for tmpl in _TEMPLATES:
system_prompt = _prompt(
name=str(tmpl["name"]),
role=str(tmpl["role"]),
persona=str(tmpl["persona"]),
tools=list(tmpl["tool_assignments"]), # type: ignore[arg-type]
rules=[(r["condition"], r["action"]) for r in tmpl["escalation_rules"]], # type: ignore[index]
)
conn.execute(
sa.text(
"INSERT INTO agent_templates "
"(id, name, role, description, category, persona, system_prompt, "
"model_preference, tool_assignments, escalation_rules, is_active, sort_order) "
"VALUES "
"(:id, :name, :role, :description, :category, :persona, :system_prompt, "
":model_preference, CAST(:tool_assignments AS jsonb), "
"CAST(:escalation_rules AS jsonb), :is_active, :sort_order)"
),
{
"id": tmpl["id"],
"name": tmpl["name"],
"role": tmpl["role"],
"description": tmpl["description"],
"category": tmpl["category"],
"persona": tmpl["persona"],
"system_prompt": system_prompt,
"model_preference": tmpl["model_preference"],
"tool_assignments": json.dumps(tmpl["tool_assignments"]),
"escalation_rules": json.dumps(tmpl["escalation_rules"]),
"is_active": True,
"sort_order": tmpl["sort_order"],
},
)
def downgrade() -> None:
op.drop_table("agent_templates")

View File

@@ -186,6 +186,89 @@ class Agent(Base):
return f"<Agent id={self.id} name={self.name!r} tenant_id={self.tenant_id}>" return f"<Agent id={self.id} name={self.name!r} tenant_id={self.tenant_id}>"
class AgentTemplate(Base):
"""
Pre-built AI employee templates available to all tenants.
Templates are NOT tenant-scoped (no tenant_id, no RLS). Any authenticated
portal user can browse templates. Only tenant admins can deploy them.
Deploying a template creates an independent Agent snapshot — subsequent
changes to the template do not affect deployed agents.
"""
__tablename__ = "agent_templates"
id: Mapped[uuid.UUID] = mapped_column(
UUID(as_uuid=True),
primary_key=True,
default=uuid.uuid4,
)
name: Mapped[str] = mapped_column(String(255), nullable=False)
role: Mapped[str] = mapped_column(String(255), nullable=False)
description: Mapped[str] = mapped_column(
Text,
nullable=False,
default="",
comment="2-3 sentence card preview description shown in the template gallery",
)
category: Mapped[str] = mapped_column(
String(100),
nullable=False,
default="general",
comment="Template category: support | sales | operations | finance | general",
)
persona: Mapped[str] = mapped_column(
Text,
nullable=False,
default="",
comment="Paragraph describing the agent's communication style and personality",
)
system_prompt: Mapped[str] = mapped_column(
Text,
nullable=False,
default="",
comment="Full system prompt including AI transparency clause",
)
model_preference: Mapped[str] = mapped_column(
String(50),
nullable=False,
default="quality",
comment="quality | balanced | economy | local",
)
tool_assignments: Mapped[list[Any]] = mapped_column(
JSON,
nullable=False,
default=list,
comment="JSON array of tool name strings",
)
escalation_rules: Mapped[list[Any]] = mapped_column(
JSON,
nullable=False,
default=list,
comment="JSON array of {condition, action} escalation rule objects",
)
is_active: Mapped[bool] = mapped_column(
Boolean,
nullable=False,
default=True,
comment="Inactive templates are hidden from the gallery",
)
sort_order: Mapped[int] = mapped_column(
Integer,
nullable=False,
default=0,
comment="Display order in the template gallery (ascending)",
)
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
nullable=False,
server_default=func.now(),
)
def __repr__(self) -> str:
return f"<AgentTemplate id={self.id} name={self.name!r} category={self.category!r}>"
class ChannelConnection(Base): class ChannelConnection(Base):
""" """
Links a messaging platform workspace to a Konstruct tenant. Links a messaging platform workspace to a Konstruct tenant.

View File

@@ -0,0 +1 @@
# Konstruct system prompt utilities

View File

@@ -0,0 +1,68 @@
"""
System prompt builder for Konstruct AI employees.
Assembles a coherent system prompt from wizard inputs (name, role, persona,
tools, escalation rules) and always appends the mandatory AI transparency
clause. The transparency clause is non-negotiable and cannot be omitted.
"""
from __future__ import annotations
# Non-negotiable AI transparency clause (Phase 1 architectural decision).
# Agents must always disclose their AI nature when directly asked.
AI_TRANSPARENCY_CLAUSE = (
"When directly asked if you are an AI, always disclose that you are an AI assistant."
)
def build_system_prompt(
name: str,
role: str,
persona: str = "",
tool_assignments: list[str] | None = None,
escalation_rules: list[dict[str, str]] | None = None,
) -> str:
"""
Build a system prompt for an AI employee from wizard inputs.
Args:
name: The agent's display name (e.g. "Mara").
role: The agent's role title (e.g. "Customer Support Rep").
persona: Optional paragraph describing the agent's communication style
and personality traits.
tool_assignments: Optional list of tool names available to the agent.
Omitted from the prompt when empty or None.
escalation_rules: Optional list of escalation rule dicts, each with
"condition" and "action" keys. Omitted when empty/None.
Returns:
A complete system prompt string, always ending with the AI transparency
clause.
"""
sections: list[str] = []
# --- Identity header ---
identity = f"You are {name}, {role}."
if persona:
identity = f"{identity}\n\n{persona}"
sections.append(identity)
# --- Tools section (omitted when empty) ---
effective_tools = tool_assignments if tool_assignments else []
if effective_tools:
tool_lines = "\n".join(f"- {tool}" for tool in effective_tools)
sections.append(f"You have access to the following tools:\n{tool_lines}")
# --- Escalation rules section (omitted when empty) ---
effective_rules = escalation_rules if escalation_rules else []
if effective_rules:
rule_lines = "\n".join(
f"- If {rule.get('condition', '')}: {rule.get('action', '')}"
for rule in effective_rules
)
sections.append(f"Escalation rules:\n{rule_lines}")
# --- AI transparency clause (always present, non-negotiable) ---
sections.append(AI_TRANSPARENCY_CLAUSE)
return "\n\n".join(sections)

View File

@@ -0,0 +1,168 @@
"""
Unit tests for shared.prompts.system_prompt_builder.
Tests verify:
- Full prompt with all fields produces expected sections
- Minimal prompt (name + role only) still includes AI transparency clause
- Empty tools and escalation_rules omit those sections
- AI transparency clause is always present regardless of inputs
"""
from __future__ import annotations
import pytest
from shared.prompts.system_prompt_builder import build_system_prompt
AI_TRANSPARENCY_CLAUSE = "When directly asked if you are an AI, always disclose that you are an AI assistant."
class TestBuildSystemPromptFull:
"""Test build_system_prompt with all fields populated."""
def test_contains_name(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Customer Support",
persona="Friendly and helpful",
tool_assignments=["knowledge_base_search"],
escalation_rules=[{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"}],
)
assert "You are Mara" in prompt
def test_contains_role(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Customer Support",
persona="Friendly and helpful",
tool_assignments=["knowledge_base_search"],
escalation_rules=[{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"}],
)
assert "Customer Support" in prompt
def test_contains_persona(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Customer Support",
persona="Friendly and helpful",
tool_assignments=["knowledge_base_search"],
escalation_rules=[{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"}],
)
assert "Friendly and helpful" in prompt
def test_contains_tool(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Customer Support",
persona="Friendly and helpful",
tool_assignments=["knowledge_base_search"],
escalation_rules=[{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"}],
)
assert "knowledge_base_search" in prompt
def test_contains_escalation_rule(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Customer Support",
persona="Friendly and helpful",
tool_assignments=["knowledge_base_search"],
escalation_rules=[{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"}],
)
assert "billing_dispute AND attempts > 2" in prompt
def test_contains_ai_transparency_clause(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Customer Support",
persona="Friendly and helpful",
tool_assignments=["knowledge_base_search"],
escalation_rules=[{"condition": "billing_dispute AND attempts > 2", "action": "handoff_human"}],
)
assert AI_TRANSPARENCY_CLAUSE in prompt
class TestBuildSystemPromptMinimal:
"""Test build_system_prompt with only name and role provided."""
def test_minimal_contains_name(self) -> None:
prompt = build_system_prompt(name="Alex", role="Sales Assistant")
assert "You are Alex" in prompt
def test_minimal_contains_role(self) -> None:
prompt = build_system_prompt(name="Alex", role="Sales Assistant")
assert "Sales Assistant" in prompt
def test_minimal_contains_ai_transparency_clause(self) -> None:
prompt = build_system_prompt(name="Alex", role="Sales Assistant")
assert AI_TRANSPARENCY_CLAUSE in prompt
def test_minimal_is_string(self) -> None:
prompt = build_system_prompt(name="Alex", role="Sales Assistant")
assert isinstance(prompt, str)
assert len(prompt) > 0
class TestBuildSystemPromptEmptySections:
"""Test that empty tools and escalation_rules omit those sections."""
def test_empty_tools_omits_tools_section(self) -> None:
prompt = build_system_prompt(
name="Bob",
role="Office Manager",
persona="Organized and efficient",
tool_assignments=[],
escalation_rules=[],
)
# Should not contain a tools header/section
assert "tools:" not in prompt.lower() or "tools" not in prompt.split("\n")[0]
def test_empty_escalation_omits_escalation_section(self) -> None:
prompt = build_system_prompt(
name="Bob",
role="Office Manager",
persona="Organized and efficient",
tool_assignments=[],
escalation_rules=[],
)
# Should not contain an escalation section
assert "Escalation" not in prompt
def test_none_tools_omits_tools_section(self) -> None:
prompt = build_system_prompt(
name="Bob",
role="Office Manager",
tool_assignments=None,
escalation_rules=None,
)
assert "Escalation" not in prompt
def test_empty_still_has_ai_transparency(self) -> None:
prompt = build_system_prompt(
name="Bob",
role="Office Manager",
tool_assignments=[],
escalation_rules=[],
)
assert AI_TRANSPARENCY_CLAUSE in prompt
class TestBuildSystemPromptAIClauseAlwaysPresent:
"""AI transparency clause must always be present — non-negotiable."""
def test_ai_clause_present_full_args(self) -> None:
prompt = build_system_prompt(
name="Mara",
role="Support",
persona="Helpful",
tool_assignments=["kb_search"],
escalation_rules=[{"condition": "x", "action": "handoff_human"}],
)
assert AI_TRANSPARENCY_CLAUSE in prompt
def test_ai_clause_present_name_role_only(self) -> None:
prompt = build_system_prompt(name="Z", role="Y")
assert AI_TRANSPARENCY_CLAUSE in prompt
def test_ai_clause_present_with_persona_only(self) -> None:
prompt = build_system_prompt(name="Sam", role="Analyst", persona="Detail-oriented")
assert AI_TRANSPARENCY_CLAUSE in prompt