Short-Term Memory with LangChain

Let the agent continue one conversation without forcing the client to replay every previous turn.

This recipe keeps the LangGraph orchestration from the prior recipe in this series and adds the next production capability: short-term memory inside one conversation thread.

The boundary matters. Short-term memory is not a user profile and it is not durable personalization. It is the recent thread state that lets a user ask “what about the second one?” and get an answer that knows what “the second one” refers to.

Durable memory is a separate concern, covered in a later recipe in this series. This cookbook intentionally uses process-local memory so the first memory concept is easy to inspect, test, and reset.

What you will buildWhat you will build

A production-shape FastAPI service that extends the orchestrated book agent from the prior recipe:

• Inherited orchestration: the direct / deliberate LangGraph route still chooses the response path.
• Thread identity: every agent call includes a stable thread_id.
• Short-term state: the service remembers recent user and assistant turns for that thread.
• Bounded context: only the most recent messages are injected into the next Nebius call.
• Reset: DELETE /threads/{thread_id} clears local thread state.
• Network-free tests: tests mock Nebius with respx.

POST /agent/run ──► thread_id ──► load recent turns ──► LangGraph route ──► Nebius stream ──► SSE
                         ▲                                                        │
                         └──────────────── save user + assistant turn ◄───────────┘

DELETE /threads/{thread_id} ──► clear local thread state

PrerequisitesPrerequisites

• Python 3.12+
• uv
• A Nebius API key from the Nebius console
• Docker (optional)

Run the cookbookRun the cookbook

cp .env.example .env
# Open .env and fill NEBIUS_API_KEY.

While .env is open, replace the two Nebius defaults. The repo’s .env.example still ships the older api.studio.nebius.ai hostname and a chat model that Token Factory no longer serves, so the first request fails without this:

NEBIUS_BASE_URL=https://api.tokenfactory.nebius.com/v1/
NEBIUS_MODEL=Qwen/Qwen3.5-397B-A17B

Then install and start:

uv sync
make dev

Start a thread:

curl -N -X POST http://localhost:8000/agent/run \
  -H 'content-type: application/json' \
  -d '{"thread_id":"demo-thread","prompt":"Recommend three books about product strategy."}'

Follow up with the same thread_id:

curl -N -X POST http://localhost:8000/agent/run \
  -H 'content-type: application/json' \
  -d '{"thread_id":"demo-thread","prompt":"Which of those is best for a founder?"}'

The second request receives the first turn as thread context. The SSE stream includes memory status:

event: status
data: {"phase":"memory_loaded","threadId":"demo-thread","messages":2}

event: status
data: {"phase":"routed","route":"deliberate","contextNeed":"curated_recommendation"}

event: token
data: {"text":"..."}

event: status
data: {"phase":"memory_saved","threadId":"demo-thread","messages":4}

Reset the thread:

curl -X DELETE http://localhost:8000/threads/demo-thread

Walk-throughWalk-through

API contractAPI contract

POST /agent/runPOST /agent/run

Runs the agent and streams named SSE events.

Request:

{
  "thread_id": "demo-thread",
  "prompt": "Which of those is best for a founder?",
  "temperature": 0.4,
  "max_tokens": 1024,
  "history": []
}

Fields:

SSE events:

DELETE /threads/{thread_id}DELETE /threads/{thread_id}

Clears process-local memory for a thread.

Response:

{
  "threadId": "demo-thread",
  "deleted": true
}

How it worksHow it works

The request model makes thread_id required. That is the only API shape change from the prior recipe in this series.

class AgentRunRequest(BaseModel):
    """Payload for POST /agent/run."""

    thread_id: str = Field(..., min_length=1, max_length=120, pattern="^[A-Za-z0-9_.:-]+$")
    prompt: str = Field(..., min_length=1, max_length=8_000)
    temperature: float = Field(default=0.4, ge=0.0, le=2.0)
    max_tokens: int = Field(default=1024, ge=1, le=8192)
    history: list[ChatHistoryMessage] = Field(default_factory=list, max_length=12)

