Building a Production Support Agent with Memory

pip install fastapi uvicorn openai kronvex httpx python-dotenv

KRONVEX_API_KEY=kx_live_your_key_here
KRONVEX_AGENT_ID=agent_support_001
OPENAI_API_KEY=sk-your_openai_key
# Optional — your RAG endpoint
RAG_ENDPOINT=https://your-rag-api.example.com/search

import os
import httpx
from kronvex import KronvexClient

kx = KronvexClient(api_key=os.getenv("KRONVEX_API_KEY"))
AGENT_ID = os.getenv("KRONVEX_AGENT_ID")
RAG_ENDPOINT = os.getenv("RAG_ENDPOINT")


async def get_rag_chunks(query: str, top_k: int = 3) -> str:
    """Retrieve relevant doc chunks from your vector DB."""
    if not RAG_ENDPOINT:
        return ""
    async with httpx.AsyncClient() as client:
        r = await client.post(RAG_ENDPOINT, json={
            "query": query, "top_k": top_k
        }, timeout=4.0)
        chunks = r.json().get("results", [])
        return "\n\n".join(c.get("text", "") for c in chunks)


async def build_context(user_id: str, message: str) -> str:
    """Build the full context block: RAG docs + user memories."""
    sections = []

    # 1. RAG — shared product knowledge
    try:
        docs = await get_rag_chunks(message)
        if docs:
            sections.append(f"[PRODUCT DOCUMENTATION]\n{docs}")
    except Exception:
        pass  # RAG failure is non-fatal

    # 2. Kronvex — user-specific memory
    try:
        ctx = kx.inject_context(
            message=message,
            agent_id=user_id,
            threshold=0.65,
            top_k=5
        )
        if ctx.memories_used > 0:
            sections.append(ctx.context_block)
    except Exception:
        pass  # Memory failure is non-fatal

    return "\n\n".join(sections)


async def store_interaction(
    user_id: str,
    user_message: str,
    agent_response: str,
    session_id: str
):
    """Store the interaction as episodic memory after responding."""
    # Only store substantive exchanges (skip "thanks", "ok", etc.)
    if len(user_message) < 20:
        return

    summary = (
        f"User said: {user_message[:200]}\n"
        f"Agent replied: {agent_response[:300]}"
    )
    kx.remember(
        content=summary,
        agent_id=user_id,
        memory_type="episodic",
        session_id=session_id,
        ttl_days=90
    )

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from openai import AsyncOpenAI
from context import build_context, store_interaction
import asyncio, os

app = FastAPI()
oai = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))

SYSTEM_BASE = """You are a helpful customer support agent for Kronvex.
You have access to product documentation and the user's conversation history.
Be concise, direct, and personal. If you have past context about this user, use it.
Do NOT repeat information the user already knows."""


class ChatRequest(BaseModel):
    user_id: str
    message: str
    session_id: str = "default"
    conversation: list[dict] = []  # last N turns


class ChatResponse(BaseModel):
    reply: str
    memories_used: int = 0
    rag_used: bool = False


@app.post("/chat", response_model=ChatResponse)
async def chat(req: ChatRequest):
    # 1. Build context (RAG + memory) — non-blocking
    context_block = await build_context(req.user_id, req.message)

    # 2. Compose system prompt
    system = SYSTEM_BASE
    if context_block:
        system += f"\n\n{context_block}"

    # 3. Build messages array
    messages = [{"role": "system", "content": system}]
    messages.extend(req.conversation[-6:])  # last 3 turns
    messages.append({"role": "user", "content": req.message})

    # 4. Call LLM
    completion = await oai.chat.completions.create(
        model="gpt-4o",
        messages=messages,
        max_tokens=600,
        temperature=0.4
    )
    reply = completion.choices[0].message.content

    # 5. Store interaction async — don't block the response
    asyncio.create_task(
        store_interaction(req.user_id, req.message, reply, req.session_id)
    )

    return ChatResponse(
        reply=reply,
        memories_used=1 if context_block else 0,
        rag_used="PRODUCT DOCUMENTATION" in context_block
    )


if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)

def store_user_profile(user_id: str, user_data: dict):
    """Call this on signup / first login."""
    facts = [
        (f"User plan: {user_data['plan']}",         "semantic"),
        (f"User name: {user_data['name']}",          "semantic"),
        (f"Joined: {user_data['created_at']}",       "semantic"),
        (f"Company: {user_data.get('company','—')}", "semantic"),
    ]
    for content, mtype in facts:
        kx.remember(
            content=content,
            agent_id=user_id,
            memory_type=mtype,
            pinned=True  # pinned = never expires, always recalled
        )

uvicorn main:app --reload

# First message — agent has no memory yet
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user_alice",
    "session_id": "sess_001",
    "message": "I keep hitting rate limits on the /recall endpoint"
  }'

# New session — agent remembers from last time
curl -X POST http://localhost:8000/chat \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "user_alice",
    "session_id": "sess_002",
    "message": "Still having that issue"
  }'
# → Agent responds: "Still hitting the /recall rate limits?
#   Alice, you're on the Starter plan (1k req/day limit)..."