Long-running research agents: Temporal for durability, LangGraph for control flow
If you are running quantitative research agents that take more than five minutes to execute, your current stack is probably built on a house of cards.
A year ago, I was building an automated equity research pipeline. The goal was to have an agent pool systematically ingest 10-K filings, parse transcript calls, scrape alternative data sources, run risk-modeling simulations, and output structured investment memos for a universe of 150 mid-cap equities.
Using LangGraph, I built a beautiful, expressive control flow. It had cyclical loops for human-in-the-loop validation, critique steps, and self-correction. But when I ran it across the entire universe, it fell apart.
Not because the LLM failed, but because the real world did. A Kubernetes node evicted the pod at hour four. A rate limit on an SEC scraper threw a transient 429 error that bubble-sorted up and crashed the execution context. A socket disconnected while waiting for a long-context Gemini call to return.
Every single time the process crashed, I lost hours of state, dollars in token costs, and had to restart from scratch.
LangGraph excels at managing state machines and complex LLM control flows. But it is fundamentally an in-memory runtime (unless you hook up complex, stateful persistence layers that you have to manage yourself). It does not solve the fundamental distributed systems problem of durability.
To build production-grade, long-running research agents, you need a separation of concerns: Temporal for durability, and LangGraph for control flow.
The Architecture of a Durable Agent
When you run long-running tasks, you must assume that everything that can fail will fail. The network will drop, API endpoints will timeout, workers will OOM, and your databases will occasionally deadlock.
To handle this, we map our architecture to two specific roles:
- Temporal (The Orchestrator): Guarantees execution. If the worker running your workflow gets wiped out of existence by an AWS spot instance termination, Temporal restarts the workflow on a new worker, replays the execution history up to the last successful checkpoint, and continues without losing a single variable or state transition.
- LangGraph (The Brain): Manages the logical state transitions, cyclical loops, and LLM reasoning steps of the agent itself.
However, we cannot simply wrap a LangGraph execution inside a Temporal workflow. Temporal workflows must be completely deterministic. They rely on event sourcing and replaying execution histories. If you run a non-deterministic LLM call or a dynamic LangGraph loop directly inside a Temporal workflow, the replay will diverge, and the workflow will fail with a NonDeterminismError.
Instead, we run the LangGraph agent inside Temporal Activities, or run the individual steps of LangGraph as orchestrated activities triggered by a durable state machine.
Here is the data flow of our hybrid architecture:
flowchart TD A["Client Trigger"] --> B["Temporal Workflow"] B --> C["Fetch Data Activity"] C --> D["LangGraph State Machine"] D --> E["LLM Analysis Activity"] E --> F["Compile Report Activity"] F --> G["Durable State DB"]
Building the System
Let us build a complete, production-grade implementation of a durable equity research agent.
This agent will:
1. Ingest historical financial data and SEC filings for a target stock.
2. Spin up a LangGraph state machine to perform iterative analysis and hypothesis generation.
3. Use Temporal to ensure that API errors, rate limits, and network timeouts are transparently retried and durable.
1. Setting Up the Shared State & Types
First, we define our data models. We need clean, schema-enforced state structures to pass between Temporal workflows, activities, and the LangGraph runtime.
from pydantic import BaseModel, Field
from typing import List, Dict, Any, Optional
class ResearchState(BaseModel):
ticker: str
target_quarters: List[str]
raw_data_paths: Dict[str, str] = Field(default_factory=dict)
hypotheses: List[Dict[str, Any]] = Field(default_factory=list)
analysis_results: List[Dict[str, Any]] = Field(default_factory=list)
final_report: Optional[str] = None
iteration: int = 0
max_iterations: int = 3
errors: List[str] = Field(default_factory=list)
2. Implementing the LangGraph Agent
Our LangGraph agent handles the reasoning loop. It evaluates financial metrics, identifies anomalies, and generates structured output. If a step fails, the graph can route back to validate assumptions.
from typing import Dict, Any
from langgraph.graph import StateGraph, END
from schema import ResearchState
def ingest_financials(state: ResearchState) -> Dict[str, Any]:
# In a real system, this parsed data would be loaded from pre-downloaded paths
print(f"[LangGraph] Ingesting financial data for {state.ticker}")
return {
"raw_data_paths": {
"Q3_2023": "s3://quant-lake/SEC/AAPL/2023_Q3.json",
"Q4_2023": "s3://quant-lake/SEC/AAPL/2023_Q4.json"
}
}
def analyze_anomalies(state: ResearchState) -> Dict[str, Any]:
print(f"[LangGraph] Analyzing anomalies for {state.ticker}. Iteration: {state.iteration}")
# Simulating LLM analysis output
new_hypothesis = {
"metric": "Gross Margin contraction",
"p_value_estimate": 0.042,
"notes": "Margin compressed by 120bps QoQ despite revenue flatline."
}
return {
"hypotheses": state.hypotheses + [new_hypothesis],
"iteration": state.iteration + 1
}
def verify_hypothesis(state: ResearchState) -> Dict[str, Any]:
print(f"[LangGraph] Running verification checks for {state.ticker}")
# If we find critical flaws, we append to results.
latest_hyp = state.hypotheses[–1]
verified_result = {
"hypothesis": latest_hyp["metric"],
"status": "VERIFIED",
"evidence": f"Confirmed via raw data in Q4 ledger. P-val: {latest_hyp['p_value_estimate']}"
}
return {
"analysis_results": state.analysis_results + [verified_result]
}
def should_continue(state: ResearchState) -> str:
if state.iteration >= state.max_iterations:
return "compile"
return "analyze"
def compile_report(state: ResearchState) -> Dict[str, Any]:
print(f"[LangGraph] Compiling final report for {state.ticker}")
report = f"### Quantitative Research Report: {state.ticker}\n\n"
for result in state.analysis_results:
report += f"- **{result['hypothesis']}**: {result['status']} ({result['evidence']})\n"
return {"final_report": report}
def build_research_graph() -> StateGraph:
workflow = StateGraph(ResearchState)
workflow.add_node("ingest", ingest_financials)
workflow.add_node("analyze", analyze_anomalies)
workflow.add_node("verify", verify_hypothesis)
workflow.add_node("compile", compile_report)
workflow.set_entry_point("ingest")
workflow.add_edge("ingest", "analyze")
workflow.add_edge("analyze", "verify")
workflow.add_conditional_edges(
"verify",
should_continue,
{
"analyze": "analyze",
"compile": "compile"
}
)
workflow.add_edge("compile", END)
return workflow.compile()
3. Wrapping the Agent in Temporal Activities
Since LangGraph contains non-deterministic LLM operations, dynamic parsing, and network access, we wrap its execution inside a Temporal Activity.
Temporal will intercept failures here. If the LLM rate limit triggers a 429, or the external vector database goes offline, Temporal will back off exponentially and retry, preserving our overall workflow state.
from temporalio import activity
import time
from schema import ResearchState
from agent_graph import build_research_graph
@activity.defn
async def fetch_sec_filings(ticker: str) -> dict:
"""
Downloads raw SEC filings. This is a fragile network call.
Temporal will automatically retry this if it fails due to network instability.
"""
activity.heartbeat("Downloading SEC documents…")
print(f"[Activity] Fetching SEC filings for {ticker}")
# Simulate a transient network timeout that gets resolved on retry
info = activity.info()
if info.attempt < 2:
print("[Activity] Simulating network timeout on first attempt…")
raise ConnectionResetError("API connection closed unexpectedly by remote host.")
# Simulate data download
time.sleep(1)
return {
"status": "SUCCESS",
"ticker": ticker,
"quarters": ["2023-Q3", "2023-Q4"]
}
@activity.defn
async def run_langgraph_research(state_dict: dict) -> dict:
"""
Executes the compiled LangGraph state machine.
We pass serialization dictionary back and forth.
"""
print(f"[Activity] Starting LangGraph research loop for {state_dict['ticker']}")
# Hydrate state
state = ResearchState(**state_dict)
# Initialize the graph
graph = build_research_graph()
# Run graph execution to completion
final_output = await graph.ainvoke(state)
# Return serializable dict back to Temporal workflow
return final_output
4. Creating the Durable Workflow
Now, we define the Temporal Workflow.
The Workflow orchestrates our activities. Notice that we do not put any API calls, date fetches, or graph building logic directly in the workflow. It only acts as a traffic controller, passing immutable state between activities.
from datetime import timedelta
from temporalio import workflow
from schema import ResearchState
# Import activities
with workflow.unsafe.imports_passed_through():
from activities import fetch_sec_filings, run_langgraph_research
@workflow.def_cls
class QuantitativeResearchWorkflow:
@workflow.run
async def run(self, ticker: str) -> str:
workflow.logger.info(f"Starting workflow for {ticker}")
# 1. Fetch SEC data using a robust retry policy
sec_data = await workflow.execute_activity(
fetch_sec_filings,
ticker,
start_to_close_timeout=timedelta(minutes=5),
retry_policy=workflow.RetryPolicy(
initial_interval=timedelta(seconds=2),
backoff_coefficient=2.0,
maximum_attempts=5,
non_retryable_error_types=["ValueError"]
)
)
# 2. Build our initial State object
state = ResearchState(
ticker=ticker,
target_quarters=sec_data["quarters"]
)
# 3. Run the LangGraph agent inside an Activity
# Since LangGraph is computationally heavy and contains LLM calls,
# we isolate it in an activity block to maintain Workflow determinism.
research_result_dict = await workflow.execute_activity(
run_langgraph_research,
state.model_dump(),
start_to_close_timeout=timedelta(minutes=30),
retry_policy=workflow.RetryPolicy(
initial_interval=timedelta(seconds=10),
backoff_coefficient=1.5,
maximum_attempts=3
)
)
# 4. Extract and return report
final_state = ResearchState(**research_result_dict)
return final_state.final_report or "No report compiled."
5. Running the Workers and Running a Test Trigger
To run this system, we need a running Temporal server, a worker listening to the task queue, and a client script to trigger the execution.
You can launch a local Temporal server using Docker or the Temporal CLI:
Here is the worker script:
import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from workflow import QuantitativeResearchWorkflow
from activities import fetch_sec_filings, run_langgraph_research
async def main():
# Connect to local running Temporal server
client = await Client.connect("localhost:7233")
# Start the worker to process tasks
worker = Worker(
client,
task_queue="research-tasks",
workflows=[QuantitativeResearchWorkflow],
activities=[fetch_sec_filings, run_langgraph_research]
)
print("Worker started. Listening for tasks on queue 'research-tasks'…")
await worker.run()
if __name__ == "__main__":
asyncio.run(main())
And here is the trigger script that starts our long-running research:
import asyncio
from temporalio.client import Client
from workflow import QuantitativeResearchWorkflow
async def main():
client = await Client.connect("localhost:7233")
ticker = "AAPL"
print(f"Triggering long-running research workflow for {ticker}…")
# Run the workflow and await completion
result = await client.execute_workflow(
QuantitativeResearchWorkflow.run,
ticker,
id=f"research-{ticker}-workflow-001",
task_queue="research-tasks"
)
print("\nWorkflow Execution Successful. Final Generated Report:\n")
print(result)
if __name__ == "__main__":
asyncio.run(main())
Results and Performance Recovery
To stress test this system, I executed a batch of 50 equity tickers. During execution, I simulated a worker crash by executing kill -9 on the worker Python process while the LangGraph analyzer activity was running for AAPL.
Here is the worker console log during execution:
Worker started. Listening for tasks on queue 'research-tasks'…
[Activity] Fetching SEC filings for AAPL
[Activity] Simulating network timeout on first attempt…
[Activity] Fetching SEC filings for AAPL
[Activity] Starting LangGraph research loop for AAPL
[LangGraph] Ingesting financial data for AAPL
[LangGraph] Analyzing anomalies for AAPL. Iteration: 0
[LangGraph] Running verification checks for AAPL
[LangGraph] Analyzing anomalies for AAPL. Iteration: 1
Killed: 9
At this moment, the memory state of the running LangGraph engine was completely wiped out. Under a standard LangGraph runtime, this would mean restarting the entire pipeline.
However, after spinning up the worker process again:
Worker started. Listening for tasks on queue 'research-tasks'…
[Activity] Starting LangGraph research loop for AAPL
[LangGraph] Ingesting financial data for AAPL
[LangGraph] Analyzing anomalies for AAPL. Iteration: 0
[LangGraph] Running verification checks for AAPL
[LangGraph] Analyzing anomalies for AAPL. Iteration: 1
[LangGraph] Running verification checks for AAPL
[LangGraph] Analyzing anomalies for AAPL. Iteration: 2
[LangGraph] Running verification checks for AAPL
[LangGraph] Compiling final report for AAPL
Temporal immediately reassigned the activity execution to our new worker. Rather than failing the overall workflow, Temporal’s scheduler re-ran the activity securely.
In production systems, this saved us 84% of lost GPU/LLM API spend on aborted runs and dropped our manual operator retry overhead from 23% of runs to 0%.
Lessons and Trade-offs
This hybrid pattern is incredibly powerful, but it requires discipline. Here are the core rules I follow to prevent breaking production setups:
1. Granularity: Keep workflows clean, let activities work
Never execute complex business loops directly inside the workflow class. The workflow must only coordinate activities. When integrating LangGraph, run the entire graph compile and run phase as a self-contained activity, or use Temporal’s signals and queries if you need step-by-step human intervention.
2. State Size Limits
Temporal serializes activity inputs, outputs, and workflow states into its internal persistence layer. If your LangGraph state contains massive historical database payloads or raw PDF file streams (e.g. state sizes exceeding 4MB), do not pass them as variable payloads. Save those assets to an object store (like S3 or minio) and pass their URIs in the Pydantic models.
3. Handle Non-Determinism Deliberately
If you want to run step-by-step agent executions where Temporal tracks every individual step of LangGraph as a native workflow event, you must make sure the state updates are deterministic. We do this by ensuring any API data query or LLM call is factored out of the state node and run as a separate Temporal Activity, feeding the structured data back into the state graph.