LangGraph Node in Machine Learning

from typing import Annotated, Literal, TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


# 1. Define the shared state schema
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]
    task_type: str
    final_answer: str


# 2. Define node functions
def classifier_node(state: AgentState) -> dict:
    """Classify the user query into a task type."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    response = llm.invoke([
        {"role": "system", "content": "Classify the query as 'code', 'research', or 'general'. Reply with just the category."},
        state["messages"][-1]
    ])
    return {"task_type": response.content.strip().lower()}


def code_agent(state: AgentState) -> dict:
    """Handle code-related queries."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm.invoke([
        {"role": "system", "content": "You are an expert programmer. Provide clean, production-ready code."},
        *state["messages"]
    ])
    return {"final_answer": response.content, "messages": [response]}


def research_agent(state: AgentState) -> dict:
    """Handle research queries with citations."""
    llm = ChatOpenAI(model="gpt-4o", temperature=0.3)
    response = llm.invoke([
        {"role": "system", "content": "You are a research analyst. Provide thorough analysis with sources."},
        *state["messages"]
    ])
    return {"final_answer": response.content, "messages": [response]}


def general_agent(state: AgentState) -> dict:
    """Handle general conversational queries."""
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    response = llm.invoke(state["messages"])
    return {"final_answer": response.content, "messages": [response]}


# 3. Define the routing function for conditional edges
def route_by_task(state: AgentState) -> Literal["code_agent", "research_agent", "general_agent"]:
    task = state.get("task_type", "general")
    if task == "code":
        return "code_agent"
    elif task == "research":
        return "research_agent"
    return "general_agent"


# 4. Build the graph
graph = StateGraph(AgentState)

graph.add_node("classifier", classifier_node)
graph.add_node("code_agent", code_agent)
graph.add_node("research_agent", research_agent)
graph.add_node("general_agent", general_agent)

graph.add_edge(START, "classifier")
graph.add_conditional_edges("classifier", route_by_task)
graph.add_edge("code_agent", END)
graph.add_edge("research_agent", END)
graph.add_edge("general_agent", END)

# 5. Compile and run
app = graph.compile()

result = app.invoke({
    "messages": [{"role": "user", "content": "Write a Python function to calculate compound interest in INR"}]
})
print(result["final_answer"])

from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode
from langgraph.types import interrupt, Command


# Define tools
@tool
def search_database(query: str) -> str:
    """Search the internal product database."""
    # Simulated database search
    return f"Found 3 products matching '{query}': Widget A (INR 499), Widget B (INR 999), Widget C (INR 1,499)"


@tool
def place_order(product_name: str, quantity: int) -> str:
    """Place an order for a product. Requires human approval."""
    # This tool has side effects -- we want human approval
    return f"Order placed: {quantity}x {product_name}"


# State schema
class State(TypedDict):
    messages: Annotated[list, add_messages]


# Node functions
def agent(state: State) -> dict:
    llm = ChatOpenAI(model="gpt-4o").bind_tools([search_database, place_order])
    response = llm.invoke(state["messages"])
    return {"messages": [response]}


def human_review(state: State) -> dict:
    """Pause for human approval before executing side-effect tools."""
    last_message = state["messages"][-1]
    tool_calls = getattr(last_message, "tool_calls", [])

    # Check if any tool call requires approval
    needs_approval = any(tc["name"] == "place_order" for tc in tool_calls)

    if needs_approval:
        # Interrupt execution and wait for human input
        decision = interrupt(
            {"question": "Approve this order?", "tool_calls": tool_calls}
        )
        if decision != "approved":
            return {"messages": [{"role": "tool", "content": "Order cancelled by human reviewer.", "tool_call_id": tool_calls[0]["id"]}]}

    return {}  # No changes, proceed to tool execution


def should_continue(state: State) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "human_review"
    return END


# Build graph
graph = StateGraph(State)
graph.add_node("agent", agent)
graph.add_node("human_review", human_review)
graph.add_node("tools", ToolNode([search_database, place_order]))

graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"human_review": "human_review", END: END})
graph.add_edge("human_review", "tools")
graph.add_edge("tools", "agent")

# Compile with persistence for human-in-the-loop
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# Run with a thread ID for persistence
config = {"configurable": {"thread_id": "order-session-1"}}

# First invocation -- will pause at human_review if order is requested
result = app.invoke(
    {"messages": [{"role": "user", "content": "Find me a widget under INR 1000 and place an order for 2"}]},
    config=config
)

# Resume after human approval
result = app.invoke(Command(resume="approved"), config=config)
print(result["messages"][-1].content)

from typing import Annotated, Literal, TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import create_react_agent
from langgraph.checkpoint.postgres import PostgresSaver
import psycopg


# Shared state for the supervisor graph
class SupervisorState(TypedDict):
    messages: Annotated[list, add_messages]
    next_agent: str
    iteration_count: int


# --- Define specialist agents as subgraphs ---

@tool
def query_sql(sql: str) -> str:
    """Execute a read-only SQL query against the analytics database."""
    return f"Query result for: {sql} -> 42 rows returned"

