Description

%load_ext autoreload
%autoreload 2

import os
import sys
import importlib


def _require_import(module_name: str):
    try:
        return importlib.import_module(module_name)
    except ModuleNotFoundError as e:
        raise RuntimeError(
            f"""Missing Python package {module_name!r}.

This tutorial is meant to be run from `tutorials/LangChain_LangGraph` with its pinned dependencies.

Quick fixes:
- Docker (recommended): `cd tutorials/LangChain_LangGraph && docker compose up --build`
- Local venv: `cd tutorials/LangChain_LangGraph && pip install -r requirements.txt`
"""
        ) from e


langchain = _require_import("langchain")
langchain_core = _require_import("langchain_core")
langgraph = _require_import("langgraph")

import logging
import platform

logging.basicConfig(
    level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s - %(message)s"
)
_LOG = logging.getLogger("learn_langchain.examples")

_LOG.info("python=%s", sys.version.split()[0])
_LOG.info("platform=%s", platform.platform())
_LOG.info("langchain=%s", getattr(langchain, "__version__", "unknown"))
_LOG.info("langchain_core=%s", getattr(langchain_core, "__version__", "unknown"))
_LOG.info("langgraph=%s", getattr(langgraph, "__version__", "unknown"))
_LOG.info("LLM_PROVIDER=%s", os.getenv("LLM_PROVIDER", "(unset)"))

try:
    import deepagents  # type: ignore
    from deepagents import CompiledSubAgent, create_deep_agent  # type: ignore
    from deepagents.backends import (  # type: ignore
        CompositeBackend,
        FilesystemBackend,
        StateBackend,
        StoreBackend,
    )

    print("deepagents:", getattr(deepagents, "__version__", "(unknown)"))
except Exception as e:  # pragma: no cover
    raise RuntimeError(
        "Deep Agents section requires `deepagents`.\n"
        "Run this notebook via the provided Docker setup or install tutorial deps.\n"
        f"Import error: {type(e).__name__}: {str(e)[:200]}"
    )

from pathlib import Path
from typing import Any

from langchain_core.messages import AIMessage, ToolMessage
import json

# Message and tool call extraction utilities.


def _all_tool_calls(messages: list[Any]) -> list[dict[str, Any]]:
    """
    Collect tool call dicts emitted by `AIMessage` objects in `messages`.
    """
    calls: list[dict[str, Any]] = []
    for m in messages:
        if isinstance(m, AIMessage):
            for tc in m.tool_calls or []:
                if isinstance(tc, dict):
                    calls.append(tc)
    return calls


def _tool_outputs(messages: list[Any], tool_name: str) -> list[Any]:
    """
    Return tool message contents for `tool_name`.
    """
    outs: list[Any] = []
    for m in messages:
        if isinstance(m, ToolMessage) and getattr(m, "name", None) == tool_name:
            outs.append(m.content)
    return outs

# Text conversion and formatting utilities.


def _as_text(x: Any) -> str:
    """
    Convert `x` to a readable string for printing.
    """
    if isinstance(x, str):
        return x
    return (
        json.dumps(x, indent=2, ensure_ascii=False)
        if isinstance(x, (dict, list))
        else str(x)
    )


def _read_file_text(read_result: Any) -> str:
    """
    Extract plain text from a `read_file` tool output.
    """
    if read_result is None:
        return ""
    if isinstance(read_result, str):
        return read_result
    if isinstance(read_result, list):
        parts = [str(x) for x in read_result]
        return "\n".join(parts)
    if isinstance(read_result, dict):
        content = read_result.get("content")
        if isinstance(content, list):
            return "\n".join([str(x) for x in content])
        if isinstance(content, str):
            return content
        text = read_result.get("text")
        if isinstance(text, str):
            return text
    return _as_text(read_result)


def _extract_bullets(text: str) -> list[str]:
    """
    Extract Markdown-style bullet lines from `text`.
    """
    lines = []
    for ln in text.splitlines():
        s = ln.strip()
        if s.startswith("- ") or s.startswith("* "):
            lines.append("- " + s[2:].strip())
            continue
        if s.startswith("•"):
            lines.append("- " + s.lstrip("•").strip())
            continue
        if len(s) >= 3 and s[0].isdigit() and s[1] in (".", ")"):
            lines.append("- " + s[2:].strip())
            continue
        if (
            len(s) >= 4
            and s[0].isdigit()
            and s[1].isdigit()
            and s[2] in (".", ")")
        ):
            lines.append("- " + s[3:].strip())
            continue
    return lines

# Dataset context building utility for prompts.


