Build a Support Agent

In this tutorial you will build a customer support agent that remembers customer preferences, searches a knowledge base using semantic similarity, and logs every reasoning step for compliance review.

By the end you will have an agent that:

• Persists customer interaction history using HatiData memory
• Searches knowledge base articles with hybrid SQL (semantic_match + JOIN_VECTOR)
• Records chain-of-thought reasoning to an immutable audit ledger
• Replays and verifies decision traces on demand

Prerequisites

• Python 3.10 or later
• A HatiData account and API key (sign up)
• hatidata Python SDK installed

pip install hatidata

Set your credentials as environment variables:

export HATIDATA_API_KEY="hd_your_api_key"
export HATIDATA_ORG="your-org-slug"

---

Step 1: Set Up HatiData

Create a client and verify the connection before defining any schema.

import os
from hatidata import HatiDataClient

client = HatiDataClient(
    api_key=os.environ["HATIDATA_API_KEY"],
    org=os.environ["HATIDATA_ORG"],
)

# Verify connectivity
status = client.ping()
print(f"Connected to HatiData: {status}")

The client handles connection pooling, JWT refresh, and retry logic automatically.

---

Step 2: Create the Schema

Define three tables that represent the support agent's data layer: customers, tickets, and a knowledge base.

client.execute("""
    CREATE TABLE IF NOT EXISTS customers (
        customer_id   TEXT PRIMARY KEY,
        name          TEXT NOT NULL,
        email         TEXT NOT NULL,
        tier          TEXT DEFAULT 'free',
        created_at    TIMESTAMPTZ DEFAULT now()
    )
""")

client.execute("""
    CREATE TABLE IF NOT EXISTS support_tickets (
        ticket_id     TEXT PRIMARY KEY,
        customer_id   TEXT NOT NULL REFERENCES customers(customer_id),
        subject       TEXT NOT NULL,
        body          TEXT NOT NULL,
        status        TEXT DEFAULT 'open',
        resolution    TEXT,
        created_at    TIMESTAMPTZ DEFAULT now(),
        resolved_at   TIMESTAMPTZ
    )
""")

client.execute("""
    CREATE TABLE IF NOT EXISTS knowledge_base (
        article_id    TEXT PRIMARY KEY,
        title         TEXT NOT NULL,
        content       TEXT NOT NULL,
        category      TEXT NOT NULL,
        updated_at    TIMESTAMPTZ DEFAULT now()
    )
""")

print("Schema created.")

Seed a few knowledge base articles for the agent to search against:

articles = [
    (
        "kb-001",
        "How to reset your password",
        "Visit /settings/security and click Reset Password. A confirmation link is emailed within 2 minutes.",
        "account",
    ),
    (
        "kb-002",
        "Billing cycle and invoice dates",
        "Invoices are generated on the 1st of each month. Enterprise customers receive PDF invoices by email.",
        "billing",
    ),
    (
        "kb-003",
        "API rate limits by tier",
        "Free: 100 req/min. Growth: 1000 req/min. Enterprise: unlimited with burst allowance.",
        "api",
    ),
    (
        "kb-004",
        "Data export formats",
        "Export your data as Parquet, CSV, or JSON from the dashboard under Settings > Export.",
        "data",
    ),
]

client.executemany(
    "INSERT INTO knowledge_base (article_id, title, content, category) VALUES (?, ?, ?, ?)",
    articles,
)
print(f"Seeded {len(articles)} knowledge base articles.")

---

Step 3: Store Customer Memories

HatiData memory lets the agent persist structured facts about each customer. Memories are stored alongside embeddings so they can be retrieved by semantic similarity later.

from hatidata.memory import MemoryClient

memory = MemoryClient(client)

def remember_interaction(customer_id: str, summary: str, metadata: dict):
    """Store a customer interaction as a searchable memory."""
    memory.store(
        agent_id="support-agent-v1",
        content=summary,
        metadata={
            "customer_id": customer_id,
            "type": "interaction",
            **metadata,
        },
    )

# Store two past interactions for an existing customer
remember_interaction(
    customer_id="cust-123",
    summary="Customer asked about billing cycle. Prefers email summaries over in-app notifications.",
    metadata={"topic": "billing", "sentiment": "neutral"},
)

remember_interaction(
    customer_id="cust-123",
    summary="Customer reported API rate limit errors during a batch job running at 02:00 UTC.",
    metadata={"topic": "api", "sentiment": "frustrated"},
)

print("Customer memories stored.")

Retrieve relevant memories when the agent handles a new ticket:

def recall_customer_context(customer_id: str, query: str) -> list[dict]:
    """Retrieve memories relevant to the current customer query."""
    results = memory.search(
        agent_id="support-agent-v1",
        query=query,
        filters={"customer_id": customer_id},
        top_k=5,
    )
    return results

context = recall_customer_context("cust-123", "billing invoice question")
for mem in context:
    print(f"  [{mem['score']:.2f}] {mem['content']}")

memory.search runs a hybrid query: vector ANN retrieval narrows the candidate set, then the query engine applies the customer_id filter and re-ranks by exact cosine similarity.

---

Step 4: Search with Hybrid SQL

HatiData extends SQL with semantic_match() for embedding-based ranking and JOIN_VECTOR for combining semantic scores with structured predicates.

def find_relevant_articles(query: str, category: str | None = None) -> list[dict]:
    """
    Search the knowledge base using semantic similarity.
    Optionally restrict by category for higher precision.
    """
    category_filter = f"AND category = '{category}'" if category else ""

    rows = client.query(f"""
        SELECT
            article_id,
            title,
            content,
            category,
            semantic_match(content, '{query}') AS relevance_score
        FROM knowledge_base
        WHERE semantic_match(content, '{query}') > 0.65
        {category_filter}
        ORDER BY relevance_score DESC
        LIMIT 3
    """)

    return [dict(row) for row in rows]


def find_articles_with_ticket_context(ticket_body: str) -> list[dict]:
    """
    Join knowledge base articles against a ticket using vector similarity.
    Returns articles ranked by relevance to the ticket content.
    """
    rows = client.query(f"""
        SELECT
            kb.article_id,
            kb.title,
            kb.category,
            v.similarity
        FROM knowledge_base kb
        JOIN_VECTOR (
            source       => '{ticket_body}',
            target_column => content,
            metric       => 'cosine'
        ) AS v ON kb.article_id = v.row_id
        WHERE v.similarity > 0.60
        ORDER BY v.similarity DESC
        LIMIT 5
    """)

    return [dict(row) for row in rows]


# Agent handling an incoming billing inquiry
ticket_body = "I have not received my invoice for last month. When do invoices get sent?"

articles = find_articles_with_ticket_context(ticket_body)
print("Relevant articles:")
for a in articles:
    print(f"  [{a['similarity']:.2f}] {a['title']} (category: {a['category']})")

JOIN_VECTOR lets you combine the semantic ranking with any SQL predicate — for example, restricting to articles updated within the last 90 days.

---

Step 5: Log Reasoning Steps

Every decision the agent makes is recorded as a chain-of-thought step in HatiData's immutable ledger. Steps are cryptographically hash-chained, creating a tamper-evident audit trail.

from hatidata.cot import CotClient, StepType

cot = CotClient(client)

