LangGraph Stateful Agents

LangGraph Stateful Agent Workflows

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    iteration: int

llm_with_tools = llm.bind_tools(tools)

def call_agent(state: AgentState) -> AgentState:
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response], "iteration": state["iteration"]+1}

def should_continue(state: AgentState) -> str:
    last_msg = state["messages"][-1]
    if hasattr(last_msg, "tool_calls") and last_msg.tool_calls:
        return "tools"
    return END

tool_node = ToolNode(tools)

graph = StateGraph(AgentState)
graph.add_node("agent", call_agent)
graph.add_node("tools", tool_node)
graph.set_entry_point("agent")
graph.add_conditional_edges("agent", should_continue,
    {"tools": "tools", END: END})
graph.add_edge("tools", "agent")  # loop back

app = graph.compile()
result = app.invoke({"messages":[HumanMessage("Analyse sales trends")],
                     "iteration":0})