Deep Agent – Persistent File System

OSS Product Help LangChain
9

Hi @lreal I hope this sheds more light to your matter:

Deep Agents’ built‑in file system is a virtual FS stored in the graph state under files. Without a checkpointer you typically only see the final state (last write wins when sub‑agents write the same filename). With a checkpointer attached and streaming enabled, you can record every intermediate version of improved_code.txt during the single run.

Key points:

  • Virtual FS lives in state: Deep Agents’ file tools (write_file, read_file, ls, edit_file) operate on a virtual FS backed by LangGraph state, not the host disk. See the built‑in components docs (File System Tools) and README.

  • Attach a checkpointer: A checkpointer persists state snapshots at each step so you can retrieve the evolving files contents during a run.

  • Stream values to collect history: While the graph runs, stream values and append each files["improved_code.txt"] snapshot to a list.

  • Optional on‑disk persistence: Use a SQLite/Postgres checkpointer for durable history across processes; InMemorySaver suffices if you only need history for the current run.

from deepagents import create_deep_agent
from langgraph.checkpoint.memory import InMemorySaver
# Optional durable alternative:
# from langgraph.checkpoint.sqlite import SqliteSaver

# Build your agent exactly as you already do ...
# judge_agent = create_deep_agent(...)

# 1) Attach a checkpointer
checkpointer = InMemorySaver()
# For durability across sessions, prefer:
# checkpointer = SqliteSaver.from_conn_string("sqlite:///deep_agent_history.db")
judge_agent.checkpointer = checkpointer

# 2) Use a stable thread_id to keep the run’s history together
config = {"configurable": {"thread_id": "code-improver-1"}}

# 3) Stream the run and record each intermediate filesystem snapshot
history = []  # will hold all versions of improved_code.txt
for snapshot in judge_agent.stream(
    {"messages": [{"role": "user", "content": "Start the code improvement process."}]},
    config=config,
    stream_mode="values",
):
    files = snapshot.get("files", {})
    if "improved_code.txt" in files:
        history.append(files["improved_code.txt"])  # append each version

# 4) Optionally, also get the final aggregated result
final_result = judge_agent.invoke(
    {"messages": [{"role": "user", "content": "Summarize the final decision."}]},
    config=config,
)

# Now `history` contains every intermediate version the sub‑agents wrote,
# and `final_result.get("files", {}).get("improved_code.txt")` is the last version.

  • When both sub‑agents write the same filename, the virtual FS reflects the latest write at that step. The checkpointer + streaming pattern above preserves the sequence of writes so you don’t lose intermediate versions.

  • If you prefer to avoid last‑write‑wins entirely, give each sub‑agent a distinct filename (e.g., improved_code.optimized.txt and improved_code.readable.txt) and let the supervisor write the final merged output to improved_code.txt.

References:

1 Like