from langgraph.graph import StateGraph
from langgraph.types import CachePolicy
from langgraph.cache.sqlite import SqliteCache

graph = StateGraph(State)
graph.add_node(
    "expensive_fetch",
    fetch_fn,
    cache_policy=CachePolicy(ttl=3600, key_func=lambda s: s["query"]),
)
# RedisCache(url=...) for distributed workers
compiled = graph.compile(cache=SqliteCache("cache.db"))

# defer=True — node execution is deferred until the run is about to end,
# i.e. after every other upstream node has completed
graph.add_node("aggregate", aggregate_fn, defer=True)

# Model middleware — no subclassing required.
# create_react_agent is @deprecated since v1.0; use create_agent from langchain.agents.
# The legacy pre_model_hook/post_model_hook are now before_model/after_model middleware.
from langchain.agents import create_agent

agent = create_agent(
    model=model,
    tools=tools,
    middleware=[compress_history, redact_pii],  # before_model / after_model hooks
    system_prompt="...",                          # prompt= renamed to system_prompt
)

from langgraph.graph import StateGraph, START, END
from langgraph.types import Command
from typing import TypedDict, Annotated, Literal
from operator import add

class State(TypedDict):
    input: str
    results: Annotated[list[str], add]

def supervisor(state) -> Command[Literal["worker", END]]:
    if not state.get("results"):
        return Command(update={"input": state["input"]}, goto="worker")
    return Command(goto=END)

def worker(state) -> dict:
    return {"results": [f"Processed: {state['input']}"]}

graph = StateGraph(State)
graph.add_node("supervisor", supervisor)
graph.add_node("worker", worker)
graph.add_edge(START, "supervisor")
graph.add_edge("worker", "supervisor")
app = graph.compile()

# If this crashes at step 5 of 10, all progress is lost
result = app.invoke(initial_state)

import logging

async def run_with_recovery(workflow_id: str, initial_state: dict):
    """Run workflow with automatic recovery from checkpoint."""
    config = {"configurable": {"thread_id": workflow_id}}

    try:
        state = app.get_state(config)
        if state.values:
            logging.info(f"Resuming workflow {workflow_id}")
            return app.invoke(None, config=config)  # None = resume from checkpoint
    except Exception:
        pass  # No existing checkpoint

    logging.info(f"Starting new workflow {workflow_id}")
    return app.invoke(initial_state, config=config)

# Get all checkpoints for a workflow
config = {"configurable": {"thread_id": "analysis-123"}}
for checkpoint in app.get_state_history(config):
    print(f"Step: {checkpoint.metadata['step']}")
    print(f"Node: {checkpoint.metadata['source']}")
    print(f"State: {checkpoint.values}")

# Rollback to previous checkpoint
history = list(app.get_state_history(config))
previous = history[1]  # One step back
app.update_state(config, previous.values)

app = workflow.compile()  # No checkpointer — progress lost on crash
result = app.invoke(state)  # Can't resume if interrupted

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.postgres import PostgresSaver

# Development: In-memory (fast, no setup)
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

# Production: PostgreSQL (shared, durable)
# from_conn_string is a @contextmanager — setup() MUST be called once
with PostgresSaver.from_conn_string("postgresql://...") as checkpointer:
    checkpointer.setup()
    app = workflow.compile(checkpointer=checkpointer)

    # Invoke with thread_id for resumability
    config = {"configurable": {"thread_id": "analysis-123"}}
    result = app.invoke(initial_state, config=config)

# User preferences stored in thread-1 state
# When user starts thread-2, preferences are lost!
with PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer:
    checkpointer.setup()
    app = workflow.compile(checkpointer=checkpointer)

from langgraph.checkpoint.postgres import PostgresSaver
from langgraph.store.postgres import PostgresStore

# from_conn_string is a @contextmanager — setup() MUST be called once
# Checkpointer = SHORT-TERM (thread-scoped)
# Store = LONG-TERM (cross-thread, namespaced)
with (
    PostgresSaver.from_conn_string(DATABASE_URL) as checkpointer,
    PostgresStore.from_conn_string(DATABASE_URL) as store,
):
    checkpointer.setup()
    store.setup()

    # Compile with BOTH
    app = workflow.compile(checkpointer=checkpointer, store=store)

from langgraph.store.base import BaseStore

async def agent_with_memory(state: AgentState, *, store: BaseStore):
    user_id = state["user_id"]

    # Read cross-thread memory
    memories = await store.aget(namespace=("users", user_id), key="preferences")
    if memories and memories.value.get("prefers_concise"):
        state["system_prompt"] += "\nBe concise."

    # Write cross-thread memory
    await store.aput(
        namespace=("users", user_id),
        key="last_topic",
        value={"topic": state["current_topic"], "timestamp": datetime.now().isoformat()}
    )
    return state

