Langchain Deep Agents

START -> planner -> (for each subtask) subagent_dispatcher -> virtual_fs_write
                                             |
                                             v
                                        reflection -> planner (replan)
                                                   -> END (done)
                                                   -> interrupt (escalate)

PLANNER_SYSTEM = """You are the planner for a Deep Agent.

Your job is to decompose the user's goal into subtasks and assign each to a
specialized subagent. You do NOT execute subtasks yourself.

Available subagent roles: {roles_list}

Return a JSON plan:
{{
  "subtasks": [
    {{"role": "research-specialist",
      "instruction": "Find the most recent SEC 10-K filing for ACME.",
      "expected_artifact": "acme_10k_summary.md",
      "read_files": []}}
  ],
  "reasoning": "Why this decomposition."
}}

Do not write file contents. Do not answer the user directly.
"""

from langchain_core.messages import SystemMessage, HumanMessage
from langgraph.prebuilt import create_react_agent
from langchain_anthropic import ChatAnthropic

SUBAGENT_PROMPTS = {
    "research-specialist": (
        "You are a research specialist. Given an instruction, produce a "
        "fact-dense summary with inline citations. Return ONLY the summary. "
        "Do NOT plan, do NOT delegate."
    ),
    "code-writer": (
        "You are a code writer. Given a spec, produce runnable Python. "
        "Return ONLY code in one fenced block. Do NOT explain."
    ),
    "critic": (
        "You are a critic. Given an artifact and a spec, list concrete defects. "
        "Return a JSON list of {line, issue, severity}. Do NOT rewrite the artifact."
    ),
}

def build_subagent(role: str, model):
    return create_react_agent(
        model=model,
        tools=ROLE_TOOLS[role],
        # KEY: prompt parameter replaces default state_modifier; override=True
        # means no parent-prompt composition.
        prompt=SystemMessage(content=SUBAGENT_PROMPTS[role]),
    )

def invoke_subagent(subagent, instruction: str, read_files: dict):
    # Build messages from scratch — do not reuse parent message list.
    context = "\n\n".join(f"# {name}\n{content}" for name, content in read_files.items())
    messages = [HumanMessage(content=f"{context}\n\n## Task\n{instruction}")]
    result = subagent.invoke({"messages": messages})
    return result["messages"][-1].content

from typing import TypedDict, Annotated
from operator import or_  # merge dict state updates

class DeepAgentState(TypedDict):
    messages: list
    plan: dict
    files: Annotated[dict, or_]  # {name: {content, written_at_step, status}}
    step: int
    reflection_depth: int

MAX_FILE_AGE_STEPS = 20
INLINE_SIZE_LIMIT_BYTES = 100 * 1024  # 100 KB — above this, spill to disk; keeps state < 500 KB (P51)

def cleanup_node(state: DeepAgentState) -> dict:
    current_step = state["step"]
    kept = {}
    for name, entry in state["files"].items():
        age = current_step - entry["written_at_step"]
        if entry.get("status") == "done" and age > 3:
            continue  # evict completed
        if age > MAX_FILE_AGE_STEPS:
            continue  # evict stale
        kept[name] = entry
    return {"files": kept}

def write_artifact(state, name: str, content: str) -> dict:
    if len(content.encode("utf-8")) > INLINE_SIZE_LIMIT_BYTES:
        # Spill to disk; keep only a pointer in state.
        path = f"/tmp/deep_agent/{state['step']}_{name}"
        with open(path, "w") as f: f.write(content)
        entry = {"content": None, "disk_path": path, "written_at_step": state["step"], "status": "active"}
    else:
        entry = {"content": content, "written_at_step": state["step"], "status": "active"}
    return {"files": {name: entry}}

REFLECTION_SYSTEM = """You are the reflection node of a Deep Agent.

Input: the current plan, the artifacts produced so far (by filename + status),
and the step count. Output a decision:

{"decision": "continue"}           # more subtasks remain, proceed
{"decision": "replan", "reason": "..."}  # current plan is wrong, replanner
{"decision": "end", "final_answer": "..."}  # we are done
{"decision": "escalate", "question": "..."}  # ask the human

Do not rewrite files. Do not invent facts. Base the decision only on what is
present in files and plan.
"""

MAX_REFLECTION_DEPTH = 5

def reflection_node(state: DeepAgentState, model) -> dict:
    if state["reflection_depth"] >= MAX_REFLECTION_DEPTH:
        return {"decision": "escalate", "question": "Max reflection depth reached."}
    # Feed summaries, not full file contents, to the reflection model.
    file_summary = {name: {"status": e["status"], "bytes": len(e.get("content") or "")}
                    for name, e in state["files"].items()}
    msg = HumanMessage(content=f"Plan: {state['plan']}\nFiles: {file_summary}\nStep: {state['step']}")
    result = model.invoke([SystemMessage(content=REFLECTION_SYSTEM), msg])
    return {"decision": result.content, "reflection_depth": state["reflection_depth"] + 1}

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, END

checkpointer = MemorySaver()
graph = StateGraph(DeepAgentState)
graph.add_node("planner", planner_node)
graph.add_node("subagent_dispatcher", dispatch_node)
graph.add_node("cleanup", cleanup_node)
graph.add_node("reflection", reflection_node)
graph.set_entry_point("planner")
graph.add_edge("planner", "subagent_dispatcher")
graph.add_edge("subagent_dispatcher", "cleanup")
graph.add_edge("cleanup", "reflection")
graph.add_conditional_edges("reflection", route_after_reflection,
    {"continue": "planner", "replan": "planner", "end": END, "escalate": END})

# Only interrupt on user-facing boundary — the checkpoint taken here is the
# one the user sees and can resume from.
app = graph.compile(checkpointer=checkpointer, interrupt_after=["reflection"])