LangGraph in Python for Agentic LLM Workflows

Introduction to LangGraph

LangGraph is a framework for building stateful, graph-based workflows for large language model applications. It extends LangChain-style orchestration with explicit nodes, edges, state transitions, conditional routing, and multi-step execution patterns.

For production agentic systems, LangGraph is useful because it makes the workflow structure visible. Instead of hiding behavior inside one large prompt, teams can model the application as a controlled graph with auditable steps, state boundaries, routing decisions, and execution paths.

The following example shows a minimal LangGraph setup with a graph, model, and prompt template.

import langgraph as lg
from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate

# Initialize LangGraph components
graph = lg.Graph()
llm = ChatOpenAI()
prompt = ChatPromptTemplate.from_template("Hello, {name}!")

Core Concepts of LangGraph

LangGraph models an LLM application as a graph of nodes and edges. Nodes represent tasks, tools, agents, or processing steps. Edges define how state moves from one step to another. This structure is useful for applications that require routing, retries, parallel execution, memory, or multi-agent collaboration.

The following example defines two graph nodes and connects them with an edge.

# Define nodes in the graph
@graph.node
def greeting(state):
    name = state['name']
    response = llm.invoke(prompt.format(name=name))
    return {"greeting": response}

@graph.node
def farewell(state):
    return {"farewell": "Goodbye, " + state['name'] + "!"}

# Define edges
graph.add_edge('greeting', 'farewell')

State Management in LangGraph

State management is one of the key reasons to use LangGraph for agentic workflows. Each node can read from and write to shared state, which allows the application to preserve context, intermediate results, routing decisions, and tool outputs across multiple steps.

The following example initializes state and runs it through the graph.

# Initialize state
initial_state = {"name": "Alice"}

# Run the graph
final_state = graph.run(initial_state)

print(final_state)
# Output: {'name': 'Alice', 'greeting': 'Hello, Alice!', 'farewell': 'Goodbye, Alice!'}

Conditional Flows

Conditional routing allows a graph to choose different execution paths based on state, model output, validation results, tool responses, or policy decisions. This is important for agentic systems where not every request should follow the same path.

The following example routes execution based on a simple mood field.

@graph.node
def check_mood(state):
    mood = state.get('mood', 'neutral')
    if mood == 'happy':
        return {'next': 'positive_response'}
    else:
        return {'next': 'neutral_response'}

graph.add_edge('check_mood', 'positive_response', condition=lambda x: x['next'] == 'positive_response')
graph.add_edge('check_mood', 'neutral_response', condition=lambda x: x['next'] == 'neutral_response')

Parallel Execution

LangGraph can run independent nodes in parallel, which is useful when multiple retrieval calls, tool calls, evaluations, or enrichment steps do not depend on each other.

The following example executes two independent tasks before moving to the final step.

@graph.node
def task_a(state):
    return {"result_a": "Task A completed"}

@graph.node
def task_b(state):
    return {"result_b": "Task B completed"}

graph.add_edge('start', ['task_a', 'task_b'])
graph.add_edge(['task_a', 'task_b'], 'end')

result = graph.run({"start": True})
print(result)
# Output: {'start': True, 'result_a': 'Task A completed', 'result_b': 'Task B completed'}

Error Handling

Error handling is required for production agent workflows because model calls, tools, APIs, and routing decisions can fail. A reliable graph should define fallback paths, error handlers, retry logic, and escalation rules.

The following example routes a failed operation to an error handler.

@graph.node
def risky_operation(state):
    if state.get('safe', False):
        return {"result": "Operation successful"}
    else:
        raise ValueError("Unsafe operation")

@graph.node
def error_handler(state, error):
    return {"error_message": str(error)}

graph.add_edge('risky_operation', 'success', condition=lambda x: 'result' in x)
graph.add_edge('risky_operation', 'error_handler', on_error=True)

result = graph.run({"safe": False})
print(result)
# Output: {'safe': False, 'error_message': 'Unsafe operation'}

Integrating External APIs

