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. The application addresses the graph with user_id.
  • 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 graph_id.

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

Set up the user, thread, and account graph

Use stable identifiers from your application. This example uses one user, one conversation thread, and one account Context Graph.

import json
import os
import time
from zep_cloud.client import Zep
from zep_cloud.types import Message
client = Zep(api_key=os.environ["ZEP_API_KEY"])
USER_ID = "user-alice"
THREAD_ID = "support-case-8472"
ACCOUNT_ID = "acme"
ACCOUNT_GRAPH_ID = f"account-{ACCOUNT_ID}"
client.user.add(
    user_id=USER_ID,
    first_name="Alice",
    last_name="Smith",
    email="[email protected]",
)
client.thread.create(thread_id=THREAD_ID, user_id=USER_ID)
client.graph.create(graph_id=ACCOUNT_GRAPH_ID)

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.add(
    graph_id=ACCOUNT_GRAPH_ID,
    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.add(
    graph_id=ACCOUNT_GRAPH_ID,
    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},
)
account_episode = client.graph.add(
    graph_id=ACCOUNT_GRAPH_ID,
    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"},
)

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(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(
        graph_id=ACCOUNT_GRAPH_ID,
        query="Which account uses regional failover?",
        scope="edges",
        search_filters={"episode_uuids": [account_episode.uuid_]},
        limit=10,
    )
    if account_results.edges:
        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(
        THREAD_ID,
        messages=[
            Message(name="Alice Smith", role="user", content=user_message),
        ],
        return_context=True,
    )
    account_results = client.graph.search(
        graph_id=ACCOUNT_GRAPH_ID,
        query=user_message,
        scope="edges",
        limit=10,
    )
    account_context = "\n".join(
        edge.fact for edge in (account_results.edges or [])
    )
    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