LangGraph I

# script_id: day_053_langgraph_graphs/d48_loop_sketch
# fragment
messages, steps, errors = [task], 0, 0          # hidden state: three local variables
while True:
    reply = model(messages); steps += 1         # "think"
    messages.append(reply)
    if not reply.tool_calls:                    # transition: done
        return reply.text
    if steps >= MAX_STEPS or errors >= MAX_ERRORS:
        return partial_answer(messages)         # transition: budget stop
    for call in reply.tool_calls:               # "act"
        result, failed = run(call)
        messages.append(result); errors += failed

uv add "langgraph==1.2.20" "langchain==1.4.9" "langchain-openai==1.7.4"

# script_id: day_053_langgraph_graphs/graph_toy
# The smallest useful graph: a node in a loop, a conditional edge, an append reducer.
# Install: uv add "langgraph==1.2.20". Env: none. Services: none.
import operator
from typing import Annotated, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END, START, StateGraph

class State(TypedDict):
    steps: Annotated[list[str], operator.add]   # reducer: returned lists are appended

def work(state: State) -> dict:
    return {"steps": ["work"]}                   # a partial update, never the whole state

def keep_going(state: State) -> str:
    return "work" if len(state["steps"]) < 3 else END

builder = StateGraph(State)
builder.add_node("work", work)
builder.add_edge(START, "work")
builder.add_conditional_edges("work", keep_going, ["work", END])   # list = every possible target
graph = builder.compile(checkpointer=InMemorySaver())             # saves state per thread
cfg = {"configurable": {"thread_id": "toy-1"}, "recursion_limit": 25}
print(graph.invoke({"steps": []}, cfg))

{'steps': ['work', 'work', 'work']}

# script_id: day_053_langgraph_graphs/break_it
# Three classic mistakes: no reducer, no exit (caught by recursion_limit), no thread_id.
# Install: uv add "langgraph==1.2.20". Env: none. Services: none.
import operator
from typing import Annotated, TypedDict

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph

class Appending(TypedDict):
    steps: Annotated[list[str], operator.add]

class Replacing(TypedDict):
    steps: list[str]                             # no reducer: the last write wins

def build(schema, checkpointer=None):
    b = StateGraph(schema)
    b.add_node("work", lambda s: {"steps": ["work"]})
    b.add_edge(START, "work")
    b.add_conditional_edges("work", lambda s: "work" if len(s["steps"]) < 3 else END, ["work", END])
    return b.compile(checkpointer=checkpointer)

seen = []                                        # 1 + 2: the list never grows, so the loop never exits
try:
    for values in build(Replacing).stream({"steps": []}, {"recursion_limit": 5}, stream_mode="values"):
        seen.append(values["steps"])
except GraphRecursionError as exc:
    print("states seen:", seen)
    print("GraphRecursionError:", str(exc).splitlines()[0])

try:                                             # 3: a checkpointer needs a thread_id
    build(Appending, InMemorySaver()).invoke({"steps": []})
except ValueError as exc:
    print("ValueError:", exc)

states seen: [[], ['work'], ['work'], ['work'], ['work'], ['work']]
GraphRecursionError: Recursion limit of 5 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
ValueError: Checkpointer requires one or more of the following 'configurable' keys: thread_id, checkpoint_ns, checkpoint_id

# script_id: day_053_langgraph_graphs/atlas/graph
# atlas/graph.py: the Atlas investigator as create_agent and as a hand-built StateGraph.
# Install: uv add "langgraph==1.2.20" "langchain==1.4.9" "langchain-openai==1.7.4"
# Env: OPENAI_API_KEY (optional: without it a scripted offline model runs), ATLAS_MODEL, ORBIT_DATA
# Services: none. Reads datasets/orbit/issues/ (orbit.db read-only, issues.jsonl) and docs/*.md
import json
import os
import re
import sqlite3
from contextlib import closing
from dataclasses import dataclass
from pathlib import Path
from typing import Literal

from langchain.tools import tool

