Deep Agent – Persistent File System

OSS Product Help LangChain
1

How can we make the file system in Deep Agent persist, so it retains history updates throughout a single run?

2

Hi @lreal

I think you are looking for checkpointer funability - Overview

Am I right?

3

We have two sub-agents writing in parallel to improved_code.txt, but only the supervisor’s version is saved. How can we use a checkpointer to keep the full history instead of just the final overwrite? Could you show this in code?

thanks in advance

4

I don’t know how your subagents work. Could you please share the code?

5

Here’s a minimal version of the code:


class CodeImproverState(TypedDict):
    messages: Annotated[list, operator.add]
    optimized_code: str
    readable_code: str
    final_code: str


def run_python_code(code: str):
    try:
        namespace = {}
        exec(code, namespace)
        return {"success": True, "output": "Code executed successfully"}
    except Exception as e:
        return {"success": False, "error": str(e)}


optimized_writer_instructions = """You are a code optimization expert.
Your task: Write optimized code to `improved_code.txt`
You can iterate and improve your code multiple times before handing off.
Each time you write, update the file with your improvements."""

readable_writer_instructions = """You are a code readability expert.
Your task: Write readable code to `improved_code.txt`
You can iterate and improve your code multiple times before handing off.
Each time you write, update the file with your improvements."""

judge_instructions = """You are the supervisor.
1. Call both sub-agents - they will write to `improved_code.txt` multiple times
2. After they finish, read `improved_code.txt` 
3. Write your final decision to `improved_code.txt`"""


optimized_writer = {
    "name": "Optimized Writer",
    "prompt": optimized_writer_instructions,
    "tools": [],
}

readable_writer = {
    "name": "Readable Writer",
    "prompt": readable_writer_instructions,
    "tools": [],
}

judge_agent = create_deep_agent(
    tools=[run_python_code],
    instructions=judge_instructions,
    model="gpt-4o-mini",
    subagents=[optimized_writer, readable_writer],
    state_schema=CodeImproverState,
).with_config({"recursion_limit": 200})

I used a checkpointer, but when the sub-agent runs multiple times, it doesn’t show the history.

6

Great, thank you. Let me see how it works.

7

Any updates on that?

8

Not yet, I may investigate further tomorrow and be back with some more results if any

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
Deep Agents: Custom state fields not accessible in ToolRuntime.state
10

That was helpful, thanks.

1 Like
11

Hi @lreal

I am happy it helped :worried: