Skip to navigation
Get Started

Unify customer and account context

Combine personal context with shared account records, events, documents, and conversations

A customer task can require personal context and information that belongs to the customer’s account. Keep each source in its correct scope:

  • A user graph stores a person’s conversations, activity, and preferences. Zep creates the user graph when you create the user. The application addresses the graph with the graph_uuid from the user create response.
  • A shared account Context Graph stores account records and support events. The Context Graph also stores contracts, product documents, and other shared data. The application addresses the graph with the uuid from the graph create response.

This recipe creates both scopes and retrieves from them for one authorized support task.

Set up the user, thread, and account graph

Zep generates the UUID of each user, thread, and graph. Store each UUID in your application database next to your own identifier, for example in zep_user_uuid, zep_thread_uuid, and zep_account_graph_uuid columns. Use the stored UUIDs on all later calls. This example uses one user, one conversation thread, and one account Context Graph.

To protect a create call against retries, you can send an idempotency key with the call. See Create a graph.

import json
import os
import time
from zep_cloud.client import Zep
from zep_cloud.types import AddMessage, SearchFilters
client = Zep(api_key=os.environ["ZEP_API_KEY"])
APP_USER_ID = "user-alice"
ACCOUNT_ID = "acme"
user = client.user.create(
    first_name="Alice",
    last_name="Smith",
    email="[email protected]",
    metadata={"app_user_id": APP_USER_ID},
)
zep_user_uuid = user.uuid_
thread = client.thread.create(user_uuid=zep_user_uuid)
zep_thread_uuid = thread.uuid_
account_graph = client.graph.create(
    name=f"Account {ACCOUNT_ID}",
    description="Shared account records, support events, and product documents",
)
zep_account_graph_uuid = account_graph.uuid_

Add shared account sources

Add each shared source to the account Context Graph. The example combines an account record, a support event, and a product document.

client.graph.episode.add(
    zep_account_graph_uuid,
    type="json",
    data=json.dumps({
        "account_id": ACCOUNT_ID,
        "plan": "Enterprise",
        "region": "eu-west",
        "renewal_date": "2026-11-15",
    }),
    source_description="Account record from the CRM",
    metadata={"source": "crm", "account_id": ACCOUNT_ID},
)
client.graph.episode.add(
    zep_account_graph_uuid,
    type="json",
    data=json.dumps({
        "case_id": "CASE-8472",
        "status": "investigating",
        "service": "Atlas Gateway",
        "event": "Regional failover validation failed",
    }),
    source_description="Support case event",
    metadata={"source": "support", "account_id": ACCOUNT_ID},
)
product_doc = client.graph.episode.add(
    zep_account_graph_uuid,
    type="text",
    data=(
        "Atlas Gateway Enterprise accounts can use regional failover. "
        "Support must confirm backup-region validation before a cutover."
    ),
    source_description="Atlas Gateway support guide",
    metadata={"source": "product_docs", "product": "atlas-gateway"},
)
account_episode_uuid = product_doc.episode.uuid_

Wait until the account context is searchable

Zep processes episodes asynchronously. Poll the last submitted account episode, and then retry a query that is specific to the imported data. The ingestion status guide explains the processing order and production alternatives to polling.

deadline = time.monotonic() + 300
while True:
    episode = client.graph.episode.get(zep_account_graph_uuid, account_episode_uuid)
    if episode.processed:
        break
    if time.monotonic() >= deadline:
        raise TimeoutError("The account context did not finish processing.")
    time.sleep(5)
search_deadline = time.monotonic() + 300
while True:
    account_results = client.graph.search_edges(
        zep_account_graph_uuid,
        query="Which account uses regional failover?",
        filters=SearchFilters(episode_uuids=[account_episode_uuid]),
        limit=10,
    )
    if account_results.items:
        break
    if time.monotonic() >= search_deadline:
        raise TimeoutError("The account context is not searchable.")
    time.sleep(5)

Retrieve personal and account context

Authorize the account in your application before you request its Context Graph. Then record the current user message, retrieve its Context Block in the same request, and search the authorized account graph.

def retrieve_support_context(
    user_message: str,
    authorized_account_ids: set[str],
) -> dict[str, str]:
    if ACCOUNT_ID not in authorized_account_ids:
        raise PermissionError("The user is not authorized for this account.")
    memory_response = client.thread.add_messages(
        zep_thread_uuid,
        messages=[
            AddMessage(name="Alice Smith", role="user", content=user_message),
        ],
        return_context=True,
    )
    account_results = client.graph.search_edges(
        zep_account_graph_uuid,
        query=user_message,
        limit=10,
    )
    account_context = "\n".join(
        edge.fact for edge in (account_results.items or []) if edge.fact
    )
    return {
        "user_context": memory_response.context,
        "account_context": account_context,
    }

Put the two context values in the model request as untrusted reference data. Keep stable application instructions in a higher-priority message. Do not put retrieved context in a developer or system message.

After the model returns a response, add the assistant message to the thread so that later tasks can retrieve it as agent memory.

Apply governance

Your application must validate the relationship between the user and the account. You can also use policy-based access control to limit an API key to permitted account graphs and sources.

Retain the source episode references from account search results when you must trace retrieved facts to their sources.

Next steps