Thread 1 (chat-001)    Thread 2 (chat-002)
┌─────────────────┐    ┌─────────────────┐
│ Checkpointer    │    │ Checkpointer    │
│ - msg history   │    │ - msg history   │
│ - workflow pos  │    │ - workflow pos  │
└─────────────────┘    └─────────────────┘
         ↓ shared ↓
┌─────────────────────────────────────┐
│     Store (cross-thread)            │
│  namespace=("users", "alice")       │
│  - preferences, last_topic          │
└─────────────────────────────────────┘

@entrypoint()
def my_workflow(data: str) -> str:
    result = expensive_task(data).result()
    return result
    # If interrupted, all progress lost — no checkpointer

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import InMemorySaver

checkpointer = InMemorySaver()

@entrypoint(checkpointer=checkpointer)
def my_workflow(data: str) -> str:
    result = expensive_task(data).result()
    return result

# Invoke with thread_id for resumability
config = {"configurable": {"thread_id": "session-123"}}
result = my_workflow.invoke("input", config)

from langgraph.types import interrupt, Command

@entrypoint(checkpointer=checkpointer)
def approval_workflow(request: dict) -> dict:
    result = analyze_request(request).result()

    approved = interrupt({"question": "Approve?", "details": result})

    if approved:
        return execute_action(result).result()
    return {"status": "rejected"}

# Resume after human review
for chunk in approval_workflow.stream(Command(resume=True), config):
    print(chunk)

from langgraph.graph import StateGraph

def node_a(state):
    return {"data": process(state["input"])}

def node_b(state):
    return {"result": transform(state["data"])}

graph = StateGraph(State)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge("a", "b")
app = graph.compile()

from langgraph.func import entrypoint, task

@task
def process_data(input: str) -> str:
    return process(input)

@task
def transform_data(data: str) -> str:
    return transform(data)

@entrypoint()
def workflow(input: str) -> str:
    data = process_data(input).result()
    return transform_data(data).result()

@task
def plan(topic: str) -> list[str]:
    return planner.invoke(f"Create outline for: {topic}")

@task
def write_section(section: str) -> str:
    return llm.invoke(f"Write section: {section}")

@entrypoint()
def report_workflow(topic: str) -> str:
    sections = plan(topic).result()
    section_futures = [write_section(s) for s in sections]  # Fan-out
    completed = [f.result() for f in section_futures]         # Fan-in
    return "\n\n".join(completed)

import { entrypoint, task, MemorySaver } from "@langchain/langgraph";

const processData = task("processData", async (data: string) => transform(data));

const workflow = entrypoint(
  { name: "myWorkflow", checkpointer: new MemorySaver() },
  async (input: string) => {
    const result = await processData(input);
    return result;
  }
);

from langgraph.graph import StateGraph

def node_a(state):
    return {"data": process(state["input"])}

def node_b(state):
    return {"result": transform(state["data"])}

# Verbose for simple linear flow
graph = StateGraph(State)
graph.add_node("a", node_a)
graph.add_node("b", node_b)
graph.add_edge("a", "b")
app = graph.compile()

from langgraph.func import entrypoint, task

@task
def process_data(input: str) -> str:
    return process(input)

@task
def transform_data(data: str) -> str:
    return transform(data)

@entrypoint()
def workflow(input: str) -> str:
    data = process_data(input).result()  # Clean, simple
    return transform_data(data).result()

@task
def process(data: str) -> str:
    return transform(data)

@entrypoint()
def workflow(data: str) -> str:
    result = process(data)  # result is a Future, not a string!
    return result  # Returns Future object, not the processed string

@task
def fetch_source_a(query: str) -> dict:
    return api_a.search(query)

@task
def fetch_source_b(query: str) -> dict:
    return api_b.search(query)

@entrypoint()
def parallel_search(query: str) -> dict:
    # Launch in parallel — futures start immediately
    future_a = fetch_source_a(query)
    future_b = fetch_source_b(query)

    # Block on both results
    results = [future_a.result(), future_b.result()]
    return {"combined": results}

@task
def process_item(item: dict) -> dict:
    return transform(item)

@entrypoint()
def batch_workflow(items: list[dict]) -> list[dict]:
    futures = [process_item(item) for item in items]  # All launch in parallel
    return [f.result() for f in futures]  # Collect all results

app = workflow.compile(interrupt_before=["publish"])

# Human reviews...
state.values["approved"] = True
app.update_state(config, state.values)
result = app.invoke(None, config)
# What if human rejects? No path!

