I am trying to create a chat agent, give me python code for local testing. I already have

Thanks for reaching out, Aditya. I'm an AI assistant and I'm here to help. Give me a moment while I research this topic.

I am creating a suport agent, which will require context awarenedd and memoery retention

Thanks — I can help build a support agent with context awareness and memory. Quick clarifying questions so I give you the right code: 1. Do you want short-term memory only (conversation history during a session) or long-term memory across sessions (persisted per user)? 2. If long-term, do you prefer an in-memory/local file solution for testing or a real database (Postgres, Redis, etc.)? 3. Do you need retrieval-augmented responses from a knowledge base (RAG)? If yes, what form is your KB in (files, search index, vector DB)? 4. Should the agent store structured user data (name, tickets, preferences) via

userdata

, or just keep raw message history? 5. Any preferred LLM model and whether tools/plugins (OpenAI, Letta) are available in your environment? Answer these and I’ll provide a focused Python example (session-only, local-persisted, or DB-backed) that includes chat context,

userdata

, and an example RAG hook if you want it. Sources: Pipeline nodes and hooks | LiveKit docs | External data and RAG | LiveKit docs | Context Variables | LiveKit docs | Agents and handoffs | LiveKit docs

I will be using OpenAi models gpt-4.1-mini, It will have access to tools and Rag aswell.

Great — below is a concise, runnable Python example that shows a support agent using OpenAI

gpt-4.1-mini

, with: • session-level chat context (multi-turn), • structured

userdata

for short- and long-term memory (local JSON file persistence for testing), • a simple RAG hook that looks up a local knowledge base (text files) and injects results before agent replies, • a sample tool integration pattern (tool is a placeholder you can implement to call external APIs). Adjust paths, model name, and tool implementations as needed. Note: install required packages (

livekit-agents

,

livekit-plugins-openai

) and set your OpenAI API key in the environment.

import asyncio
import json
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Optional

from livekit.agents import AgentSession, ChatContext, ChatMessage, Agent
from livekit.plugins import openai

# ---------- Config ----------
MODEL_NAME = "gpt-4.1-mini"
USERDATA_STORE = Path("./userdata_store.json")
KB_DIR = Path("./kb")  # local knowledge base: plain text files
# ----------------------------

# Initialize LLM plugin
llm = openai.LLM(model=MODEL_NAME)

# Structured session userdata for memory
@dataclass
class SupportSessionData:
    user_id: str
    user_name: Optional[str] = None
    last_ticket_id: Optional[str] = None
    preferences: dict = None

    def __post_init__(self):
        if self.preferences is None:
            self.preferences = {}

# Simple local persistence for long-term memory (file-based for testing)
def load_userdata(user_id: str) -> SupportSessionData:
    if USERDATA_STORE.exists():
        all_data = json.loads(USERDATA_STORE.read_text())
        if user_id in all_data:
            return SupportSessionData(**all_data[user_id])
    return SupportSessionData(user_id=user_id)

def save_userdata(data: SupportSessionData):
    all_data = {}
    if USERDATA_STORE.exists():
        all_data = json.loads(USERDATA_STORE.read_text())
    all_data[data.user_id] = asdict(data)
    USERDATA_STORE.write_text(json.dumps(all_data, indent=2))

# Simple RAG lookup: returns relevant snippets from local KB files
def rag_lookup(query: str, max_files: int = 3) -> str:
    if not KB_DIR.exists():
        return ""
    results = []
    for p in KB_DIR.glob("*.txt"):
        text = p.read_text()
        if query.lower() in text.lower():
            # take first 500 chars as snippet
            results.append(f"From {p.name}:\n{text[:500]}\n---")
            if len(results) >= max_files:
                break
    return "\n".join(results)

# Placeholder tool example: search tickets API, call external system, etc.
async def ticket_lookup_tool(ticket_id: str) -> str:
    # Replace with real API calls in production.
    await asyncio.sleep(0.1)
    return f"Ticket {ticket_id} status: OPEN (sample data)"

