Build a Per-Person Context Assembler: fuse email, chat, and meeting transcripts

Level: intermediate
Time to complete: 35 min
Prerequisites: 4

What you will buildThe same fusion stack the enterprise systems ship - registry, router, rank fusion, rerank, dedup, fact currency - shrunk to one function. About 35 minutes.

What You Will Build

You will build one function that, given a single correspondent, assembles one grounded context from three different sources - their email history, the chat corpora where they appear, and the transcripts of meetings held with them - and hands it to a drafting step.

This is not a toy. It is the same fusion stack that Azure AI Search agentic retrieval, Microsoft 365 Copilot, and the Zep memory engine use, shrunk to something you can hold in your head: a registry of per-source retrievers, a router, rank fusion, a reranker, a dedup pass, and a fact-currency layer. By the end you will see a draft built from the whole relationship surface a commitment from a meeting months ago that the live email thread never mentioned.

The code here is illustrative pseudocode in Python. It is written to be read, not pasted into production - the retriever, embedding, and reranker calls are stubs you wire to whatever stack you already run.

Prerequisites

An embedding model and an LLM you can call, and a vector store you can query per source.
Three sources of text about a person that you can search: an email store, a chat store, and a set of meeting transcripts.
A way to persist a small facts table per correspondent (any database row or JSON file will do).
Comfort reading Python-shaped pseudocode. Nothing here depends on a specific framework.

Tip

If you only have one source today, build the whole pipeline anyway with that one source registered. The entire point is that adding the second and third source later is a registration, not a rewrite. Starting with one is the correct first step, not a compromise.

Mental Model

You are not building one big index. You are building a registry of per-source retrievers and a fusion pipeline on top of them. Every source is a plug behind one uniform interface. The assembler never knows or cares which source it is talking to, so adding a new source later means registering one more retriever, not touching the assembler.

That is the pluggable-provider idea, and it is exactly what the enterprise systems ship as connectors and knowledge sources. The two things that separate a contemporary assembler from a naive one are a router (do not query every source every time) and a currency layer (when sources disagree about a person, resolve to the fact that is true now). Everything in between - fusion, rerank, dedup - is about turning three overlapping piles of text into one clean, non-repeating context.

Steps

Eight steps. The first two are free; the fusion and currency logic unlock with Supporter. Build them in order - each stage consumes the output of the one before it.

Register each source behind one interface

Define a single retriever shape and implement it once per source. The assembler depends only on the shape, never on the source.

# every provider returns the same record shape
# retrieve(correspondent, query, k) -> [ {text, source, timestamp, score, msg_id} ]

class Retriever:
    source = "base"
    def retrieve(self, correspondent, query, k):
        raise NotImplementedError

class EmailRetriever(Retriever):
    source = "email"
    def retrieve(self, correspondent, query, k):
        return email_index.search(person=correspondent, query=query, k=k)

REGISTRY = [EmailRetriever(), ChatRetriever(), TranscriptRetriever()]

You know it worked when you can loop for r in REGISTRY and every one returns the same record shape, so the caller cannot tell email from transcript.

One uniform interface, three source plugs behind a registry. The assembler never knows which source it is talking to.

Route, do not blindly fan out

The naive version queries all three sources every time. The contemporary version asks a cheap router which sources are even worth hitting for this person and this query. A published router cut up to eighty percent of the traffic and half the latency at equal accuracy versus querying everything.

For a first build, fan out to all registered sources and leave the router as a clearly marked seam. It starts to matter at fleet scale, not on one correspondent.

def select_sources(correspondent, query, registry):
    # TODO: replace with a learned router at fleet scale (see RAGRoute).
    # For one correspondent, fan-out to all registered sources is fine.
    return registry

def gather(correspondent, query, k=20):
    chosen = select_sources(correspondent, query, REGISTRY)
    return {r.source: r.retrieve(correspondent, query, k) for r in chosen}

You know it worked when gather returns one ranked list per selected source, keyed by source name.

What You Will Build

Prerequisites

Mental Model

Steps

1Register each source behind one interface

2Route, do not blindly fan out

Sources

Get the next tutorial as it ships.

Register each source behind one interface

Route, do not blindly fan out