The context assembler

1 min read
On trails:AGCTXstep 4/11

A context assembler is the code that builds the prompt fresh on every loop of the agent from the five context typesAnatomy of contextFive kinds of context, told apart by where they come from and how long they live.Opens in a new tab. Nothing is carried over by accident: each loop decides again what goes in.

Five sources of context (static, dynamic, episodic, retrieved, stateful) feed the context assembler, which builds the prompt for one model call. After the call, the loop starts again at the assembler. 1 Static: persona, rules 2 Dynamic: the trigger 3 Episodic: fetch, truncate 4 Retrieved: just in time 5 Stateful: query the store Context assembler Model call one window every loop starts here again
The same five steps, every loop.

For every loop it grabs the static context, grabs the dynamic one (the trigger), fetches the episodic history and truncates it, retrieves context for this query, and queries for the stateful record.

In LangGraph

Here the assembler is a plain function. Everything the model sees is in the one list it returns, which makes it the one place to look when an answer is wrong. The history is truncated to the last ten messages, so what falls off is a decision made here (see overflow managementContext overflowWhen the history outgrows the window: cut the oldest, or summarize them.Opens in a new tab).

Python
from langchain_core.messages import SystemMessagefrom langgraph.graph import StateGraph, MessagesState, START, END STATIC = "You are a billing support agent. Never reveal internal notes." class State(MessagesState):    user_id: str def assemble(state: State) -> list:    query = state["messages"][-1]                       # dynamic: what triggered this loop    history = state["messages"][:-1][-10:]              # episodic: fetched, then truncated    docs = retrieve(query.content)                      # retrieved: fetched just in time    profile = load_profile(state["user_id"])            # stateful: pulled from a store    return [        SystemMessage(STATIC),                          # static: same every time        SystemMessage(f"Profile: {profile}\n\nDocs:\n{docs}"),        *history,        query,    ] def call_model(state: State) -> dict:    return {"messages": [model.invoke(assemble(state))]} graph = (    StateGraph(State)    .add_node("call_model", call_model)    .add_edge(START, "call_model")    .add_edge("call_model", END)    .compile())

Because it is ordinary code, the assembler can be tested without calling a model: feed it a state, check the list. Long documents sit before the query, as Claude’s docs advise for large inputs: data first, question last (see rerankingContext rerankingModels read the middle of a long context worst, so put what matters at the edges.Opens in a new tab).

Where it matters

In LangChain’s create_agent the same job is done by middleware that edits the request just before each model call. Whichever way it is wired, log what was assembled (see traceabilityTraceabilityLog the whole reasoning trace, so any answer can be explained afterwards.Opens in a new tab): an agent that misbehaves is usually an agent that was shown the wrong thing.

Sources