Autonomous Multi-Agent Workflows: Designing State Machines with LangGraph

from typing import TypedDict, Annotated, Sequence
import operator
from langgraph.graph import StateGraph, END
from langchain_core.messages import BaseMessage, HumanMessage

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
    next_step: str
    compliance_approved: bool

# Initialize StateGraph
workflow = StateGraph(AgentState)

# Define Nodes
def supervisor_router(state: AgentState):
    last_msg = state["messages"][-1].content
    if "APPROVE" in last_msg:
        return {"next_step": "finalize", "compliance_approved": True}
    return {"next_step": "researcher"}

def researcher_node(state: AgentState):
    # Execute web search / database tools
    return {"messages": [HumanMessage(content="Extracted facts for review.")]}

def compliance_node(state: AgentState):
    # Verify citations & check PII rules
    return {"messages": [HumanMessage(content="APPROVE: Verified citations.")]}

workflow.add_node("supervisor", supervisor_router)
workflow.add_node("researcher", researcher_node)
workflow.add_node("compliance", compliance_node)

workflow.set_entry_point("supervisor")
workflow.add_conditional_edges("supervisor", lambda x: x["next_step"], {
    "researcher": "researcher",
    "finalize": END
})
workflow.add_edge("researcher", "compliance")
workflow.add_edge("compliance", "supervisor")

app = workflow.compile()