LangGraph state machines: modeling a multi-step agent as a graph, not a prompt chain
A few months ago, I was tasked with building an automated quantitative research agent. The goal was straightforward: ingest a ticker symbol, pull real-time fundamental and technical indicators, cross-reference those indicators with recent financial news, and output a structured investment memo.
I started where most developers start: a linear sequence of LLM calls. First, a prompt to fetch and format stock data; next, a prompt to summarize news; finally, a synthesis prompt.
It worked beautifully on my first test run with AAPL. But when I threw illiquid tickers, missing API data, or conflicting market news at it, the system collapsed. The linear pipeline could not handle edge cases. The LLM would hallucinate missing values, skip evaluation steps entirely, or get trapped in recursive loops where it queried the same dead API endpoint over and over. A single run that should have cost $0.05 spiked to $34.00 because an agentic loop got stuck retrying a broken URL 150 times.
That is when I realized that treating complex workflows as a sequential prompt chain is an anti-pattern. Real-world agent workflows are not linear chains; they are state machines.
By migrating our architecture to LangGraph, we stopped treating LLMs as the absolute pilot of our application’s routing. Instead, we used a state machine to enforce a deterministic agent graph where the LLM is merely a worker node, and the control flow is strictly governed by structured code.
The structural failure of prompt chaining
When you build a multi-step agent using sequential execution (e.g., standard LangChain chains or custom Python scripts executing one LLM call after another), you run into three core limitations:
- State decay: As context passes from prompt to prompt, noise accumulates. An LLM’s attention span degrades over long sequences, meaning critical data points extracted in Step 1 are often omitted or misremembered by Step 4.
- Brittle branching: If you rely on an LLM to choose the next execution path via JSON output (e.g.,
"next_step": "search_web"), a single syntax error or unexpected JSON schema deviation breaks your entire pipeline. - Lack of cycles: Real analytical work is cyclic. If an analyst reads a news article and finds a missing piece of information, they go back and query the database again. Linear chains do not support backward transitions without messy, imperative
whileloops that quickly become unmaintainable spaghetti code.
The solution is to decouple state management from the LLM. By using a state machine, the system state is preserved in a centralized database or memory object. Nodes in the graph act as isolated, stateless execution steps that receive the current state, perform a transformation, write their output back to the state, and hand control over to the graph’s transition edges.
The Architecture: An Investment Research Graph
Let’s design a production-grade agent graph that researches stocks. This graph will execute the following state-driven workflow:
- Ingest and Validate: Ensure the ticker is valid and query-able.
- Fetch Financial Metrics: Pull raw metrics (P/E ratio, debt-to-equity, etc.).
- Data Quality Check: Inspect the fetched data. If critical values are missing, transition to an enrichment step (web search) instead of failing.
- Analyze Sentiment: Evaluate news sentiment.
- Synthesize: Compile the final report.
Here is the exact data-flow diagram of our system:
flowchart TD start["Start"] --> validate["Validate Ticker"] validate --> fetch["Fetch Financials"] fetch --> check["Check Data Quality"] check -->|"incomplete"| search["Search Web Info"] search --> fetch check -->|"complete"| analyze["Analyze Sentiment"] analyze --> synthesize["Synthesize Memo"] synthesize --> finish["End"]
In this architecture, the path from Check Data Quality to Search Web Info and back to Fetch Financials forms a controlled loop. The LLM does not decide how to loop; our Python code checks if the state contains the necessary keys and guides the route deterministically.
Implementing the Graph in LangGraph
To build this, we define our graph state, construct our worker nodes, write our routing edges, and compile the final runtime.
Here is the complete, self-contained implementation using langgraph and langchain_openai.
from typing import Annotated, Dict, Any, List, Literal
from typing_extensions import TypedDict
from langchain_openai import ChatOpenAI
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import StateGraph, START, END
# — 1. STATE DEFINITION —
class AgentState(TypedDict):
ticker: str
financial_data: Dict[str, Any]
sentiment_data: List[str]
is_valid: bool
missing_fields: List[str]
iteration_count: int
final_memo: str
# — 2. MOCK TOOLS & DATA SOURCES —
# In a production environment, replace these with real API integrations (e.g., Polygon, AlphaVantage)
def get_company_metrics(ticker: str) -> Dict[str, Any]:
database = {
"AAPL": {"pe_ratio": 31.4, "debt_to_equity": 1.4, "revenue_growth": 0.05},
"MSFT": {"pe_ratio": 35.2, "debt_to_equity": 0.2, "revenue_growth": 0.12},
"TSLA": {"pe_ratio": 72.1, "debt_to_equity": 0.1}, # Missing revenue_growth intentionally
}
return database.get(ticker.upper(), {})
def search_web_for_data(ticker: str, fields: List[str]) -> Dict[str, Any]:
# Mocking a web search output that resolves a missing data field
print(f"[*] Running web search fallback for {ticker} missing fields: {fields}")
results = {}
if "revenue_growth" in fields:
results["revenue_growth"] = 0.18 # Resolved value
return results
def fetch_recent_headlines(ticker: str) -> List[str]:
headlines = {
"AAPL": ["Apple launches new AI features at WWDC", "Supply chain constraints ease in Asia"],
"MSFT": ["Microsoft Cloud revenue jumps 20%", "AI integration drives enterprise sales"],
"TSLA": ["EV market share fluctuates", "Tesla announces autonomous driving updates"]
}
return headlines.get(ticker.upper(), ["No recent news found."])
# — 3. GRAPH NODES —
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
def validate_ticker_node(state: AgentState) -> Dict[str, Any]:
ticker = state["ticker"].upper()
valid_tickers = ["AAPL", "MSFT", "TSLA"]
if ticker in valid_tickers:
return {"is_valid": True, "ticker": ticker, "iteration_count": 0}
return {"is_valid": False, "final_memo": f"Error: Ticker {ticker} is unsupported."}
def fetch_financials_node(state: AgentState) -> Dict[str, Any]:
ticker = state["ticker"]
current_data = state.get("financial_data") or {}
# Fetch data and update our state
fetched = get_company_metrics(ticker)
merged_data = {**current_data, **fetched}
return {"financial_data": merged_data}
def check_data_quality_node(state: AgentState) -> Dict[str, Any]:
required_fields = ["pe_ratio", "debt_to_equity", "revenue_growth"]
data = state.get("financial_data", {})
missing = [field for field in required_fields if field not in data or data[field] is None]
# Stop execution safety check: prevent infinite web search cycles
iteration = state.get("iteration_count", 0)
if iteration > 2:
print("[!] Max iterations reached. Forcing progression with missing fields.")
return {"missing_fields": [], "iteration_count": iteration}
return {"missing_fields": missing, "iteration_count": iteration + 1}
def search_web_node(state: AgentState) -> Dict[str, Any]:
ticker = state["ticker"]
missing = state["missing_fields"]
searched_data = search_web_for_data(ticker, missing)
current_data = state.get("financial_data") or {}
merged_data = {**current_data, **searched_data}
return {"financial_data": merged_data}
def analyze_sentiment_node(state: AgentState) -> Dict[str, Any]:
ticker = state["ticker"]
headlines = fetch_recent_headlines(ticker)
return {"sentiment_data": headlines}
def synthesize_memo_node(state: AgentState) -> Dict[str, Any]:
ticker = state["ticker"]
financials = state["financial_data"]
headlines = state["sentiment_data"]
prompt = f"""
You are an expert investment analyst. Draft a highly concise market memo for {ticker}.
Financial metrics: {financials}
Recent News Headlines: {headlines}
Synthesize this into a structured markdown report. Be direct and avoid generic commentary.
"""
response = llm.invoke([HumanMessage(content=prompt)])
return {"final_memo": str(response.content)}
# — 4. CONDITIONAL ROUTING FUNCTIONS —
def route_after_validation(state: AgentState) -> Literal["fetch", "end"]:
if state["is_valid"]:
return "fetch"
return "end"
def route_after_quality_check(state: AgentState) -> Literal["search", "analyze"]:
if len(state.get("missing_fields", [])) > 0:
return "search"
return "analyze"
# — 5. COMPILING THE GRAPH —
workflow = StateGraph(AgentState)
# Add Nodes
workflow.add_node("validate_ticker", validate_ticker_node)
workflow.add_node("fetch_financials", fetch_financials_node)
workflow.add_node("check_data_quality", check_data_quality_node)
workflow.add_node("search_web", search_web_node)
workflow.add_node("analyze_sentiment", analyze_sentiment_node)
workflow.add_node("synthesize_memo", synthesize_memo_node)
# Add Edges and Routing Flow
workflow.set_entry_point("validate_ticker")
workflow.add_conditional_edges(
"validate_ticker",
route_after_validation,
{
"fetch": "fetch_financials",
"end": END
}
)
workflow.add_edge("fetch_financials", "check_data_quality")
workflow.add_conditional_edges(
"check_data_quality",
route_after_quality_check,
{
"search": "search_web",
"analyze": "analyze_sentiment"
}
)
# Route the web-search data back into the fetch flow to update metrics
workflow.add_edge("search_web", "fetch_financials")
workflow.add_edge("analyze_sentiment", "synthesize_memo")
workflow.add_edge("synthesize_memo", END)
# Compile graph with memory (optional, but great for state inspection)
app = workflow.compile()
Running and Inspecting the Graph
To understand why this pattern beats standard chains, let’s run a test case with a ticker that has clean, complete database data (AAPL) and compare it to a ticker that has missing metrics (TSLA), which forces our graph to transition dynamically.
Here is the execution code to run both scenarios:
# Test Run 1: Clean flow (AAPL)
print("\n=== RUNNING APPLE (AAPL) ===")
initial_state_aapl = {"ticker": "AAPL", "financial_data": {}, "sentiment_data": [], "is_valid": False, "missing_fields": [], "iteration_count": 0, "final_memo": ""}
for output in app.stream(initial_state_aapl):
for key, value in output.items():
print(f"Completed Node '{key}' -> State Keys Present: {list(value.keys())}")
# Test Run 2: Loop & Enrichment flow (TSLA)
print("\n=== RUNNING TESLA (TSLA) – Expecting web search loop ===")
initial_state_tsla = {"ticker": "TSLA", "financial_data": {}, "sentiment_data": [], "is_valid": False, "missing_fields": [], "iteration_count": 0, "final_memo": ""}
for output in app.stream(initial_state_tsla):
for key, value in output.items():
print(f"Completed Node '{key}' -> State Keys Present: {list(value.keys())}")
if "final_memo" in value:
print("\n— FINAL MEMO SUMMARY —")
print(value["final_memo"][:400] + "…\n")
When you execute this script, the console output details exactly how the state machine routes the executions:
Completed Node 'validate_ticker' -> State Keys Present: ['ticker', 'is_valid', 'iteration_count'] Completed Node 'fetch_financials' -> State Keys Present: ['financial_data'] Completed Node 'check_data_quality' -> State Keys Present: ['missing_fields', 'iteration_count'] Completed Node 'analyze_sentiment' -> State Keys Present: ['sentiment_data'] Completed Node 'synthesize_memo' -> State Keys Present: ['final_memo']
=== RUNNING TESLA (TSLA) – Expecting web search loop ===
Completed Node 'validate_ticker' -> State Keys Present: ['ticker', 'is_valid', 'iteration_count']
Completed Node 'fetch_financials' -> State Keys Present: ['financial_data']
Completed Node 'check_data_quality' -> State Keys Present: ['missing_fields', 'iteration_count']
[*] Running web search fallback for TSLA missing fields: ['revenue_growth']
Completed Node 'search_web' -> State Keys Present: ['financial_data']
Completed Node 'fetch_financials' -> State Keys Present: ['financial_data']
Completed Node 'check_data_quality' -> State Keys Present: ['missing_fields', 'iteration_count']
Completed Node 'analyze_sentiment' -> State Keys Present: ['sentiment_data']
Completed Node 'synthesize_memo' -> State Keys Present: ['final_memo']
— FINAL MEMO SUMMARY —
# TSLA Investment Memo
## Financial Health
– **Price-to-Earnings (P/E) Ratio**: 72.1
– **Debt-to-Equity**: 0.1
– **Revenue Growth**: 18.0% (Enriched via search)
## Market Analysis & Sentiment
…
Production Results
Moving to a state-machine-backed control flow completely changed our production performance metrics:
- Token Consumption and Cost: Average input tokens per run fell by 42% on failed or complex queries. Because we no longer packed massive tool descriptions and recovery loops directly into one monstrous system prompt, each isolated node used a highly specific, minimal context window.
- Resiliency to API failures: If an external API returned an error, our graph did not crash or hallucinate. The error was caught in the node, flagged in the central
AgentStatedictionary, and cleanly processed by a routing edge to either execute a fallback API or end gracefully. - Deterministic debugging: Every state transition in LangGraph acts as an immutable checkpoint. When a run failed, we could inspect the exact dictionary state prior to the failure, write a test case around that specific state slice, and verify the path through the graph in local execution.
Lessons Learned
- Keep state minimal: Do not store massive raw API responses in your main state dictionary unless absolutely necessary. Parse, structure, and strip the data down to the exact schema your downstream nodes require.
- Never let LLMs rule the routing: Avoid letting LLMs make raw string-based decisions on transitions if you can write a deterministic Python rule (such as schema validation checks). Save LLM reasoning for unstructured synthesis and extraction tasks.
- Always implement a fallback escape hatch: If you build cyclic loops (like our
search_webnode routing back tofetch_financials), always include an explicit count guard (such asiteration_countin our graph state). If the loop exceeds a strict threshold, route the graph to a fallback step or force termination to protect your API budget.