Integrations

Persistent Memory for Pipecat Voice Agents - Unison Integration

Give Pipecat voice agents persistent memory: recall before the call primes the system prompt, the transcript is ingested when the call ends - via the pipecat-unison integration.

pipecat-unison

Learn how to give Pipecat voice agents persistent team memory with Unison.

Voice agents forget everything between calls unless you persist the transcript and recall it next time. The pipecat-unison integration wires both ends into a Pipecat pipeline: recall primes the system prompt before the conversation starts, and the transcript is ingested when the call ends. The pattern below is what it does under the hood.

A tiny Unison client

import httpx, os

BRAIN = os.environ.get("UNISON_API_URL", "https://brain.unisonlabs.ai")
HEADERS = {"Authorization": f"Bearer {os.environ['UNISON_TOKEN']}"}

def recall(query: str) -> str:
    ctx = httpx.get(f"{BRAIN}/v1/brain/context", params={"q": query}, headers=HEADERS).json()
    return "" if ctx["weakEvidence"] else ctx["contextMd"]

def remember(turns: list[dict], source_ref: str) -> None:
    httpx.post(f"{BRAIN}/v1/brain/ingest", headers=HEADERS, json={
        "items": [{"type": "conversation", "turns": turns, "sourceRef": source_ref, "visibility": "private"}]
    })

Prime the prompt, ingest the transcript

from pipecat.pipeline.pipeline import Pipeline
from pipecat.pipeline.task import PipelineTask
from pipecat.services.openai import OpenAILLMService, OpenAILLMContext

caller_id = "+15551234567"
memory = recall(f"previous calls and preferences for {caller_id}")

context = OpenAILLMContext(messages=[
    {"role": "system", "content": f"You are a helpful voice assistant.\n\nWhat we already know:\n{memory}"},
])

llm = OpenAILLMService(api_key=os.environ["OPENAI_API_KEY"])
pipeline = Pipeline([transport.input(), stt, context_aggregator.user(), llm, tts, transport.output()])
task = PipelineTask(pipeline)

# when the call ends, persist the transcript so the next call recalls it
@transport.event_handler("on_client_disconnected")
async def on_call_end(_transport, _client):
    turns = [m for m in context.get_messages() if m["role"] in ("user", "assistant")]
    remember(turns, source_ref=f"call-{caller_id}")

recall primes the system prompt before the first turn; remember ingests the transcript when the caller hangs up. For caller-scoped memory — each phone number remembers its own history while the workspace shares product knowledge — use a brain:act-as service key with the caller id as actor (details). The pipecat-unison repo ships the maintained, drop-in version of this wiring.

On this page