Pipecat + ContextDB
Give Pipecat voice agents memory across every call.
Pipecat gives you composable voice pipelines with function calling on the LLM service. ContextDB adds the two functions every consequential agent needs: one that recalls the caller's context, and one that refuses to act on an unconfirmed maybe.
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 checks trusted memory first and returns
either the booking or the question to ask. The model speaks whichever
comes back.
# pipecat: register memory functions on the llm
from pipecat.services.llm_service import FunctionCallParams
from contextdb_cloud_client import CloudClient
async def book_visit(params: FunctionCallParams):
day = params.arguments["day"]
async with CloudClient(BASE_URL, api_key=KEY) as cdb:
evidence = await cdb.recall_for_action(
CALLER_ID, f"book service visit on {day}"
)
if not evidence:
await params.result_callback(
{"status": "needs_confirmation",
"say": "I don't have a confirmed day on file. "
f"Should I book {day}?"}
)
return
booking = await calendar.book(CALLER_ID, day)
await params.result_callback(
{"status": "booked", "reference": booking.ref}
)
llm.register_function("book_visit", book_visit)
After the run
Save a confirmed detail when the caller says yes.
A yes in the conversation should do two things: confirm the pending fact so future calls can act on it, and record the decision with evidence. Both happen in one call.
async def confirm_day(params: FunctionCallParams):
async with CloudClient(BASE_URL, api_key=KEY) as cdb:
pending = await cdb.pending_confirmations(CALLER_ID)
target = next(
(m for m in pending
if params.arguments["day"] in m.content), None
)
if target:
# graduates the fact; recorded in the decision log
await cdb.confirm(CALLER_ID, target.id)
await params.result_callback({"status": "confirmed"})
return
await params.result_callback({"status": "nothing_pending"})
llm.register_function("confirm_day", confirm_day)
# at pipeline end: store durable facts, honestly sourced
await cdb.remember(
CALLER_ID, "gate code is 4417 for technician access",
source="user_stated", confidence=0.95,
)
Action map
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 | recall_for_action → act, ask, or abstain |
| Caller says yes | Confirmation function handler | confirm → fact graduates, decision recorded |
| Pipeline ends | End-of-run handler | remember with source and confidence |
Questions
Pipecat-specific questions.
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 the Apache-2.0 SDK, so pipelines can develop offline and deploy hosted. 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.
Add two memory functions to your Pipecat agent.
The quickstart covers the recall, gate, and confirm calls above.