MODEL = os.environ.get("ATLAS_MODEL", "gpt-6-luna")
DATA_DIR = Path(os.environ.get("ORBIT_DATA", "datasets/orbit"))
PRICES_PER_1M = {"input": 0.10, "output": 0.50}   # gpt-6-luna USD, as of 2026-11, verify at the pricing page
SYSTEM = ("You investigate GitHub issues for Orbit, a Python background-jobs library. Search past "
          "issues and read the most relevant doc before answering. Cite issue ids and doc ids.")

@dataclass(frozen=True)
class AgentConfig:                     # built on Day 48 (compact): the budgets become routing rules
    max_steps: int = 8                 # model calls per run
    max_seconds: float = 60            # Exercise 2 adds this check to the router
    max_tokens: int = 100_000          # cumulative input + output tokens
    max_errors: int = 3                # failed tool calls

CONFIG = AgentConfig()

# script_id: day_053_langgraph_graphs/atlas/graph
@tool
def search_issues(query: str, state: Literal["open", "closed"] | None = None, limit: int = 5) -> str:
    """Search past Orbit issues by keywords. Returns JSON rows with id, title and state."""
    # built on Day 35: a fixed, parameterised query on a read-only connection (the model never writes SQL)
    words = re.findall(r"[a-z0-9-]{3,}", query.lower())[:6] or [query.lower()]
    score = " + ".join(["(lower(title || ' ' || body) LIKE ?)"] * len(words))
    likes = [f"%{w}%" for w in words]
    sql = (f"SELECT id, title, state FROM issues WHERE ({score}) > 0 {'AND state = ?' if state else ''} "
           f"ORDER BY ({score}) DESC, id LIMIT ?")
    uri = f"file:{DATA_DIR / 'issues' / 'orbit.db'}?mode=ro"
    with closing(sqlite3.connect(uri, uri=True)) as db:
        rows = db.execute(sql, likes + ([state] if state else []) + likes + [limit]).fetchall()
    return json.dumps([{"id": i, "title": t, "state": s} for i, t, s in rows])

@tool
def get_doc_section(doc_id: str, heading: str | None = None) -> str:
    """Read an Orbit doc by id (e.g. 'retries-and-backoff'), optionally only one '## heading' section."""
    if not re.fullmatch(r"[a-z0-9-]+", doc_id):          # blocks '../' paths from a confused model
        raise ValueError(f"bad doc_id {doc_id!r}: use ids like 'retries-and-backoff'")
    path = DATA_DIR / "docs" / f"{doc_id}.md"
    if not path.exists():
        raise LookupError(f"no doc {doc_id!r}")
    text = path.read_text(encoding="utf-8")
    if heading:
        sections = re.split(r"(?m)^## ", text)
        text = next((s for s in sections if s.lower().startswith(heading.lower())), text)
    return json.dumps({"doc_id": doc_id, "text": text[:2000]})

TOOLS = [search_issues, get_doc_section]

def load_task(issue_id: int) -> str:
    """The issue as the agent's task. Issue text is untrusted input (injection defences: Day 69)."""
    for line in (DATA_DIR / "issues" / "issues.jsonl").open(encoding="utf-8"):
        issue = json.loads(line)
        if issue["id"] == issue_id:
            return f"{issue['title']}\n\nOrbit issue #{issue_id}:\n{issue['body'][:1500]}"
    raise LookupError(f"no issue {issue_id}")

# script_id: day_053_langgraph_graphs/atlas/graph
from langchain_core.language_models import BaseChatModel
from langchain_core.messages import AIMessage, ToolMessage
from langchain_core.outputs import ChatGeneration, ChatResult

