Pipecat + ContextDB

Check confirmed customer memory inside Pipecat booking functions.

Pipecat provides composable voice pipelines with function calling on the LLM service. Register recall and action-decision handlers there, keep identity in authenticated pipeline state, then write sourced facts when the pipeline ends.

Register a recall function and an evaluate_action function on your LLM service, then close each decision with report_execution. recall_for_action is a compatibility retrieval helper, not host authorization. Examples use the public Cloud client against ContextDB Cloud. The local engine uses pycontextdb. This independent guide has no vendor partnership or endorsement.

Credential-free simulation

Run the complete booking flow locally.

The simulation uses the real Pipecat handlers with mocked ContextDB transport. It shows recall, ask, confirmation, re-evaluation, booking, and terminal memory formation in order.

git clone https://github.com/atomsai/contextdb-pipecat-example.git
cd contextdb-pipecat-example
python -m venv .venv
source .venv/bin/activate
python -m pip install -e ".[dev]"
python -m contextdb_pipecat.simulate

Inspect the runnable example and tests → Try a memory scenario in Playground →

In the pipeline

Check the customer's current details inside the booking function.

The model calls book_visit when the conversation sounds like a booking. The handler derives the caller from authenticated pipeline state, checks host permissions and the current slot, then handles every ContextDB outcome. The model speaks the returned result.

# pipecat: register memory functions on the llm
from pipecat.services.llm_service import FunctionCallParams
from contextdb_cloud_client import CloudClient

def register_booking_function(llm, session):
    identity = require_authenticated_pipeline_session(session)
    caller = identity.customer_id

    async def book_visit(params: FunctionCallParams):
        day = params.arguments["day"]
        slot = await calendar.get_slot(day)
        authorized = await calendar.can_book(
            actor_id=identity.actor_id,
            customer_id=caller,
            slot=slot,
        )
        if not authorized:
            await params.result_callback({"status": "forbidden"})
            return

        async with CloudClient(BASE_URL, api_key=KEY) as cdb:
            decision = await cdb.evaluate_action(
                caller, f"book service visit on {day}"
            )

            if decision.outcome == "act":
                try:
                    booking = await calendar.book(
                        caller, day, expected_version=slot.version
                    )
                except SlotChanged:
                    await cdb.report_execution(
                        caller, decision.decision_id, "appointment.book", "failed",
                        idempotency_key=f"pipecat-receipt-{decision.decision_id}",
                        error_code="current_state_changed",
                    )
                    await params.result_callback({"status": "state_changed"})
                    return
                await cdb.report_execution(
                    caller, decision.decision_id, "appointment.book", "succeeded",
                    idempotency_key=f"pipecat-receipt-{decision.decision_id}",
                    external_ref=booking.ref,
                )
                await params.result_callback(
                    {"status": "booked", "reference": booking.ref}
                )
                return

            if decision.outcome == "ask":
                await cdb.report_execution(
                    caller, decision.decision_id, "appointment.book", "skipped",
                    idempotency_key=f"pipecat-receipt-{decision.decision_id}",
                )
                await params.result_callback(
                    {"status": "needs_confirmation",
                     "say": f"Please confirm that I should book {day}."}
                )
                return

            if decision.outcome == "abstain":
                await cdb.report_execution(
                    caller, decision.decision_id, "appointment.book", "skipped",
                    idempotency_key=f"pipecat-receipt-{decision.decision_id}",
                )
                await params.result_callback({"status": "abstained"})
                return

            raise RuntimeError("unknown ContextDB action outcome")

    llm.register_function("book_visit", book_visit)

After the run

Save a confirmed detail when the caller says yes.

Bind the caller's explicit yes to one pending memory ID. The confirm call records that attestation. It does not authorize a booking, and the later booking still needs a fresh action decision and host checks.

def register_confirmation_function(llm, session):
    identity = require_authenticated_pipeline_session(session)
    caller = identity.customer_id

    async def confirm_day(params: FunctionCallParams):
        attestation = await require_spoken_attestation(session, params)
        pending = await confirmation_store.get(
            caller, attestation.pending_memory_id
        )
        if pending is None:
            await params.result_callback({"status": "nothing_pending"})
            return
        async with CloudClient(BASE_URL, api_key=KEY) as cdb:
            await cdb.confirm(
                caller,
                pending.memory_id,
                idempotency_key=f"pipecat-confirm-{pending.memory_id}",
            )
        await params.result_callback({"status": "confirmed"})

    llm.register_function("confirm_day", confirm_day)

async def on_pipeline_end(session):
    identity = require_authenticated_pipeline_session(session)
    user_turn = require_user_turn(session.final_turn)
    async with CloudClient(BASE_URL, api_key=KEY) as cdb:
        await cdb.remember(
            identity.customer_id,
            user_turn.content,
            source="user_stated",
            confidence=0.95,
            idempotency_key=f"pipecat-{session.id}-final-user-turn",
        )

Pipecat pipeline order

Add memory at four points in the Pipecat pipeline.

Moment	Pipecat surface	ContextDB call
Pipeline starts	Before the context aggregator	`recall` → caller snapshot in system prompt
Consequential function call	Registered function handler	`evaluate_action` → branch, then `report_execution`
Caller says yes	Confirmation function handler	Host authenticates and retains the end-user attestation, then `confirm` records the scoped memory update
Pipeline ends	End-of-run handler	`remember` with source and confidence

Pipecat implementation

Where memory belongs in a Pipecat pipeline.

Should memory be a frame processor instead of a function?

Recall for conversational grounding can be a processor that enriches context frames. The action gate should stay inside the function handlers guarding consequential calls, because that is where the decision belongs and where the ask can be spoken naturally.

Does this couple my pipeline to ContextDB?

The coupling is two registered functions and one HTTPS client. The same trust model also runs locally through pycontextdb, so pipelines can develop offline before using Cloud. See open SDK vs Cloud.

What about interruptions and mid-call corrections?

Corrections are new statements: store them with their own source and confidence. The gate reads current evidence at action time, so a correction made ten seconds ago is what the booking checks against.

Run the Pipecat memory example.

The linked example covers conversational recall and sourced writes. The authenticated, host-owned evaluate_action and report_execution boundary is shown on this page for you to add.

Run the Pipecat example