### Install LangGraph

Source: https://docs.langchain.com/oss/python/langgraph/overview/index

Instructions for installing the LangGraph library using pip or uv. This is the initial step before using the framework.

```bash
pip install -U langgraph
```

```bash
uv add langgraph
```

--------------------------------

### Install LangGraph Dependencies

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

Installs the necessary LangGraph libraries, including langchain_core, langchain-anthropic, and langgraph, to get started with building LLM-powered applications.

```bash
pip install langchain_core langchain-anthropic langgraph

```

--------------------------------

### LangGraph Hello World Example

Source: https://docs.langchain.com/oss/python/langgraph/overview/index

A simple Python example demonstrating how to create and compile a basic LangGraph using a mock LLM. It defines a state, adds a node, connects edges, and invokes the compiled graph.

```python
from langgraph.graph import StateGraph, MessagesState, START, END

def mock_llm(state: MessagesState):
    return {"messages": [{"role": "ai", "content": "hello world"}]}

graph = StateGraph(MessagesState)
graph.add_node(mock_llm)
graph.add_edge(START, "mock_llm")
graph.add_edge("mock_llm", END)
graph = graph.compile()

graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})
```

--------------------------------

### LLM with Structured Output and Tool Binding (Python)

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

Demonstrates augmenting an LLM for structured output using Pydantic models and binding tools for function calling. It shows how to invoke the LLM to get structured data and trigger tool execution.

```python
# Schema for structured output
from pydantic import BaseModel, Field


class SearchQuery(BaseModel):
    search_query: str = Field(None, description="Query that is optimized web search.")
    justification: str = Field(
        None, description="Why this query is relevant to the user's request."
    )


# Augment the LLM with schema for structured output
structured_llm = llm.with_structured_output(SearchQuery)

# Invoke the augmented LLM
output = structured_llm.invoke("How does Calcium CT score relate to high cholesterol?")

# Define a tool
def multiply(a: int, b: int) -> int:
    return a * b

# Augment the LLM with tools
llm_with_tools = llm.bind_tools([multiply])

# Invoke the LLM with input that triggers the tool call
msg = llm_with_tools.invoke("What is 2 times 3?")

# Get the tool call
msg.tool_calls

```

--------------------------------

### Define LLM with Tools and Build LangGraph Workflow

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This Python snippet showcases how to define tools, bind them to an LLM, and construct a LangGraph workflow. The workflow includes nodes for LLM calls and tool execution, with conditional routing to manage the flow based on LLM decisions. It utilizes `MessagesState` for managing conversation history and tool interactions.

```python
from langgraph.graph import MessagesState, StateGraph, END
from langchain.messages import SystemMessage, HumanMessage, ToolMessage
from typing import Literal
from IPython.display import Image

# Assume 'llm' and 'tools' are defined elsewhere
# Example: 
# from langchain.tools import tool
# @tool
# def add(a: int, b: int) -> int:
#     return a + b
# 
# @tool
# def multiply(a: int, b: int) -> int:
#     return a * b
# 
# @tool
# def divide(a: int, b: int) -> int:
#     return a / b
# 
# tools = [add, multiply, divide]
# llm = ChatOpenAI(model="gpt-4o") # Replace with your LLM

tools_by_name = {tool.name: tool for tool in tools}
llm_with_tools = llm.bind_tools(tools)

# Nodes
def llm_call(state: MessagesState):
    """LLM decides whether to call a tool or not"""
    return {
        "messages": [
            llm_with_tools.invoke(
                [
                    SystemMessage(
                        content="You are a helpful assistant tasked with performing arithmetic on a set of inputs."
                    )
                ]
                + state["messages"]
            )
        ]
    }

def tool_node(state: dict):
    """Performs the tool call"""
    result = []
    for tool_call in state["messages"][-1].tool_calls:
        tool = tools_by_name[tool_call["name"]]
        observation = tool.invoke(tool_call["args"])
        result.append(ToolMessage(content=observation, tool_call_id=tool_call["id"]))
    return {"messages": result}

# Conditional edge function to route to the tool node or end based upon whether the LLM made a tool call
def should_continue(state: MessagesState) -> Literal["tool_node", END]:
    """Decide if we should continue the loop or stop based upon whether the LLM made a tool call"""
    messages = state["messages"]
    last_message = messages[-1]

    # If the LLM makes a tool call, then perform an action
    if last_message.tool_calls:
        return "tool_node"

    # Otherwise, we stop (reply to the user)
    return END

# Build workflow
agent_builder = StateGraph(MessagesState)

# Add nodes
agent_builder.add_node("llm_call", llm_call)
agent_builder.add_node("tool_node", tool_node)

# Add edges to connect nodes
agent_builder.add_edge(START, "llm_call")
agent_builder.add_conditional_edges(
    "llm_call",
    should_continue,
    ["tool_node", END]
)
agent_builder.add_edge("tool_node", "llm_call")

# Compile the agent
agent = agent_builder.compile()

# Show the agent graph (optional)
# display(Image(agent.get_graph(xray=True).draw_mermaid_png()))

# Invoke the agent
# messages = [HumanMessage(content="Add 3 and 4.")]
# messages = agent.invoke({"messages": messages})
# for m in messages["messages"]:
#     m.pretty_print()

```