def approval_gate(state) -> str:
    if not state.get("human_reviewed"):
        return state  # Pauses here due to interrupt_before

    if state["approved"]:
        return {"next": "publish"}
    else:
        return {"next": "revise"}

# Compile with interrupt
app = workflow.compile(interrupt_before=["approval_gate"])

# Step 1: Run until approval gate
config = {"configurable": {"thread_id": "doc-123"}}
result = app.invoke({"topic": "AI"}, config=config)

# Step 2: Human reviews
state = app.get_state(config)
print(f"Draft: {state.values['draft']}")

# Step 3: Human decides
app.update_state(config, {
    "approved": True,
    "feedback": "Looks good",
    "human_reviewed": True,
})

# Step 4: Resume
result = app.invoke(None, config=config)

app = workflow.compile(interrupt_before=["first_review", "final_review"])

def review(state):
    feedback = get_human_feedback()
    state["feedback"] = feedback
    # Proceeds regardless of whether human approved

from langgraph.types import interrupt, Command

async def run_with_feedback(initial_state: dict):
    config = {"configurable": {"thread_id": str(uuid.uuid4())}}

    while True:
        result = app.invoke(initial_state, config=config)

        if "__interrupt__" not in result:
            return result  # Completed without interrupt

        info = result["__interrupt__"][0].value
        print(f"Output: {info.get('output', 'N/A')}")
        feedback = input("Approve? (yes/no/feedback): ")

        if feedback.lower() == "yes":
            return app.invoke(Command(resume={"approved": True}), config=config)
        elif feedback.lower() == "no":
            return {"status": "rejected"}
        else:
            # Incorporate feedback and retry
            initial_state = None
            app.invoke(
                Command(resume={"approved": False, "feedback": feedback}),
                config=config
            )

def get_valid_age(state):
    prompt = "What is your age?"
    while True:
        answer = interrupt(prompt)
        if isinstance(answer, int) and 0 < answer < 150:
            return {"age": answer}
        prompt = f"'{answer}' is not valid. Enter 1-150."

def approval_node(state):
    try:
        response = interrupt({"question": "Approve?"})  # interrupt raises!
    except Exception:
        response = {"approved": False}  # Catches the interrupt exception — broken

from langgraph.types import interrupt, Command

def approval_node(state):
    """Conditionally interrupt based on risk level."""
    if state["risk_level"] == "high":
        response = interrupt({
            "question": "High-risk action. Approve?",
            "action": state["proposed_action"],
            "risk_level": state["risk_level"],
        })
        if not response.get("approved"):
            return {"status": "rejected", "action": None}

    return {"status": "approved", "action": state["proposed_action"]}

config = {"configurable": {"thread_id": "workflow-123"}}

# Initial run — stops at interrupt
result = graph.invoke(initial_state, config)

# Check for interrupt
if "__interrupt__" in result:
    info = result["__interrupt__"][0].value
    print(f"Question: {info['question']}")

    # Resume with user response
    final = graph.invoke(Command(resume={"approved": True}), config)

async def run_agents(state):
    tasks = [agent.analyze(state["content"]) for agent in agents]
    results = await asyncio.gather(*tasks)  # One exception crashes everything
    return {"results": results}

import asyncio

async def parallel_with_isolation(agents: list, content: str, timeout: int = 30):
    """Run agents with per-agent timeout and error isolation."""
    async def run_with_timeout(agent):
        try:
            return await asyncio.wait_for(agent.analyze(content), timeout=timeout)
        except asyncio.TimeoutError:
            return {"agent": agent.name, "error": "timeout"}
        except Exception as e:
            return {"agent": agent.name, "error": str(e)}

    tasks = [run_with_timeout(a) for a in agents]
    results = await asyncio.gather(*tasks, return_exceptions=True)

    successes = [r for r in results if not isinstance(r, Exception) and "error" not in r]
    failures = [r for r in results if isinstance(r, Exception) or "error" in r]

    return {"successes": successes, "failures": failures}

def process_all(state):
    results = []
    for task in state["tasks"]:
        result = process(task)  # Sequential!
        results.append(result)
    return {"results": results}

from langgraph.graph import StateGraph, START, END
from langgraph.types import Send
from typing import TypedDict, Annotated
from operator import add

class OverallState(TypedDict):
    subjects: list[str]
    jokes: Annotated[list[str], add]  # Accumulates from parallel branches

class JokeState(TypedDict):
    subject: str

def continue_to_jokes(state: OverallState) -> list[Send]:
    """Fan-out: create parallel branch for each subject."""
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]