LangGraph can orchestrate external API calls inside LLM workflows. In production, API access should be controlled with authentication, rate limits, error handling, input validation, and observability.

The following example retrieves weather data and passes it into a report-generation node.

import requests

@graph.node
def fetch_weather(state):
    city = state['city']
    api_key = "your_api_key_here"
    url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}"
    response = requests.get(url)
    data = response.json()
    return {"weather": data['weather'][0]['description']}

@graph.node
def generate_weather_report(state):
    weather = state['weather']
    report = llm.invoke(f"Generate a weather report for {weather} conditions.")
    return {"report": report}

graph.add_edge('fetch_weather', 'generate_weather_report')

Use Case: Customer Support Chatbot

A customer support workflow is a natural LangGraph use case because requests often need classification, routing, retrieval, escalation, and response generation. The workflow should also preserve audit logs and escalation paths for unresolved or high-risk cases.

The following example routes product and shipping inquiries to different handlers.

@graph.node
def classify_inquiry(state):
    inquiry = state['user_input']
    classification = llm.invoke(f"Classify this inquiry: {inquiry}")
    return {"inquiry_type": classification}

@graph.node
def handle_product_inquiry(state):
    inquiry = state['user_input']
    response = llm.invoke(f"Provide product information for: {inquiry}")
    return {"bot_response": response}

@graph.node
def handle_shipping_inquiry(state):
    inquiry = state['user_input']
    response = llm.invoke(f"Provide shipping information for: {inquiry}")
    return {"bot_response": response}

graph.add_edge('classify_inquiry', 'handle_product_inquiry', condition=lambda x: 'product' in x['inquiry_type'].lower())
graph.add_edge('classify_inquiry', 'handle_shipping_inquiry', condition=lambda x: 'shipping' in x['inquiry_type'].lower())

result = graph.run({"user_input": "When will my order arrive?"})
print(result['bot_response'])

Persistent State Across Conversations

Persistent state allows a workflow to retain relevant context across turns. In production systems, this should be designed carefully so that memory is scoped, summarized, secured, and not allowed to leak sensitive information between users or sessions.

The following example uses a simple conversation state object.

class ConversationState:
    def __init__(self):
        self.history = []

    def add_message(self, role, content):
        self.history.append({"role": role, "content": content})

    def get_context(self):
        return "\n".join([f"{msg['role']}: {msg['content']}" for msg in self.history[-5:]])

conv_state = ConversationState()

@graph.node
def process_user_input(state):
    user_input = state['user_input']
    conv_state.add_message("user", user_input)
    context = conv_state.get_context()
    response = llm.invoke(f"Given this context:\n{context}\nRespond to: {user_input}")
    conv_state.add_message("assistant", response)
    return {"bot_response": response}

# Usage
for user_input in ["Hello!", "What's the weather like?", "Thank you!"]:
    result = graph.run({"user_input": user_input})
    print(f"User: {user_input}")
    print(f"Bot: {result['bot_response']}")

Dynamic Graph Modification

Dynamic graph behavior allows the workflow to adapt based on query complexity, policy checks, tool results, or user intent. This flexibility should be used carefully because dynamic behavior can make testing and auditing harder.

The following example routes a query to either a simple or detailed response path.

@graph.node
def assess_complexity(state):
    query = state['user_query']
    complexity = llm.invoke(f"Assess the complexity of this query: {query}")
    return {"complexity": complexity}

@graph.node
def simple_response(state):
    query = state['user_query']
    return {"response": llm.invoke(f"Provide a simple answer to: {query}")}

@graph.node
def complex_response(state):
    query = state['user_query']
    return {"response": llm.invoke(f"Provide a detailed, expert-level answer to: {query}")}

@graph.node
def route_query(state):
    if state['complexity'] == 'simple':
        graph.add_edge('route_query', 'simple_response')
    else:
        graph.add_edge('route_query', 'complex_response')
    return {}

graph.add_edge('assess_complexity', 'route_query')

result = graph.run({"user_query": "What is photosynthesis?"})
print(result['response'])

