Recipe: Session Memory¶

Goal¶

Maintain conversation context across turns with AgentSession, using provider-native message history for full tool-call context preservation.

Basic example¶

from doc_processing_toolkit import Agent

agent = Agent(model="claude-sonnet-4-5", instructions="Remember context.")

async with agent.session() as session:
    await session.run("My name is Alice and I love pizza.")
    r1 = await session.run("What is my name?")
    r2 = await session.run("What food do I like?")

print(r1.output)
print(r2.output)

print(session.metadata["run_count"])
print(session.metadata["total_input_tokens"])
print(session.metadata["total_output_tokens"])
print(session.metadata["total_cost"])

Session with history configuration¶

Control what is stored in session history to manage token usage:

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="You are a research assistant.",
    tools=[...],
    session_history_config={
        "mode": "no_reasoning",    # Drop thinking blocks to save tokens
        "preserve_media": True,    # Keep images/documents across turns
    },
)

async with agent.session() as session:
    await session.run("Analyze this document and summarize.")
    await session.run("What were the key findings?")

Available modes: full, messages_only, no_tool_results, no_reasoning, compact_tool_results, text_summary.

Session with compaction¶

For long-running sessions, enable compaction to prevent context overflow:

agent = Agent(
    model="claude-sonnet-4-5",
    instructions="You are a research assistant.",
    tools=[...],
    compaction=True,
    compaction_config={
        "strategy": "sliding_window",
        "keep_recent_messages": 8,
        "sliding_window_include_truncation_notice": True,
    },
)

async with agent.session() as session:
    for task in task_list:
        result = await session.run(task)
        print(f"Compactions so far: {session.metadata['compaction_count']}")

Persistence¶

Serialize and restore session state for cross-process continuation:

# Save
data = session.to_dict()
import json
with open("session_state.json", "w") as f:
    json.dump(data, f)

# Restore
with open("session_state.json") as f:
    data = json.load(f)
restored = AgentSession.from_dict(data, agent=agent)
result = await restored.run("Continue the analysis.")

The serialized state includes provider-native message history, metadata (tokens, cost, compaction counts), and session configuration.

What to validate¶

Follow-up answers reference earlier turns
session.get_history() has all turns (including tool calls and results)
each turn's final assistant reply is in that history — check the last entry after a run is an assistant/model message carrying result.output, and that it appears exactly once
Cumulative session metadata changes across runs
Compaction metadata accumulates when enabled

Notes¶

Session history stores provider-native messages (tool calls, tool results, reasoning) — not just text pairs.
For long-running workflows, periodically serialize session state and persist externally.
Media content (images, documents) is preserved across turns as base64; consider using no_tool_results mode for media-heavy sessions.

Test sources¶

tests/e2e/test_agent_e2e.py (test_session_memory, test_session_clear, test_session_serialization)
tests/unit/test_session.py (session history normalization, filtering, serialization)