def generate_joke(state: JokeState) -> dict:
    """Worker: each runs in parallel, writes to accumulator."""
    joke = llm.invoke(f"Tell a joke about {state['subject']}")
    return {"jokes": [f"{state['subject']}: {joke.content}"]}

builder = StateGraph(OverallState)
builder.add_node("generate_topics", generate_topics)
builder.add_node("generate_joke", generate_joke)
builder.add_edge(START, "generate_topics")
builder.add_conditional_edges("generate_topics", continue_to_jokes)
builder.add_edge("generate_joke", END)  # All branches converge automatically

async def process_all(state):
    results = []
    for item in state["items"]:
        result = await process_item(item)  # Sequential await!
        results.append(result)
    return {"results": results}

import asyncio

async def parallel_map(items: list, process_fn) -> list:
    """Map: process all items concurrently."""
    tasks = [asyncio.create_task(process_fn(item)) for item in items]
    return await asyncio.gather(*tasks, return_exceptions=True)

def reduce_results(results: list) -> dict:
    """Reduce: combine all results."""
    successes = [r for r in results if not isinstance(r, Exception)]
    failures = [r for r in results if isinstance(r, Exception)]
    return {
        "total": len(results),
        "passed": len(successes),
        "failed": len(failures),
        "results": successes,
        "errors": [str(e) for e in failures],
    }

async def map_reduce_node(state: State) -> dict:
    results = await parallel_map(state["items"], process_item_async)
    return {"summary": reduce_results(results)}

# Running graph directly without Platform config
python my_graph.py  # No API server, no persistence, no multi-tenancy

{
  "dependencies": ["langchain_openai", "./my_package"],
  "graphs": {
    "my_agent": "./my_package/agent.py:graph",
    "my_workflow": "./my_package/workflow.py:app"
  },
  "env": "./.env"
}

# Local development (hot reload, in-memory)
langgraph dev

# Build Docker image for self-hosted deployment
langgraph build -t my-agent:latest

# Deploy to LangGraph Cloud
langgraph deploy

from langgraph_sdk import get_client

client = get_client(url="http://localhost:2024")

# Create assistants with different configs from the same graph
creative = await client.assistants.create(
    graph_id="my_agent",
    config={"configurable": {"temperature": 0.9, "model": "claude-sonnet-4-6"}},
    name="creative-writer",
)

precise = await client.assistants.create(
    graph_id="my_agent",
    config={"configurable": {"temperature": 0.1, "model": "claude-sonnet-4-6"}},
    name="precise-analyst",
)

# Fire-and-forget with webhook callback
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id=creative["assistant_id"],
    input={"messages": [{"role": "user", "content": "Write a story"}]},
    webhook="https://myapp.com/api/langgraph-callback",
    background=True,
)

from langgraph.func import entrypoint, task
from langgraph.checkpoint.memory import MemorySaver

@entrypoint(path="my_workflow", checkpointer=MemorySaver())
def my_workflow(inputs: dict) -> dict:
    result = analyze_task(inputs["query"]).result()
    return {"answer": result}

from langgraph_sdk import Auth

auth = Auth()

@auth.authenticate
async def authenticate(authorization: str) -> str:
    # Validate token, return user_id
    user = await verify_token(authorization)
    return user["id"]

@auth.on
async def authorize(params, user_id: str):
    # Filter resources by user ownership
    params["metadata"] = {"owner": user_id}

User: "Analyze Q1 data"     → Run starts...
User: "Actually use Q2 data" → Second run starts on same thread
# Both runs mutate state concurrently — corrupted results

from langgraph_sdk import get_client

client = get_client(url="http://localhost:2024")

# New input rejected while run is active
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "Analyze Q2"}]},
    multitask_strategy="reject",
)
# Raises 409 Conflict if a run is already active

# Cancel current run, roll back state, start fresh with new input
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "Use Q2 data instead"}]},
    multitask_strategy="rollback",
)
# Previous run cancelled, state rolled back to before it started

# Queue new input to run after current run completes
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "Also check Q3"}]},
    multitask_strategy="enqueue",
)
# Runs sequentially: current run finishes, then queued run starts

# Interrupt current run but keep its state, then continue with new input
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "Focus on revenue"}]},
    multitask_strategy="interrupt",
)
# Current run stops mid-execution, new run builds on existing state

# No multitask_strategy — concurrent messages corrupt state
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "Analyze Q2"}]},
)

# Explicit strategy prevents race conditions
run = await client.runs.create(
    thread_id=thread["thread_id"],
    assistant_id="my_agent",
    input={"messages": [{"role": "user", "content": "Analyze Q2"}]},
    multitask_strategy="reject",  # or rollback/enqueue/interrupt
)