class ScriptedChatModel(BaseChatModel):
    """Offline stand-in: search issues, read the best-matching doc, then answer. Deterministic."""
    tools_bound: bool = False

    @property
    def _llm_type(self) -> str:
        return "scripted"

    def bind_tools(self, tools, *, tool_choice=None, **kwargs):
        return self.model_copy(update={"tools_bound": tool_choice != "none"})

    def _generate(self, messages, stop=None, run_manager=None, **kwargs) -> ChatResult:
        task = next(m.text for m in messages if m.type == "human")
        done = sum(m.type == "tool" for m in messages)
        words = set(re.findall(r"[a-z]+", task.splitlines()[0].lower()))   # the title
        doc = max(sorted(p.stem for p in (DATA_DIR / "docs").glob("*.md")),
                  key=lambda d: len(set(d.split("-")) & words))
        plan = [{"name": "search_issues", "args": {"query": task.splitlines()[0]}, "id": f"call_{done}"},
                {"name": "get_doc_section", "args": {"doc_id": doc}, "id": f"call_{done}"}]
        calls = plan[done:done + 1] if self.tools_bound else []
        text = "" if calls else f"Likely cause: see doc '{doc}' and the similar issues found. (scripted)"
        n_in = sum(len(m.text) for m in messages) // 4          # about 4 characters per token
        usage = {"input_tokens": n_in, "output_tokens": 40, "total_tokens": n_in + 40}
        reply = AIMessage(content=text, tool_calls=calls, usage_metadata=usage)
        return ChatResult(generations=[ChatGeneration(message=reply)])

def make_model() -> BaseChatModel:
    if not os.environ.get("OPENAI_API_KEY"):
        return ScriptedChatModel()
    from langchain_openai import ChatOpenAI
    return ChatOpenAI(model=MODEL, use_responses_api=True)    # always pass model=; force Responses

# script_id: day_053_langgraph_graphs/atlas/graph
def envelope_err(exc: Exception) -> str:
    """Built on Day 37 (compact): a typed error the model can act on, never a stack trace."""
    kind = ("timeout" if isinstance(exc, TimeoutError) else "transient" if isinstance(exc, ConnectionError)
            else "denied" if isinstance(exc, PermissionError)
            else "permanent" if isinstance(exc, (LookupError, ValueError)) else "unknown")
    return json.dumps({"ok": False, "error": {"kind": kind, "message": str(exc)[:300],
                       "retryable": kind in ("timeout", "transient"), "hint": "fix the arguments or try another tool"}})

def usage_of(messages: list) -> tuple[int, int]:
    """(input, output) tokens summed over every AI message's usage_metadata."""
    usage = [m.usage_metadata or {} for m in messages if m.type == "ai"]
    return sum(u.get("input_tokens", 0) for u in usage), sum(u.get("output_tokens", 0) for u in usage)

def cost_usd(messages: list) -> float:
    n_in, n_out = usage_of(messages)
    return (n_in * PRICES_PER_1M["input"] + n_out * PRICES_PER_1M["output"]) / 1_000_000

# script_id: day_053_langgraph_graphs/atlas/graph
from langchain.agents import create_agent
from langchain.agents.middleware import ModelCallLimitMiddleware

def build_langchain_agent(tools=TOOLS):
    """The standard tool loop in one call. Returns a compiled LangGraph graph."""
    return create_agent(make_model(), tools=tools, system_prompt=SYSTEM,
                        middleware=[ModelCallLimitMiddleware(run_limit=CONFIG.max_steps)])

# script_id: day_053_langgraph_graphs/create_agent_run
# Run the create_agent version of Atlas on one Orbit issue and print the message trace.
# Install: as for atlas/graph. Env: OPENAI_API_KEY optional (scripted model without it), ORBIT_DATA
from atlas.graph import build_langchain_agent, cost_usd, load_task, usage_of

agent = build_langchain_agent()
print(type(agent).__name__)              # a compiled graph: invoke, stream and get_state all apply
result = agent.invoke({"messages": [{"role": "user", "content": load_task(164)}]},
                      {"recursion_limit": 25})
for m in result["messages"]:
    calls = [f"{c['name']}({c['args']})" for c in getattr(m, "tool_calls", [])]
    print(f"{m.type:>5} | {', '.join(calls) or repr(m.text[:70])}")
n_in, n_out = usage_of(result["messages"])
print(f"tokens in/out = {n_in}/{n_out}  cost = ${cost_usd(result['messages']):.5f}")