@tool
def generate_chart(data: str, chart_type: str) -> str:
    """Generate a visualization from data."""
    return f"Chart generated: {chart_type} visualization saved to /tmp/chart.png"

@tool
def write_report(content: str, title: str) -> str:
    """Write a formatted report."""
    return f"Report '{title}' written successfully ({len(content)} chars)"


llm = ChatOpenAI(model="gpt-4o")

# Each specialist is a prebuilt ReAct agent compiled as a subgraph
data_analyst = create_react_agent(llm, [query_sql], prompt="You are a data analyst. Query databases and analyze results.")
visualizer = create_react_agent(llm, [generate_chart], prompt="You are a data visualizer. Create clear, insightful charts.")
report_writer = create_react_agent(llm, [write_report], prompt="You are a report writer. Create executive summaries.")


# --- Supervisor logic ---

def supervisor(state: SupervisorState) -> dict:
    """Decide which agent to invoke next or finish."""
    llm_with_routing = ChatOpenAI(model="gpt-4o", temperature=0)
    response = llm_with_routing.invoke([
        {"role": "system", "content": (
            "You are a supervisor coordinating a data analysis team. "
            "Based on the conversation, decide the next step: "
            "'data_analyst' to query data, 'visualizer' to make charts, "
            "'report_writer' to write the report, or 'FINISH' if done. "
            "Reply with just the agent name or FINISH."
        )},
        *state["messages"]
    ])
    next_agent = response.content.strip().lower()
    return {
        "next_agent": next_agent,
        "iteration_count": state.get("iteration_count", 0) + 1
    }


def route_supervisor(state: SupervisorState) -> Literal["data_analyst", "visualizer", "report_writer", "__end__"]:
    if state["next_agent"] == "finish" or state.get("iteration_count", 0) > 10:
        return "__end__"
    return state["next_agent"]


# --- Build the supervisor graph ---

supervisor_graph = StateGraph(SupervisorState)

supervisor_graph.add_node("supervisor", supervisor)
supervisor_graph.add_node("data_analyst", data_analyst)  # Subgraph as node
supervisor_graph.add_node("visualizer", visualizer)        # Subgraph as node
supervisor_graph.add_node("report_writer", report_writer)  # Subgraph as node

supervisor_graph.add_edge(START, "supervisor")
supervisor_graph.add_conditional_edges("supervisor", route_supervisor)
supervisor_graph.add_edge("data_analyst", "supervisor")
supervisor_graph.add_edge("visualizer", "supervisor")
supervisor_graph.add_edge("report_writer", "supervisor")

# Compile with PostgreSQL persistence for production
DB_URI = "postgresql://user:pass@localhost:5432/langgraph_checkpoints"
with psycopg.Connection.connect(DB_URI) as conn:
    checkpointer = PostgresSaver(conn)
    checkpointer.setup()  # Create tables if they don't exist
    app = supervisor_graph.compile(checkpointer=checkpointer)

    # Run the full workflow
    result = app.invoke(
        {"messages": [{"role": "user", "content": "Analyze Q4 2025 sales for the India market and prepare an executive report with charts"}]},
        config={"configurable": {"thread_id": "q4-india-report-001"}}
    )
    print(result["messages"][-1].content)

import asyncio
from typing import Annotated, TypedDict
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages


class State(TypedDict):
    messages: Annotated[list, add_messages]


def chatbot(state: State) -> dict:
    llm = ChatOpenAI(model="gpt-4o", streaming=True)
    response = llm.invoke(state["messages"])
    return {"messages": [response]}


graph = StateGraph(State)
graph.add_node("chatbot", chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()


async def stream_response():
    """Stream tokens as they are generated."""
    input_msg = {"messages": [{"role": "user", "content": "Explain how UPI works in India"}]}

    # Stream mode 'messages' yields individual tokens
    async for event, metadata in app.astream(input_msg, stream_mode="messages"):
        if hasattr(event, "content") and event.content:
            print(event.content, end="", flush=True)
    print()  # newline after streaming completes

    # Stream mode 'updates' yields full node outputs
    async for update in app.astream(input_msg, stream_mode="updates"):
        for node_name, node_output in update.items():
            print(f"\n--- Node '{node_name}' completed ---")
            print(f"Output keys: {list(node_output.keys())}")


asyncio.run(stream_response())

# langgraph.yaml -- LangGraph Platform deployment config
# Used when deploying to LangGraph Cloud or self-hosted LangGraph Platform

graphs:
  my_agent:
    path: ./src/agent.py:graph
    # The compiled graph object to serve

env:
  OPENAI_API_KEY: ${OPENAI_API_KEY}
  DATABASE_URL: ${DATABASE_URL}

# Persistence configuration
checkpointer:
  backend: postgres
  connection_string: ${DATABASE_URL}

# Resource limits
recursion_limit: 50
timeout_seconds: 300

# Authentication (LangGraph Platform)
auth:
  type: api_key
  header: x-api-key