import httpx

# Manual API calls lose type safety and streaming support
response = httpx.post(
    "http://localhost:2024/runs",
    json={"input": {"messages": [{"role": "user", "content": "hello"}]}},
)

from langgraph.pregel.remote import RemoteGraph

remote = RemoteGraph(
    "my_agent",
    url="http://localhost:2024",
)

# Invoke like a local graph
result = remote.invoke(
    {"messages": [{"role": "user", "content": "Analyze this data"}]},
    config={"configurable": {"thread_id": "remote-123"}},
)

from langgraph.pregel.remote import RemoteGraph

remote = RemoteGraph(
    "my_agent",
    url="http://localhost:2024",
)

result = await remote.ainvoke(
    {"messages": [{"role": "user", "content": "Analyze this data"}]},
    config={"configurable": {"thread_id": "remote-123"}},
)

# Stream updates in real-time
async for chunk in remote.astream(
    {"messages": [{"role": "user", "content": "Write a report"}]},
    config={"configurable": {"thread_id": "stream-456"}},
    stream_mode="updates",
):
    print(chunk)

# Stream events for fine-grained control
async for event in remote.astream_events(
    {"messages": [{"role": "user", "content": "Summarize"}]},
    config={"configurable": {"thread_id": "events-789"}},
    version="v2",
):
    if event["event"] == "on_chat_model_stream":
        print(event["data"]["chunk"].content, end="")

from langgraph.graph import StateGraph, START, END
from langgraph.pregel.remote import RemoteGraph

remote_analyzer = RemoteGraph("analyzer", url="http://analyzer:2024")

builder = StateGraph(OrchestratorState)
builder.add_node("analyze", remote_analyzer)  # Remote graph as local node
builder.add_node("summarize", summarize_node)
builder.add_edge(START, "analyze")
builder.add_edge("analyze", "summarize")
builder.add_edge("summarize", END)

def route(state) -> str:
    if state["quality_score"] >= 0.8:
        return "publish"
    elif state["retry_count"] < 3:
        return "retry"
    # No fallback — if quality < 0.8 AND retries exhausted, returns None → runtime error

workflow.add_conditional_edges("check", route)
# No explicit mapping — unclear what routes exist

from langgraph.graph import END

def route_based_on_quality(state: WorkflowState) -> str:
    if state["quality_score"] >= 0.8:
        return "publish"
    elif state["retry_count"] < 3:
        return "retry"
    else:
        return "manual_review"  # Always have a fallback

workflow.add_conditional_edges(
    "quality_check",
    route_based_on_quality,
    {
        "publish": "publish_node",
        "retry": "generator",
        "manual_review": "review_queue",
    }
)

Sequential:    A → B → C              (simple edges)
Branching:     A → (B or C)           (conditional edges)
Looping:       A → B → A              (retry logic)
Convergence:   (A or B) → C           (multiple inputs)
Diamond:       A → (B, C) → D         (parallel then merge)

def child_node(state: ChildState) -> Command[Literal["escalate"]]:
    if state["needs_escalation"]:
        return Command(goto="escalate")  # Looks for "escalate" in child graph — fails!

from langgraph.types import Command
from typing_extensions import Literal

def child_node(state: ChildState) -> Command[Literal["supervisor"]]:
    if state["needs_escalation"]:
        return Command(
            update={"escalation_reason": "Complexity exceeded threshold"},
            goto="supervisor",
            graph=Command.PARENT,  # Navigate to parent graph
        )
    return Command(update={"result": "handled"}, goto="next_step")

from langgraph.graph import StateGraph, START, END
from typing import TypedDict, Annotated
from operator import add

class ParentState(TypedDict):
    query: str
    escalation_reason: str
    results: Annotated[list[str], add]  # Reducer required for shared keys

class ChildState(TypedDict):
    query: str
    escalation_reason: str

# Child subgraph
child_builder = StateGraph(ChildState)
child_builder.add_node("analyze", child_node)
child_builder.add_edge(START, "analyze")
child_graph = child_builder.compile()

# Parent graph
parent_builder = StateGraph(ParentState)
parent_builder.add_node("child_agent", child_graph)
parent_builder.add_node("supervisor", supervisor_node)
parent_builder.add_node("specialist", specialist_node)
parent_builder.add_edge(START, "child_agent")
parent_builder.add_edge("supervisor", "specialist")
parent_builder.add_edge("specialist", END)

app = parent_builder.compile()

