DOCSEARCHbyAlgolia
Submit Docs

LangGraph

repository·main·Indexed 12 days ago

https://github.com/langchain-ai/langgraph

A low-level orchestration framework for building, managing, and deploying long-running, stateful agents. It provides tools for complex workflows requiring durability, human-in-the-loop interaction, and memory management, including checkpoint savers like PostgresSaver and a conformance test suite for validating custom checkpointer implementations.

Tokens
50.3K
Snippets
151
Records
200
Agent score
99%

What's inside LangGraph

  1. What is LangGraph and why use it?

    main

    LangGraph is a low-level orchestration framework designed for building, managing, and deploying long-running, stateful agents. Unlike simple chains, LangGraph provides infrastructure for complex workflows including:

    • Durable execution: Agents can persist through failures and resume from their last state.
    • Human-in-the-loop: Allows for inspecting and modifying agent state during execution for human oversight.
    • Comprehensive memory: Supports both short-term working memory (for reasoning) and long-term persistent memory (across sessions).
    • Debugging and Observability: Integrates with LangSmith to trace execution paths, capture state transitions, and provide runtime metrics.
    • Production-ready deployment: Designed for scalable, stateful, long-running workflows.

  2. What is the LangGraph Python SDK?

    main

    The LangGraph Python SDK is a client library used to connect to a running LangGraph API server. It allows Python applications to manage assistants and threads, and to stream runs from the server.

    If you are running a server locally via langgraph-cli, the SDK defaults to http://localhost:8123. For remote servers, you must specify the URL during client initialization.

  4. What is LangGraph and when to use it

    main

    LangGraph is a low-level orchestration framework designed for building, managing, and deploying long-running, stateful agents. It provides infrastructure for:

    • Durable execution
    • Streaming
    • Human-in-the-loop interactions
    • Persistence and memory

    When to use LangGraph vs LangChain

    • Use LangGraph when you have advanced needs requiring a combination of deterministic and agentic workflows, heavy customization, and carefully controlled latency.
    • Use LangChain when you want to quickly build agents and applications using pre-built agent architectures and model integrations. Note that LangChain agents are actually built on top of LangGraph to provide their durability and streaming capabilities.

  5. What is LangGraph Checkpoint Conformance?

    main
    LangGraph Checkpoint Conformance is a test suite used to validate that a BaseCheckpointSaver subclass correctly implements the LangGraph checkpoint storage contract. It verifies critical behaviors such as blob round-trips, metadata preservation, namespace isolation, and incremental channel updates to ensure your checkpointer is production-ready.

  6. How LangGraph Checkpoints and Threads work

    main

    LangGraph uses checkpointers to provide a persistence layer, enabling features like human-in-the-loop, memory between interactions, and durable execution.

    Key concepts:

    • Checkpoint: A snapshot of the graph state at a specific point in time. A checkpoint tuple includes the checkpoint itself, associated configuration, metadata, and pending writes.
    • Thread: A unique ID assigned to a series of checkpoints. Threads allow for maintaining separate states for different runs, which is essential for multi-tenant applications.

    When running a graph, you must provide a thread_id in the configurable section of the config. You can optionally provide a checkpoint_id to resume a run from a specific point within a thread.

    # Required thread_id
{"configurable": {"thread_id": "1"}}

# Optional checkpoint_id to resume from a specific point
{"configurable": {"thread_id": "1", "checkpoint_id": "0c62ca34-ac19-445d-bbb0-5b4984975b2a"}}

  7. Use Agent Inbox schemas for interrupts

    main

    LangGraph Prebuilt includes schemas for integrating with Agent Inbox to handle human-in-the-loop interrupts. You can use HumanInterrupt to request specific actions from a user and HumanResponse to process their reply.

    from langgraph.types import interrupt
from langgraph.prebuilt.interrupt import HumanInterrupt, HumanResponse

def my_graph_function():
    # Extract the last tool call from the `messages` field in the state
    tool_call = state["messages"][-1].tool_calls[0]
    # Create an interrupt
    request: HumanInterrupt = {
        "action_request": {
            "action": tool_call['name'],
            "args": tool_call['args']
        },
        "config": {
            "allow_ignore": True,
            "allow_respond": True,
            "allow_edit": False,
            "allow_accept": False
        },
        "description": _generate_email_markdown(state) # Generate a detailed markdown description.
    }
    # Send the interrupt request inside a list, and extract the first response
    response = interrupt([request])[0]
    if response['type'] == "response":
        # Do something with the response
    ...

  8. Configure Serde security for checkpoint deserialization

    main

    The langgraph-checkpoint library defines a protocol for serialization/deserialization (serde). The default implementation is langgraph.checkpoint.serde.jsonplus.JsonPlusSerializer, which handles LangChain/LangGraph primitives, datetimes, and enums.

    Security Warning: By default, the serializer allows any Python type found in checkpoint data. To restrict deserialization to known-safe types, you should either:

    1. Set the environment variable LANGGRAPH_STRICT_MSGPACK=true.
    2. Pass an explicit allowed_msgpack_modules list to the JsonPlusSerializer.

  9. How Thread-Centric Streaming (v3) works

    main

    In version 3, client.threads.stream() returns a context manager that owns a single SSE (Server-Sent Events) session for a specific thread.

    Instead of opening multiple connections, you can use various typed projections—such as thread.messages, thread.tool_calls, or thread.output—which all share the same underlying connection. To maximize efficiency, you should run these consumers concurrently (e.g., using asyncio.gather) so they utilize the single SSE connection.

    from langgraph_sdk import get_client
import asyncio

client = get_client()

async with client.threads.stream(
    thread_id="my-thread",
    assistant_id="agent",
) as thread:
    # Start the run
    await thread.run.start(input={"messages": [{"role": "user", "content": "hi"}]})

    # Define consumers that share the same SSE connection
    async def get_messages():
        return [s async for s in thread.messages]

    async def get_tool_calls():
        return [c async for c in thread.tool_calls]

    # Run consumers concurrently
    messages, tool_calls = await asyncio.gather(get_messages(), get_tool_calls())

    for stream in messages:
        print(await stream.text)  # accumulated text

    final = await thread.output  # terminal state values

  10. LangGraph Ecosystem and Integrations

    main

    LangGraph can be used standalone or integrated with the LangChain ecosystem:

    • LangChain: Provides the composable components and integrations used within LangGraph workflows.
    • LangSmith: Used for agent evaluations, observability, debugging trajectories, and gaining production visibility.
    • LangSmith Deployment: A platform to deploy, scale, and share stateful agents, including visual prototyping in LangSmith Studio.
    • Deep Agents: A higher-level abstraction built on top of LangGraph for complex planning and subagent tasks.

    For JavaScript/TypeScript users, an equivalent library is available as LangGraph.js.

  12. Consume multiple projections concurrently in v3

    main

    Because all projections (e.g., messages, tool_calls) share a single SSE connection, you can consume them in parallel. Use asyncio.gather or asyncio.TaskGroup to start multiple consumers before any single projection finishes. The fan-out task will route events to all subscribers simultaneously.

    async with client.threads.stream(assistant_id="agent") as thread:
    await thread.run.start(input={"messages": [{"role": "user", "content": "hi"}]})

    async def collect_messages():
        return [s async for s in thread.messages]

    async def collect_tool_calls():
        return [c async for c in thread.tool_calls]

    messages, tool_calls = await asyncio.gather(
        collect_messages(),
        collect_tool_calls(),
    )