# Create the agent class with lifecycle hooks for RAG & memory
class SupportAgent(Agent):
    def __init__(self, session_userdata: SupportSessionData):
        instructions = (
            "You are a helpful customer support agent. Use conversation history and stored user data "
            "to personalize replies. If you find relevant KB content, cite it. If you need to call tools, "
            "respond with a tool invocation intent (handled by the host)."
        )
        super().__init__(instructions=instructions)
        self._userdata = session_userdata

    @property
    def userdata(self) -> SupportSessionData:
        return self._userdata

    # Hook: run after user turn is received but before generating a reply
    async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage):
        # 1) Simple RAG: lookup KB using user's latest message
        query = new_message.text_content()
        rag_content = rag_lookup(query)
        if rag_content:
            # Inject KB snippets into the conversation context as assistant message
            turn_ctx.add_message(role="assistant", content=f"KB_SNIPPET:\n{rag_content}")

        # 2) Example: if user mentions a ticket id pattern, run a tool to fetch status
        # (very simple detection; adapt regex for production)
        words = query.split()
        for w in words:
            if w.startswith("TICKET-"):
                status = await ticket_lookup_tool(w)
                turn_ctx.add_message(role="assistant", content=f"TOOL_RESULT:\n{status}")

        # 3) Save quick info from the user message into session userdata (demo: detect "my name is X")
        if "my name is" in query.lower():
            # naive extraction for demo
            parts = query.split("my name is")
            if len(parts) > 1:
                name = parts[1].strip().split()[0]
                self._userdata.user_name = name
                save_userdata(self._userdata)
                turn_ctx.add_message(role="assistant", content=f"Noted: user's name set to {name}.")

# Local test routine demonstrating a multi-turn session
async def run_local_test(user_id="user_123"):
    # Load persisted userdata (long-term memory for testing)
    userdata = load_userdata(user_id)

    # Create chat context (short-term session history)
    chat_ctx = ChatContext()

    # Create session with typed userdata
    session = AgentSession[SupportSessionData](llm=llm, userdata=userdata, chat_ctx=chat_ctx)

    # Start the agent instance
    agent = SupportAgent(session_userdata=userdata)
    await session.start(agent)

    # Simulate conversation turns
    turns = [
        "Hi, my name is Alice. I need help with my invoice.",
        "I have ticket TICKET-12345, what's the status?",
        "Also, can you find info about refund policy?"
    ]

    for user_text in turns:
        print(f"\nUser: {user_text}")
        result = await session.run(user_input=user_text)

        # Print assistant messages from events
        for event in result.events:
            if getattr(event, "type", None) == "message" and getattr(event.item, "role", None) == "assistant":
                print("Agent:", event.item.content)

    # At end of session, persist any updated userdata
    save_userdata(session.userdata)
    print("\nSession userdata saved:", asdict(session.userdata))

if __name__ == "__main__":
    asyncio.run(run_local_test())

If you want next: • I can switch the persistence layer to Postgres or Redis and show the DB integration. • I can show a more formal tool call pattern (structured tool returns and automatic tool invocation). • I can add vector-store RAG using a local FAISS example and retrieval before injection. Which would you like next?

I will connect to supabase (postgres) for persistant memory

Great — below is a concise Python example that replaces the local JSON persistence with Supabase (Postgres) for long‑term memory. It shows: • using OpenAI

gpt-4.1-mini

via the LiveKit OpenAI plugin, • session-level

chat_ctx

for multi-turn context, • typed

userdata

persisted to Supabase, • a simple RAG lookup against local KB files (can be swapped for a vector store), • a placeholder tool call pattern. Preconditions: • Install:

livekit-agents

,

livekit-plugins-openai

,

asyncpg

,

supabase-py

(or

supabase

), and dependencies. • Set environment variables for OpenAI and Supabase (URL + API key). • Create a Postgres table for userdata, e.g.: CREATE TABLE agent_userdata (user_id TEXT PRIMARY KEY, data JSONB); Code (adjust connection details, table names, and KB storage as needed):

import asyncio
import json
from dataclasses import dataclass, asdict
from pathlib import Path
from typing import Optional

import asyncpg
from supabase import create_client  # or use supabase-py client of your choice

from livekit.agents import AgentSession, ChatContext, ChatMessage, Agent
from livekit.plugins import openai

# ---------- Config ----------
MODEL_NAME = "gpt-4.1-mini"
KB_DIR = Path("./kb")
SUPABASE_URL = "<https://your-project.supabase.co>"
SUPABASE_KEY = "your-supabase-key"
# ----------------------------

# Initialize LLM plugin
llm = openai.LLM(model=MODEL_NAME)

# Structured session userdata
@dataclass
class SupportSessionData:
    user_id: str
    user_name: Optional[str] = None
    last_ticket_id: Optional[str] = None
    preferences: dict = None

    def __post_init__(self):
        if self.preferences is None:
            self.preferences = {}

# ---------- Supabase helpers ----------
supabase = create_client(SUPABASE_URL, SUPABASE_KEY)

async def load_userdata_from_supabase(user_id: str) -> SupportSessionData:
    # Query the userdata row
    resp = supabase.table("agent_userdata").select("data").eq("user_id", user_id).limit(1).execute()
    # supabase-py returns a tuple-like structure in some versions; adapt if needed
    rows = resp.data if hasattr(resp, "data") else resp[0]
    if rows:
        data = rows[0]["data"] if "data" in rows[0] else rows[0]
        return SupportSessionData(**data)
    return SupportSessionData(user_id=user_id)