def agent_a_node(state: AgentAState) -> Command[Literal["agent_b"]]:
    if state["needs_different_expertise"]:
        return Command(
            update={"handoff_context": state["partial_result"]},
            goto="agent_b",          # Sibling node in parent graph
            graph=Command.PARENT,    # Navigate up to parent first
        )
    return Command(update={"result": "done"}, goto=END)

class ParentState(TypedDict):
    query: str
    context: Annotated[list[str], add]  # Must have reducer for updates from children

class ResearcherState(TypedDict):
    query: str
    context: str  # Different type than parent — mapping happens at boundary

def researcher_node(state: ResearcherState) -> Command[Literal["writer"]]:
    finding = research(state["query"])
    return Command(
        update={"context": [finding]},  # Match parent's list type with reducer
        goto="writer",
        graph=Command.PARENT,
    )

def should_retry(state) -> str:
    if state.get("output"):
        return "success"
    return "retry"  # No max counter — infinite loop if LLM keeps failing

workflow.add_conditional_edges("llm_call", should_retry, {
    "success": "next_step",
    "retry": "llm_call",  # Loops forever
})

def llm_call_with_retry(state):
    try:
        result = call_llm(state["input"])
        return {"output": result, "retry_count": 0}
    except Exception as e:
        return {
            "retry_count": state.get("retry_count", 0) + 1,
            "error": str(e),
        }

def should_retry(state) -> str:
    if state.get("output"):
        return "success"
    elif state["retry_count"] < 3:
        return "retry"
    else:
        return "failed"  # Max retries exceeded

workflow.add_conditional_edges("llm_call", should_retry, {
    "success": "next_step",
    "retry": "llm_call",
    "failed": "error_handler",
})

def router(state):
    state["route_reason"] = "high score"  # State update in router
    if state["score"] > 0.8:
        return "approve"
    return "reject"

workflow.add_conditional_edges("evaluate", router)
# State update happens but routing is separate — hard to reason about

from langgraph.types import Command
from typing import Literal

def router_with_state(state) -> Command[Literal["approve", "reject"]]:
    if state["score"] > 0.8:
        return Command(
            update={"route_reason": "high score", "routed_at": time.time()},
            goto="approve"
        )
    return Command(
        update={"route_reason": "low score", "routed_at": time.time()},
        goto="reject"
    )

workflow.add_node("evaluate", router_with_state)
# No conditional edges needed — Command handles both state + routing

from sentence_transformers import SentenceTransformer
import numpy as np

embedder = SentenceTransformer("all-MiniLM-L6-v2")

ROUTE_EMBEDDINGS = {
    "technical": embedder.encode("technical code programming"),
    "business": embedder.encode("business strategy revenue"),
    "support": embedder.encode("help troubleshoot error fix"),
}

def semantic_router(state) -> str:
    query_embedding = embedder.encode(state["query"])
    similarities = {
        route: np.dot(query_embedding, emb) / (
            np.linalg.norm(query_embedding) * np.linalg.norm(emb)
        )
        for route, emb in ROUTE_EMBEDDINGS.items()
    }
    best = max(similarities, key=similarities.get)
    return "general" if similarities[best] < 0.3 else best

class AgentState(TypedDict):
    messages: list  # No reducer — each node REPLACES entire message history
    user_id: str

def agent_node(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # Overwrites all previous messages!

from langgraph.graph import MessagesState
from langgraph.graph.message import add_messages
from typing import Annotated

# Option 1: Extend built-in MessagesState (recommended)
class AgentState(MessagesState):
    user_id: str
    context: dict

# Option 2: Manual add_messages reducer
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]  # Smart append/update by ID
    user_id: str

def agent_node(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response]}  # Appends, doesn't overwrite

from pydantic import BaseModel

# Using Pydantic for ALL internal state — unnecessary runtime validation overhead
class WorkflowState(BaseModel):
    input: str
    intermediate_result: str = ""
    agent_responses: list[dict] = []
    # Every node update triggers Pydantic validation

from typing import TypedDict, Annotated
from operator import add
from pydantic import BaseModel, Field

# Internal state: TypedDict (no runtime overhead)
class WorkflowState(TypedDict):
    input: str
    output: str
    agent_responses: Annotated[list[dict], add]

# Boundary validation: Pydantic (validates external data)
class WorkflowInput(BaseModel):
    input: str = Field(description="User input", min_length=1)
    config: dict = Field(default_factory=dict)

class WorkflowOutput(BaseModel):
    output: str
    confidence: float = Field(ge=0, le=1)

class State(TypedDict):
    config: dict  # No reducer — later nodes overwrite entire dict
    status: str   # No reducer — same issue

def node_a(state):
    return {"config": {"key_a": "value_a"}}

