📖 KDD 2026 Hands-On Tutorial

Building Agentic AI Systems with LangGraph

From LLM Pipelines to Multi-Agent Decision Workflows

⏱ 3 Hours total👥 Data Scientists, ML Engineers, AI Researchers

Abstract

Recent advances in large language models have enabled agentic AI systems that can reason, plan, call tools, and coordinate with other agents. This hands-on tutorial introduces LangGraph for designing stateful agentic workflows — from basic reasoning pipelines to multi-agent decision assistants.

What You'll Build

✓

Tool-using agents that interact with APIs and datasets

✓

Multi-agent systems where specialized agents collaborate

✓

Reasoning loops and self-refinement pipelines

✓

A complete multi-agent decision assistant

Tutorial Outline

1

Introduction to Agentic AI

Why agents? Pipelines vs loops, core capabilities

⏱ 30 min

2

LangGraph Fundamentals

StateGraph, nodes, edges, conditional routing

⏱ 40 min

3

Tool-Using Agents

@tool decorator, ToolNode, free APIs

⏱ 40 min

4

Multi-Agent Collaboration

Supervisor/worker, debate system

⏱ 40 min

5

Decision Intelligence Agents

Parallel evaluators, synthesis

⏱ 20 min

6

Scaling Agentic Systems

Production safeguards, evaluation

⏱ 10 min

Prerequisites

🔹 Python 3.10+

🔹 Basic ML / LLM familiarity

📦 Install: pip install langgraph langchain langchain-groq

About the Instructor

Tutorial Instructor

AI Researcher & LangGraph Practitioner · KDD 2026

LangGraphMulti-Agent AIKDD 2026Agentic SystemsLLMOps

Researcher and practitioner specialising in agentic AI systems, multi-agent architectures, and production LLM deployments. This tutorial distills real-world lessons from building LangGraph-based systems into a practical 3-hour hands-on session designed for data scientists and ML engineers.

What We'll Cover Together

Why simple prompt engineering is insufficient for complex, multi-step tasks
How to design stateful agent graphs using LangGraph's StateGraph API
Integrating free, no-API-key tools like DuckDuckGo, Wikipedia, and yfinance
Orchestrating multiple specialized agents under a supervisor
Production safeguards: checkpointing, recursion limits, and observability

1

Introduction to Agentic AI

⏱ 30 min

Learning Objectives

Identify the limitations of prompt engineering for multi-step, tool-dependent tasks
Contrast LLM pipelines with agent workflows and select the right approach for a given task
List and describe the four core agent capabilities: reasoning, planning, tool use, and self-refinement

Why Prompt Engineering Alone Isn't Enough

Simple prompt engineering works well for isolated tasks, but breaks down on multi-step problems that require memory across turns, conditional branching based on intermediate results, use of external tools (APIs, databases, calculators), and self-correction when initial outputs are wrong.

Memory across turns
Conditional branching on intermediate results
External tool use (APIs, databases, calculators)
Self-correction when initial outputs are wrong

From Pipelines to Agent Workflows

An LLM pipeline is a linear chain: input → model → output. An agent workflow adds a control loop: the model can observe its environment, decide on an action, execute it, and observe the result — repeating until the task is complete.

Left: a simple LLM pipeline — one-shot, no memory. Right: an agent workflow cycling through Reason → Act → Observe until done.

python

from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage

llm = ChatGroq(model="llama-3.3-70b-versatile")

# Pipeline: one-shot
response = llm.invoke([HumanMessage("Summarise quantum computing in 2 sentences")])
print(response.content)

Core Agent Capabilities

Reasoning: Chain-of-thought planning before acting
Planning: Breaking goals into sub-tasks
Tool use: Calling external APIs, databases, and code interpreters
Self-refinement: Critiquing and improving its own outputs

When to Use Agents (vs Pipelines)

Not every problem needs an agent. Use a pipeline when the task is deterministic and single-step. Use an agent when the task requires multiple steps, tool use, or self-correction.

