Back to blogs

Part 1: Beyond Sequential Chains - Your First Agentic Workflow with LangGraph

March 29, 2024

Part 1: Beyond Sequential Chains - Your First Agentic Workflow with LangGraph

The human brain is an intricate network, not a linear assembly line. Its intelligence isn't just about processing information from A to B, but about dynamic state management, decision-making, and iterative refinement. As someone deeply invested in AI Research, particularly in replicating cognitive architectures and exploring Mixture-of-Experts (MoE) systems, I find myself constantly pushing against the limitations of current LLM frameworks.

Why Linear Chains Fail the Brain Test

For a while, tools like LangChain Expression Language (LCEL) offered a neat way to compose LLM calls into sequential pipelines. You define a prompt, pipe it to an LLM, maybe pipe the output to a parser, and then to another LLM. It's clean, direct, and performs well for simple, single-shot tasks.

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser
 
# A simple LCEL chain
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant."),
    ("user", "{question}")
])
model = ChatOpenAI(temperature=0)
output_parser = StrOutputParser()
 
chain = prompt | model | output_parser
 
# print(chain.invoke({"question": "What is the capital of France?"}))

This is efficient for its purpose. But our brains don't work like this. When faced with a complex problem, we don't just run a single, predefined sequence. We iterate. We reflect. We ask clarifying questions. We might branch our thought process, explore different avenues, and then converge. Crucially, we maintain state – our current understanding, past attempts, and the problem context – which evolves with each step.

LCEL chains are fundamentally stateless and unidirectional. They can't loop back, make dynamic choices based on intermediate outputs, or hold onto a persistent context across multiple "turns" of thought. For building truly agentic systems – the kind that could eventually contribute to a sophisticated MoE architecture where experts dynamically engage and disengage – we need something more. We need graphs.

Embracing Non-Linearity: Why LangGraph (Conceptually) Matters

While my preference leans heavily towards raw APIs and custom, high-performance TypeScript implementations over bloated Python frameworks, the conceptual leap that LangGraph introduces is undeniably important. It acknowledges that for agentic workflows, you need:

  1. State Management: A way to maintain and update context across multiple steps.
  2. Non-Linear Execution: The ability to branch, loop, and make decisions about the next step based on the current state.

LangGraph, at its core, provides a structure to define stateful, cyclical computations. It's an abstraction layer for building finite state machines, which is a powerful paradigm for managing complex agent behavior. While I'd ultimately prefer to implement these concepts with custom, minimal Python or Rust/Go services interacting via Redis for state, LangGraph offers a relatively accessible starting point for exploring these architectures. It's less of the "kitchen sink" approach of vanilla LangChain and more focused on the graph execution engine, which is a step in the right direction.

Let's dive into building a simple, stateful agent that can decide whether to answer a query directly or ask for clarification, demonstrating its core capabilities.

Building Your First Agentic Workflow with LangGraph

Our goal is a simple agent that:

  1. Receives a user query.
  2. Decides: Is the query clear enough to answer directly?
  3. If yes, Answers the query.
  4. If no, Asks for Clarification.
  5. If clarification is asked, it then waits for updated input and loops back to Decide.

This requires state to remember the initial query, any clarifications made, and the current response.

1. Defining the Graph State

First, we need to define the schema for our graph's state. This is how information will be passed and updated between nodes. We'll use a TypedDict for clarity and type-safety.

from typing import TypedDict, Annotated, List, Union
import operator
 
# Define the state schema
class AgentState(TypedDict):
    """
    Represents the state of our agent's workflow.
    This state is passed between nodes and updated.
    """
    query: str
    clarifications: Annotated[List[str], operator.add] # Append new clarifications
    answer: Union[str, None]
    needs_clarification: bool # Flag to decide next step

Annotated[List[str], operator.add] is a neat trick here. It tells LangGraph that when a node returns a new list for clarifications, it should append to the existing list, rather than overwriting it. This is crucial for maintaining historical context.

2. Crafting the Agent Nodes

Each step in our agent's thought process will be a "node" in the graph. These are simply Python functions that take the current AgentState as input and return an updated AgentState (or a partial state to merge).

We'll need an LLM, so let's set that up.

import os
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
 
# Ensure your OPENAI_API_KEY is set as an environment variable
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
 
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2) # Using a faster, cheaper model

Now, let's define our nodes:

Node 1: decide_action This node acts as a router. It uses an LLM to determine if the query is clear enough or requires more information.

def decide_action(state: AgentState) -> AgentState:
    """
    Decides whether the query needs clarification or can be answered directly.
    """
    print("---DECIDE ACTION NODE---")
    query = state["query"]
    
    # Simple prompt to let the LLM decide
    decision_prompt = ChatPromptTemplate.from_messages([
        ("system", 
         "You are an intelligent router. Based on the user's query, decide if it is clear enough "
         "to provide a direct answer, or if it requires more clarification. "
         "Respond 'CLARIFY' if more information is needed, otherwise respond 'ANSWER'.\n\n"
         "Current query: {query}\n"
         "Existing clarifications (if any): {clarifications}"
        ),
        ("user", "Decision for: {query}")
    ])
    
    decision_chain = decision_prompt | llm | StrOutputParser()
    
    raw_decision = decision_chain.invoke({
        "query": query,
        "clarifications": "\n".join(state.get("clarifications", []))
    })
    
    decision = raw_decision.strip().upper()
    print(f"Decision: {decision}")
    
    return {"needs_clarification": (decision == "CLARIFY")}

Node 2: answer_query If the query is clear, this node attempts to provide a direct answer.

def answer_query(state: AgentState) -> AgentState:
    """
    Answers the user's query directly.
    """
    print("---ANSWER QUERY NODE---")
    query = state["query"]
    
    answer_prompt = ChatPromptTemplate.from_messages([
        ("system", 
         "You are a helpful assistant. Provide a concise and accurate answer to the following query. "
         "If clarifications were previously provided, use them to inform your answer.\n\n"
         "Query: {query}\n"
         "Clarifications: {clarifications}"
        ),
        ("user", "{query}")
    ])
    
    answer_chain = answer_prompt | llm | StrOutputParser()
    
    response = answer_chain.invoke({
        "query": query,
        "clarifications": "\n".join(state.get("clarifications", []))
    })
    
    print(f"Answer: {response}")
    return {"answer": response}

Node 3: ask_clarification If the query is unclear, this node generates a clarifying question.

def ask_clarification(state: AgentState) -> AgentState:
    """
    Asks the user for clarification based on the current query.
    """
    print("---ASK CLARIFICATION NODE---")
    query = state["query"]
    
    clarification_prompt = ChatPromptTemplate.from_messages([
        ("system", 
         "You are a polite assistant. Formulate a concise and clear question to get more information "
         "from the user regarding their query. Focus on what's missing.\n\n"
         "Original Query: {query}\n"
         "Previous Clarifications (if any): {clarifications}"
        ),
        ("user", "What clarification is needed for: {query}")
    ])
    
    clarification_chain = clarification_prompt | llm | StrOutputParser()
    
    clarifying_question = clarification_chain.invoke({
        "query": query,
        "clarifications": "\n".join(state.get("clarifications", []))
    })
    
    print(f"Clarification needed: {clarifying_question}")
    return {"clarifications": [clarifying_question], "answer": None} # Clear previous answer if any

3. Assembling the Graph with StatefulGraph

Now, we bring it all together using StatefulGraph. This is where the non-linear magic happens.

from langgraph.graph import StateGraph, END
 
# Create the graph instance
workflow = StateGraph(AgentState)
 
# Add the nodes
workflow.add_node("decide_action", decide_action)
workflow.add_node("answer_query", answer_query)
workflow.add_node("ask_clarification", ask_clarification)
 
# Set the entry point (where the graph starts)
workflow.set_entry_point("decide_action")
 
# Define edges: how nodes connect
# If 'decide_action' determines no clarification is needed, go to 'answer_query'
# Otherwise, go to 'ask_clarification'
workflow.add_conditional_edges(
    "decide_action",       # Source node
    lambda state: "clarify" if state["needs_clarification"] else "answer", # A function that determines the next node
    {
        "clarify": "ask_clarification",
        "answer": "answer_query"
    }
)
 
# After answering, the workflow is complete
workflow.add_edge("answer_query", END)
 
# After asking for clarification, we loop back to 'decide_action'
# The user provides new input, and the graph effectively restarts from 'decide_action'
# with the updated state. For demonstration, we'll manually feed updated state.
# For a real agent, this would typically involve waiting for user input.
workflow.add_edge("ask_clarification", "decide_action") # Loop back! This is the key.
 
