Knowledge base

User asks: "How do I set up my first dashboard?"
         ↓
Support AI agent retrieves two sources in parallel:
  1. Real-time session context  ←  EventBuffer / RedisEventBuffer
  2. Relevant golden paths      ←  GET /knowledge-base/{product_id}?query=...
         ↓
Both are injected into the LLM context window
         ↓
Support AI agent answers with structured, product-specific guidance

GET /knowledge-base/{product_id}?query=<text>&top_k=<int>

Authorization: Bearer <api_key>

import httpx
from autoplay_sdk import AsyncConnectorClient, ActionsPayload

CONNECTOR_URL = "https://your-connector.onrender.com"
API_KEY = "unkey_xxxx..."
PRODUCT_ID = "acme-corp"


async def build_context(user_message: str, session_buffer: list[str]) -> str:
    """Fetch golden paths and combine with real-time session context."""
    async with httpx.AsyncClient() as client:
        resp = await client.get(
            f"{CONNECTOR_URL}/knowledge-base/{PRODUCT_ID}",
            params={"query": user_message, "top_k": 3},
            headers={"Authorization": f"Bearer {API_KEY}"},
        )
        resp.raise_for_status()
        kb = resp.json()

    golden_paths = "\n\n".join(
        f"## {r['title']}\n" + "\n".join(f"- {s}" for s in r["steps"])
        for r in kb["results"]
    )

    session_context = "\n".join(session_buffer) if session_buffer else "No session activity yet."

    return f"""
## What the user is doing right now
{session_context}

## Relevant adoption guidance
{golden_paths}
""".strip()


async def answer(user_message: str, session_buffer: list[str]) -> str:
    context = await build_context(user_message, session_buffer)
    # Pass `context` as part of your system prompt to the LLM
    ...

from autoplay_sdk import RedisEventBuffer

buffer = RedisEventBuffer(
    redis_url="redis://...",
    session_id=current_session_id,
)

# Drain the buffer and fetch knowledge base in parallel
session_payloads, kb_response = await asyncio.gather(
    buffer.drain(),
    fetch_knowledge_base(user_message),
)

session_text = "\n".join(p.to_text() for p in session_payloads)