async def save_userdata_to_supabase(data: SupportSessionData):
    payload = asdict(data)
    # Upsert row by user_id
    supabase.table("agent_userdata").upsert({"user_id": data.user_id, "data": payload}).execute()

# ---------- Simple RAG and tool ----------
def rag_lookup(query: str, max_files: int = 3) -> str:
    if not KB_DIR.exists():
        return ""
    results = []
    for p in KB_DIR.glob("*.txt"):
        text = p.read_text()
        if query.lower() in text.lower():
            results.append(f"From {p.name}:\n{text[:500]}\n---")
            if len(results) >= max_files:
                break
    return "\n".join(results)

async def ticket_lookup_tool(ticket_id: str) -> str:
    await asyncio.sleep(0.1)
    return f"Ticket {ticket_id} status: OPEN (sample)"

# ---------- Agent with RAG & Supabase-backed userdata ----------
class SupportAgent(Agent):
    def __init__(self, session_userdata: SupportSessionData):
        instructions = (
            "You are a helpful support agent. Use conversation history and stored user data "
            "to personalize replies. If you find KB results, cite snippets. For external lookups, "
            "emit tool invocation intents (host will call tools)."
        )
        super().__init__(instructions=instructions)
        self._userdata = session_userdata

    @property
    def userdata(self) -> SupportSessionData:
        return self._userdata

    async def on_user_turn_completed(self, turn_ctx: ChatContext, new_message: ChatMessage):
        query = new_message.text_content()
        # RAG injection
        rag_content = rag_lookup(query)
        if rag_content:
            turn_ctx.add_message(role="assistant", content=f"KB_SNIPPET:\n{rag_content}")

        # Simple ticket detection & tool call
        for token in query.split():
            if token.startswith("TICKET-"):
                status = await ticket_lookup_tool(token)
                turn_ctx.add_message(role="assistant", content=f"TOOL_RESULT:\n{status}")

        # Naive name capture demo
        if "my name is" in query.lower():
            parts = query.split("my name is")
            if len(parts) > 1:
                name = parts[1].strip().split()[0]
                self._userdata.user_name = name
                # Persist immediately to Supabase
                await save_userdata_to_supabase(self._userdata)
                turn_ctx.add_message(role="assistant", content=f"Noted: user's name set to {name}.")

# ---------- Local test with Supabase persistence ----------
async def run_local_test(user_id="user_123"):
    # Load persisted userdata from Supabase
    userdata = await load_userdata_from_supabase(user_id)

    chat_ctx = ChatContext()
    session = AgentSession[SupportSessionData](llm=llm, userdata=userdata, chat_ctx=chat_ctx)

    agent = SupportAgent(session_userdata=userdata)
    await session.start(agent)

    turns = [
        "Hi, my name is Alice. I need help with my invoice.",
        "I have ticket TICKET-12345, what's the status?",
        "Also, can you find info about refund policy?"
    ]

    for user_text in turns:
        print(f"\nUser: {user_text}")
        result = await session.run(user_input=user_text)
        for event in result.events:
            if getattr(event, "type", None) == "message" and getattr(event.item, "role", None) == "assistant":
                print("Agent:", event.item.content)

    # Save any final userdata
    await save_userdata_to_supabase(session.userdata)
    print("\nSession userdata saved:", asdict(session.userdata))

if __name__ == "__main__":
    asyncio.run(run_local_test())

Next steps I can provide: • a schema and migrations for the Postgres table, • an example using a vector DB (pgvector in Supabase) for RAG retrieval, • a more robust tool invocation pattern (structured tool outputs and automatic tool runner). Which of those would you like next?

what will be the endpoints where i can recieve messages from? and I will need to send message response to a custom endpoint

Thanks — I’ll explain the typical endpoints/mechanisms where an agent can receive user messages and how you can send replies to a custom endpoint, then ask one quick clarifying question so I can give code tailored to your setup. What you can receive messages from • Frontend chat (recommended): user text sent over the lk.chat text stream (your frontend SDK posts messages to the room). The agent listens to that text stream and processes incoming messages. • WebRTC audio → STT: live audio from participants is transcribed and delivered as text to the agent. • External services / webhooks: your external systems can POST user messages into your agent host (see Programmatic Input below). • Programmatic input API: call into the agent runtime (e.g.,

session.generate_reply(user_input=...)

or a

text_input_cb

) to inject messages from other services or queues. How to send agent responses to a custom endpoint • Tool-based HTTP call (best practice): implement a tool that makes an HTTP POST to your custom endpoint whenever the agent decides to forward a reply. Tools are invoked from agent logic or lifecycle hooks. ◦ Example pattern: define an async tool