The memory store is small by design. It keeps recent messages per thread, trims old messages, and returns copies so route handlers cannot mutate internal state accidentally.

@dataclass
class ThreadMemoryStore:
    max_messages_per_thread: int = 12
    _threads: dict[str, list[dict[str, str]]] = field(default_factory=dict)
    _lock: asyncio.Lock = field(default_factory=asyncio.Lock)

    async def get_history(self, thread_id: str) -> list[dict[str, str]]:
        async with self._lock:
            return list(self._threads.get(thread_id, []))

    async def append_turn(self, thread_id: str, *, user: str, assistant: str) -> int:
        async with self._lock:
            messages = self._threads.setdefault(thread_id, [])
            messages.extend(
                [
                    {"role": "user", "content": user},
                    {"role": "assistant", "content": assistant},
                ]
            )
            del messages[: max(0, len(messages) - self.max_messages_per_thread)]
            return len(messages)

The route composes stored thread memory with any client-provided history before invoking the inherited LangGraph agent.

stored_history = await memory.get_history(payload.thread_id)
history = [*stored_history, *(item.model_dump() for item in payload.history)]

async for event in agent.run(
    payload.prompt,
    options=AgentRunOptions(
        temperature=payload.temperature,
        max_tokens=payload.max_tokens,
        history=history,
    ),
    cancel_event=cancel_event,
):
    yield _sse(event.name, event.data)

After the stream completes, the route saves the new turn. It stores only the assistant answer text, not the operational metrics footer.

retained = await memory.append_turn(
    payload.thread_id,
    user=payload.prompt,
    assistant="".join(assistant_chunks).strip(),
)

yield _sse(
    "status",
    {
        "phase": "memory_saved",
        "threadId": payload.thread_id,
        "messages": retained,
    },
)

The inherited prompt builder already knows how to use recent history. If history exists, it wraps the current prompt with conversation context:

if history:
    recent = history[-6:]
    context = "\n".join(
        f"{item['role']}: {item['content'][:800]}"
        for item in recent
        if item.get("content")
    )
    prompt = (
        "Recent conversation context:\n"
        f"{context}\n\nCurrent user request:\n{prompt}\n\n"
        "Resolve references like 'that topic' from the recent conversation."
    )

Memory stays outside the FastAPI transport layer, and the route stays explicit about when memory is loaded and saved.

ConfigurationConfiguration

Design decisionsDesign decisions

Thread memory is scoped by thread_id. The server does not try to infer continuity from IP address, cookies, or prompt content. The client or application layer owns the conversation id.

Memory is process-local in this recipe. That keeps the first memory cookbook runnable in minutes and makes the concept easy to inspect. It also means memory disappears on restart and is not shared across replicas. A later recipe in this series introduces durable user memory with Postgres.

Recall is bounded. The store keeps a maximum of 12 messages per thread and the prompt builder uses only recent history. Unbounded transcript replay is a hidden latency and cost bug.

The route emits memory phases. memory_loaded and memory_saved make memory behavior observable in the SSE stream and web playground.

Failure modesFailure modes

Test itTest it

uv run pytest
uv run ruff check
uv run ruff format --check

The tests mock the Nebius streaming endpoint with respx. They verify that a second request with the same thread_id receives stored context and that DELETE /threads/{thread_id} clears it.

Ship itShip it

Before running this in production:

• Derive thread_id from an authenticated session or server-created conversation id.
• Do not let users read or delete arbitrary thread ids without authorization.
• Move thread state to a shared store before running multiple replicas.
• Add summarization once long threads exceed the prompt budget.
• Keep memory writes after successful model completion so failed runs do not pollute context.
• Redact sensitive content before storing thread messages if your product accepts PII.

Going furtherGoing further

• Add summarization once thread state grows beyond the context budget.
• Store checkpoints in Postgres for durable thread continuity.
• Derive thread_id from an authenticated session rather than trusting arbitrary client input.
• Add UI affordances for starting and clearing conversations.
• See the next recipe in this series, on long-term memory, for durable user memory across threads.
• LangChain short-term memory docs