def build_dataset_context() -> str:
    """
    Build a compact dataset context string for prompts, if `DATASET_META` exists.
    """
    meta = globals().get("DATASET_META", None)
    if not isinstance(meta, dict):
        return "Dataset context: (not loaded)."
    cols = meta.get("columns", [])
    sample_rows = meta.get("sample_rows", [])
    time_col = meta.get("time_col", None)
    freq = meta.get("freq", None)
    parts = [
        f"path={meta.get('path')}",
        f"tool_path={meta.get('tool_path')}",
        f"n_rows={meta.get('n_rows')}",
        f"n_cols={meta.get('n_cols')}",
        f"time_col={time_col}",
        f"freq={freq}",
        f"columns={cols}",
        f"sample_rows={sample_rows}",
    ]
    return "Dataset context:\n" + "\n".join(parts)


# Build dataset context string for use in agent prompts.
DATASET_CONTEXT = build_dataset_context()
print("Utilities loaded and DATASET_CONTEXT ready")

llm_da = get_chat_model()
agent = create_deep_agent(model=llm_da)
out = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "In 3 bullets, explain what a deep agent is useful for.",
            }
        ]
    }
)
print("state keys:", sorted(list(out.keys()))[:20])
print(
    "final message preview:", getattr(out["messages"][-1], "content", "")[:240]
)

agent = create_deep_agent(model=get_chat_model())
prompt = (
    "Before doing anything else, call write_todos with 5 EDA tasks for THIS dataset. "
    "Mark the first task as in_progress.\n\n"
    f"{DATASET_CONTEXT}\n\nIf you need raw rows, use read_file {DATASET_META.get('tool_path')}."
)
out = agent.invoke({"messages": [{"role": "user", "content": prompt}]})
todos = out.get("todos", None)
print("todos channel present:", todos is not None)
print(_as_text(todos)[:800])

root = Path(".").resolve()
Path("workspace").mkdir(parents=True, exist_ok=True)

# Create an agent with filesystem backend for tool interaction.
agent = create_deep_agent(
    model=get_chat_model(),
    backend=FilesystemBackend(root_dir=str(root), virtual_mode=True),
)
print("Agent created with FilesystemBackend")

# Invoke agent with filesystem tools: write, read, and summarize.
prompt = (
    "Use filesystem tools:\n"
    "1) write_file /workspace/notes.md with EXACTLY 6 lines, each starting with '- ', of EDA checks for THIS dataset\n"
    "2) read_file /workspace/notes.md\n"
    "3) Then respond."
    "\n\n"
    f"{DATASET_CONTEXT}\n\nIf you need raw rows, use read_file {DATASET_META.get('tool_path')}."
)
out = agent.invoke({"messages": [{"role": "user", "content": prompt}]})

# Extract outputs and verify file was created.
reads = _tool_outputs(out["messages"], "read_file")
print("n read_file tool outputs:", len(reads))
paths = sorted([str(p) for p in Path("workspace").rglob("notes.md")])
print("notes.md paths on disk:", paths)

if paths:
    read_txt = Path(paths[0]).read_text(encoding="utf-8")
else:
    read_txt = (
        _read_file_text(reads[-1])
        if reads
        else _as_text(getattr(out["messages"][-1], "content", "(missing)"))
    )

# Extract and display key bullets from the response.
bullets = _extract_bullets(read_txt)
print("two bullets:", bullets[:2])
if len(bullets) < 2:
    print("raw preview:", read_txt[:400])

from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore

# Set up checkpointer for persistent, thread-scoped agent state.
ckpt = MemorySaver()

# Define note paths for different backends.
NOTE_STATE = "/workspace/notes.md"
NOTE_FS = "/workspace/notes.md"
NOTE_STORE = "/memories/notes.md"


# Helper to invoke agent with thread_id for persistence scoping.
def run(agent, thread_id: str, user_msg: str):
    """
    Invoke `agent` with `thread_id` for persistence/scoping tests.
    """
    return agent.invoke(
        {"messages": [{"role": "user", "content": user_msg}]},
        config={"configurable": {"thread_id": thread_id}},
    )


# Demo 1: State backend (thread-scoped, in-memory).
agent_state = create_deep_agent(model=get_chat_model(), checkpointer=ckpt)

# Write file in thread A and read from both threads.
run(
    agent_state,
    "STATE_A",
    f"write_file {NOTE_STATE} with 'hello from STATE thread A'",
)
outA = run(agent_state, "STATE_A", f"read_file {NOTE_STATE}")
outB = run(
    agent_state, "STATE_B", f"read_file {NOTE_STATE} (if missing, say so)"
)