--------------------------------

### Define and Compile a Router Workflow in Python

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This snippet demonstrates how to define a workflow with a router in LangGraph. It adds nodes, defines conditional edges based on a routing function, compiles the workflow, and visualizes it using a mermaid diagram. Finally, it invokes the workflow with input and prints the output.

```python
from langgraph.graph import START, END
from IPython.display import Image, display

# Assume llm_call_1, llm_call_2, llm_call_3, llm_call_router, route_decision, router_builder are defined elsewhere

# Add nodes
router_builder.add_node("llm_call_1", llm_call_1)
router_builder.add_node("llm_call_2", llm_call_2)
router_builder.add_node("llm_call_3", llm_call_3)
router_builder.add_node("llm_call_router", llm_call_router)

# Add edges to connect nodes
router_builder.add_edge(START, "llm_call_router")
router_builder.add_conditional_edges(
    "llm_call_router",
    route_decision,
    {  # Name returned by route_decision : Name of next node to visit
        "llm_call_1": "llm_call_1",
        "llm_call_2": "llm_call_2",
        "llm_call_3": "llm_call_3",
    },
)
router_builder.add_edge("llm_call_1", END)
router_builder.add_edge("llm_call_2", END)
router_builder.add_edge("llm_call_3", END)

# Compile workflow
router_workflow = router_builder.compile()

# Show the workflow
display(Image(router_workflow.get_graph().draw_mermaid_png()))

# Invoke
state = router_workflow.invoke({"input": "Write me a joke about cats"})
print(state["output"])
```

--------------------------------

### Tool Definitions for Agents in Python

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This Python code defines three tools: multiply, add, and divide, using the `@tool` decorator. Each tool is documented with its purpose, arguments, and return type, making them available for use by LangChain agents.

```python
from langchain.tools import tool


# Define tools
@tool
def multiply(a: int, b: int) -> int:
    """Multiply `a` and `b`. 

    Args:
        a: First int
        b: Second int
    """
    return a * b


@tool
def add(a: int, b: int) -> int:
    """Adds `a` and `b`. 

    Args:
        a: First int
        b: Second int
    """
    return a + b


@tool
def divide(a: int, b: int) -> float:
    """Divide `a` and `b`. 

    Args:
        a: First int
        b: Second int
    """
    return a / b
```

--------------------------------

### Initialize Anthropic LLM and Set API Key

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

Initializes the ChatAnthropic LLM client after securely retrieving the ANTHROPIC_API_KEY from environment variables or user input. This LLM is configured with the 'claude-sonnet-4-5-20250929' model.

```python
import os
import getpass

from langchain_anthropic import ChatAnthropic

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")


_set_env("ANTHROPIC_API_KEY")

llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")

```

--------------------------------

### Evaluator-Optimizer Workflow in Python

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This Python code sets up an evaluator-optimizer workflow. It defines a state with joke-related fields and uses an LLM with structured output for evaluation. Nodes for joke generation and evaluation are created, with conditional edges to route based on the evaluation outcome.