CompiledStateGraph
human | 'ORB-1001 "max number of clients reached" after scaling out workers (no'
   ai | search_issues({'query': 'ORB-1001 "max number of clients reached" after scaling out workers (no failover involved)'})
 tool | '[{"id": 161, "title": "Redis failover causes reconnect storm: thousand'
   ai | get_doc_section({'doc_id': 'scaling-workers'})
 tool | '{"doc_id": "scaling-workers", "text": "---\\nid: scaling-workers\\ntitle'
   ai | "Likely cause: see doc 'scaling-workers' and the similar issues found. "
tokens in/out = 1740/120  cost = $0.00023

# script_id: day_053_langgraph_graphs/atlas/graph
from typing import Annotated, TypedDict

from langchain_core.messages import AnyMessage, HumanMessage, SystemMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]   # reducer: append (or replace by message id)
    steps: int                                            # no reducer: the last write wins
    errors: int
    stop_reason: str | None

def call_model(state: AgentState, llm) -> dict:
    reply = llm.invoke([SystemMessage(SYSTEM), *state["messages"]])
    return {"messages": [reply], "steps": state.get("steps", 0) + 1,
            "stop_reason": None if reply.tool_calls else "done"}

def run_tools(state: AgentState, tools_by_name: dict) -> dict:
    errors, results = state.get("errors", 0), []
    for call in state["messages"][-1].tool_calls:          # answer EVERY call, failures included
        try:
            content = tools_by_name[call["name"]].invoke(call["args"])
        except Exception as exc:                           # unknown tool, bad arguments, tool crash
            errors += 1
            content = envelope_err(exc)
        results.append(ToolMessage(content=content, tool_call_id=call["id"], name=call["name"]))
    return {"messages": results, "errors": errors}

FINALIZE_PROMPT = "Stop using tools. Give your best partial answer now and say what is still unknown."

def finalize(state: AgentState, llm) -> dict:
    skipped = [ToolMessage(content=json.dumps({"ok": False, "error": {"kind": "skipped", "message": "budget used up"}}),
                           tool_call_id=c["id"], name=c["name"])
               for c in getattr(state["messages"][-1], "tool_calls", [])]
    reply = llm.invoke([SystemMessage(SYSTEM), *state["messages"], *skipped, HumanMessage(FINALIZE_PROMPT)])
    reason = state.get("stop_reason") or stop_check(state) or "max_steps"
    return {"messages": [*skipped, reply], "stop_reason": reason}

# script_id: day_053_langgraph_graphs/atlas/graph
from langgraph.graph import END, START, StateGraph

def stop_check(state: AgentState, config: AgentConfig = CONFIG) -> str | None:
    """Built on Day 48's StopChecker: pure, cheapest first, unit-testable without a model."""
    if state.get("errors", 0) >= config.max_errors:
        return "max_errors"
    if state.get("steps", 0) >= config.max_steps:
        return "max_steps"
    if sum(usage_of(state["messages"])) >= config.max_tokens:
        return "max_tokens"
    return None

def route_after_model(state: AgentState) -> str:
    if not state["messages"][-1].tool_calls:
        return END                                         # the model answered
    return "finalize" if stop_check(state) else "run_tools"

def route_after_tools(state: AgentState) -> str:
    return "finalize" if stop_check(state) else "call_model"

def build_agent_graph(tools, *, checkpointer=None):
    llm = make_model()
    act, final = llm.bind_tools(tools), llm.bind_tools(tools, tool_choice="none")
    by_name = {t.name: t for t in tools}
    g = StateGraph(AgentState)
    g.add_node("call_model", lambda s: call_model(s, act))       # node names are a contract:
    g.add_node("run_tools", lambda s: run_tools(s, by_name))     # Days 54-55 and 61 build on them
    g.add_node("finalize", lambda s: finalize(s, final))
    g.add_edge(START, "call_model")
    g.add_conditional_edges("call_model", route_after_model, ["run_tools", "finalize", END])
    g.add_conditional_edges("run_tools", route_after_tools, ["call_model", "finalize"])
    g.add_edge("finalize", END)
    return g.compile(checkpointer=checkpointer)    # validates the wiring once; run it many times

