LangGraph — Stateful Agent Workflows

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langchain_core.tools import tool
import json

# -------------------------------------------------------
# 1. Define the shared state
# -------------------------------------------------------
class AgentState(TypedDict):
    # add_messages is a reducer: new messages are appended, not overwritten
    messages: Annotated[list, add_messages]
    # Other state fields can be plain values (last write wins)
    iteration: int
    final_answer: str | None

# -------------------------------------------------------
# 2. Define tools
# -------------------------------------------------------
@tool
def search_web(query: str) -> str:
    """Search the internet for current information on a topic."""
    # Replace with real search API
    return f"Top results for '{query}': [result 1], [result 2], [result 3]"

@tool
def run_python(code: str) -> str:
    """Execute Python code and return stdout + any errors."""
    import io, sys, contextlib
    output = io.StringIO()
    try:
        with contextlib.redirect_stdout(output):
            exec(code, {})
        return output.getvalue() or "Code ran successfully (no output)"
    except Exception as e:
        return f"Error: {type(e).__name__}: {e}"

tools = [search_web, run_python]
tools_by_name = {t.name: t for t in tools}

# -------------------------------------------------------
# 3. Initialize the LLM with tools bound
# -------------------------------------------------------
llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0)
llm_with_tools = llm.bind_tools(tools)

# -------------------------------------------------------
# 4. Define nodes
# -------------------------------------------------------
def call_model(state: AgentState) -> dict:
    """The reasoning node — calls the LLM."""
    response = llm_with_tools.invoke(state["messages"])
    return {
        "messages": [response],
        "iteration": state.get("iteration", 0) + 1,
    }

def execute_tools(state: AgentState) -> dict:
    """The action node — executes tool calls from the last AI message."""
    last_message = state["messages"][-1]
    tool_results = []

    for tool_call in last_message.tool_calls:
        tool = tools_by_name[tool_call["name"]]
        result = tool.invoke(tool_call["args"])
        tool_results.append(
            ToolMessage(
                content=str(result),
                tool_call_id=tool_call["id"],
                name=tool_call["name"],
            )
        )

    return {"messages": tool_results}

# -------------------------------------------------------
# 5. Define routing logic (conditional edges)
# -------------------------------------------------------
def should_continue(state: AgentState) -> str:
    """Decide what to do after the model responds."""
    last_message = state["messages"][-1]

    # If the model wants to use tools, go to tool execution
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "execute_tools"

    # If we've iterated too many times, stop
    if state.get("iteration", 0) >= 10:
        return END

    # Otherwise, we're done
    return END

# -------------------------------------------------------
# 6. Build the graph
# -------------------------------------------------------
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("call_model", call_model)
workflow.add_node("execute_tools", execute_tools)

# Set entry point
workflow.set_entry_point("call_model")

# Add conditional edge from call_model
workflow.add_conditional_edges(
    "call_model",
    should_continue,
    {
        "execute_tools": "execute_tools",
        END: END,
    }
)

# After tools execute, always go back to the model
workflow.add_edge("execute_tools", "call_model")

# Compile the graph
app = workflow.compile()

# -------------------------------------------------------
# 7. Run the agent
# -------------------------------------------------------
initial_state = {
    "messages": [HumanMessage(content="Search for the latest LangGraph release notes and summarize the key new features.")],
    "iteration": 0,
    "final_answer": None,
}

for event in app.stream(initial_state, stream_mode="values"):
    last_msg = event["messages"][-1]
    if hasattr(last_msg, "content") and isinstance(last_msg.content, str):
        print(f"[{type(last_msg).__name__}]: {last_msg.content[:200]}")

from langgraph.checkpoint.memory import MemorySaver

# The interrupt_before parameter causes the graph to pause
# before executing the specified node
checkpointer = MemorySaver()

app_with_human = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["execute_tools"],  # pause before executing any tool
)

# Thread ID lets you resume the same conversation
thread_config = {"configurable": {"thread_id": "my-thread-1"}}

# Start the run — it will pause before tool execution
initial_state = {
    "messages": [HumanMessage(content="Delete all files in /tmp/old_logs/")],
    "iteration": 0,
    "final_answer": None,
}

# Run until the first interrupt
for event in app_with_human.stream(initial_state, config=thread_config):
    last_msg = event["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        print("\n--- HUMAN APPROVAL REQUIRED ---")
        for tc in last_msg.tool_calls:
            print(f"Tool: {tc['name']}")
            print(f"Args: {tc['args']}")
        print("-------------------------------")

# Human inspects the tool call and approves (or rejects)
approval = input("Approve this action? (yes/no): ")

if approval.lower() == "yes":
    # Resume from the checkpoint
    for event in app_with_human.stream(None, config=thread_config):
        print(event["messages"][-1].content)
else:
    print("Action rejected by human reviewer.")

from langgraph.graph import StateGraph, END
from typing import Literal

class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str  # which worker to call next

# Specialized worker agents (each is its own compiled graph or function)
def research_agent(state: SupervisorState) -> dict:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def coding_agent(state: SupervisorState) -> dict:
    # This agent specializes in writing and running code
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def supervisor_node(state: SupervisorState) -> dict:
    """Decides which worker to call next, or whether to finish."""
    system_prompt = """You are a supervisor managing a research agent and a coding agent.
Given the conversation, decide who should act next.
Respond with JSON: {"next": "research_agent" | "coding_agent" | "FINISH"}"""

    response = llm.invoke([
        {"role": "system", "content": system_prompt},
        *state["messages"]
    ])

    decision = json.loads(response.content)
    return {"next_agent": decision["next"]}

def route_from_supervisor(state: SupervisorState) -> str:
    return state["next_agent"]

# Build the supervisor graph
graph = StateGraph(SupervisorState)
graph.add_node("supervisor", supervisor_node)
graph.add_node("research_agent", research_agent)
graph.add_node("coding_agent", coding_agent)

graph.set_entry_point("supervisor")
graph.add_conditional_edges(
    "supervisor",
    route_from_supervisor,
    {
        "research_agent": "research_agent",
        "coding_agent": "coding_agent",
        "FINISH": END,
    }
)
graph.add_edge("research_agent", "supervisor")
graph.add_edge("coding_agent", "supervisor")

supervisor_app = graph.compile()