Conversational AI Platform Architecture

/** Assistant configuration — the declarative definition of a conversational agent. */

export interface AssistantConfig {
  /** Unique identifier */
  id: string;

  /** Human-readable name */
  name: string;

  /** System prompt (supports {{variable}} templates) */
  systemPrompt: string;

  /** LLM model to use */
  model: string;

  /** Available tools from the tool registry */
  tools: string[];

  /** Memory strategy */
  memory: {
    strategy: "full" | "sliding-window" | "summarize";
    /** Max messages before summarization (for summarize strategy) */
    maxMessages?: number;
    /** Max tokens in context window budget for history */
    maxHistoryTokens?: number;
  };

  /** Guardrail policies to enforce */
  guardrails: {
    /** Block topics (e.g., ["competitor-pricing", "legal-advice"]) */
    blockedTopics: string[];
    /** PII detection and redaction */
    piiHandling: "block" | "redact" | "allow";
    /** Maximum response length in tokens */
    maxResponseTokens: number;
    /** Custom safety prompt appended to system prompt */
    safetyInstructions?: string;
  };

  /** Metadata for analytics */
  metadata: {
    team: string;
    environment: "staging" | "production";
    version: string;
  };
}

"""Multi-turn memory management with rolling summarization."""

from __future__ import annotations

from dataclasses import dataclass
from anthropic import Anthropic

client = Anthropic()


@dataclass
class ConversationMemory:
    """Manages conversation history with summarization."""

    summary: str = ""
    recent_messages: list[dict] = None
    max_recent: int = 10
    total_turns: int = 0

    def __post_init__(self):
        if self.recent_messages is None:
            self.recent_messages = []

    def add_message(self, role: str, content: str):
        """Add a message and trigger summarization if needed."""
        self.recent_messages.append({"role": role, "content": content})
        self.total_turns += 1

        if len(self.recent_messages) > self.max_recent:
            self._summarize_oldest()

    def _summarize_oldest(self):
        """Summarize the oldest half of messages into the running summary."""
        cutoff = len(self.recent_messages) // 2
        to_summarize = self.recent_messages[:cutoff]
        self.recent_messages = self.recent_messages[cutoff:]

        messages_text = "\n".join(
            f"{m['role']}: {m['content']}" for m in to_summarize
        )

        response = client.messages.create(
            model="claude-haiku-4-5-20251001",
            max_tokens=512,
            messages=[{
                "role": "user",
                "content": (
                    "Summarize this conversation segment concisely, "
                    "preserving key facts, decisions, and user preferences. "
                    "Previous summary: "
                    f"{self.summary or 'None'}\n\n"
                    f"New messages:\n{messages_text}"
                ),
            }],
        )
        self.summary = response.content[0].text

    def get_context(self) -> list[dict]:
        """Build the message list for the LLM call."""
        context = []
        if self.summary:
            context.append({
                "role": "user",
                "content": (
                    f"[Conversation summary so far: {self.summary}]"
                ),
            })
            context.append({
                "role": "assistant",
                "content": "I understand the conversation context. How can I help?",
            })
        context.extend(self.recent_messages)
        return context

"""Input and output guardrail middleware."""

from __future__ import annotations

import re
from dataclasses import dataclass
from enum import Enum
from typing import Callable


class GuardrailResult(str, Enum):
    PASS = "pass"
    WARN = "warn"
    BLOCK = "block"


@dataclass
class GuardrailCheck:
    result: GuardrailResult
    reason: str
    modified_content: str | None = None  # For redaction


# ---- Input Guardrails ----

def check_pii(text: str, handling: str) -> GuardrailCheck:
    """Detect and handle PII in user input."""
    # Simple regex patterns for common PII
    patterns = {
        "email": r"[\w.+-]+@[\w-]+\.[\w.-]+",
        "phone": r"\b\d{3}[-.]?\d{3}[-.]?\d{4}\b",
        "ssn": r"\b\d{3}-\d{2}-\d{4}\b",
    }

    found = {}
    for pii_type, pattern in patterns.items():
        matches = re.findall(pattern, text)
        if matches:
            found[pii_type] = matches

    if not found:
        return GuardrailCheck(GuardrailResult.PASS, "No PII detected")

    if handling == "block":
        return GuardrailCheck(
            GuardrailResult.BLOCK,
            f"PII detected: {', '.join(found.keys())}. Please remove personal information.",
        )

    if handling == "redact":
        redacted = text
        for pattern in patterns.values():
            redacted = re.sub(pattern, "[REDACTED]", redacted)
        return GuardrailCheck(
            GuardrailResult.WARN,
            "PII redacted from input",
            modified_content=redacted,
        )

    return GuardrailCheck(GuardrailResult.PASS, "PII handling set to allow")


def check_blocked_topics(
    text: str, blocked: list[str]
) -> GuardrailCheck:
    """Check if input references blocked topics."""
    text_lower = text.lower()
    for topic in blocked:
        if topic.lower() in text_lower:
            return GuardrailCheck(
                GuardrailResult.BLOCK,
                f"This topic ({topic}) is outside my scope. "
                "Please contact the appropriate team.",
            )
    return GuardrailCheck(GuardrailResult.PASS, "No blocked topics")