send_to_webhook(content)

that posts JSON to your endpoint and returns a confirmation string. • Direct outbound HTTP in lifecycle hooks: inside

on_user_turn_completed

or after generating a reply, call your endpoint directly (use

aiohttp

or

httpx

). • Frontend forwarding: let the agent publish replies to

lk.chat

and have your frontend capture those messages and forward them to your endpoint (useful if your endpoint expects activity from the client). • Job/room metadata for dynamic endpoints: pass target endpoint URLs or API keys via

job.metadata

or

room.metadata

so the agent knows where to post. Security notes (important) • Never hardcode API keys/URLs in agent code; pass them via secure metadata or environment variables. • Sign or verify webhook payloads if your endpoint requires authenticity. Quick example sketch (tool approach) • Agent detects a reply to send externally → invokes

send_to_webhook(content)

tool → tool POSTS to your endpoint → agent continues. Clarifying question Which integration flow do you want code for? 1) The agent calls your endpoint directly from the server (tool or lifecycle hook), or 2) Your frontend forwards agent messages to your endpoint, or 3) An external system posts messages into the agent via an HTTP webhook (so I should show how to accept incoming webhooks and inject them into the session)? Tell me 1, 2, or 3 and I’ll provide a concise code example (Supabase + gpt-4.1-mini compatible) for that flow. Sources: Text and transcriptions | LiveKit docs | External data and RAG | LiveKit docs

what will the memory footprint for this asgent? Will it use as much resources as and Voice agent? And what about token cacning?

Thanks — I’ll answer the three parts you asked about: memory footprint (chat vs voice), resource differences, and token caching + practical mitigation steps. Acknowledging: you’re building a support chat agent (gpt-4.1-mini) with tools, RAG, and Supabase persistence. 1) Memory footprint — chat agent vs voice agent • Chat agent: lightweight. It mainly handles text I/O, LLM calls, session state, and any RAG/tool logic. Per‑session RAM is small (typical tens of MBs) since there’s no audio pipeline. • Voice agent: heavier. STT/TTS, audio buffering, VAD, and real‑time media pipelines add substantial memory and CPU (can be hundreds of MBs per active voice session). • Conclusion: your chat support agent will use far fewer resources than a comparable voice agent. 2) Other resource differences • CPU: chat agent spends CPU on LLM calls and any RAG retrievals; voice agents also need continuous audio encoding/decoding and STT/TTS compute. • Network: chat sends/receives text and LLM payloads. Voice also streams audio and may send audio chunks to STT services, increasing bandwidth. • Startup / concurrency: voice sessions often require longer-lived processes and lower concurrency per host; chat sessions are more scalable horizontally. • Persistence (Supabase): DB reads/writes add I/O but are small compared to LLM request cost. Use batched or async writes to avoid blocking. 3) Token caching, prompt size, and mitigation strategies • Token usage: each LLM request includes prompt tokens (system + messages + injected RAG snippets). Large conversation history or large KB snippets increase token count and cost. • Token caching: there’s no automatic global token cache for dynamic user context. Use these strategies: ◦ Summarize older turns: periodically compress older chat history into a short summary and store that summary in user memory; include the summary instead of full history in prompts. ◦ Selective context windowing: keep only the most recent N turns + key facts from

userdata

when building prompts. ◦ Store extracted facts in structured

userdata

(Supabase) and pass only those facts into prompts rather than full raw history. ◦ Precompute RAG retrieval results for frequent queries or index KB with pgvector so you retrieve concise passages rather than entire docs. ◦ Prewarm static context: load static instructions or non-user-specific data at start rather than embedding repeatedly in every prompt. ◦ Rate-limit or debounce LLM calls from rapid user typing to avoid duplicate requests. Practical implementation tips for your stack • Persist key facts to Supabase (JSONB). On each turn, read only needed fields; avoid rehydrating full conversation unless required. • Use pgvector in Supabase for RAG so retrieval returns short, relevant snippets — reduces tokens. • Implement an async write pattern: update Supabase in background after sending reply to avoid latency spikes. • Summarization workflow: after N messages, call the LLM to summarize and store summary in Supabase; trim chat_ctx accordingly. • If you need per-host capacity planning: estimate memory per chat session in tens of MBs, per voice session up to hundreds of MBs — run load tests to tune. If you want, I can: • Provide a code snippet to summarize and compress conversation history into Supabase. • Show how to use pgvector in Supabase for RAG and return short passages. • Suggest a simple capacity estimate formula based on expected concurrent sessions. Which of those would you like next? Sources: examples/voice_agents | livekit/livekit_composite | DeepWiki | External data and RAG | LiveKit docs