Haystack Tutorial (Python): persisting agent state for intermediate developers

This tutorial shows how to persist agent state in Haystack so a multi-step workflow can survive process restarts, retries, and long-running conversations. You need this when your agent is tracking tool outputs, intermediate decisions, or conversation history and you can’t afford to lose that state between requests.

What You'll Need

• •Python 3.10+
• •haystack-ai
• •An OpenAI API key
• •A working internet connection for model calls
• •Basic familiarity with Haystack pipelines and components

Install the dependencies:

pip install haystack-ai openai

Set your API key in the environment:

export OPENAI_API_KEY="your-key-here"

Step-by-Step

1. •Start by defining a small state object that can be serialized cleanly. For intermediate systems, keep state explicit: user input, tool results, and the latest answer.

from dataclasses import dataclass, asdict
import json
from pathlib import Path

STATE_FILE = Path("agent_state.json")

@dataclass
class AgentState:
    user_message: str = ""
    tool_result: str = ""
    final_answer: str = ""

def save_state(state: AgentState) -> None:
    STATE_FILE.write_text(json.dumps(asdict(state), indent=2))

def load_state() -> AgentState:
    if not STATE_FILE.exists():
        return AgentState()
    data = json.loads(STATE_FILE.read_text())
    return AgentState(**data)

2. •Next, build a simple Haystack pipeline that uses an LLM to produce an answer from the current state. This example keeps the state outside the pipeline, which is the practical pattern when you want persistence across runs.

from haystack import Pipeline, component
from haystack.components.builders import PromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator

template = """
You are an assistant.
User message: {{ user_message }}
Tool result: {{ tool_result }}

Return a concise final answer.
"""

@component
class StateLoader:
    @component.output_types(user_message=str, tool_result=str)
    def run(self, user_message: str, tool_result: str):
        return {"user_message": user_message, "tool_result": tool_result}

pipe = Pipeline()
pipe.add_component("prompt_builder", PromptBuilder(template=template))
pipe.add_component("llm", OpenAIChatGenerator(model="gpt-4o-mini"))
pipe.connect("prompt_builder.prompt", "llm.messages")

3. •Add a deterministic “tool” step and persist the result before calling the model. In real systems this could be a database lookup, policy engine call, or internal service request.

def fake_tool_lookup(user_message: str) -> str:
    if "policy" in user_message.lower():
        return "Policy status: active. Renewal date: 2026-03-01."
    return "No matching records found."

state = load_state()
state.user_message = "Check my policy status"
state.tool_result = fake_tool_lookup(state.user_message)
save_state(state)

result = pipe.run({
    "prompt_builder": {
        "user_message": state.user_message,
        "tool_result": state.tool_result,
    }
})

state.final_answer = result["llm"]["replies"][0].text
save_state(state)
print(state.final_answer)

4. •Now simulate a restart by loading the saved file and continuing from where you left off. This is the core persistence pattern: reconstruct context from disk instead of keeping it only in memory.

reloaded_state = load_state()

print("User message:", reloaded_state.user_message)
print("Tool result:", reloaded_state.tool_result)
print("Final answer:", reloaded_state.final_answer)

# Example continuation after restart
reloaded_state.user_message = "What is my renewal date?"
reloaded_state.tool_result = fake_tool_lookup(reloaded_state.user_message)
save_state(reloaded_state)

5. •If you want to persist more than one turn, store a list of messages instead of a single string. That gives you a simple conversation log you can replay into prompts later.

from typing import List

@dataclass
class ConversationState:
    messages: List[dict]

def save_conversation(state: ConversationState) -> None:
    STATE_FILE.write_text(json.dumps({"messages": state.messages}, indent=2))

def load_conversation() -> ConversationState:
    if not STATE_FILE.exists():
        return ConversationState(messages=[])
    data = json.loads(STATE_FILE.read_text())
    return ConversationState(messages=data["messages"])

conv = load_conversation()
conv.messages.append({"role": "user", "content": "Show my policy status"})
conv.messages.append({"role": "assistant", "content": "Policy status is active."})
save_conversation(conv)

Testing It

Run the script once and confirm that agent_state.json is created with the expected fields. Then stop the process, rerun it, and verify that load_state() restores the previous values instead of starting empty.

If you’re using the LLM step, check that final_answer changes based on user_message and tool_result. For production work, also test failure cases like missing files, partial writes, and corrupted JSON.

Next Steps

• •Replace local JSON persistence with Redis or Postgres for multi-instance deployments.
• •Move from single-state fields to a message ledger plus metadata for auditability.
• •Wrap the persistence layer behind a repository interface so your Haystack pipeline stays clean.

Keep learning

• •The complete AI Agents Roadmap — my full 8-step breakdown
• •Free: The AI Agent Starter Kit — PDF checklist + starter code
• •Work with me — I build AI for banks and insurance companies

By Cyprian Aarons, AI Consultant at Topiax.