# script_id: day_053_langgraph_graphs/stream_demo
# Run the StateGraph agent on one issue, streaming per-node diffs; then trip the recursion fuse.
# Install: as for atlas/graph. Env: OPENAI_API_KEY optional (scripted model without it), ORBIT_DATA
from langgraph.errors import GraphRecursionError

from atlas.graph import TOOLS, build_agent_graph, cost_usd, load_task

graph = build_agent_graph(TOOLS)
start = {"messages": [{"role": "user", "content": load_task(164)}],
         "steps": 0, "errors": 0, "stop_reason": None}
cfg = {"recursion_limit": 25}           # > 2 * max_steps + 1: the budget stops first; the limit is a fuse
final = None
for mode, chunk in graph.stream(start, cfg, stream_mode=["updates", "values"]):
    if mode == "values":
        final = chunk                    # the full state after each step; keep the last one
        continue
    for node, update in chunk.items():   # the diff a node returned
        calls = [c["name"] for m in update.get("messages", []) for c in getattr(m, "tool_calls", [])]
        print(f"{node:<10} steps={update.get('steps', '-')} errors={update.get('errors', '-')} "
              f"calls={calls} stop_reason={update.get('stop_reason')}")
print(f"stop_reason={final['stop_reason']} steps={final['steps']} cost=${cost_usd(final['messages']):.5f}")
try:
    graph.invoke(start, {"recursion_limit": 3})          # far too small, on purpose
except GraphRecursionError as exc:
    print("GraphRecursionError:", str(exc).splitlines()[0])

call_model steps=1 errors=- calls=['search_issues'] stop_reason=None
run_tools  steps=- errors=0 calls=[] stop_reason=None
call_model steps=2 errors=- calls=['get_doc_section'] stop_reason=None
run_tools  steps=- errors=0 calls=[] stop_reason=None
call_model steps=3 errors=- calls=[] stop_reason=done
stop_reason=done steps=3 cost=$0.00023
GraphRecursionError: Recursion limit of 3 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.

# script_id: day_053_langgraph_graphs/command_toy
# A node that routes itself: Command(goto=..., update=...) instead of a separate router function.
# Install: uv add "langgraph==1.2.20". Env: none. Services: none.
from typing import Literal, TypedDict

from langgraph.graph import END, START, StateGraph
from langgraph.types import Command

class State(TypedDict):
    errors: int
    stop_reason: str | None

def flaky_tool(state: State) -> Command[Literal["flaky_tool", "finalize"]]:
    errors = state["errors"] + 1                       # pretend every attempt fails
    if errors >= 3:
        return Command(goto="finalize", update={"errors": errors, "stop_reason": "max_errors"})
    return Command(goto="flaky_tool", update={"errors": errors})

builder = StateGraph(State)
builder.add_node("flaky_tool", flaky_tool)
builder.add_node("finalize", lambda s: {})
builder.add_edge(START, "flaky_tool")                  # no edge out of flaky_tool: Command routes it
builder.add_edge("finalize", END)
print(builder.compile().invoke({"errors": 0, "stop_reason": None}, {"recursion_limit": 25}))

{'errors': 3, 'stop_reason': 'max_errors'}

# script_id: day_053_langgraph_graphs/parity_bench
# Same 5 issues, model, tools and budgets; three runtimes: own loop, create_agent, StateGraph.
# Install: as for atlas/graph. Env: OPENAI_API_KEY optional (scripted model without it), ORBIT_DATA
# Writes reports/d53_parity.json. Without a key this checks control-flow parity, not answer quality.
import inspect
import json
from pathlib import Path

from langchain_core.messages import HumanMessage, SystemMessage

import atlas.graph as ag