def node_b(state):
    return {"config": {"key_b": "value_b"}}
    # config is now {"key_b": "value_b"} — key_a is LOST

from typing import Annotated

def merge_dicts(a: dict, b: dict) -> dict:
    """Custom reducer that deep-merges dictionaries."""
    return {**a, **b}

def last_value(a, b):
    """Keep only the latest value."""
    return b

class State(TypedDict):
    config: Annotated[dict, merge_dicts]   # Merges updates from all nodes
    status: Annotated[str, last_value]     # Explicit: keeps latest value

def node_a(state):
    return {"config": {"key_a": "value_a"}}

def node_b(state):
    return {"config": {"key_b": "value_b"}}
    # config is now {"key_a": "value_a", "key_b": "value_b"}

from langgraph.types import RemainingSteps

def agent_node(state: WorkflowState, remaining: RemainingSteps):
    if remaining.steps < 5:
        return {"action": "summarize_and_exit"}
    return {"action": "continue"}

from typing import TypedDict

class WorkflowState(TypedDict):
    findings: list[dict]  # No reducer — each node REPLACES the list

def agent_a(state):
    return {"findings": [{"source": "a", "result": "..."}]}

def agent_b(state):
    return {"findings": [{"source": "b", "result": "..."}]}
    # agent_a's findings are LOST

from typing import TypedDict, Annotated
from operator import add

class WorkflowState(TypedDict):
    input: str
    output: str
    findings: Annotated[list[dict], add]  # Accumulates across nodes
    metadata: dict

def agent_a(state):
    return {"findings": [{"source": "a", "result": "..."}]}

def agent_b(state):
    return {"findings": [{"source": "b", "result": "..."}]}
    # Both findings are preserved: [agent_a's, agent_b's]

from dataclasses import dataclass

@dataclass
class ContextSchema:
    llm_provider: str = "anthropic"
    temperature: float = 0.7

graph = StateGraph(WorkflowState, context_schema=ContextSchema)

def my_node(state: WorkflowState, context: ContextSchema):
    return {"output": call_llm(state["input"], context.temperature)}

from langgraph.cache.memory import InMemoryCache
from langgraph.types import CachePolicy

builder.add_node("embed", embed_node, cache_policy=CachePolicy(ttl=300))
graph = builder.compile(cache=InMemoryCache())

def process_items(state):
    for item in state["items"]:
        result = process(item)  # Client sees nothing until ALL items done
    return {"results": results}

from langgraph.config import get_stream_writer

def node_with_progress(state):
    writer = get_stream_writer()

    for i, item in enumerate(state["items"]):
        writer({
            "type": "progress",
            "current": i + 1,
            "total": len(state["items"]),
            "status": f"Processing {item}",
        })
        result = process(item)

    writer({"type": "complete", "message": "All items processed"})
    return {"results": results}

# Consume custom events
for mode, chunk in graph.stream(inputs, stream_mode=["updates", "custom"]):
    if mode == "custom":
        if chunk.get("type") == "progress":
            print(f"Progress: {chunk['current']}/{chunk['total']}")
    elif mode == "updates":
        print(f"State updated: {list(chunk.keys())}")

@app.post("/stream")
async def stream_workflow(request: WorkflowRequest):
    async def event_generator():
        async for mode, chunk in graph.astream(
            request.inputs, stream_mode=["updates", "custom"]
        ):
            yield f"data: {json.dumps({'mode': mode, 'data': chunk})}\n\n"

    return StreamingResponse(event_generator(), media_type="text/event-stream")

for chunk in graph.stream(inputs):  # defaults to "values" — full state every step
    print(chunk)  # Massive output, hard to parse

async for mode, chunk in graph.astream(
    inputs,
    stream_mode=["updates", "custom", "messages"]
):
    match mode:
        case "updates":
            update_ui_state(chunk)
        case "custom":
            show_progress(chunk)
        case "messages":
            append_to_chat(chunk)

for namespace, chunk in graph.stream(inputs, subgraphs=True, stream_mode="updates"):
    print(f"[{'/'.join(namespace) or 'root'}] {chunk}")

for msg, meta in graph.stream(inputs, stream_mode="messages"):
    print(msg.content, end="")  # Shows tokens from ALL LLM calls — mixed output

for msg, meta in graph.stream(inputs, stream_mode="messages"):
    if meta["langgraph_node"] == "writer_agent":
        print(msg.content, end="", flush=True)

model = init_chat_model("claude-sonnet-4-6", tags=["main_response"])

for msg, meta in graph.stream(inputs, stream_mode="messages"):
    if "main_response" in meta.get("tags", []):
        print(msg.content, end="", flush=True)