# Extract and display reads from each thread.
state_reads_a = _tool_outputs(outA["messages"], "read_file")
state_reads_b = _tool_outputs(outB["messages"], "read_file")
print(
    "StateBackend thread A read:",
    _as_text(state_reads_a[-1])[:120] if state_reads_a else "(no tool output)",
)
print(
    "StateBackend thread B read:",
    _as_text(state_reads_b[-1])[:120] if state_reads_b else "(no tool output)",
)

# Demo 2: Filesystem backend (disk-backed, cross-thread).
root = Path("tmp_runs/deepagents/fs_root").resolve()
root.mkdir(parents=True, exist_ok=True)

# Create agent with filesystem backend for persistent, cross-thread access.
agent_fs = create_deep_agent(
    model=get_chat_model(),
    checkpointer=ckpt,
    backend=FilesystemBackend(root_dir=str(root), virtual_mode=True),
)

# Write file in thread A, read from both threads.
run(agent_fs, "FS_A", f"write_file {NOTE_FS} with 'hello from FS thread A'")
outA = run(agent_fs, "FS_A", f"read_file {NOTE_FS}")
outB = run(agent_fs, "FS_B", f"read_file {NOTE_FS}")

# Extract and display reads from each thread (both should see the file).
fs_reads_a = _tool_outputs(outA["messages"], "read_file")
fs_reads_b = _tool_outputs(outB["messages"], "read_file")
print(
    "FilesystemBackend thread A read:",
    _as_text(fs_reads_a[-1])[:120] if fs_reads_a else "(no tool output)",
)
print(
    "FilesystemBackend thread B read:",
    _as_text(fs_reads_b[-1])[:120] if fs_reads_b else "(no tool output)",
)
print("fs root_dir:", root)

# Demo 3: Store backend via CompositeBackend (cross-thread under /memories/).
store = InMemoryStore()

# Define composite backend that routes /memories/* to StoreBackend, else to StateBackend.
composite_backend = lambda rt: CompositeBackend(
    default=StateBackend(rt),
    routes={"/memories/": StoreBackend(rt)},
)

# Create agent with store backend for cross-thread memory access.
agent_store = create_deep_agent(
    model=get_chat_model(),
    checkpointer=ckpt,
    backend=composite_backend,
    store=store,
)

# Write file to /memories/ in thread A, read from both threads.
run(
    agent_store,
    "STORE_A",
    f"write_file {NOTE_STORE} with 'hello from STORE thread A'",
)
outA = run(agent_store, "STORE_A", f"read_file {NOTE_STORE}")
outB = run(agent_store, "STORE_B", f"read_file {NOTE_STORE}")

# Extract and display reads from each thread (both should see the file in /memories/).
store_reads_a = _tool_outputs(outA["messages"], "read_file")
store_reads_b = _tool_outputs(outB["messages"], "read_file")
print(
    "StoreBackend thread A read:",
    _as_text(store_reads_a[-1])[:120] if store_reads_a else "(no tool output)",
)
print(
    "StoreBackend thread B read:",
    _as_text(store_reads_b[-1])[:120] if store_reads_b else "(no tool output)",
)

from langgraph.checkpoint.memory import MemorySaver

# Set up checkpointer for stateful agent execution.
ckpt = MemorySaver()

# Define a subagent that produces executive EDA profiles.
profiler_subagent = {
    "name": "profile-agent",
    "description": "Produces an executive EDA profile: summary + next analyses.",
    "system_prompt": (
        "You are an EDA profiling specialist.\n"
        "Use only the dataset context provided in the task message.\n"
        "Do not claim to have read files unless the message includes their contents.\n"
        "Output:\n"
        "- 2-bullet executive summary\n"
        "- 2 next analyses\n"
        "Keep it concise."
    ),
    "tools": [],
}

# Set up workspace and create agent with subagent delegation.
root = Path(".").resolve()
Path("workspace").mkdir(parents=True, exist_ok=True)
dataset_on_disk = Path("workspace/data/T1_slice.csv")
print(
    "dataset exists on disk:", dataset_on_disk.exists(), "path:", dataset_on_disk
)

agent = create_deep_agent(
    model=get_chat_model(),
    checkpointer=ckpt,
    backend=FilesystemBackend(root_dir=str(root), virtual_mode=True),
    subagents=[profiler_subagent],  # type: ignore[list-item]
    name="main-agent",
)
print("Agent created with profiler subagent")

# Invoke agent with task() delegation to the subagent.
prompt = (
    "Delegate to the profile-agent using task(), then present the final result.\n"
    "Important: delegate / subagent."
    "\n\n"
    f"{DATASET_CONTEXT}\n\nIf you need raw rows, use read_file {DATASET_META.get('tool_path')}."
)
out = agent.invoke(
    {"messages": [{"role": "user", "content": prompt}]},
    config={"configurable": {"thread_id": "DA5"}},
)