def run_loop(task: str, config: ag.AgentConfig = ag.CONFIG) -> dict:
    """Built on Day 48 (compact): the same stop rules as a while loop over the same chat model."""
    llm, by_name = ag.make_model().bind_tools(ag.TOOLS), {t.name: t for t in ag.TOOLS}
    state = {"messages": [HumanMessage(task)], "steps": 0, "errors": 0}
    while True:
        reply = llm.invoke([SystemMessage(ag.SYSTEM), *state["messages"]])
        state["messages"].append(reply)
        state["steps"] += 1
        if not reply.tool_calls:
            return {**state, "stop_reason": "done"}
        if reason := ag.stop_check(state, config):
            return {**state, "stop_reason": reason}       # (Day 48 also writes a partial answer here)
        update = ag.run_tools(state, by_name)
        state["messages"] += update["messages"]
        state["errors"] = update["errors"]
        if reason := ag.stop_check(state, config):
            return {**state, "stop_reason": reason}

# script_id: day_053_langgraph_graphs/parity_bench
lc_agent, graph = ag.build_langchain_agent(), ag.build_agent_graph(ag.TOOLS)
RUNNERS = {
    "own loop": run_loop,
    "create_agent": lambda t: lc_agent.invoke({"messages": [HumanMessage(t)]}, {"recursion_limit": 25}),
    "StateGraph": lambda t: graph.invoke({"messages": [HumanMessage(t)], "steps": 0, "errors": 0,
                                          "stop_reason": None}, {"recursion_limit": 25}),
}
LOC = {"own loop": [run_loop], "create_agent": [ag.build_langchain_agent],
       "StateGraph": [ag.call_model, ag.run_tools, ag.finalize, ag.route_after_model,
                      ag.route_after_tools, ag.build_agent_graph]}
set_file = Path("agent_set.json")                      # Day 48 writes a list of {"issue_id", ...}
issues = ([p["issue_id"] for p in json.loads(set_file.read_text())] if set_file.exists()
          else [10, 22, 164, 186, 204])                # stand-in ids if Day 48 hasn't run
rows = []
for issue_id in issues:
    for name, run in RUNNERS.items():
        msgs = (out := run(ag.load_task(issue_id)))["messages"]
        docs = [c["args"].get("doc_id") for m in msgs for c in getattr(m, "tool_calls", [])
                if c["name"] == "get_doc_section"]
        rows.append({"issue": issue_id, "runner": name, "steps": sum(m.type == "ai" for m in msgs),
                     "stop_reason": out.get("stop_reason")              # create_agent keeps no reason
                     or ("max_steps" if sum(m.type == "ai" for m in msgs) >= ag.CONFIG.max_steps else "done"),
                     "tokens": sum(ag.usage_of(msgs)),
                     "usd": ag.cost_usd(msgs), "doc": docs[-1] if docs else None})
print(f"{'runtime':<13} {'steps':>5} {'tokens':>7} {'usd':>9} {'LoC':>4}  stop reasons")
for name in RUNNERS:
    mine = [r for r in rows if r["runner"] == name]
    loc = sum(len(inspect.getsource(fn).splitlines()) for fn in LOC[name])
    print(f"{name:<13} {sum(r['steps'] for r in mine):>5} {sum(r['tokens'] for r in mine):>7} "
          f"{sum(r['usd'] for r in mine):>9.5f} {loc:>4}  {sorted({r['stop_reason'] for r in mine})}")
same = sum(len({r["doc"] for r in rows if r["issue"] == i}) == 1 for i in issues)
print(f"same doc read by all three runtimes: {same}/{len(issues)}")
Path("reports").mkdir(exist_ok=True)
Path("reports/d53_parity.json").write_text(json.dumps(rows, indent=2))
print("wrote reports/d53_parity.json")

runtime       steps  tokens       usd  LoC  stop reasons
own loop         15    7328   0.00097   17  ['done']
create_agent     15    7328   0.00097    4  ['done']
StateGraph       15    7328   0.00097   40  ['done']
same doc read by all three runtimes: 5/5
wrote reports/d53_parity.json