def handle_ticket(ticket_id: str, customer_id: str, ticket_body: str) -> str:
    """
    Handle a support ticket with full chain-of-thought logging.
    Returns the drafted resolution.
    """
    session_id = f"ticket-{ticket_id}"

    # Observation: incoming ticket
    cot.log_step(
        agent_id="support-agent-v1",
        session_id=session_id,
        step_type=StepType.OBSERVATION,
        content=f"Received ticket from customer {customer_id}: {ticket_body[:120]}",
        metadata={"ticket_id": ticket_id},
    )

    # Reasoning: retrieve customer context
    memories = recall_customer_context(customer_id, ticket_body)
    memory_summary = "; ".join(m["content"][:80] for m in memories[:2])

    cot.log_step(
        agent_id="support-agent-v1",
        session_id=session_id,
        step_type=StepType.REASONING,
        content=f"Retrieved {len(memories)} past interactions. Context: {memory_summary}",
    )

    # Tool call: knowledge base search
    articles = find_relevant_articles(ticket_body)
    top_article = articles[0] if articles else None

    cot.log_step(
        agent_id="support-agent-v1",
        session_id=session_id,
        step_type=StepType.TOOL_CALL,
        content=(
            f"Searched knowledge base. Top match: '{top_article['title']}' "
            f"(score {top_article['relevance_score']:.2f})"
        ) if top_article else "No relevant articles found. Escalating.",
        metadata={"article_ids": [a["article_id"] for a in articles]},
    )

    # Conclusion: draft resolution
    if top_article:
        resolution = (
            f"Based on our knowledge base: {top_article['content']} "
            "If you need further help, reply to this ticket."
        )
    else:
        resolution = "I have escalated your ticket to a specialist. Expect a response within 4 hours."

    cot.log_step(
        agent_id="support-agent-v1",
        session_id=session_id,
        step_type=StepType.CONCLUSION,
        content="Resolution drafted. Ticket will be marked resolved.",
        metadata={"resolution_length": len(resolution)},
    )

    # Persist the interaction so future tickets benefit from this context
    remember_interaction(
        customer_id=customer_id,
        summary=(
            f"Ticket {ticket_id}: {ticket_body[:80]} — resolved using "
            f"article {top_article['article_id'] if top_article else 'escalation'}"
        ),
        metadata={
            "ticket_id": ticket_id,
            "topic": top_article["category"] if top_article else "unknown",
        },
    )

    return resolution


resolution = handle_ticket(
    ticket_id="tkt-9901",
    customer_id="cust-123",
    ticket_body="I have not received my invoice for last month. When do invoices get sent?",
)
print(f"\nResolution:\n{resolution}")

---

Step 6: Replay and Verify

After handling tickets you can replay any session to audit the agent's decisions step by step. HatiData verifies the cryptographic hash chain automatically.

def audit_ticket_session(ticket_id: str):
    """Replay and verify the chain-of-thought trace for a given ticket."""
    session_id = f"ticket-{ticket_id}"

    trace = cot.replay_session(
        agent_id="support-agent-v1",
        session_id=session_id,
    )

    print(f"\nAudit trace for session: {session_id}")
    print(f"Steps recorded : {len(trace.steps)}")
    print(f"Chain integrity: {'VALID' if trace.chain_valid else 'BROKEN'}")
    print()

    for i, step in enumerate(trace.steps, 1):
        print(f"  Step {i} [{step.step_type}] at {step.timestamp}")
        print(f"    {step.content[:120]}")
        print(f"    Hash: {step.hash[:16]}...")

audit_ticket_session("tkt-9901")

The chain_valid flag is False if any step was modified after recording. Each step's hash incorporates the previous step's hash, so a single tampered record breaks the entire chain from that point forward.

---

What You Built

Capability	HatiData Feature
Remember customer history	MemoryClient.store() + MemoryClient.search()
Semantic knowledge base search	semantic_match() SQL function
Structured + vector join	JOIN_VECTOR clause
Immutable reasoning audit	CotClient.log_step()
Tamper-evident verification	CotClient.replay_session()

---

Related Concepts

• Persistent Memory — How HatiData stores and indexes agent memories
• Hybrid SQL — Full semantic_match and JOIN_VECTOR syntax
• Chain-of-Thought Ledger — Hash-chained reasoning traces explained
• Build a Fraud Detection Agent — Next tutorial: semantic triggers and audit trails