# Verify tool calls were made and display final response.
calls = _all_tool_calls(out["messages"])
print("tool calls:", [c.get("name") for c in calls])
print(
    "final message preview:", getattr(out["messages"][-1], "content", "")[:240]
)

from langchain.agents import create_agent
from langgraph.checkpoint.memory import MemorySaver

# Set up checkpointer for stateful agent execution.
ckpt = MemorySaver()

# Define system prompt for hypothesis generation.
specialized_prompt = (
    "You are a strict hypothesis generator for EDA.\n"
    "Use only the dataset context provided in the task message.\n"
    "Do not claim to have read files unless the message includes their contents.\n"
    "Return:\n"
    "- 2 plausible hypotheses (bullets)\n"
    "- For each: 1 concrete test/plot\n"
    "Be concise."
)

# Create compiled agent using the specialized prompt.
compiled_worker = create_agent(
    model=get_chat_model(), tools=[], system_prompt=specialized_prompt
)

# Wrap compiled agent as a CompiledSubAgent for task delegation.
hypothesis_agent = CompiledSubAgent(
    name="hypothesis-agent",
    description="Generates hypotheses and concrete tests/plots for EDA.",
    runnable=compiled_worker,
)

# Set up workspace and create main agent with hypothesis subagent.
root = Path(".").resolve()
Path("workspace").mkdir(parents=True, exist_ok=True)
dataset_on_disk = Path("workspace/data/T1_slice.csv")
print(
    "dataset exists on disk:", dataset_on_disk.exists(), "path:", dataset_on_disk
)

agent = create_deep_agent(
    model=get_chat_model(),
    checkpointer=ckpt,
    backend=FilesystemBackend(root_dir=str(root), virtual_mode=True),
    subagents=[hypothesis_agent],
    name="main-agent",
)
print("Agent created with hypothesis subagent")

# Invoke agent with task() delegation to hypothesis subagent.
prompt = (
    "Delegate to the hypothesis-agent using task(), then summarize results.\n"
    "Important: delegate / hypothesis."
    "\n\n"
    f"{DATASET_CONTEXT}\n\nIf you need raw rows, use read_file {DATASET_META.get('tool_path')}."
)
out = agent.invoke(
    {"messages": [{"role": "user", "content": prompt}]},
    config={"configurable": {"thread_id": "DA6"}},
)

# Verify tool calls and display final response.
calls = _all_tool_calls(out["messages"])
print("tool calls:", [c.get("name") for c in calls])
print(
    "final message preview:", getattr(out["messages"][-1], "content", "")[:240]
)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

# Import HITL (human-in-the-loop) interrupt configuration.
try:
    from langchain.agents.middleware.human_in_the_loop import InterruptOnConfig
except ModuleNotFoundError as e:  # pragma: no cover
    raise RuntimeError(
        """This Deep Agents HITL demo needs the tutorial dependencies.

Run it from `tutorials/LangChain_LangGraph` with `requirements.txt` installed (or via Docker).
"""
    ) from e

# Set up checkpointer and workspace for HITL demo.
ckpt = MemorySaver()
thread_id = "DA7_NOTEBOOK"
root = Path(".").resolve()
Path("workspace").mkdir(parents=True, exist_ok=True)

# Create agent with HITL interrupts on edit_file operations.
agent = create_deep_agent(
    model=get_chat_model(),
    checkpointer=ckpt,
    backend=FilesystemBackend(root_dir=str(root), virtual_mode=True),
    interrupt_on={
        "edit_file": InterruptOnConfig(allowed_decisions=["approve", "reject"])
    },
)
print("Agent created with HITL interrupt on edit_file")

# Step 1: Create a file (no interrupt expected).
out_step1 = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "write_file /workspace/notes.md with 'line1\\nline2\\n'",
            }
        ]
    },
    config={"configurable": {"thread_id": thread_id}},
)
print("Step 1 completed: file created")

# Step 2: Attempt file edit (expect HITL interrupt), then resume with approval.
out_step2 = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "edit_file /workspace/notes.md replace 'line2' with 'LINE2_EDITED' then read_file /workspace/notes.md",
            }
        ]
    },
    config={"configurable": {"thread_id": thread_id}},
)