```python
from typing import Literal
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display
from langchain_core.pydantic_v1 import BaseModel, Field

# Assume llm is an LLM instance

# Graph state
class State(BaseModel):
    joke: str
    topic: str
    feedback: str
    funny_or_not: str


# Schema for structured output to use in evaluation
class Feedback(BaseModel):
    grade: Literal["funny", "not funny"] = Field(
        description="Decide if the joke is funny or not.",
    )
    feedback: str = Field(
        description="If the joke is not funny, provide feedback on how to improve it.",
    )


# Augment the LLM with schema for structured output
evaluator = llm.with_structured_output(Feedback)


# Nodes
def llm_call_generator(state: State):
    """LLM generates a joke"""

    if state.get("feedback"):
        msg = llm.invoke(
            f"Write a joke about {state['topic']} but take into account the feedback: {state['feedback']}"
        )
    else:
        msg = llm.invoke(f"Write a joke about {state['topic']}")
    return {"joke": msg.content}

def llm_call_evaluator(state: State):
    """LLM evaluates the joke"""

    grade = evaluator.invoke(f"Grade the joke {state['joke']}")
    return {"funny_or_not": grade.grade, "feedback": grade.feedback}


# Conditional edge function to route back to joke generator or end based upon feedback from the evaluator
def route_joke(state: State):
    """Route back to joke generator or end based upon feedback from the evaluator"""

    if state["funny_or_not"] == "funny":
        return "Accepted"
    elif state["funny_or_not"] == "not funny":
        return "Rejected + Feedback"


# Build workflow
optimizer_builder = StateGraph(State)

# Add the nodes
optimizer_builder.add_node("llm_call_generator", llm_call_generator)
optimizer_builder.add_node("llm_call_evaluator", llm_call_evaluator)

# Add edges to connect nodes
optimizer_builder.add_edge(START, "llm_call_generator")
optimizer_builder.add_edge("llm_call_generator", "llm_call_evaluator")
optimizer_builder.add_conditional_edges(
    "llm_call_evaluator",
    route_joke,
    {
        "Accepted": END,
        "Rejected + Feedback": "llm_call_generator",
    },
)

# Compile the workflow
optimizer_workflow = optimizer_builder.compile()

# Show the workflow
display(Image(optimizer_workflow.get_graph().draw_mermaid_png()))

# Invoke
state = optimizer_workflow.invoke({"topic": "Cats"})
print(state["joke"])
```

--------------------------------

### Define Pydantic Models for Structured Output in Python

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This Python snippet defines Pydantic models for structured output, specifically for creating a report plan. The `Section` model includes a name and description, while the `Sections` model is a list of `Section` objects. These models are used to augment an LLM for generating structured plans.

```python
from typing import Annotated, List
import operator
from pydantic import BaseModel, Field

# Schema for structured output to use in planning
class Section(BaseModel):
    name: str = Field(
        description="Name for this section of the report.",
    )
    description: str = Field(
        description="Brief overview of the main topics and concepts to be covered in this section.",
    )


class Sections(BaseModel):
    sections: List[Section] = Field(
        description="Sections of the report.",
    )


# Augment the LLM with schema for structured output
# Assume llm and planner are defined elsewhere
# planner = llm.with_structured_output(Sections)
```

--------------------------------

### Parallel LLM Execution in LangGraph (Python)

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

Demonstrates parallel execution of LLM calls using LangGraph's StateGraph. It defines a state, parallel nodes for generating a joke, story, and poem, and an aggregator node to combine the outputs. This approach increases speed by running LLM tasks concurrently.

```python
# Graph state
class State(TypedDict):
    topic: str
    joke: str
    story: str
    poem: str
    combined_output: str


# Nodes
def call_llm_1(state: State):
    """First LLM call to generate initial joke"""

    msg = llm.invoke(f"Write a joke about {state['topic']}")
    return {"joke": msg.content}


def call_llm_2(state: State):
    """Second LLM call to generate story"""

    msg = llm.invoke(f"Write a story about {state['topic']}")
    return {"story": msg.content}

def call_llm_3(state: State):
    """Third LLM call to generate poem"""

    msg = llm.invoke(f"Write a poem about {state['topic']}")
    return {"poem": msg.content}

def aggregator(state: State):
    """Combine the joke and story into a single output"""

    combined = f"Here's a story, joke, and poem about {state['topic']}!\n\n"
    combined += f"STORY:\n{state['story']}\n\n"
    combined += f"JOKE:\n{state['joke']}\n\n"
    combined += f"POEM:\n{state['poem']}"
    return {"combined_output": combined}


# Build workflow
parallel_builder = StateGraph(State)

# Add nodes
parallel_builder.add_node("call_llm_1", call_llm_1)
parallel_builder.add_node("call_llm_2", call_llm_2)
parallel_builder.add_node("call_llm_3", call_llm_3)
parallel_builder.add_node("aggregator", aggregator)

# Add edges to connect nodes
parallel_builder.add_edge(START, "call_llm_1")
parallel_builder.add_edge(START, "call_llm_2")
parallel_builder.add_edge(START, "call_llm_3")
parallel_builder.add_edge("call_llm_1", "aggregator")
parallel_builder.add_edge("call_llm_2", "aggregator")
parallel_builder.add_edge("call_llm_3", "aggregator")
parallel_builder.add_edge("aggregator", END)
parallel_workflow = parallel_builder.compile()

# Show workflow
display(Image(parallel_workflow.get_graph().draw_mermaid_png()))

# Invoke
state = parallel_workflow.invoke({"topic": "cats"})
print(state["combined_output"])

```

