Voice Memory Agent

A voice-first AI agent stress test with long-term memory that chains STT, memory search, LLM, TTS, memory store, and graph update in a 12-step pipeline exercising 7 distinct integration categories: Deepgram STT, AssemblyAI STT, Zep memory, Mem0 memory, Neo4j graph, ElevenLabs TTS, and PlayHT TTS.

Environment variables

This example requires OPENAI_API_KEY, WAXELL_API_KEY, and WAXELL_API_URL. Use --dry-run to skip real API calls.

Architecture

Key Code

STT Provider Selection with @decision

Weighted scoring picks the best STT provider using confidence and feature scores.

@waxell.decision(name="stt_provider_selection", options=["deepgram", "assemblyai"])
def select_stt_provider(dg_confidence: float, aai_confidence: float) -> dict:
    dg_weighted = dg_confidence * 0.6 + 0.85 * 0.4   # word-level timestamps
    aai_weighted = aai_confidence * 0.6 + 0.90 * 0.4  # sentiment + chapters
    chosen = "deepgram" if dg_weighted >= aai_weighted else "assemblyai"
    return {"chosen": chosen, "reasoning": f"Weighted: deepgram={dg_weighted:.3f}, assemblyai={aai_weighted:.3f}"}

Multi-Source Memory Retrieval

Three @retrieval-decorated functions gather context from Zep, Mem0, and Neo4j.

@waxell.retrieval(source="zep")
def search_zep_memory(query: str, zep_client, session_id: str) -> list[dict]:
    results = zep_client.memory.search(session_id=session_id, query=query)
    return [{"text": r.message.content[:200], "score": r.score} for r in results[:5]]

@waxell.retrieval(source="mem0")
def search_mem0_memory(query: str, mem0_client, user_id: str) -> list[dict]:
    results = mem0_client.search_memory(query=query, user_id=user_id)
    return [{"text": m["data"], "score": m.get("score", 0)} for m in results[:5]]

@waxell.retrieval(source="neo4j")
def search_neo4j_graph(query: str, neo4j_driver) -> list[dict]:
    graph_result = neo4j_driver.run_query(query)
    return [{"text": f"{n.labels[0]}:{n.properties['name']}", "id": n.id} for n in graph_result.nodes]

What this demonstrates

• 12-step voice pipeline -- STT, memory search, LLM generation, TTS synthesis, memory store, and graph update in one trace.
• 7 integration categories -- Deepgram, AssemblyAI, Zep, Mem0, Neo4j, ElevenLabs, and PlayHT each with appropriate tool_type.
• @retrieval across 3 sources -- Zep conversation history, Mem0 user facts, and Neo4j knowledge graph.
• @decision with weighted scoring -- STT provider selection using confidence and feature weights.
• Memory write-back -- new conversation turns stored to Zep, user facts to Mem0, and entities to Neo4j.
• Stress test -- generates rich, multi-integration telemetry data for dashboards and policy evaluation.

Run it

# Dry-run mode (no API key needed)
cd dev/waxell-dev
python -m app.demos.voice_memory_agent --dry-run

# Live mode
export OPENAI_API_KEY="sk-..."
python -m app.demos.voice_memory_agent

Source

dev/waxell-dev/app/demos/voice_memory_agent.py