Pipeline: Summarisation, translation, classification
Agent: Research assistant, data analyst, code debugger
Rule of thumb: if it needs a tool or a loop, it needs an agent

Build a Simple Agent Wrapper ▶

Starter code

from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage, SystemMessage

llm = ChatGroq(model="llama-3.3-70b-versatile")

# TODO: Build a simple ReAct-style loop that:
# 1. Asks the LLM what action to take
# 2. Prints the reasoning
# 3. Stops when the LLM says "DONE"
question = "What are the top 3 benefits of using LangGraph?"

2

LangGraph Fundamentals

⏱ 40 min

Learning Objectives

Define StateGraph, nodes, edges, and shared state in LangGraph
Construct a graph with conditional routing using add_conditional_edges
Compile and invoke a LangGraph workflow with a typed state dictionary

What is LangGraph?

LangGraph is a framework built on LangChain for creating stateful, multi-actor AI workflows as graphs. Unlike linear chains, LangGraph lets you define workflows with cycles, conditional branching, and explicit shared state.

Core Concepts

Agent node runs the LLM → conditional edge checks for tool calls → ToolNode executes and loops back → END when finished.

State: A typed dictionary shared across all nodes. Nodes read from and write to state.
Nodes: Python functions that receive the current state and return updates to it.
Edges: Connections between nodes — unconditional or conditional.
Graph: The assembled StateGraph defining the full workflow.

python

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

# 1. Define state schema
class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str

# 2. Define nodes
def reasoning_node(state: AgentState) -> dict:
    result = llm.invoke(state["messages"])
    return {"messages": [result], "next_step": "tools"}

def tool_node(state: AgentState) -> dict:
    tool_result = run_tools(state["messages"][-1])
    return {"messages": [tool_result], "next_step": "end"}

# 3. Build graph
workflow = StateGraph(AgentState)
workflow.add_node("reason", reasoning_node)
workflow.add_node("tools", tool_node)

# 4. Add edges
workflow.set_entry_point("reason")
workflow.add_conditional_edges(
    "reason",
    lambda s: s["next_step"],
    {"tools": "tools", "end": END}
)
workflow.add_edge("tools", "reason")

# 5. Compile and run
app = workflow.compile()
result = app.invoke({"messages": [HumanMessage("What is 2+2?")], "next_step": ""})

State Transitions & Conditional Routing

python

def should_continue(state: AgentState) -> str:
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "continue"
    return "end"

workflow.add_conditional_edges("agent", should_continue, {
    "continue": "tools",
    "end": END
})

Build a Simple Reasoning Graph ▶

Starter code

from langgraph.graph import StateGraph, END
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
import operator

class QAState(TypedDict):
    messages: Annotated[list, operator.add]

llm = ChatGroq(model="llama-3.3-70b-versatile")