from langgraph.config import get_stream_writer

def call_custom_llm(state):
    writer = get_stream_writer()
    for chunk in your_streaming_client.generate(state["prompt"]):
        writer({"type": "llm_token", "content": chunk.text})
    return {"response": full_response}

result = graph.invoke({"input": "hello"})
output = result["output"]  # Any — no type info
interrupts = result.get("__interrupt__")  # magic key

from langgraph.types import GraphOutput

result = graph.invoke({"input": "hello"}, version="v2")
result.value       # OutputT — auto-coerced to Pydantic/dataclass if typed
result.interrupts  # tuple[Interrupt, ...] — replaces __interrupt__ key

async for part in graph.astream(
    inputs,
    stream_mode=["updates", "messages"],
    version="v2",
):
    # part is a typed StreamPart discriminated union
    if part["type"] == "updates":
        handle_state_update(part["data"])  # dict[str, Any]
    elif part["type"] == "messages":
        msg, metadata = part["data"]       # tuple[AnyMessage, dict]
    elif part["type"] == "custom":
        show_progress(part["data"])        # Any — StreamWriter content

class MyState(BaseModel):
    answer: str

result = compiled.invoke({...}, version="v2")
assert isinstance(result.value, MyState)  # auto-coerced — not a plain dict

def call_agent(state: SharedState):
    # Unnecessary transformation when schemas match
    input = {"messages": state["messages"], "context": state["context"]}
    output = agent_subgraph.invoke(input)
    return {"messages": output["messages"], "context": output["context"]}

from langgraph.graph.message import add_messages

class SharedState(TypedDict):
    messages: Annotated[list, add_messages]
    context: dict

# Build subgraph with SAME state
agent_builder = StateGraph(SharedState)
agent_builder.add_node("think", think_node)
agent_builder.add_node("act", act_node)
agent_builder.add_edge(START, "think")
agent_builder.add_edge("think", "act")
agent_builder.add_edge("act", END)
agent_subgraph = agent_builder.compile()

# Add compiled subgraph directly as node — no wrapper needed
parent_builder = StateGraph(SharedState)
parent_builder.add_node("agent_team", agent_subgraph)
parent_builder.add_edge(START, "agent_team")
parent_builder.add_edge("agent_team", END)

def call_analysis(state: ParentState):
    result = analysis_subgraph.invoke(state)  # Schema mismatch! KeyError
    return result  # Wrong schema returned to parent

class ParentState(TypedDict):
    query: str
    analysis_result: dict

class AnalysisState(TypedDict):
    input_text: str
    findings: list[str]
    score: float

# Build and compile subgraph
analysis_builder = StateGraph(AnalysisState)
analysis_builder.add_node("analyze", analyze_node)
analysis_builder.add_node("score", score_node)
analysis_builder.add_edge(START, "analyze")
analysis_builder.add_edge("analyze", "score")
analysis_builder.add_edge("score", END)
analysis_subgraph = analysis_builder.compile()

def call_analysis(state: ParentState) -> dict:
    # Map parent → subgraph state
    subgraph_input = {"input_text": state["query"], "findings": [], "score": 0.0}

    # Invoke subgraph
    subgraph_output = analysis_subgraph.invoke(subgraph_input)

    # Map subgraph → parent state
    return {
        "analysis_result": {
            "findings": subgraph_output["findings"],
            "score": subgraph_output["score"],
        }
    }

parent_builder.add_node("analysis", call_analysis)

def call_subgraph(state: ParentState):
    result = subgraph.invoke({"query": state["query"]})  # No config — breaks tracing
    return {"result": result["output"]}

from langgraph.config import get_runnable_config

def call_subgraph_with_mapping(state: ParentState) -> dict:
    # 1. Extract relevant data
    subgraph_input = {
        "query": state["user_query"],
        "context": state.get("context", {}),
        "history": [],
    }

    # 2. Propagate config for tracing/checkpointing
    config = get_runnable_config()
    result = subgraph.invoke(subgraph_input, config)

    # 3. Transform output back
    return {
        "subgraph_result": result["output"],
        "metadata": {"subgraph": "analysis", "steps": result.get("step_count", 0)},
    }

# Parent-only (recommended) — propagates to all subgraphs
parent = parent_builder.compile(checkpointer=PostgresSaver(...))

# Independent subgraph memory — for persistent agent histories
agent_subgraph = agent_builder.compile(checkpointer=True)

for namespace, chunk in graph.stream(inputs, subgraphs=True, stream_mode="updates"):
    depth = len(namespace)
    print(f"{'  ' * depth}[{'/'.join(namespace) or 'root'}] {chunk}")