--------------------------------

### Orchestrator-Worker Workflow in Python

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This Python code defines and compiles an orchestrator-worker workflow. It sets up nodes for orchestration, LLM calls, and synthesis, connecting them with edges. The workflow is then compiled and visualized, and finally invoked with an input topic.

```python
from langgraph.graph import START, END
from IPython.display import Image, display

# Assume orchestrator, llm_call, synthesizer, assign_workers are defined elsewhere
# Assume orchestrator_worker_builder is an instance of StateGraph

# Add the nodes
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)

# Add edges to connect nodes
orchestrator_worker_builder.add_edge(START, "orchestrator")
orchestrator_worker_builder.add_conditional_edges(
    "orchestrator", assign_workers, ["llm_call"]
)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.add_edge("synthesizer", END)

# Compile the workflow
orchestrator_worker = orchestrator_worker_builder.compile()

# Show the workflow
display(Image(orchestrator_worker.get_graph().draw_mermaid_png()))

# Invoke
state = orchestrator_worker.invoke({"topic": "Create a report on LLM scaling laws"})

from IPython.display import Markdown
Markdown(state["final_report"])
```

--------------------------------

### Routing LLM Workflows with LangGraph (Python)

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

Illustrates how to implement conditional routing in LangGraph for LLM tasks. It defines a router function that uses an LLM with structured output to decide which subsequent node (for story, joke, or poem generation) to execute based on user input. This enables specialized processing paths.

```python
from typing_extensions import Literal
from langchain.messages import HumanMessage, SystemMessage


# Schema for structured output to use as routing logic
class Route(BaseModel):
    step: Literal["poem", "story", "joke"] = Field(
        None, description="The next step in the routing process"
    )


# Augment the LLM with schema for structured output
router = llm.with_structured_output(Route)


# State
class State(TypedDict):
    input: str
    decision: str
    output: str


# Nodes
def llm_call_1(state: State):
    """Write a story"""

    result = llm.invoke(state["input"])
    return {"output": result.content}


def llm_call_2(state: State):
    """Write a joke"""

    result = llm.invoke(state["input"])
    return {"output": result.content}

def llm_call_3(state: State):
    """Write a poem"""

    result = llm.invoke(state["input"])
    return {"output": result.content}

def llm_call_router(state: State):
    """Route the input to the appropriate node"""

    # Run the augmented LLM with structured output to serve as routing logic
    decision = router.invoke(
        [
            SystemMessage(
                content="Route the input to story, joke, or poem based on the user's request."
            ),
            HumanMessage(content=state["input"]),
        ]
    )

    return {"decision": decision.step}


# Conditional edge function to route to the appropriate node
def route_decision(state: State):
    # Return the node name you want to visit next
    if state["decision"] == "story":
        return "llm_call_1"
    elif state["decision"] == "joke":
        return "llm_call_2"
    elif state["decision"] == "poem":
        return "llm_call_3"


# Build workflow
router_builder = StateGraph(State)


```

--------------------------------

### LangGraph Joke Generation Workflow (Python)

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

Implements a joke generation workflow using LangGraph's StateGraph API. It defines nodes for generating, checking, improving, and polishing a joke, connecting them with conditional edges to create a dynamic process.

