Building Agentic Workflows in Python: Beyond Basic RAG with LangGraph

If you built an AI feature recently, chances are you used standard Retrieval-Augmented Generation (RAG): fetch relevant documents from a vector database, pass them to a Large Language Model (LLM) along with a prompt, and return the answer.

While basic RAG works well for straightforward static lookup, real-world engineering demands more:

What happens when the vector search returns irrelevant documents?
What if the query requires multi-step calculations or multi-source lookups?
What if the LLM hallucinates and needs to self-correct before presenting a response to the user?

This is where Agentic AI comes in. Instead of a linear pipeline (Input -> Retrieve -> Generate), agentic architectures treat LLMs as reasoning engines within a stateful graph. They evaluate outputs, decide which tools to call, and dynamically loop back to correct mistakes.

In this guide, we'll build a self-correcting, stateful agentic system using Python and LangGraph.

Why LangGraph for Agentic Control?

Standard chains (like basic LangChain or LlamaIndex) operate as Directed Acyclic Graphs (DAGs) — they move in one direction.

True agentic reasoning, however, requires cyclical loops:

      [ User Query ]
            │
            ▼
     [ Reason / Plan ]
            │
    ┌───────┴───────┐
    ▼               ▼
[Call Tool]    [Evaluate Output] ──(Is valid?)──► [Return Result]
    │               │
    └───────┬───────┘ (If invalid / missing data)
            │
            ▼
     [Self-Correct Loop]

LangGraph models your application as a state machine:

State: A shared data structure representing the current system status.
Nodes: Python functions that process the state and return updates.
Edges: Control logic (conditional routes) that dictate which node runs next based on current state.

Step-by-Step Implementation

Let's build a self-evaluating RAG agent with a fallback search tool. If the retrieved context doesn't properly answer the question, the agent will dynamically switch tools to query external data.

1. Prerequisites & Setup

Install the required packages:

pip install langgraph langchain-openai langchain-community tavily-python

Set your environment variables:

import os

os.environ["OPENAI_API_KEY"] = "your-openai-key"
os.environ["TAVILY_API_KEY"] = "your-tavily-key"  # For fallback search

2. Define the Agent State

The state acts as a shared ledger updated by each node in the graph.

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    # 'add_messages' appends new messages rather than overwriting existing ones
    messages: Annotated[Sequence[BaseMessage], add_messages]
    search_needed: bool
    iterations: int

3. Define Nodes (Tools & Reasoning)

Now, we write the functions that modify our AgentState.

from langchain_openai import ChatOpenAI
from langchain_community.tools import TavilySearchResults
from langchain_core.messages import SystemMessage, HumanMessage

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
web_search_tool = TavilySearchResults(max_results=2)

def agent_node(state: AgentState):
    """Generates an answer or decides to use a tool."""
    messages = state["messages"]
    
    system_prompt = SystemMessage(
        content="You are a helpful technical assistant. If you lack context or information, "
                "flag that you need a search."
    )
    
    response = llm.invoke([system_prompt] + list(messages))
    return {"messages": [response], "iterations": state.get("iterations", 0) + 1}

def search_evaluator_node(state: AgentState):
    """Evaluates whether the last response requires external search grounding."""
    last_message = state["messages"][-1].content
    
    eval_prompt = f"""
    Analyze the following response:
    '{last_message}'
    
    Does this response express uncertainty, missing information, or a need to search external sources?
    Reply with ONLY 'YES' if search is needed, or 'NO' if the answer is complete.
    """
    
    evaluation = llm.invoke([HumanMessage(content=eval_prompt)]).content.strip().upper()
    needs_search = "YES" in evaluation
    
    return {"search_needed": needs_search}

def fallback_search_node(state: AgentState):
    """Executes external web search if internal knowledge wasn't sufficient."""
    user_query = state["messages"][0].content
    search_results = web_search_tool.invoke({"query": user_query})
    
    tool_message = HumanMessage(
        content=f"Web Search Results for context:\n{search_results}\n\nPlease synthesize a complete answer using these results."
    )
    return {"messages": [tool_message], "search_needed": False}

4. Wire the Graph with Conditional Routing

Now we construct the graph structure, binding the nodes with conditional edges.

Initialize Graph: StateGraph setup. Define the state graph boundary using StateGraph(AgentState).
Add Nodes: Register functions. Add agent, evaluator, and search nodes to the workflow graph.
Add Conditional Edges: Route execution flow. Define routes that inspect search_needed and iteration counts to prevent infinite loops.
Compile Workflow: Execution ready. Compile the graph into a runnable executable pipeline.

from langgraph.graph import StateGraph, END

# Router decision logic
def route_search(state: AgentState):
    # Guard against infinite loops: maximum 3 retries
    if state.get("iterations", 0) > 3:
        return END
        
    if state.get("search_needed"):
        return "fallback_search"
    return END

# Construct the graph
builder = StateGraph(AgentState)

# Add nodes
builder.add_node("agent", agent_node)
builder.add_node("evaluator", search_evaluator_node)
builder.add_node("fallback_search", fallback_search_node)

# Set entry point
builder.set_entry_point("agent")

# Connect workflow
builder.add_edge("agent", "evaluator")

# Add conditional routing based on evaluator feedback
builder.add_conditional_edges(
    "evaluator",
    route_search,
    {
        "fallback_search": "fallback_search",
        END: END
    }
)

# Loop back to agent after fallback search adds data to state
builder.add_edge("fallback_search", "agent")

# Compile the execution runnable
app = builder.compile()

5. Running the Agentic Pipeline

from langchain_core.messages import HumanMessage

# Run query requiring updated web context
inputs = {
    "messages": [HumanMessage(content="What are the key architectural features of Python 3.13?")],
    "iterations": 0
}

for step in app.stream(inputs):
    for node_name, output in step.items():
        print(f"--- Node Executed: {node_name} ---")
        if "messages" in output:
            print(f"Latest Content: {output['messages'][-1].content[:150]}...\n")
        elif "search_needed" in output:
            print(f"Search Required: {output['search_needed']}\n")

Key Takeaways for Production Agentic Systems

Concept	Traditional Pipelines	Agentic Workflows (LangGraph)
Execution Path	Fixed DAG (linear / branching)	Cyclic (dynamic loops, retries)
Error Handling	Hard failures / Exception blocks	Self-evaluating correction loops
State Management	Passed through arguments	Centralized, append-only state log
Control	Implicit in code flow	Explicit state machine graph logic

Always Set Loop Guardrails: Agents can easily fall into infinite retry loops when confused. Enforce a strict max_iterations counter in state.
Keep Evaluator Prompts Small: Use lightweight models (e.g., gpt-4o-mini) for evaluation nodes to keep latency low and cost minimal.
Decouple Decision from Execution: Separate the evaluator node from the action node to test and benchmark each step independently.