Multi-Agent Collaboration

LangGraph can coordinate multiple specialized agents, such as researchers, writers, reviewers, validators, and tool executors. Multi-agent workflows need clear ownership of state, termination conditions, and validation gates to avoid loops or conflicting outputs.

The following example chains researcher, writer, and editor nodes.

@graph.node
def researcher(state):
    topic = state['topic']
    research = llm.invoke(f"Conduct research on: {topic}")
    return {"research": research}

@graph.node
def writer(state):
    research = state['research']
    article = llm.invoke(f"Write an article based on this research: {research}")
    return {"article": article}

@graph.node
def editor(state):
    article = state['article']
    edited_article = llm.invoke(f"Edit and improve this article: {article}")
    return {"final_article": edited_article}

graph.add_edge('researcher', 'writer')
graph.add_edge('writer', 'editor')

result = graph.run({"topic": "Artificial Intelligence in Healthcare"})
print(result['final_article'])

Use Case: Recipe Generator

A recipe generator is a simple example of a workflow that extracts constraints, checks available inputs, and generates a response. The same pattern applies to enterprise workflows that require constraint extraction before generation.

The following example extracts preferences, checks ingredients, and generates a recipe.

@graph.node
def get_preferences(state):
    preferences = state['user_preferences']
    return {"dietary_restrictions": llm.invoke(f"Extract dietary restrictions from: {preferences}")}

@graph.node
def inventory_check(state):
    ingredients = state['available_ingredients']
    return {"usable_ingredients": llm.invoke(f"List ingredients that can be used in a recipe: {ingredients}")}

@graph.node
def generate_recipe(state):
    restrictions = state['dietary_restrictions']
    ingredients = state['usable_ingredients']
    recipe = llm.invoke(f"Generate a recipe considering these restrictions: {restrictions} and using these ingredients: {ingredients}")
    return {"recipe": recipe}

graph.add_edge('get_preferences', 'inventory_check')
graph.add_edge('inventory_check', 'generate_recipe')

result = graph.run({
    "user_preferences": "vegetarian, no nuts",
    "available_ingredients": "tomatoes, pasta, garlic, olive oil, basil"
})
print(result['recipe'])

Debugging and Logging

Debugging and logging are mandatory for production LangGraph systems. Teams need visibility into state transitions, node execution, tool calls, errors, retries, latency, and final outputs.

The following example adds basic logging around graph execution.

import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@graph.node
def log_state(state):
    logger.info(f"Current state: {state}")
    return {}

@graph.node
def process_data(state):
    data = state['input_data']
    result = llm.invoke(f"Process this data: {data}")
    logger.info(f"Processed data: {result}")
    return {"processed_data": result}

graph.add_edge('log_state', 'process_data')
graph.add_edge('process_data', 'log_state')

graph.run({"input_data": "Sample input for processing"})

Additional Resources

For more information on LangGraph and its applications, consider exploring the following resources:

1. LangChain Documentation: https://python.langchain.com/docs/get_started/introduction
2. “Large Language Models and Graph-based Reasoning” (ArXiv:2307.05722): https://arxiv.org/abs/2307.05722
3. “Graph-based Reasoning with Large Language Models” (ArXiv:2305.15117): https://arxiv.org/abs/2305.15117

These resources provide useful background on LangGraph, graph-based reasoning, and structured orchestration patterns for LLM applications.

Closing Thoughts

LangGraph is valuable when an LLM application needs explicit workflow control, persistent state, conditional routing, tool orchestration, and multi-agent coordination. It is especially useful for agentic systems where the application must move through defined steps rather than rely on a single prompt-response cycle.

The production decision is straightforward: use simple chains for simple tasks, use RAG when answers need grounding, and use LangGraph when the workflow needs stateful execution, branching, retries, parallel work, or controlled agent collaboration. The more autonomy the workflow has, the more important logging, evaluation, guardrails, and governance become.

Related Reading

• LangChain and LangGraph Basics
• Agents and Multi-Agent Frameworks in Python
• Diffusion Models in Python