# Compile the graph
app = workflow.compile()

4. Executing the Agent

To run this, we'll manually simulate the user providing new input after a clarification is requested.

# Initial query
initial_state = {
    "query": "I need help with my Python code.",
    "clarifications": [],
    "answer": None,
    "needs_clarification": False # Initialized, will be updated by node
}
 
print("\n--- First Execution ---")
# Invoke the graph with the initial state
# The `stream` method is efficient for long-running agents, yielding state updates
for s in app.stream(initial_state):
    print(s)
    print("---")
 
# Let's assume the last state yielded is the final one
final_state_after_first_run = list(app.stream(initial_state))[-1]
# If clarification was asked, manually update the query for the next turn
if final_state_after_first_run.get("ask_clarification") or final_state_after_first_run.get("decide_action", {}).get("needs_clarification"):
    print("\n--- Agent requested clarification. User provides more info. ---")
    
    # Simulate user updating the query based on clarification
    updated_query = "My Python code is a web scraper using BeautifulSoup and I'm getting a 'NoneType' error when trying to find an element."
    
    # Create new state for the next turn, merging updated query and existing clarifications
    # In a real app, this would be a new user message. Here we manually update `query`
    # LangGraph state update implicitly merges dictionaries.
    next_state = {
        **initial_state, # Preserve original structure
        "query": updated_query,
        "clarifications": final_state_after_first_run.get("ask_clarification", {}).get("clarifications", []) # Carry over the clarifying question asked
    }
 
    print("\n--- Second Execution with updated query ---")
    for s in app.stream(next_state):
        print(s)
        print("---")
 
    # Get the final answer
    final_state = list(app.stream(next_state))[-1]
    print(f"\nFinal Answer: {final_state.get('answer_query', {}).get('answer')}")
 
else:
    final_state = final_state_after_first_run
    print(f"\nFinal Answer: {final_state.get('answer_query', {}).get('answer')}")

Visualizing the Graph (Optional but Recommended)

For more complex graphs, visualization is essential. If you have graphviz installed (pip install pygraphviz and sudo apt-get install graphviz on Linux), you can render your graph.

from IPython.display import Image, display
 
# Requires pygraphviz
# If you don't have graphviz, skip this or install it:
# pip install pygraphviz
# sudo apt-get install graphviz (on Debian/Ubuntu)
 
try:
    display(Image(app.get_graph().draw_png()))
except Exception as e:
    print(f"Could not draw graph: {e}. Make sure pygraphviz and graphviz are installed.")

What I Learned

This exercise, even with a framework like LangGraph, reinforces critical principles for building intelligent agents:

  1. State is King: Without explicit state management, dynamic, iterative behavior is impossible. The AgentState TypedDict and the operator.add annotation for lists are elegant solutions for managing evolving context.
  2. Graphs Enable True Agency: Moving beyond linear chains to a graph structure unlocks the ability for agents to make decisions, branch their logic, and loop back – mimicking a more natural, cognitive problem-solving process. This is the bedrock for sophisticated MoE architectures where expert routing and iterative refinement are paramount.
  3. Modular Design via Nodes: Each node is a focused, single-responsibility function. This promotes clean code, testability, and easier debugging. It's a pattern competitive programmers appreciate: break down complex problems into manageable, composable units.
  4. Conditional Routing is Powerful: add_conditional_edges is the engine of dynamic behavior. It allows the agent to react to the current state and choose the most appropriate next step, crucial for building adaptive systems.
  5. Still Room for Optimization: While LangGraph is a significant step up from vanilla LCEL for complex agent flows, it's still an abstraction. For maximum performance and minimal overhead (which is always my goal), one might consider implementing a custom graph execution engine with raw API calls, explicit state storage (e.g., Redis for inter-service communication), and potentially even compile-to-native solutions for the "routing" logic. This would allow fine-grained control over resource allocation and latency, especially critical for high-throughput AI research.

This is just Part 1. The journey towards building agents that truly "think" – with intricate MoE structures, long-term memory, and complex reasoning loops – has just begun. The fundamental concepts of state, nodes, and dynamic graph traversal are indispensable building blocks for that ambitious future. Next, we'll explore more complex decision points and how to integrate external tools into these agentic loops.