def answer_node(state: QAState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

# TODO: Build and compile the graph
# 1. Create StateGraph(QAState)
# 2. Add "answer" node
# 3. Set entry point to "answer"
# 4. Add edge from "answer" to END
# 5. Compile and invoke with a question

3

Tool-Using Agents

⏱ 40 min

Learning Objectives

Define custom tools using the @tool decorator and bind them to an LLM
Build a tool-using agent that invokes external APIs and computation tools
Route tool call results back through the agent's reasoning loop using conditional edges

Defining Tools for LangGraph

Tools are Python functions decorated with @tool that give the agent external capabilities. All tools below are completely free — no API keys required.

Agent generates a tool_call → ToolNode dispatches to the right tool → result returned as ToolMessage → Agent continues or finishes.

python

# pip install langchain duckduckgo-search yfinance wikipedia
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for current information using DuckDuckGo (free, no API key)."""
    from duckduckgo_search import DDGS
    results = DDGS().text(query, max_results=3)
    return "\n\n".join([f"{r['title']}\n{r['body']}" for r in results])

@tool
def lookup_wikipedia(topic: str) -> str:
    """Look up a topic on Wikipedia and return a summary."""
    import wikipedia
    try:
        return wikipedia.summary(topic, sentences=4)
    except wikipedia.DisambiguationError as e:
        return f"Ambiguous. Suggestions: {e.options[:3]}"
    except Exception as e:
        return f"Not found: {e}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression safely."""
    try:
        result = eval(expression, {"__builtins__": {}}, {})
        return str(result)
    except Exception as e:
        return f"Error: {e}"

@tool
def get_stock_price(ticker: str) -> str:
    """Get the latest stock price using yfinance (free, no API key)."""
    import yfinance as yf
    stock = yf.Ticker(ticker)
    price = stock.fast_info.get("lastPrice", "N/A")
    return f"{ticker}: ${price}"

tools = [search_web, lookup_wikipedia, calculate, get_stock_price]

Building a Tool-Using Agent Graph

python

from langgraph.graph import StateGraph, END
from langgraph.prebuilt import ToolNode
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage
from typing import TypedDict, Annotated
import operator

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

llm = ChatGroq(model="llama-3.3-70b-versatile").bind_tools(tools)

def agent_node(state: AgentState) -> dict:
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

def should_use_tools(state: AgentState) -> str:
    last = state["messages"][-1]
    if hasattr(last, "tool_calls") and last.tool_calls:
        return "tools"
    return "end"

workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.add_node("tools", ToolNode(tools))

workflow.set_entry_point("agent")
workflow.add_conditional_edges("agent", should_use_tools, {
    "tools": "tools",
    "end": END
})
workflow.add_edge("tools", "agent")

app = workflow.compile()

result = app.invoke({
    "messages": [HumanMessage("What is the current price of AAPL stock, and what is 1500 * 1.08?")]
})
print(result["messages"][-1].content)

Build a Data Analysis Agent ▶

Starter code

from langchain_core.tools import tool
import pandas as pd

@tool
def load_dataset(filename: str) -> str:
    """Load a CSV dataset and return basic statistics."""
    df = pd.read_csv(filename)
    return df.describe().to_string()

@tool
def filter_data(filename: str, column: str, condition: str) -> str:
    """Filter a dataset by a condition on a column."""
    df = pd.read_csv(filename)
    filtered = df[df[column].astype(str).str.contains(condition)]
    return filtered.head(10).to_string()

# TODO: Build a LangGraph agent that uses these tools
# to answer questions about a CSV dataset
data_tools = [load_dataset, filter_data]

4

Multi-Agent Collaboration

⏱ 40 min

Learning Objectives

Design specialized agent roles in a supervisor/worker architecture
Implement a supervisor node that routes tasks to worker agents based on state
Build and run a complete multi-agent pipeline with at least three specialized agents

Why Multiple Agents?

A single agent trying to do everything becomes a "god agent" — complex, unreliable, and hard to debug. Specialized agents are better at their jobs, easier to test, and can work in parallel.

Supervisor/Worker: A coordinator agent delegates tasks to specialized workers
Pipeline: Agents pass work sequentially, each adding value
Debate/Adversarial: Agents argue opposite positions to find the best answer
Parallel: Multiple agents work simultaneously on different subtasks

Supervisor Architecture

Supervisor routes to a specialist (solid), worker returns result (dashed), loop continues until FINISH.

python

from langgraph.graph import StateGraph, END
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage, SystemMessage
from typing import TypedDict, Annotated
import operator

class CollaborativeState(TypedDict):
    messages: Annotated[list, operator.add]
    task: str
    research_result: str
    analysis_result: str
    final_report: str
    next_agent: str

llm = ChatGroq(model="llama-3.3-70b-versatile")

def supervisor_node(state: CollaborativeState) -> dict:
    system = SystemMessage("""You are a supervisor coordinating a research team.
Decide which agent should act next:
- 'researcher': for gathering information
- 'analyst': for analyzing gathered data
- 'writer': for composing the final report
- 'FINISH': when the report is complete
Respond with only the agent name.""")
    response = llm.invoke([system] + state["messages"])
    return {"next_agent": response.content.strip()}

def researcher_node(state: CollaborativeState) -> dict:
    prompt = f"Research the following topic thoroughly: {state['task']}"
    result = llm.invoke([HumanMessage(prompt)])
    return {"research_result": result.content, "messages": [result]}

def analyst_node(state: CollaborativeState) -> dict:
    prompt = f"Analyze this research and extract key insights:\n{state['research_result']}"
    result = llm.invoke([HumanMessage(prompt)])
    return {"analysis_result": result.content, "messages": [result]}

def writer_node(state: CollaborativeState) -> dict:
    prompt = f"Write a report based on:\nResearch: {state['research_result']}\nAnalysis: {state['analysis_result']}"
    result = llm.invoke([HumanMessage(prompt)])
    return {"final_report": result.content, "messages": [result]}

workflow = StateGraph(CollaborativeState)
workflow.add_node("supervisor", supervisor_node)
workflow.add_node("researcher", researcher_node)
workflow.add_node("analyst",    analyst_node)
workflow.add_node("writer",     writer_node)

workflow.set_entry_point("supervisor")
workflow.add_conditional_edges("supervisor", lambda s: s["next_agent"], {
    "researcher":"researcher", "analyst":"analyst",
    "writer":"writer", "FINISH":END
})
workflow.add_edge("researcher", "supervisor")
workflow.add_edge("analyst",    "supervisor")
workflow.add_edge("writer",     "supervisor")
app = workflow.compile()

Implement a Collaborative Reasoning System ▶

Starter code

from typing import TypedDict

class DebateState(TypedDict):
    question: str
    pro_argument: str
    con_argument: str
    final_verdict: str
    round: int

# TODO: Implement:
# 1. pro_agent_node: argues in favour of the topic
# 2. con_agent_node: argues against the topic
# 3. judge_node: evaluates both arguments and gives a verdict
# 4. Wire them: pro -> con -> judge

5

Decision Intelligence Agents

⏱ 20 min

Learning Objectives

Build parallel evaluator agents that score options on qualitative, quantitative, and risk dimensions
Combine multi-dimensional scores from parallel agents into a unified decision recommendation
Generate a structured recommendation with a confidence score and key trade-offs

Decision-Support Architecture

Decision intelligence agents support human decision-making — not replace it. They gather evidence, evaluate options from multiple angles, and present structured recommendations with reasoning.

Three parallel evaluators feed their scores into the Synthesizer which produces a final recommendation with confidence.

python

from langgraph.graph import StateGraph, END
from langchain_groq import ChatGroq
from langchain_core.messages import HumanMessage
from typing import TypedDict, List
import json

class DecisionState(TypedDict):
    decision_question: str
    options: List[str]
    qualitative_scores: dict
    quantitative_scores: dict
    risk_assessment: dict
    final_recommendation: str
    confidence: float

llm = ChatGroq(model="llama-3.3-70b-versatile")

def qualitative_evaluator(state: DecisionState) -> dict:
    prompt = f"""Evaluate options for: {state['decision_question']}
Options: {state['options']}
Score each (1-10) on: strategic fit, stakeholder acceptance,
implementation complexity, long-term sustainability.
Return as JSON: {{"option_name": {{"strategic_fit": X, ...}}}}"""
    result = llm.invoke([HumanMessage(prompt)])
    return {"qualitative_scores": json.loads(result.content)}

def quantitative_evaluator(state: DecisionState) -> dict:
    prompt = f"""Estimate numerical metrics for: {state['decision_question']}
Options: {state['options']}
Metrics: ROI (%), cost ($K), time to value (months). Return as JSON."""
    result = llm.invoke([HumanMessage(prompt)])
    return {"quantitative_scores": json.loads(result.content)}

def risk_assessor(state: DecisionState) -> dict:
    prompt = f"""Assess risks for: {state['decision_question']}
Options: {state['options']}
Top 3 risks per option with probability and impact. Return as JSON."""
    result = llm.invoke([HumanMessage(prompt)])
    return {"risk_assessment": json.loads(result.content)}

def recommendation_synthesizer(state: DecisionState) -> dict:
    prompt = f"""Synthesize this decision analysis:
Question: {state['decision_question']}
Qualitative: {json.dumps(state['qualitative_scores'], indent=2)}
Quantitative: {json.dumps(state['quantitative_scores'], indent=2)}
Risks: {json.dumps(state['risk_assessment'], indent=2)}

Provide: 1) Recommended option with rationale 2) Key trade-offs
3) Critical success factors 4) Confidence score (0-1)"""
    result = llm.invoke([HumanMessage(prompt)])
    return {"final_recommendation": result.content, "confidence": 0.85}

workflow = StateGraph(DecisionState)
workflow.add_node("qualitative",  qualitative_evaluator)
workflow.add_node("quantitative", quantitative_evaluator)
workflow.add_node("risks",        risk_assessor)
workflow.add_node("recommend",    recommendation_synthesizer)

workflow.set_entry_point("qualitative")
workflow.add_edge("qualitative",  "quantitative")
workflow.add_edge("quantitative", "risks")
workflow.add_edge("risks",        "recommend")
workflow.add_edge("recommend",    END)
decision_app = workflow.compile()

6

Scaling Agentic Systems

⏱ 10 min

Learning Objectives

Identify common failure modes in production agents and apply safeguards to prevent them
Apply evaluation metrics to measure agent task completion, reasoning quality, and robustness
Configure checkpointing, recursion limits, and observability for a production-ready agent

Reliability Challenges

Agentic systems face unique failure modes:

Hallucination cascades: One incorrect reasoning step compounds into wrong answers
Tool reliability: External APIs fail; agents must handle errors gracefully
Infinite loops: Agents without termination conditions can run indefinitely
Context window limits: Long reasoning chains exhaust the model's context

python

MAX_STEPS = 10

def safe_agent_node(state) -> dict:
    if state["step_count"] >= MAX_STEPS:
        return {"messages": ["Max steps reached, stopping."], "step_count": state["step_count"]+1}
    try:
        response = llm.invoke(state["messages"])
        return {"messages": [response], "step_count": state["step_count"]+1}
    except Exception as e:
        return {"errors": [str(e)], "step_count": state["step_count"]+1}

# LangGraph built-in recursion limit:
config = {"recursion_limit": 25}
result = app.invoke(initial_state, config=config)

Evaluation Methods

Task completion rate: Did the agent complete the assigned goal?
Tool call accuracy: Did it call the right tools with the right arguments?
Reasoning quality: Was the chain of thought coherent and correct?
Efficiency: Did it reach the goal in a reasonable number of steps?
Robustness: Does it handle edge cases and errors gracefully?

python

def evaluate_agent(agent_app, test_cases):
    results = []
    for case in test_cases:
        output = agent_app.invoke(case["input"])
        result = {
            "input":   case["input"],
            "expected": case["expected"],
            "actual":   output,
            "passed":   case["check_fn"](output, case["expected"])
        }
        results.append(result)
    pass_rate = sum(r["passed"] for r in results) / len(results)
    print(f"Pass rate: {pass_rate:.1%}")
    return results

Production Considerations

Recursion limit, Checkpointer, Human Review, LangSmith observability, Error Handler, and Rate Limiter guard production agents.

Checkpointing: Use LangGraph's built-in checkpointer (SQLite or PostgreSQL) to persist state
Human-in-the-loop: Add interrupt points where humans can review agent decisions
Observability: Use LangSmith to trace every step, measure latency, catch regressions
Rate limiting: Implement retries with exponential backoff for external tool calls
Cost management: Track token usage per workflow; cache repeated tool calls

python

from langgraph.checkpoint.sqlite import SqliteSaver

memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "user_session_123"}}
result = app.invoke(initial_state, config=config)