# Check if interrupt occurred and inspect its payload.
has_intr = "__interrupt__" in out_step2
print("interrupted:", has_intr)
if has_intr:
    intr0 = out_step2["__interrupt__"][0]
    print(
        "interrupt payload preview:",
        _as_text(getattr(intr0, "value", intr0))[:240],
    )
    # Resume execution by approving the edit.
    out_step2 = agent.invoke(
        Command(resume={"decisions": [{"type": "approve"}]}),
        config={"configurable": {"thread_id": thread_id}},
    )
    print("Step 2 resumed with approval")

# Step 3: Verify edited file content via fresh read.
out_step3 = agent.invoke(
    {"messages": [{"role": "user", "content": "read_file /workspace/notes.md"}]},
    config={"configurable": {"thread_id": thread_id}},
)

# Extract and display the final file content.
read_outs = _tool_outputs(out_step3["messages"], "read_file")
paths = sorted([str(p) for p in Path("workspace").rglob("notes.md")])
print("notes.md paths on disk:", paths)

if paths:
    read_txt = Path(paths[0]).read_text(encoding="utf-8")
else:
    read_txt = (
        _read_file_text(read_outs[-1])
        if read_outs
        else _as_text(getattr(out_step3["messages"][-1], "content", ""))
    )

print("final file content:")
print(read_txt.replace("\\n", "\\\\n")[:200])

from langgraph.checkpoint.memory import MemorySaver

# Set up checkpointer for persistent agent state during sandbox tests.
ckpt = MemorySaver()
thread_id = "DA8_NOTEBOOK"

# Create sandbox root directory and a secret outside of it (for path traversal test).
root = Path("tmp_runs/deepagents/sandbox_root").resolve()
root.mkdir(parents=True, exist_ok=True)

outside_secret = Path("tmp_runs/deepagents/secret_outside_sandbox.txt")
outside_secret.parent.mkdir(parents=True, exist_ok=True)
outside_secret.write_text("SUPER_SECRET=do_not_leak\\n", encoding="utf-8")

# Create agent with filesystem backend constrained to sandbox_root.
backend = FilesystemBackend(root_dir=str(root), virtual_mode=True)
agent = create_deep_agent(
    model=get_chat_model(), checkpointer=ckpt, backend=backend
)
print("Agent created with sandboxed filesystem backend")

# Helper function to debug actual tool calls made by the model.
def _print_tool_calls(state: dict, label: str) -> None:
    """
    Print tool call names and args emitted by the model.
    """
    calls = _all_tool_calls(state.get("messages", []))
    simplified = [{"name": c.get("name"), "args": c.get("args")} for c in calls]
    print(f"{label} tool_calls:", simplified)


# Test 1: Safe file write and read inside the sandbox.
out_ok = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Call write_file on /workspace/ok.txt with content 'safe'. Then call read_file on /workspace/ok.txt. Do not read any other paths.",
            }
        ]
    },
    config={"configurable": {"thread_id": thread_id}},
)

# Extract and display the safe read result.
ok_reads = _tool_outputs(out_ok["messages"], "read_file")
ok_txt = (
    _read_file_text(ok_reads[-1])
    if ok_reads
    else _as_text(getattr(out_ok["messages"][-1], "content", ""))
)
print("ok read preview:", ok_txt[:80])
_print_tool_calls(out_ok, "ok")

# Test 2: Attempt path traversal escape outside sandbox.
out_env = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Call read_file on /workspace/../tmp_runs/deepagents/secret_outside_sandbox.txt exactly. Return the raw result (error or content). Do not substitute any other path.",
            }
        ]
    },
    config={"configurable": {"thread_id": thread_id}},
)

# Check if path traversal was blocked.
escape_reads = _tool_outputs(out_env["messages"], "read_file")
print(
    "escape attempt output preview:",
    _as_text(escape_reads[-1])[:160] if escape_reads else "(no tool output)",
)
_print_tool_calls(out_env, "escape")

# Test 3: Attempt to read a host system file.
out_hosts = agent.invoke(
    {
        "messages": [
            {
                "role": "user",
                "content": "Call read_file on /etc/hosts exactly. Return the raw result (error or content). Do not substitute any other path.",
            }
        ]
    },
    config={"configurable": {"thread_id": thread_id}},
)

# Check if host system file access was blocked.
hosts_reads = _tool_outputs(out_hosts["messages"], "read_file")
print(
    "/etc/hosts attempt output preview:",
    _as_text(hosts_reads[-1])[:160] if hosts_reads else "(no tool output)",
)
_print_tool_calls(out_hosts, "hosts")

# Display final sandbox state.
print("sandbox root_dir:", root)
print(
    "sandbox files:",
    sorted([str(p.relative_to(root)) for p in root.rglob("*") if p.is_file()])[
        :20
    ],
)