```python
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from IPython.display import Image, display


# Graph state
class State(TypedDict):
    topic: str
    joke: str
    improved_joke: str
    final_joke: str


# Nodes
def generate_joke(state: State):
    """First LLM call to generate initial joke"""

    msg = llm.invoke(f"Write a short joke about {state['topic']}")
    return {"joke": msg.content}


def check_punchline(state: State):
    """Gate function to check if the joke has a punchline"""

    # Simple check - does the joke contain "?" or "!"
    if "?" in state["joke"] or "!" in state["joke"]:
        return "Pass"
    return "Fail"


def improve_joke(state: State):
    """Second LLM call to improve the joke"""

    msg = llm.invoke(f"Make this joke funnier by adding wordplay: {state['joke']}")
    return {"improved_joke": msg.content}


def polish_joke(state: State):
    """Third LLM call for final polish"""
    msg = llm.invoke(f"Add a surprising twist to this joke: {state['improved_joke']}")
    return {"final_joke": msg.content}


# Build workflow
workflow = StateGraph(State)

# Add nodes
workflow.add_node("generate_joke", generate_joke)
workflow.add_node("improve_joke", improve_joke)
workflow.add_node("polish_joke", polish_joke)

# Add edges to connect nodes
workflow.add_edge(START, "generate_joke")
workflow.add_conditional_edges(
    "generate_joke", check_punchline, {"Fail": "improve_joke", "Pass": END}
)
workflow.add_edge("improve_joke", "polish_joke")
workflow.add_edge("polish_joke", END)

# Compile
chain = workflow.compile()

# Show workflow
display(Image(chain.get_graph().draw_mermaid_png()))

# Invoke
state = chain.invoke({"topic": "cats"})
print("Initial joke:")
print(state["joke"])
print("\n--- --- ---\n")
if "improved_joke" in state:
    print("Improved joke:")
    print(state["improved_joke"])
    print("\n--- --- ---\n")

    print("Final joke:")
    print(state["final_joke"])
else:
    print("Final joke:")
    print(state["joke"])

```

--------------------------------

### Define StateGraph for Orchestrator-Worker Pattern in Python

Source: https://docs.langchain.com/oss/python/langgraph/overview/oss/python/langgraph/workflows-agents_agents

This Python snippet defines the state structure for an orchestrator-worker workflow using LangGraph's `StateGraph`. It includes types for the overall graph state (`State`) and the worker's state (`WorkerState`), specifying fields like `topic`, `sections`, `completed_sections`, and `final_report`. It also defines functions for the orchestrator, worker, synthesizer, and the conditional edge logic for assigning workers.

```python
from langgraph.graph import StateGraph
from langgraph.types import Send
from typing import Annotated, List
import operator
from pydantic import BaseModel, Field
from langchain_core.messages import SystemMessage, HumanMessage

# Assume llm is defined elsewhere

# Graph state
class State(TypedDict):
    topic: str  # Report topic
    sections: list[Section]  # List of report sections
    completed_sections: Annotated[
        list, operator.add
    ]  # All workers write to this key in parallel
    final_report: str  # Final report


# Worker state
class WorkerState(TypedDict):
    section: Section
    completed_sections: Annotated[list, operator.add]


# Nodes
def orchestrator(state: State):
    """Orchestrator that generates a plan for the report"""

    # Generate queries
    # Assume planner is defined elsewhere and uses Sections schema
    report_sections = planner.invoke(
        [
            SystemMessage(content="Generate a plan for the report."),
            HumanMessage(content=f"Here is the report topic: {state['topic']}"),
        ]
    )

    return {"sections": report_sections.sections}


def llm_call(state: WorkerState):
    """Worker writes a section of the report"""

    # Generate section
    section = llm.invoke(
        [
            SystemMessage(
                content="Write a report section following the provided name and description. Include no preamble for each section. Use markdown formatting."
            ),
            HumanMessage(
                content=f"Here is the section name: {state['section'].name} and description: {state['section'].description}"
            ),
        ]
    )

    # Write the updated section to completed sections
    return {"completed_sections": [section.content]}


def synthesizer(state: State):
    """Synthesize full report from sections"""

    # List of completed sections
    completed_sections = state["completed_sections"]

    # Format completed section to str to use as context for final sections
    completed_report_sections = "\n\n---\n\n".join(completed_sections)

    return {"final_report": completed_report_sections}


# Conditional edge function to create llm_call workers that each write a section of the report
def assign_workers(state: State):
    """Assign a worker to each section in the plan"""

    # Kick off section writing in parallel via Send() API
    return [Send("llm_call", {"section": s}) for s in state["sections"]]


# Build workflow
orchestrator_worker_builder = StateGraph(State)

# Add nodes
orchestrator_worker_builder.add_node("orchestrator", orchestrator)
orchestrator_worker_builder.add_node("llm_call", llm_call)
orchestrator_worker_builder.add_node("synthesizer", synthesizer)

# Add edges
orchestrator_worker_builder.add_edge("orchestrator", assign_workers)
orchestrator_worker_builder.add_edge("llm_call", "synthesizer")
orchestrator_worker_builder.set_entry_point("orchestrator")

# Compile workflow
# orchestrator_worker_workflow = orchestrator_worker_builder.compile()

```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.